ci added

.gitignore

@@ -1,4 +1,5 @@
 ENV
 *.pyc
+.venv/*
 src/.venv/*
 

.gitlab-ci.yml

@@ -0,0 +1,55 @@
+stages:
+  - dockerize
+  - deploy
+
+variables:
+  IMAGE_NAME: $CI_REGISTRY/$CI_PROJECT_PATH
+
+dockerize:
+  image: registry.hottis.de/dockerized/docker-bash:latest
+  stage: dockerize
+  tags:
+    - hottis
+    - linux
+    - docker
+  script:
+    - docker build --tag $IMAGE_NAME:${CI_COMMIT_SHORT_SHA} .
+    - docker login -u gitlab-ci-token -p $CI_JOB_TOKEN $CI_REGISTRY;
+    - docker push $IMAGE_NAME:${CI_COMMIT_SHORT_SHA}
+    - if [ "$CI_COMMIT_TAG" != "" ]; then
+        docker tag $IMAGE_NAME:${CI_COMMIT_SHORT_SHA} $IMAGE_NAME:${CI_COMMIT_TAG};
+        docker push $IMAGE_NAME:${CI_COMMIT_TAG};
+      fi
+
+deploy:
+  image: registry.hottis.de/dockerized/docker-bash:latest
+  stage: deploy
+  tags:
+    - saerbeck-deployment-only
+  only:
+    - tags
+  variables:
+    GIT_STRATEGY: none
+    CONTAINER_NAME: preprocessor
+  script:
+    - docker stop $CONTAINER_NAME || echo "container not running, never mind"
+    - docker rm $CONTAINER_NAME || echo "container not existing, never mind"
+    - docker login -u gitlab-ci-token -p $CI_JOB_TOKEN $CI_REGISTRY 
+    - docker run 
+      -d
+      --name $CONTAINER_NAME
+      --restart always
+      --network internal-network
+      -e "APPLICATION_TENANT=$APPLICATION_TENANT"
+      -e "MQTT_LOGIN=$MQTT_LOGIN"
+      -e "MQTT_PASSWORD=$MQTT_PASSWORD"
+      -e "MQTT_BROKER=$MQTT_BROKER"
+      -e "PGHOST=timescaledb"
+      -e "PGUSER=$PGUSER"
+      -e "PGPASSWORD=$PGPASSWORD"
+      -e "PGDATABASE=$PGDATABASE"
+      $IMAGE_NAME:$CI_COMMIT_TAG
+    - docker network connect external-network $CONTAINER_NAME
+  environment:
+    name: production
+

Dockerfile

@@ -0,0 +1,37 @@
+FROM python:3.10-bullseye
+
+ENV APPLICATION_TENANT "-"
+
+ENV MQTT_LOGIN "-"
+ENV MQTT_PASSWORD "-"
+ENV MQTT_BROKER "-"
+ENV MQTT_PORT "8883"
+ENV MQTT_CA ""
+
+ENV PGHOST ""
+ENV PGPORT "5432"
+ENV PGUSER "-"
+ENV PGPASSWORD "-"
+ENV PGSSLMODE "disable"
+ENV PGDATABASE "-"
+
+ARG APP_DIR="/opt/app"
+
+RUN \
+    apt update && \
+    apt upgrade -y && \
+    apt install -y libpq-dev && \
+    mkdir -p ${APP_DIR}
+
+COPY src/requirements.txt ${APP_DIR}/
+COPY src/preprocess.py ${APP_DIR}/
+
+WORKDIR ${APP_DIR}
+
+RUN \
+    pip install -r requirements.txt
+
+USER nobody
+
+CMD [ "python", "preprocess.py" ]
+

docs/decoding.txt

@@ -0,0 +1,21 @@
+http://wiki.dragino.com/xwiki/bin/view/Main/User%20Manual%20for%20LoRaWAN%20End%20Nodes/LDDS75%20-%20LoRaWAN%20Distance%20Detection%20Sensor%20User%20Manual/
+
+
+Value, Size (bytes)
+Battery (mV), 2
+Distance (mm), 2
+Interrupt, 1
+Temperature, 2
+Status, 1
+
+
+payload = 'DQYCjAAAAAE='
+frame = base64.b64decode(payload)
+
+battery = struct.unpack('<H', frame[0:2])[0]
+distance = struct.unpack('<H', frame[2:4])[0]
+interrupt: frame[4:5]
+temperature: frame[5:7]
+status: struct.unpack('<H', frame[7:8])[0]
+
+

schema/create-schema.sql

@@ -0,0 +1,31 @@
+create database level_monitoring;
+create extension timescaledb;
+
+
+
+create table application_t (
+    id serial not null primary key,
+    device_id VARCHAR(32) NOT NULL UNIQUE,
+    label varchar(16) not null unique,
+    sensor_type varchar(16) not null,
+    ground_level numeric(10, 0) not null
+);
+
+create table measurement_t (
+    time timestamp without time zone not null,
+    application_name varchar(16) not null,
+    raw_level numeric(10, 0),
+    level numeric(10, 0),
+    status varchar(16),
+    battery float
+);
+
+select create_hypertable('measurement_t', 'time');
+
+insert into application_t (device_id, label, sensor_type, ground_level)
+    values('eui-a84041a2c18341d6', 'deilbach', 'LDDS75', 300);
+
+create user preprocessor password 'geheim';
+grant select on application_t to preprocessor;
+grant insert on measurement_t to preprocessor;
+

schema/extend-schema.sql

@@ -1,30 +0,0 @@
--- extend the schema monitoring from badesee application
-
-
-create table level_sensor_t (
-    id serial not null primary key,
-    sensor_type varchar(16) not null,
-    ground_level numeric(10, 0) not null,
-    device integer not null references device_id(id),
-    unique(device)
-);
-
-create table level_measurement_t (
-    time timestamp without time zone not null,
-    device_name varchar(16) not null,
-    raw_level numeric(10, 0),
-    level numeric(10, 0),
-    status varchar(16),
-    battery float
-);
-
-select create_hypertable('level_measurement_t', 'time');
-
-insert into device_t (device_id, label) values('eui-a84041a2c18341d6', 'deilbach');
-insert into level_sensor_t (sensor_type, ground_level, device) 
-    values('LDDS75', 300, (select id from device_t where device_id='eui-a84041a2c18341d6'));
-
-create user level_preprocessor password 'geheim';
-grant select on device_t, level_sensor_t to level_preprocessor;
-grant insert on level_measurement_t to level_preprocessor;
-

src/preprocess.py

@@ -10,13 +10,8 @@ import psycopg2
 import psycopg2.extras
 
 
-oldDevice = {}
 
-class DeviceNotFoundException (Exception): 
+class ApplicationNotFoundException (Exception): 
-    def __init__(self, deviceId):
-        self.deviceId = deviceId
-
-class UnknownSensorException (Exception): 
     def __init__(self, deviceId):
         self.deviceId = deviceId
 
@@ -29,35 +24,25 @@ class DbOp(object):
         conn.autocommit = False
         return conn
 
-    def getDevice(self, deviceId):
+    def getApplication(self, deviceId):
         try:
             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 })
+                    cur.execute("select label, sensor_type, ground_level from application_t where device_id = %(deviceId)s", { 'deviceId': deviceId })
                     res = cur.fetchone()
                     if res is None:
-                        raise DeviceNotFoundException(deviceId)
+                        raise ApplicationNotFoundException(deviceId)
-                    device_label = res[1]
+                    application = {
-                    id = res[0]
+                        'label': res[0],
-                logger.debug(f"{device_label=}")
+                        'sensor_type': res[1],
-                with conn.cursor() as cur:
+                        'ground_level': res[2]
-                    cur.execute("select sensor_type, ground_level from level_sensor_t where device = %(id)s", { 'id': id })
+                    }
-                    res = cur.fetchone()
+                logger.debug(f"{application=}")
-                    if res is None:
+            return application
-                        raise UnknownSensorException(deviceId)
-                    sensor_type = res[0]
-                    ground_level = res[1]
-                logger.debug(f"{sensor_type=}, {ground_level=}")
-                device = {
-                    'label': device_label,
-                    'sensor_type': sensor_type,
-                    'ground_level': ground_level
-                }
-            return device
         except Exception as e:
             logger.error(f"Error getting device: {e}")
-            raise DeviceNotFoundException(deviceId)
+            raise ApplicationNotFoundException(deviceId)
         finally:
             if conn:
                 conn.close()
@@ -68,9 +53,18 @@ class DbOp(object):
             conn = self.__getConn()
             with conn:
                 with conn.cursor() as cur:
-                    for entry in measurement['measurements']:
+                    cur.execute("""
-                        cur.execute("insert into measurement_t (time, device_name, sensor_name, temperature) values (now(), %(dname)s, %(sname)s, %(tempc)s)",
+                        insert into measurement_t (time, application_name, raw_level, level, status, battery) 
-                                    { 'dname': measurement['label'], 'sname': entry['label'], 'tempc': entry['value'] })
+                            values (now(), %(application_name)s, %(raw_level)s, %(level)s, %(status)s, %(battery)s)
+                        """,
+                        { 
+                            'application_name': measurement['application_name'], 
+                            'raw_level': measurement['raw_level'],
+                            'level': measurement['level'],
+                            'status': measurement['status'],
+                            'battery': measurement['battery']
+                        }
+                    )
         except Exception as e:
             logger.error(f"Error storing measurement: {e}")
         finally:
@@ -94,78 +88,27 @@ def mqttOnMessageCallback(client, userdata, message):
         device_id = parse_payload['end_device_ids']['device_id']
 
         dbh = DbOp(config)
-        device = dbh.getDevice(device_id)
+        application = dbh.getApplication(device_id)
-
-        global oldDevice
-        if (device != oldDevice):
-            logger.info("device configuration in database has changed")
-            oldDevice = device
-            sendSetupMessage = True
-        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'])
-        status = struct.unpack('<H', frame[0:2])[0]
-        logger.debug(f"{frame=}, {status=}")
 
-        if (status == 4):
+        battery = struct.unpack('>H', frame[0:2])[0]
-            logger.info(f"Start up message received from {device_id}, {device['label']}")
+        distance = struct.unpack('>H', frame[2:4])[0]
-            for i in range(0, num_of_sensors):
+        status = struct.unpack('?', frame[7:8])[0]
-                start_index = 2 + (i * 8)
+        logger.debug(f"{frame=}, {battery=}, {distance=}, {status=}")
-                end_index = start_index + 8
-                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)
-            sendSetupMessage = True
-        else:
-            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)
-                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['measurements'].append({
-                    "address": received_sensor_address,
-                    "value": value,
-                    "label": sensors[received_sensor_address][1]
-                })
-            dbh.storeMeasurement(measurement)
 
-        if (sendSetupMessage):
+        measurement = {
-            sendSetupMessage = False
+            'application_name': application['label'],
-            setupMessage = bytes()
+            'raw_level': distance,
-            null = 0
+            'level': application['ground_level'] - distance,
-            null = null.to_bytes(1, byteorder='big')
+            'status': 'Ok' if status else 'No sensor',
-            for sk, sv in sensors.items():
+            'battery': battery / 100
-                logger.debug(f"{sk=}, {sv=}")
+        }
-                setupMessage += struct.pack('<Q', sk) + sv[0].to_bytes(1, byteorder='big') + bytes(sv[1], 'ASCII') + null
+        logger.debug(f"{measurement=}")
-            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"
-                    }
-                ]
-            }))
 
-    except UnknownSensorException as e:
+        dbh.storeMeasurement(measurement)
-        logger.error(f"unknown sensor in message {e.sensorAddress}")
+
-    except DeviceNotFoundException as e:
+    except ApplicationNotFoundException as e:
         logger.error(f"message from unknown device {e.deviceId}")
     except Exception as e:
         logger.error(f"unable to parse message {payload}, {e}")
@@ -206,6 +149,7 @@ client.tls_set(
     ciphers=None
 )
 client.connect(config["MQTT_BROKER"], int(config["MQTT_PORT"]))
+#client.connect('172.16.2.16', 1883)
 logger.info("mqtt loop starting")
 client.loop_forever()