database intro

schema/create-schema.sql

@@ -1,10 +1,10 @@
-CREATE TABLE device_t {
+CREATE TABLE device_t (
     id SERIAL NOT NULL PRIMARY KEY,
     device_id VARCHAR(32) NOT NULL UNIQUE,
     label VARCHAR(16) NOT NULL
-};
+);
 
-CREATE TABLE sensor_t {
+CREATE TABLE sensor_t (
     id SERIAL NOT NULL PRIMARY KEY,
     address NUMERIC NOT NULL UNIQUE,
     label VARCHAR(5) NOT NULL,
@@ -12,14 +12,28 @@ CREATE TABLE sensor_t {
     device INTEGER NOT NULL REFERENCES device_t(id),
     unique (label, device),
     unique (index, device)
-};
+);
 
-CREATE TABLE measurement_t {
+CREATE TABLE measurement_t (
     time TIMESTAMP WITHOUT TIME ZONE NOT NULL,
     device_name VARCHAR(16) NOT NULL,
     sensor_name VARCHAR(5) NOT NULL,
     temperature DOUBLE PRECISION
-};
+);
 
 SELECT create_hypertable('measurement_t', 'time');
 
+insert into device_t (device_id, label) values('eui-43fa12f400006c88', 'badesee');
+insert into sensor_t (address, label, index, device) 
+  values (13258914387362694952, '0,5m ', 0, 1), 
+         (12970366982842161448, '2,0m ', 1, 1), 
+         (10664523975231507496, '3,0m ', 2, 1), 
+         (15276209993662477608, '4,0m ', 3, 1);
+
+select d.label as device_label,
+       s.label as label,
+       s.address as address,
+       s.index as index
+  from device_t d, sensor_t s
+  where d.id = s.device;
+  

src/preprocess.py

@@ -6,36 +6,37 @@ import ssl
 import json
 import base64
 import struct
+import psycopg2
+import psycopg2.extras
 
-
+#DEVICES = {
-DEVICES = {
+#    'eui-43fa12f400006c88': {
-    'eui-43fa12f400006c88': {
+#        'label': 'Badesee',
-        'label': 'Badesee',
+#        'sensors': [
-        'sensors': [
+#            { 
-            { 
+#                'address': 0xb8012062f611c728,
-                'address': 0xb8012062f611c728,
+#                'label': '0,5m ',
-                'label': '0,5m ',
+#                'index': 0
-                'index': 0
+#            },
-            },
+#            {
-            {
+#                'address': 0xb400000d0ac31928,
-                'address': 0xb400000d0ac31928,
+#                'label': '2,0m ',
-                'label': '2,0m ',
+#                'index': 1
-                'index': 1
+#            },
-            },
+#            {
-            {
+#                'address': 0x9400000d6a4f8c28,
-                'address': 0x9400000d6a4f8c28,
+#                'label': '3,0m ',
-                'label': '3,0m ',
+#                'index': 2
-                'index': 2
+#            },
-            },
+#            {
-            {
+#                'address': 0xd400000d6a863528,
-                'address': 0xd400000d6a863528,
+#                'label': '4,0m ',
-                'label': '4,0m ',
+#                'index': 3
-                'index': 3
+#            }
-            }
+#
-
+#        ]
-        ]
+#    }
-    }
+#}
-}
 
 oldDevice = {}
 
@@ -47,11 +48,56 @@ class UnknownSensorException (Exception):
     def __init__(self, sensorAddress):
         self.sensorAddress = sensorAddress
 
-def getDevice(deviceId):
+class DbOp(object):
+    def __init__(self, config):
+        self.db_name = config['DB_NAME']
+        self.conn = None
+
+    def __getConn(self):
+        conn = psycopg2.connect(
+            database = self.db_name,
+            sslmode = 'require'
+        )
+        conn.autocommit = False
+        return conn
+
+    def getDevice(self, deviceId):
         try:
-        return DEVICES[deviceId]
+            logger.debug("1")
-    except KeyError:
+            conn = self.__getConn()
+            with conn:
+                with conn.cursor() as cur:
+                    cur.execute("select id, label from device_t where device_id = %(deviceId)s", { 'deviceId': deviceId })
+                    res = cur.fetchone()
+                    if res is None:
                         raise DeviceNotFoundException(deviceId)
+                    device_label = res[1]
+                    id = res[0]
+                logger.debug(f"{device_label=}")
+                with conn.cursor() as cur:
+                    cur.execute("select label, address, index from sensor_t where device = %(id)s", { 'id': id })
+                    sensors = cur.fetchall()
+                logger.debug(f"{sensors=}")
+                device = {
+                    'label': device_label,
+                    'sensors': []
+                }
+                for sensor in sensors:
+                    device['sensors'].append({
+                        'label': sensor[0],
+                        'address': int(sensor[1]),
+                        'index': sensor[2]
+                    })
+            return device
+        finally:
+            if conn:
+                conn.close()
+
+#def getDevice(deviceId):
+#    try:
+#        return DEVICES[deviceId]
+#    except KeyError:
+#        raise DeviceNotFoundException(deviceId)
 
 def storeMeasurement(measurement):
     logger.info(f"Store: {measurement=}")
@@ -72,7 +118,12 @@ def mqttOnMessageCallback(client, userdata, message):
 
         sendSetupMessage = False
         device_id = parse_payload['end_device_ids']['device_id']
-        device = getDevice(device_id)
+        # device = getDevice(device_id)
+
+        dbh = DbOp(config)
+        device = dbh.getDevice(device_id)
+        logger.debug(f"{device=}")
+
         global oldDevice
         if (device != oldDevice):
             logger.info("device configuration in database has changed")
@@ -163,7 +214,8 @@ REQUIRED_CONFIG_OPTIONS = [
     'MQTT_BROKER',
     'MQTT_PORT',
     'MQTT_CA',
-    'APPLICATION_TENANT'
+    'APPLICATION_TENANT',
+    'DB_NAME'
 ]
 
 config = {}

src/requirements.txt

@@ -1,2 +1,3 @@
 loguru==0.6.0
 paho-mqtt==1.6.1
+psycopg2==2.9.5