Start recording mqtt data in rss.

1 files changed, 21 insertions, 9 deletions

diff --git a/rrd/mqtt.py b/rrd/mqtt.py
index de679ce..09589c3 100644
--- a/rrd/mqtt.py
+++ b/rrd/mqtt.py
@@ -1,7 +1,11 @@
+import datetime
 import logging
 
-import django.conf
 import paho.mqtt.client as mqtt
+import slugify
+from django.conf import settings
+
+from . import models
 
 log = logging.getLogger(__name__)
 
@@ -31,13 +35,13 @@ class MQTTClient:
     def connect(self):
         try:
             self.client.connect(
-                django.conf.settings.MQTT_SERVER,
-                django.conf.settings.MQTT_PORT,
+                settings.MQTT_SERVER,
+                settings.MQTT_PORT,
                 60,  # TODO: make the keepalive configurable
             )
         except OSError as e:
-            log.debug("Could not connect to MQTT server")
-            log.debug(e)
+            log.warning("Could not connect to MQTT server")
+            log.warning(e)
 
     def disconnect(self, reconnect: bool = True):
         self.reconnect = reconnect
@@ -46,7 +50,7 @@ class MQTTClient:
     def on_connect(self, client, userdata, flags, rc):
         log.info("Connected to MQTT")
         self.connected = True
-        client.subscribe(django.conf.settings.MQTT_TOPIC)
+        client.subscribe(settings.MQTT_TOPIC + "#")
 
     def on_disconnect(self, client, userdata, rc):
         log.info("Disconnected from MQTT")
@@ -55,9 +59,17 @@ class MQTTClient:
             self.connect()
 
     def on_message(self, client, userdata, msg):
-        print(
-            "Received msg %s %s", msg.topic, msg.payload.decode()
-        )
         log.debug(
             "Received msg %s %s", msg.topic, msg.payload.decode()
         )
+        ts = int(datetime.datetime.now().timestamp())
+        topic = msg.topic.removeprefix(settings.MQTT_TOPIC)
+        dss = models.DataSource.objects.filter(topic=topic)
+        if not dss:
+            dss = [models.DataSource.objects.create(
+                topic=topic,
+                path=slugify.slugify(topic) + ".rrd",
+            )]
+            dss[0].save()
+        for ds in dss:
+            ds.update(ts, msg.payload.decode())