database

schema/create-schema.sql

@@ -0,0 +1,25 @@
+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 {
+    id SERIAL NOT NULL PRIMARY KEY,
+    address NUMERIC NOT NULL UNIQUE,
+    label VARCHAR(5) NOT NULL,
+    index SMALLINT NOT NULL,
+    device INTEGER NOT NULL REFERENCES device_t(id),
+    unique (label, device),
+    unique (index, device)
+};
+
+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');
+

src/preprocess.py

@@ -7,6 +7,51 @@ import json
 import base64
 import struct
 
+
+DEVICES = {
+    'eui-43fa12f400006c88': {
+        'label': 'Badesee',
+        'sensors': [
+            { 
+                'address': 0xb8012062f611c728,
+                'label': '0,5m:',
+                'index': 0
+            },
+            {
+                'address': 0xb400000d0ac31928,
+                'label': '2,0m:',
+                'index': 1
+            },
+            {
+                'address': 0x9400000d6a4f8c28,
+                'label': '3,0m:',
+                'index': 2
+            },
+            {
+                'address': 0xd400000d6a863528,
+                'label': '4,0m:',
+                'index': 3
+            }
+
+        ]
+    }
+}
+
+class DeviceNotFoundException (Exception): 
+    def __init__(self, deviceId):
+        self.deviceId = deviceId
+
+class UnknownSensorException (Exception): 
+    def __init__(self, sensorAddress):
+        self.sensorAddress = sensorAddress
+
+def getDevice(deviceId):
+    try:
+        return DEVICES[deviceId]
+    except KeyError:
+        raise DeviceNotFoundException(deviceId)
+
+
 def mqttOnConnectCallback(client, userdata, flags, rc):
     logger.info("mqtt connected")
     mqtt_in_topic = f"v3/{config['APPLICATION_TENANT']}/devices/+/up"
@@ -19,36 +64,65 @@ def mqttOnMessageCallback(client, userdata, message):
         payload = message.payload
         logger.debug(f"mqtt message received: {topic} -> {payload}")
         parse_payload = json.loads(payload)
+
         device_id = parse_payload['end_device_ids']['device_id']
+        device = getDevice(device_id)
+        num_of_sensors = len(device['sensors'])
+        sensors = { x['address']:[x['index'], x['label']] for x in device['sensors']}
+
         frame = base64.b64decode(parse_payload['uplink_message']['frm_payload'])
         logger.info(f"{frame=}")
         status = struct.unpack('<H', frame[0:2])[0]
         logger.info(f"{status=}")
 
         if (status == 4):
-            logger.info(f"Start up message received from {device_id}")
+            logger.info(f"Start up message received from {device_id}, {device['label']}")
             # send downlink message with labels
-            sensor_addrs = []
-            for i in range(0, 4):
+            for i in range(0, num_of_sensors):
                 start_index = 2 + (i * 8)
                 end_index = start_index + 8
-                sensor_addrs.append(struct.unpack('<Q', frame[start_index:end_index])[0])
-                logger.debug(f"sensor {i}: 0x{sensor_addrs[i]:016x}")
+                received_sensor_address = struct.unpack('<Q', frame[start_index:end_index])[0]
+                logger.debug(f"sensor {i}: 0x{received_sensor_address:016x}")
+                if (received_sensor_address not in sensors):
+                    raise UnknownSensorException(received_sensor_address)
+            setupMessage = bytes()
+            for sk, sv in sensors.items():
+                logger.debug(f"{sk=}, {sv=}")
+                setupMessage += struct.pack('<Q', sk) + sv[0].to_bytes(1, byteorder='big') + bytes(sv[1], 'ASCII')
+            setupMessage = base64.b64encode(setupMessage).decode('ASCII')
+            logger.debug(f"about to send setup message {setupMessage}")
+            setupTopic = f"v3/{config['APPLICATION_TENANT']}/devices/{device_id}/down/push"
+            client.publish(setupTopic, json.dumps({
+                "downlinks": [
+                    {
+                        "f_port": 1,
+                        "frm_payload": setupMessage,
+                        "priority": "NORMAL"
+                    }
+                ]
+            }))
 
         else:
             logger.info(f"Regular message received from {device_id}")
             # continue to parse and process message
             sensor_addrs = []
             sensor_values = []
-            for i in range(0, 4):
+            for i in range(0, num_of_sensors):
                 addr_start_index = 2 + (i * (8 + 4))
                 addr_end_index = addr_start_index + 8
-                sensor_addrs.append(struct.unpack('<Q', frame[addr_start_index:addr_end_index])[0])
+                received_sensor_address = struct.unpack('<Q', frame[addr_start_index:addr_end_index])[0]
+                if (received_sensor_address not in sensors):
+                    raise UnknownSensorException(received_sensor_address)
+                sensor_addrs.append(received_sensor_address)
                 value_start_index = addr_end_index
                 value_end_index = value_start_index + 4
                 sensor_values.append(struct.unpack('<i', frame[value_start_index:value_end_index])[0] / 128)
                 logger.debug(f"sensor {i}: 0x{sensor_addrs[i]:016x} = {sensor_values[i]:.2f} °C")
 
+    except UnknownSensorException as e:
+        logger.error(f"unknown sensor in message {e.sensorAddress}")
+    except DeviceNotFoundException as e:
+        logger.error(f"message from unknown device {e.deviceId}")
     except Exception as e:
         logger.error(f"unable to parse message {payload}, {e}")