done so far

src/preprocess.py

@@ -9,34 +9,6 @@ import struct
 import psycopg2
 import psycopg2.extras
 
-#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
-#            }
-#
-#        ]
-#    }
-#}
 
 oldDevice = {}
 
@@ -63,7 +35,6 @@ class DbOp(object):
 
     def getDevice(self, deviceId):
         try:
-            logger.debug("1")
             conn = self.__getConn()
             with conn:
                 with conn.cursor() as cur:
@@ -89,20 +60,28 @@ class DbOp(object):
                         'index': sensor[2]
                     })
             return device
+        except Exception as e:
+            logger.error(f"Error getting device: {e}")
+            raise DeviceNotFoundException(deviceId)
         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=}")
+    def storeMeasurement(self, measurement):
+        try:
+            logger.info(f"About to store {measurement}")
+            conn = self.__getConn()
+            with conn:
+                with conn.cursor() as cur:
+                    for entry in measurement['measurements']:
+                        cur.execute("insert into measurement_t (time, device_name, sensor_name, temperature) values (%(time)s, %(dname)s, %(sname)s, %(tempc)s)",
+                                    { 'time': measurement['time'], 'dname': measurement['label'], 'sname': entry['label'], 'tempc': entry['value'] })
+        except Exception as e:
+            logger.error(f"Error storing measurement: {e}")
+        finally:
+            if conn:
+                conn.close()
     
-
 def mqttOnConnectCallback(client, userdata, flags, rc):
     logger.info("mqtt connected")
     mqtt_in_topic = f"v3/{config['APPLICATION_TENANT']}/devices/+/up"
@@ -118,11 +97,9 @@ def mqttOnMessageCallback(client, userdata, message):
 
         sendSetupMessage = False
         device_id = parse_payload['end_device_ids']['device_id']
-        # device = getDevice(device_id)
 
         dbh = DbOp(config)
         device = dbh.getDevice(device_id)
-        logger.debug(f"{device=}")
 
         global oldDevice
         if (device != oldDevice):
@@ -133,13 +110,11 @@ def mqttOnMessageCallback(client, userdata, message):
         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=}")
+        logger.debug(f"{frame=}, {status=}")
 
         if (status == 4):
             logger.info(f"Start up message received from {device_id}, {device['label']}")
-            # send downlink message with labels
             for i in range(0, num_of_sensors):
                 start_index = 2 + (i * 8)
                 end_index = start_index + 8
@@ -149,28 +124,28 @@ def mqttOnMessageCallback(client, userdata, message):
                     raise UnknownSensorException(received_sensor_address)
             sendSetupMessage = True
         else:
-            logger.info(f"Regular message received from {device_id}")
-            # continue to parse and process message
-            measurement = []
-            # sensor_addrs = []
-            # sensor_values = []
+            logger.info(f"Regular message received from {device_id}, {device['label']}")
+            measurement = {
+                'label': device['label'],
+                'time': parse_payload['received_at'],
+                'measurements': []
+            }
             for i in range(0, num_of_sensors):
                 addr_start_index = 2 + (i * (8 + 4))
                 addr_end_index = addr_start_index + 8
                 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
                 value = struct.unpack('<i', frame[value_start_index:value_end_index])[0] / 128
                 logger.debug(f"sensor {i}: 0x{received_sensor_address:016x} = {value:.2f} °C")
-                measurement.append({
+                measurement['measurements'].append({
                     "address": received_sensor_address,
                     "value": value,
                     "label": sensors[received_sensor_address][1]
                 })
-            storeMeasurement(measurement)
+            dbh.storeMeasurement(measurement)
 
         if (sendSetupMessage):
             sendSetupMessage = False
@@ -193,7 +168,6 @@ def mqttOnMessageCallback(client, userdata, message):
                 ]
             }))
 
-
     except UnknownSensorException as e:
         logger.error(f"unknown sensor in message {e.sensorAddress}")
     except DeviceNotFoundException as e: