queue sleep as variable

2 files changed, 59 insertions, 1 deletions

diff --git a/scripts/mqtt_watch.py b/scripts/mqtt_watch.py
new file mode 100644
index 0000000..e460f13
--- /dev/null
+++ b/scripts/mqtt_watch.py
@@ -0,0 +1,57 @@
+#!/usr/bin/env python3
+''' Listen for mqtt-events, and trigger for some '''
+
+import os
+import json
+from datetime import datetime
+import paho.mqtt.client as mqtt
+
+from common import dbi
+
+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']
+
+# 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('#')
+
+# 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)
+
+    print(msg.topic, 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())
+
+    #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())
+
+    #else:
+    #    return
+
+    #dbi(sql, values, verbose=True)
+
+
+
+# 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()
diff --git a/scripts/queue_runner.py b/scripts/queue_runner.py
index fd4812e..62467b5 100644
--- a/scripts/queue_runner.py
+++ b/scripts/queue_runner.py
@@ -10,6 +10,7 @@ from litequeue import SQLQueue
 from common.postgres import dbi
 
 QUEUE_DB = os.environ.get('el_QUEUE_db', 'litequeue.db')
+QUEUE_SLEEP = int(os.environ.get('el_QUEUE_sleep', 15)) # Default sleep 15 seconds when queue empty
 
 # Unlock all
 con = sqlite3.connect(QUEUE_DB)
@@ -31,7 +32,7 @@ i = 0
 while True:
     # Just sleep if queue is empty
     if q.empty() is True:
-        time.sleep(10)
+        time.sleep(QUEUE_SLEEP)
         continue
 
     # get message