server/weathermon-mqtt

   1 #!/usr/bin/lua
   2
   3 local random = math.random
   4 local json = require "json"
   5 math.randomseed(os.time())
   6
   7 local function uuid()
   8     local template ='xxxx-xxxx'
   9     return string.gsub(template, '[x]', function (c)
  10         local v = random(0, 0xf)
  11         return string.format('%x', v)
  12     end)
  13 end
  14
  15 function process_MSG(mid, topic, payload)
  16   print(topic, payload)
  17   pcall(function(topic,payload)
  18     print(payload)
  19     payload = json.decode(payload)
  20     local time = os.date(payload['Time'])
  21     if not time then time = os.date(payload['time']); end
  22     local model = payload['model']
  23     local id = payload['id']
  24     for sensor_type,sensor_data in pairs(payload) do
  25       if sensor_type ~= "Time" and sensor_type ~= "TempUnit" and sensor_type ~= "model" and sensor_type ~="id" and sensor_type ~= "time" then
  26         if model then
  27           conn:execute(string.format("CALL meteo.submit_mqtt('%s','%s','%s','%s',NULL)", topic,model,sensor_type,sensor_data))
  28           conn:commit()
  29           print(topic,model,sensor_type,sensor_data)
  30         else
  31           for param,value in pairs(sensor_data) do
  32             conn:execute(string.format("CALL meteo.submit_mqtt('%s','%s','%s','%s',NULL)", topic,sensor_type,param,value))
  33             conn:commit()
  34             print(topic,sensor_type,param,value)
  35           end
  36         end
  37       end
  38     end
  39   end, topic, payload)
  40 end
  41
  42 uci = require "uci"
  43
  44 config_name = arg[1]
  45 if not config_name then
  46   config_name = "weathermon_mqtt"
  47 end
  48
  49 mqtt_host = uci.get(config_name,"mqtt","server")
  50 if not mqtt_host then
  51   mqtt_host="127.0.0.1"
  52 end
  53
  54 mqtt_port = uci.get(config_name,"mqtt","port")
  55 if not mqtt_port then
  56   mqtt_host=1883
  57 end
  58
  59 mqtt_user = uci.get(config_name,"mqtt","username")
  60 mqtt_pwd = uci.get(config_name,"mqtt","password")
  61
  62 mqtt_id = "wm-mqtt-"..uuid()
  63
  64 db_server = uci.get(config_name,"db","server")
  65 db_name = uci.get(config_name,"db","db")
  66 db_user = uci.get(config_name,"db","username")
  67 db_pwd = uci.get(config_name,"db","password")
  68
  69 env = require("luasql.mysql").mysql()
  70 conn = env:connect(db_name,db_user,db_pwd,db_server)
  71 conn:execute("SET NAMES utf8")
  72
  73 MQTT = require "mosquitto"
  74 mqtt_client = MQTT.new(mqtt_id)
  75 if mqtt_user then
  76   mqtt_client:login_set(mqtt_user, mqtt_pwd)
  77 end
  78
  79 mqtt_client:connect("estia.rvb-home.lan",1883)
  80 mqtt_client.ON_MESSAGE = process_MSG
  81
  82 cur = conn:execute("SELECT DISTINCT topic FROM mqtt_topics WHERE topic<>''")
  83 rec = cur:fetch()
  84 while rec do
  85   mqtt_client:subscribe(rec)
  86   rec = cur:fetch()
  87 end
  88
  89 mqtt_client:loop_forever()