1 files changed, 53 insertions, 0 deletions
diff --git a/mqtt2queue.py b/mqtt2queue.py
new file mode 100644
index 0000000..102b1eb
--- /dev/null
+++ b/mqtt2queue.py
@@ -0,0 +1,53 @@
+#!/usr/bin/env python3
+
+import os
+import json
+import pickle
+from datetime import datetime
+from litequeue import SQLQueue
+import paho.mqtt.client as mqtt
+
+mqtt_server = os.environ['el_mqtt_server']
+mqtt_port = int(os.environ['el_mqtt_port'])
+keepalive = int(os.environ['el_mqtt_keepalive'])
+mqtt_topic = os.environ['el_mqtt_topic']
+
+q = SQLQueue("litequeue.db", maxsize=None)
+
+# The callback for when the client receives a CONNACK response from the server.
+def on_connect(client, userdata, flags, rc):
+    print("Connected with result code "+str(rc))
+
+    # Subscribing in on_connect() means that if we lose the connection and
+    # reconnect then subscriptions will be renewed.
+    client.subscribe(mqtt_topic)
+
+# The callback for when a PUBLISH message is received from the server.
+def on_message(client, userdata, msg):
+    name = msg.topic.split('/')[1]
+    data = json.loads(msg.payload)
+
+    if name.startswith('tmp') and 'temperature' in data and 'humidity' in data:
+        sql = "INSERT INTO mqtt_temps (name, temperature, humidity, battery, linkquality, voltage, time) VALUES(%s,%s,%s,%s,%s,%s,%s)"
+        values = (name, data['temperature'], data['humidity'], data['battery'], data['linkquality'], data['voltage'], datetime.utcnow())
+        q.put(pickle.dumps([sql, values]))
+
+    elif name == 'HAN' and 'current' in data:
+        sql = "INSERT INTO mqtt_han (name, current, power, voltage, linkquality, time) VALUES(%s,%s,%s,%s,%s,%s)"
+        values = (name, data['current'], data['power'], data['voltage'], data['linkquality'], datetime.utcnow())
+        q.put(pickle.dumps([sql, values]))
+
+
+
+# mqtt
+client = mqtt.Client()
+client.on_connect = on_connect
+client.on_message = on_message
+
+client.connect(mqtt_server, mqtt_port, keepalive)
+
+# Blocking call that processes network traffic, dispatches callbacks and
+# handles reconnecting.
+# Other loop*() functions are available that give a threaded interface and a
+# manual interface.
+client.loop_forever()