d2b53746f65ace65e7a4610bc42b0515633768f1
[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 backlogdb then
107       printLog("writing record to backlog...")
108       local backlog_con = assert(env:connect(backlogdb))
109       backlog_con:execute('BEGIN TRANSACTION')
110       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))
111       backlog_con:execute('COMMIT')
112       backlog_con:close()
113     end
114
115   end
116   
117   if logdb then
118     local log_con = assert(env:connect(logdb))
119     log_con:execute('BEGIN TRANSACTION')
120     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))
121     log_con:execute('COMMIT')
122     log_con:close()
123   end
124
125 end
126
127 function storeRecord(id,sensor,param,value) 
128
129   if not records[id] then
130     records[id] = {}
131   end  
132   
133   records[id]["timestamp"] = os.date("%Y-%m-%dT%H:%M:%S")
134   
135   if not records[id][sensor] then
136     records[id][sensor] = {}
137   end
138   
139   records[id][sensor][param] = value
140
141 end
142
143 function processJson(str)
144
145   msg=json.decode(str)
146
147   sensor={}
148
149   for key,value in pairs(msg) do
150     if value then
151       if key=="model" or key=="device" then
152         sensor_type=value
153       elseif key=="id" then
154         sensor_id=value
155       elseif key=='time' then
156         sensor_time=value
157       else
158         sensor[key]=value
159       end
160     end
161   end
162
163   if not sensor_id then
164     sensor_id = web_devid
165   end
166
167   if not (sensor_type==nil or sensor_id==nil or sensor_type=='' or sensor_id=='') then
168     local record = {}
169     for k,v in pairs(sensor) do
170       storeRecord(sensor_id,sensor_type,k,v)
171       printLog("Type = "..sensor_type..", ID = "..sensor_id..", Param = "..k..", Value = \""..v.."\"")
172       if web_url then
173         submitValue(sensor_type,sensor_id,k,v)
174       end  
175       if mqtt_client then
176         mqtt_path=string.gsub(mqtt_topic,"{(.-)}", 
177           function (name) 
178             if name=="dev" then
179               return mqtt_encode(web_devid)
180             elseif name=="type" then
181               return mqtt_encode(sensor_type)
182             elseif name=="id" then
183               return mqtt_encode(sensor_id)
184             elseif name=="param" then
185               return k
186             else
187               return '{'..name..'}'
188             end      
189           end)
190         if not mqtt_client:socket() then
191           mqtt_client:reconnect()
192         end  
193         mqtt_client:publish(mqtt_path,v,0,false)
194         mqtt_client:loop()
195       end  
196     end
197   else
198     printLog("Cannot parse sensor input: "..str)
199   end
200
201 end
202
203 function processLine(str)
204
205   msg=split(line,':')
206   msg_type=msg[1] or ''
207   msg_body=msg[2] or ''
208   if msg_type=="STATUS" then
209     printLog("Status: "..msg_body)
210   elseif msg_type=="ERROR" then
211     printLog("Error: "..msg_body)  
212   elseif msg_type=="SENSOR" then
213     printLog("SENSOR: "..msg_body)  
214     sens = split(msg_body,",")
215     sensor = {}
216     idx = 1
217     sensor_type = nil
218     sensor_id = web_devid
219     for i,rec in ipairs(sens) do
220       recrd=split(rec,'=')
221       key=recrd[1] or ''
222       value=recrd[2] or ''
223       if value then
224         if key=="TYPE" then
225           sensor_type=value
226         elseif key=="ID" then
227           sensor_id=value
228         else
229           sensor[key]=value
230         end
231       end
232     end
233     if not (sensor_type==nil or sensor_id==nil or sensor_type=='' or sensor_id=='') then
234       local record = {}
235       for k,v in pairs(sensor) do
236         storeRecord(sensor_id,sensor_type,k,v)
237         printLog("Type = "..sensor_type..", ID = "..sensor_id..", Param = "..k..", Value = "..v)
238         if web_url then 
239           submitValue(sensor_type,sensor_id,k,v)
240         end  
241         if mqtt_client then
242           mqtt_path=string.gsub(mqtt_topic,"{(.-)}", 
243             function (name) 
244               if name=="dev" then
245                 return web_devid
246               elseif name=="type" then
247                 return sensor_type
248               elseif name=="id" then
249                 return sensor_id
250               elseif name=="param" then
251                 return k
252               else
253                 return '{'..name..'}'
254               end      
255             end)
256           if not mqtt_client:socket() then
257             mqtt_client:reconnect()
258           end  
259           mqtt_client:publish(mqtt_path,v,0,false)
260           mqtt_client:loop()
261         end  
262       end
263     else
264       printLog("Cannot parse sensor input: "..msg_body)
265     end
266   elseif msg_type=="ALARM" then
267     printLog("ALARM: "..msg_body)  
268     sens = split(msg_body,",")
269     sensor = {}
270     idx = 1
271     sensor_type = nil
272     sensor_id = web_devid
273     mqtt_param = {}
274     for i,rec in ipairs(sens) do
275       recrd=split(rec,'=')
276       key=recrd[1] or ''
277       value=recrd[2] or ''
278       if value then
279         if key=="TYPE" then
280           alarm_type=value
281         elseif key=="ID" then
282           alarm_id=value
283         end
284       end
285     end
286     if not (alarm_type==nil or alarm_id==nil or alarm_type=='' or alarm_id=='') then
287       if mqtt_client then
288         mqtt_path=string.gsub(mqtt_alarm_topic,"{(.-)}", 
289           function (name) 
290             if name=="dev" then
291               return web_devid
292             elseif name=="type" then
293               return sensor_type
294             elseif name=="id" then
295               return sensor_id
296             else
297               return '{'..name..'}'
298             end      
299           end)
300         if not mqtt_client:socket() then
301           mqtt_client:reconnect()
302         end  
303         mqtt_client:publish(mqtt_path,msg_body,0,false)
304         mqtt_client:loop()
305       end
306       if alarm_exec then
307         command=alarm_exec..
308           " \""..string.gsub(alarm_type,"\"","\\\"")..
309           "\" \""..string.gsub(alarm_id,"\"","\\\"")..
310           "\" \""..string.gsub(msg_body,"\"","\\\"").."\""
311         capture(command)
312       end
313     else
314       printLog("Cannot parse alarm input: "..msg_body)
315     end
316   end
317
318 end
319
320 getConfig(arg[1])
321
322 signal.signal(signal.SIGTERM, function(signum)
323
324   printLog("Terminating...")
325   local pids = get_children()
326   for k,v in pairs(pids) do
327     printLog("Terminating subprocess "..tostring(v).."...")
328     signal.kill(v,signal.SIGTERM)
329   end
330   printLog("Exiting...")
331   os.exit(0)
332
333 end)
334
335 if backlogdb or logdb then
336   local dbdriver = require "luasql.sqlite3"
337   env = assert(dbdriver.sqlite3())
338 end
339
340 if backlogdb then
341   if not file_exists(backlogdb) then
342     touch(backlogdb)
343   end  
344   local backlog_con = assert(env:connect(backlogdb))
345   backlog_con:execute("CREATE TABLE queue(time_stamp datetime,sensor_id varchar(16),sensor varchar(16),param varchar(16),value float)")
346   backlog_con:close()
347 end
348
349 if logdb then
350   if not file_exists(logdb) then
351     touch(logdb)
352   end  
353   local log_con = assert(env:connect(logdb))
354   log_con:execute("CREATE TABLE log(time_stamp datetime,sensor_id varchar(16),sensor varchar(16),param varchar(16),value float)")
355   log_con:execute("CREATE INDEX log_idx ON log(sensor_id,sensor,param,time_stamp)")
356   log_con:close()
357 end
358
359 if web_url then
360   http = require("socket.http")
361   http.TIMEOUT = web_timeout
362 end
363
364 if mqtt_host then
365   MQTT = require "mosquitto"
366   mqtt_client = MQTT.new(mqtt_id)
367   if mqtt_user then
368     mqtt_client:login_set(mqtt_user, mqtt_passwd)
369   end
370   mqtt_client:connect(mqtt_host,mqtt_port)
371 end
372
373 if serial_port then
374   serialin=io.open(serial_port,"r")
375 elseif input_file == "-" then
376   serialin=io.stdin;
377 elseif input_file then
378   serialin=io.open(input_file,"r")
379 elseif input_exec then
380   serialin=io.popen(input_exec,"r")
381 else
382   printLog("No input selected")
383   return
384 end  
385
386 serialin:setvbuf('no')
387
388 records = {}
389
390 while 1 do
391
392   line=serialin:read("*l")
393
394   if line == nil then
395     break
396   end
397
398   pcall(function ()
399   
400     printLog("Received: "..line)
401     if startswith(line,'{') then
402       processJson(line)
403     else
404       processLine(line)
405     end
406
407     if touch_file then
408       pcall(function () 
409         touch(touch_file) 
410       end)
411     end  
412
413     if dump_file then
414       pcall(function ()
415         local f = io.open(dump_file,"w")
416         io.output(f)
417         io.write(json.encode(records))
418         io.close(f)
419       end)  
420     end
421   end)
422     
423 end