Произвольные настройки порта в конфигурационном файле.
[weathermon.git] / bin / weathermon
1 #!/usr/bin/lua
2
3 local json = require("json")
4 local signal = require("posix.signal")
5
6 require "wm_util"
7
8 function getConfig(configname)
9
10   local command,f
11
12   local uci=require("uci")
13   local cur=uci.cursor()
14   local config
15   if configname then
16     config=configname
17   else
18     config="weathermon"
19   end
20
21   web_url  = cur.get(config,"web","url")
22   web_user = cur.get(config,"web","user")
23   web_timeout = cur.get(config,"web","timeout")
24   web_pass = cur.get(config,"web","password")
25  
26   if not web_timeout then
27     web_timeout = 10
28   end
29
30   web_devid = get_devid(config)
31
32   logging = cur.get(config,"logging","enabled") 
33   touch_file = cur.get(config,"logging","touch_file") 
34
35   backlogdb = cur.get(config,"process","backlogdb")
36   logdb = cur.get(config,"process","logdb")
37
38   serial_port = cur.get(config,"serial","port")
39   serial_baud = cur.get(config,"serial","baud")
40
41   input_file = cur.get(config,"input","file")
42   input_exec = cur.get(config,"input","exec")
43   alarm_exec = cur.get(config,"alarm","exec")
44
45   if serial_port then
46
47     command = "stty -F  "..serial_port.." "..serial_baud
48     capture(command)
49
50   end
51
52   mqtt_host = cur.get(config,"mqtt","host")
53   mqtt_port = cur.get(config,"mqtt","port")
54   mqtt_id = cur.get(config,"mqtt","id")
55   mqtt_topic = cur.get(config,"mqtt","topic")
56   mqtt_alarm_topic = cur.get(config,"mqtt","alarm_topic")
57
58   mqtt_user = cur.get(config,"mqtt","user")
59   mqtt_passwd = cur.get(config,"mqtt","password")
60
61   if mqtt_host and not mqtt_id then
62     mqtt_id="weather-"..web_devid
63   end
64
65   if mqtt_host and not mqtt_port then
66     mqtt_port = 1883
67   end
68
69   if mqtt_host and not mqtt_topic then
70     mqtt_topic = 'weathermon/{dev}/{type}/{id}/{param}'
71   end
72
73   if mqtt_host and not mqtt_alarm_topic then
74     mqtt_alarm_topic = 'alarm/{dev}/{type}/{id}'
75   end
76
77   dump_file = cur.get(config,"process","dump_file")
78
79 end
80
81 function printLog(str)
82   if logging=="on" then
83     capture("logger -t weathermon "..str)
84     print(str)  
85   elseif logging=="syslog" then
86     capture("logger -t weathermon "..str)
87   elseif logging=="stdout" then 
88     print(str)  
89   end 
90 end
91
92 function submitValue(type,id,param,val)
93
94   val = tonumber(val)
95
96   if web_url and val then
97
98     local url = web_url.."?stype="..url_encode(type).."&sid="..url_encode(id).."&param="..url_encode(param).."&value="..url_encode(val)
99
100     if web_user then
101       url = url:gsub("//","//"..web_user..":"..web_pass.."@",1)
102     end
103
104     local result,code = http.request (url)
105
106     if code ~= 200 and backlog_con then
107       printLog("writing record to backlog...")
108       backlog_con:execute(string.format("INSERT INTO queue(time_stamp,sensor_id,sensor,param,value) VALUES (datetime('now','localtime'),'%s','%s','%s',%f)",id,type,param,val))
109     end
110
111   end
112   
113   if logdb then
114     log_con:execute(string.format("INSERT INTO log(time_stamp,sensor_id,sensor,param,value) VALUES (datetime('now','localtime'),'%s','%s','%s',%f)",id,type,param,val))
115   end
116
117 end
118
119 function storeRecord(id,sensor,param,value) 
120
121   if not records[id] then
122     records[id] = {}
123   end  
124   
125   records[id]["timestamp"] = os.date("%Y-%m-%dT%H:%M:%S")
126   
127   if not records[id][sensor] then
128     records[id][sensor] = {}
129   end
130   
131   records[id][sensor][param] = value
132
133 end
134
135 function processJson(str)
136
137   msg=json.decode(str)
138
139   sensor={}
140
141   for key,value in pairs(msg) do
142     if value then
143       if key=="model" or key=="device" then
144         sensor_type=value
145       elseif key=="id" then
146         sensor_id=value
147       elseif key=='time' then
148         sensor_time=value
149       else
150         sensor[key]=value
151       end
152     end
153   end
154
155   if not sensor_id then
156     sensor_id = web_devid
157   end
158
159   if not (sensor_type==nil or sensor_id==nil or sensor_type=='' or sensor_id=='') then
160     if next(sensor)==nil then
161       sensor["command"]="alarm"
162     end
163     local record = {}
164     for k,v in pairs(sensor) do
165       storeRecord(sensor_id,sensor_type,k,v)
166       printLog("Type = "..sensor_type..", ID = "..sensor_id..", Param = "..k..", Value = \""..v.."\"")
167       if web_url then
168         submitValue(sensor_type,sensor_id,k,v)
169       end  
170       if mqtt_client then
171         mqtt_path=string.gsub(mqtt_topic,"{(.-)}", 
172           function (name) 
173             if name=="dev" then
174               return mqtt_encode(web_devid)
175             elseif name=="type" then
176               return mqtt_encode(sensor_type)
177             elseif name=="id" then
178               return mqtt_encode(sensor_id)
179             elseif name=="param" then
180               return k
181             else
182               return '{'..name..'}'
183             end      
184           end)
185         if not mqtt_client:socket() then
186           mqtt_client:reconnect()
187         end  
188         mqtt_client:publish(mqtt_path,v,0,false)
189         mqtt_client:loop()
190       end  
191     end
192   else
193     printLog("Cannot parse sensor input: "..str)
194   end
195
196 end
197
198 function processLine(str)
199
200   msg=split(line,':')
201   msg_type=msg[1] or ''
202   msg_body=msg[2] or ''
203   if msg_type=="STATUS" then
204     printLog("Status: "..msg_body)
205   elseif msg_type=="ERROR" then
206     printLog("Error: "..msg_body)  
207   elseif msg_type=="SENSOR" then
208     printLog("SENSOR: "..msg_body)  
209     sens = split(msg_body,",")
210     sensor = {}
211     idx = 1
212     sensor_type = nil
213     sensor_id = web_devid
214     for i,rec in ipairs(sens) do
215       recrd=split(rec,'=')
216       key=recrd[1] or ''
217       value=recrd[2] or ''
218       if value then
219         if key=="TYPE" then
220           sensor_type=value
221         elseif key=="ID" then
222           sensor_id=value
223         else
224           sensor[key]=value
225         end
226       end
227     end
228     if not (sensor_type==nil or sensor_id==nil or sensor_type=='' or sensor_id=='') then
229       local record = {}
230       for k,v in pairs(sensor) do
231         storeRecord(sensor_id,sensor_type,k,v)
232         printLog("Type = "..sensor_type..", ID = "..sensor_id..", Param = "..k..", Value = "..v)
233         if web_url then 
234           submitValue(sensor_type,sensor_id,k,v)
235         end  
236         if mqtt_client then
237           mqtt_path=string.gsub(mqtt_topic,"{(.-)}", 
238             function (name) 
239               if name=="dev" then
240                 return web_devid
241               elseif name=="type" then
242                 return sensor_type
243               elseif name=="id" then
244                 return sensor_id
245               elseif name=="param" then
246                 return k
247               else
248                 return '{'..name..'}'
249               end      
250             end)
251           if not mqtt_client:socket() then
252             mqtt_client:reconnect()
253           end  
254           mqtt_client:publish(mqtt_path,v,0,false)
255           mqtt_client:loop()
256         end  
257       end
258     else
259       printLog("Cannot parse sensor input: "..msg_body)
260     end
261   elseif msg_type=="ALARM" then
262     printLog("ALARM: "..msg_body)  
263     sens = split(msg_body,",")
264     sensor = {}
265     idx = 1
266     sensor_type = nil
267     sensor_id = web_devid
268     mqtt_param = {}
269     for i,rec in ipairs(sens) do
270       recrd=split(rec,'=')
271       key=recrd[1] or ''
272       value=recrd[2] or ''
273       if value then
274         if key=="TYPE" then
275           alarm_type=value
276         elseif key=="ID" then
277           alarm_id=value
278         end
279       end
280     end
281     if not (alarm_type==nil or alarm_id==nil or alarm_type=='' or alarm_id=='') then
282       if mqtt_client then
283         mqtt_path=string.gsub(mqtt_alarm_topic,"{(.-)}", 
284           function (name) 
285             if name=="dev" then
286               return web_devid
287             elseif name=="type" then
288               return sensor_type
289             elseif name=="id" then
290               return sensor_id
291             else
292               return '{'..name..'}'
293             end      
294           end)
295         if not mqtt_client:socket() then
296           mqtt_client:reconnect()
297         end  
298         mqtt_client:publish(mqtt_path,msg_body,0,false)
299         mqtt_client:loop()
300       end
301       if alarm_exec then
302         command=alarm_exec..
303           " \""..string.gsub(alarm_type,"\"","\\\"")..
304           "\" \""..string.gsub(alarm_id,"\"","\\\"")..
305           "\" \""..string.gsub(msg_body,"\"","\\\"").."\""
306         capture(command)
307       end
308     else
309       printLog("Cannot parse alarm input: "..msg_body)
310     end
311   end
312
313 end
314
315 getConfig(arg[1])
316
317 signal.signal(signal.SIGTERM, function(signum)
318
319   printLog("Terminating...")
320   local pids = get_children()
321   for k,v in pairs(pids) do
322     printLog("Terminating subprocess "..tostring(v).."...")
323     signal.kill(v,signal.SIGTERM)
324   end
325   printLog("Exiting...")
326   os.exit(0)
327
328 end)
329
330 if backlogdb or logdb then
331   local dbdriver = require "luasql.sqlite3"
332   env = assert(dbdriver.sqlite3())
333 end
334
335 if backlogdb then
336   if not file_exists(backlogdb) then
337     touch(backlogdb)
338   end  
339   backlog_con = assert(env:connect(backlogdb))
340   backlog_con:execute("CREATE TABLE queue(time_stamp datetime,sensor_id varchar(16),sensor varchar(16),param varchar(16),value float)")
341 end
342
343 if logdb then
344   if not file_exists(logdb) then
345     touch(logdb)
346   end  
347   log_con = assert(env:connect(logdb))
348   log_con:execute("CREATE TABLE log(time_stamp datetime,sensor_id varchar(16),sensor varchar(16),param varchar(16),value float)")
349   log_con:execute("CREATE INDEX log_idx ON log(sensor_id,sensor,param,time_stamp)")
350 end
351
352 if web_url then
353   http = require("socket.http")
354   http.TIMEOUT = web_timeout
355 end
356
357 if mqtt_host then
358   MQTT = require "mosquitto"
359   mqtt_client = MQTT.new(mqtt_id)
360   if mqtt_user then
361     mqtt_client:login_set(mqtt_user, mqtt_passwd)
362   end
363   mqtt_client:connect(mqtt_host,mqtt_port)
364 end
365
366 if serial_port then
367   serialin=io.open(serial_port,"r")
368 elseif input_file == "-" then
369   serialin=io.stdin;
370 elseif input_file then
371   serialin=io.open(input_file,"r")
372 elseif input_exec then
373   serialin=io.popen(input_exec,"r")
374 else
375   printLog("No input selected")
376   return
377 end  
378
379 serialin:setvbuf('no')
380
381 records = {}
382
383 while 1 do
384
385   line=serialin:read("*l")
386
387   if line == nil then
388     break
389   end
390
391   pcall(function ()
392   
393     printLog("Received: "..line)
394     if startswith(line,'{') then
395       processJson(line)
396     else
397       processLine(line)
398     end
399
400     if touch_file then
401       pcall(function () 
402         touch(touch_file) 
403       end)
404     end  
405
406     if dump_file then
407       pcall(function ()
408         local f = io.open(dump_file,"w")
409         io.output(f)
410         io.write(json.encode(records))
411         io.close(f)
412       end)  
413     end
414   end)
415     
416 end