fix in deploy script

.gitignore

@@ -1,6 +1,7 @@
-config-*.json
 src/udi/udi
 src/udi/migrate_schema
 tmp/
 ENVDB
 ENVDB.cluster
+deployment/secrets.txt
+deployment/secrets

.woodpecker.yml

@@ -2,8 +2,7 @@ steps:
   build:
     image: plugins/kaniko
     settings:
-      repo:
-        from_secret: image_name
+      repo: gitea.hottis.de/wn/udi
       registry: 
         from_secret: container_registry
       tags: latest,${CI_COMMIT_SHA},${CI_COMMIT_TAG}
@@ -20,11 +19,14 @@ steps:
     secrets:
       - source: kube_config
         target: KUBE_CONFIG_CONTENT
-      - source: image_name
-        target: IMAGE_NAME
+      - source: encryption_key
+        target: ENCRYPTION_KEY
+      - source: secrets_checksum
+        target: MD5_CHECKSUM
     commands:
+      - export IMAGE_TAG=$CI_COMMIT_TAG
       - printf "$KUBE_CONFIG_CONTENT" > /tmp/kubeconfig
       - export KUBECONFIG=/tmp/kubeconfig
-      - cat $CI_WORKSPACE/deployment/deploy-yml.tmpl | sed -e 's,%IMAGE%,'$IMAGE_NAME':'$CI_COMMIT_TAG',' | kubectl apply -f - 
+      - ./deployment/deploy.sh
     when:
       - event: tag

deployment/decrypt-secrets.sh

@@ -0,0 +1,43 @@
+#!/bin/bash
+
+if [ "$ENCRYPTION_KEY" = "" ]; then
+  echo "ENCRYPTION_KEY not set"
+  exit 1
+fi
+
+if [ "$MD5_CHECKSUM" = "" ]; then
+  echo "No checksum given"
+  exit 1
+fi
+
+SECRETS_CIPHERTEXT_FILE=secrets.enc
+SECRETS_PLAINTEXT_FILE=/tmp/secrets
+TMP_FILE=`mktemp`
+POD_NAME_SUFFIX=`date +%s`
+
+cat $SECRETS_CIPHERTEXT_FILE | \
+  kubectl run openssl-$POD_NAME_SUFFIX \
+    --rm \
+    --image bitnami/debian-base-buildpack:latest \
+    --env KEY=$ENCRYPTION_KEY \
+    -i \
+    -q \
+    -- \
+    /bin/sh -c "openssl enc -aes-256-cbc -salt -pass env:KEY -a -d" > \
+    $TMP_FILE
+
+if [ `uname` = "Darwin" ]; then
+  CALCULATED_CHECKSUM=`cat $TMP_FILE | md5`
+elif [ `uname` = "Linux" ]; then
+  CALCULATED_CHECKSUM=`cat $TMP_FILE | md5sum - | awk '{print $1}'`
+fi
+
+if [ "$MD5_CHECKSUM" != "$CALCULATED_CHECKSUM" ]; then
+  echo "Invalid checksum"
+  exit 1
+fi
+
+# cat $TMP_FILE
+mv $TMP_FILE $SECRETS_PLAINTEXT_FILE
+
+

deployment/deploy-yml.tmpl

@@ -1,8 +1,7 @@
 apiVersion: v1
 kind: PersistentVolumeClaim
 metadata:
-  name: udi-archive
-  namespace: udi
+  name: %PRE%-udi-archive
 spec:
   accessModes:
     - ReadWriteOnce
@@ -14,12 +13,11 @@ spec:
 apiVersion: apps/v1
 kind: Deployment
 metadata:
-  name: udi
-  namespace: udi
+  name: %PRE%-udi
   labels:
     app: udi
   annotations:
-    secret.reloader.stakater.com/reload: "udi-conf,udi-db-cred"
+    secret.reloader.stakater.com/reload: "%PRE%-udi-conf,%PRE%-udi-db-cred,%PRE%-mqtt-password"
 spec:
   replicas: 1
   selector:
@@ -33,20 +31,18 @@ spec:
       containers:
         - name: udi
           image: %IMAGE%
-          env:
-            - name: UDI_CONF
-              valueFrom:
-                secretKeyRef:
-                  name: udi-conf
-                  key: UDI_CONF
           envFrom:
             - secretRef: 
-                name: udi-db-cred
+                name: %PRE%-udi-db-cred
+            - secretRef:
+                name: %PRE%-mqtt-password
+            - configMapRef:
+                name: %PRE%-udi-conf
           volumeMounts:
             - mountPath: /archive
               name: udi-archive
       volumes:
         - name: udi-archive
           persistentVolumeClaim:
-            claimName: udi-archive
+            claimName: %PRE%-udi-archive
 

deployment/deploy.sh

@@ -0,0 +1,92 @@
+#!/bin/bash
+
+if [ "$IMAGE_TAG" == "" ]; then
+  echo "Make sure IMAGE_TAG is set"
+  exit 1
+fi
+
+
+IMAGE_NAME=gitea.hottis.de/wn/udi
+
+CONFIG_FILE=config.json
+
+
+DEPLOYMENT_DIR=$PWD/deployment
+INSTANCES_DIR=$DEPLOYMENT_DIR/instances
+
+pushd $DEPLOYMENT_DIR > /dev/null
+./decrypt-secrets.sh || exit 1 
+. /tmp/secrets
+rm /tmp/secrets
+popd > /dev/null
+
+for NAMESPACE_DIR in `find $INSTANCES_DIR -type d -mindepth 1 -maxdepth 1`; do
+  NAMESPACE=`basename $NAMESPACE_DIR`
+  echo "Namespace: $NAMESPACE"
+
+  kubectl create namespace $NAMESPACE \
+    --dry-run=client \
+    -o yaml | \
+    kubectl -f - apply
+
+  pushd $NAMESPACE_DIR > /dev/null
+  for INSTANCE_DIR in `find . -type d -mindepth 1 -maxdepth 1`; do
+    pushd $INSTANCE_DIR > /dev/null
+    INSTANCE=`basename $INSTANCE_DIR`
+    echo "Instance: $INSTANCE"
+
+    # set secret configuration from encrypted and decrypted file
+    VARIABLE_PREFIX=`echo "$NAMESPACE""_""$INSTANCE" | tr - _`
+
+    
+    # set MQTT_PASSWORD as secret
+    MQTT_PASSWORD_VARIABLE=$VARIABLE_PREFIX"_MQTT_PASSWORD"
+    MQTT_PASSWORD="${!MQTT_PASSWORD_VARIABLE}"
+    # echo "MQTT_PASSWORD_VARIABLE: $MQTT_PASSWORD_VARIABLE"
+    # echo "MQTT_PASSWORD: $MQTT_PASSWORD"
+    kubectl create secret generic $INSTANCE-mqtt-password \
+      --from-literal=MQTT_PASSWORD="$MQTT_PASSWORD" \
+      --dry-run=client \
+      -o yaml \
+      --save-config | \
+      kubectl apply -f - -n $NAMESPACE
+
+
+    LOGIN_VARIABLE=$VARIABLE_PREFIX"_PGUSER"
+    NEW_UDI_DB_LOGIN="${!LOGIN_VARIABLE}"
+    PASSWORD_VARIABLE=$VARIABLE_PREFIX"_PGPASSWORD"
+    NEW_UDI_DB_PASSWORD="${!PASSWORD_VARIABLE}"
+    DATABASE_VARIABLE=$VARIABLE_PREFIX"_PGDATABASE"
+    NEW_UDI_DB_DATABASE="${!DATABASE_VARIABLE}"
+    NEW_UDI_DB_HOST=timescaledb.database.svc.cluster.local
+
+    kubectl create secret generic $INSTANCE-udi-db-cred \
+      --dry-run=client \
+      -o yaml \
+      --save-config \
+      --from-literal=PGUSER="$NEW_UDI_DB_LOGIN" \
+      --from-literal=PGPASSWORD="$NEW_UDI_DB_PASSWORD" \
+      --from-literal=PGDATABASE="$NEW_UDI_DB_DATABASE" \
+      --from-literal=PGHOST="$NEW_UDI_DB_HOST" \
+      --from-literal=PGSSLMODE="require" | \
+      kubectl apply -f - -n $NAMESPACE
+
+    # set configuration as configMap
+    kubectl create configmap $INSTANCE-udi-conf \
+      --from-literal=UDI_CONF="`cat $CONFIG_FILE`" \
+      --dry-run=client \
+      -o yaml \
+      --save-config | \
+      kubectl apply -f - -n $NAMESPACE
+
+    # prepare k8s deployment statement
+    cat $DEPLOYMENT_DIR/deploy-yml.tmpl | \
+      sed -e 's,%IMAGE%,'$IMAGE_NAME':'$IMAGE_TAG','g | \
+      sed -e 's,%PRE%,'$INSTANCE','g | \
+      kubectl apply -f - -n $NAMESPACE
+
+    popd > /dev/null
+  done
+  popd > /dev/null
+done
+

deployment/encrypt-secrets.sh

@@ -0,0 +1,29 @@
+#!/bin/bash
+
+if [ "$ENCRYPTION_KEY" = "" ]; then
+  echo "ENCRYPTION_KEY not set"
+  exit 1
+fi
+
+SECRETS_PLAINTEXT_FILE=secrets.txt
+SECRETS_CIPHERTEXT_FILE=secrets.enc
+
+if [ `uname` = "Darwin" ]; then
+  cat $SECRETS_PLAINTEXT_FILE | md5
+elif [ `uname` = "Linux" ]; then
+  cat $SECRETS_PLAINTEXT_FILE | md5sum - | awk '{print $1}'
+fi
+
+POD_NAME_SUFFIX=`date +%s`
+
+cat $SECRETS_PLAINTEXT_FILE | \
+  kubectl run openssl-$POD_NAME_SUFFIX \
+    --rm \
+    --image bitnami/debian-base-buildpack:latest \
+    --env KEY=$ENCRYPTION_KEY \
+    -i \
+    -q \
+    -- \
+    /bin/sh -c "openssl enc -aes-256-cbc -salt -pass env:KEY -a" > \
+    $SECRETS_CIPHERTEXT_FILE
+

deployment/instances/udi-berresheim/default/config.json

@@ -0,0 +1,23 @@
+{
+  "mqtt": {
+    "broker": "ssl://eu1.cloud.thethings.network:8883",
+    "username": "de-hottis-level-monitoring@ttn",
+    "password": "ENV",
+    "tlsEnable": "true"
+  },
+  "topicMappings": [
+    {
+      "topics": [ "v3/#" ],
+      "handler": "TTN",
+      "id": "TTN0",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+        }
+      }
+    }
+  ],
+  "archiver": {
+    "dir": "/archive"
+  }
+}

deployment/instances/udi-pg/default/config.json

@@ -0,0 +1,21 @@
+{
+  "mqtt": {
+    "broker": "ssl://eu1.cloud.thethings.network:8883",
+    "username": "com-passavant-geiger-poc@ttn",
+    "tlsEnable": "true"
+  },
+  "topicMappings": [
+    {
+      "topics": [ "v3/com-passavant-geiger-poc@ttn/devices/#" ],
+      "handler": "TTN",
+      "id": "TTN0",
+      "config": {
+        "attributes": {
+        }
+      }
+    }
+  ],
+  "archiver": {
+    "dir": "/archive"
+  }
+}

deployment/instances/udi-saerbeck/level/config.json

@@ -0,0 +1,22 @@
+{
+  "mqtt": {
+    "broker": "ssl://eu1.cloud.thethings.network:8883",
+    "username": "de-hottis-level-monitoring@ttn",
+    "password": "ENV",
+    "tlsEnable": "true"
+  },
+  "topicMappings": [
+    {
+      "topics": [ "v3/#" ],
+      "handler": "TTN",
+      "id": "TTN0",
+      "config": {
+        "attributes": {
+        }
+      }
+    }
+  ],
+  "archiver": {
+    "dir": "/archive"
+  }
+}

deployment/instances/udi-saerbeck/soil/config.json

@@ -0,0 +1,22 @@
+{
+  "mqtt": {
+    "broker": "ssl://eu1.cloud.thethings.network:8883",
+    "username": "de-hottis-lora-test1@ttn",
+    "password": "ENV",
+    "tlsEnable": "true"
+  },
+  "topicMappings": [
+    {
+      "topics": [ "v3/#" ],
+      "handler": "TTN",
+      "id": "TTN0",
+      "config": {
+        "attributes": {
+        }
+      }
+    }
+  ],
+  "archiver": {
+    "dir": "/archive"
+  }
+}

deployment/instances/udi/default/config.json

@@ -0,0 +1,135 @@
+{
+  "mqtt": {
+    "broker": "mqtt://emqx01-anonymous-cluster-internal.broker.svc.cluster.local:1883",
+    "tlsEnable": "false"
+  },
+  "topicMappings": [
+    {
+      "topics": [ "dt1/ai/periodic/1" ],
+      "handler": "DT1T",
+      "id": "DT1T.0",
+      "config": {
+        "attributes": {
+          "Application": "Temperature Wago",
+          "Device": "Freezer",
+          "HardLow": "-273",
+          "SoftLow": "-50",
+          "SoftHigh": "20",
+          "HardHigh": "100"
+        }
+      }
+    },
+    {
+      "topics": [ "dt1/ai/periodic/3" ],
+      "handler": "DT1T",
+      "id": "DT1T.1",
+      "config": {
+        "attributes": {
+          "Application": "Temperature Wago",
+          "Device": "Outdoor",
+          "HardLow": "-273",
+          "SoftLow": "-60",
+          "SoftHigh": "60",
+          "HardHigh": "100"
+        }
+      }
+    },
+    {
+      "topics": [ "IoT/PV/Values" ],
+      "handler": "PV",
+      "id": "PV",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+        }
+      }
+    },
+    {
+      "topics": [ "IoT/MBGW3/Measurement" ],
+      "handler": "MBGW3",
+      "id": "MBGW3",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+        }
+      }
+    },
+    {
+      "topics": [ "IoT/OneWireGW/Bus 1/#" ],
+      "handler": "SVER",
+      "id": "SVER0",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+          "application": "Temperature Heating",
+          "payloadRegex": "(\\d+(\\.\\d+)?)\\s*([^0-9\\s]\\S*)",
+          "deviceFrom": "topic", 
+          "devicePart": "3",
+          "valueFrom": "payload",
+          "valuePart": "1",
+          "unitFrom": "payload",
+          "unitPart": "3"
+        }
+      }
+    },
+    {
+      "topics": [ "NR/Multisensor/+/Temperatur" ],
+      "handler": "SVEJ",
+      "id": "SVEJ0",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+          "application": "Temperature Multisensor",
+          "deviceSelector": "T:2",
+          "valueSelector": "J:$.CurrentTemperature",
+          "unitSelector": "C:°C"
+        }
+      }
+    },
+    {
+      "topics": [ "NR/Multisensor/+/Feuchte" ],
+      "handler": "SVEJ",
+      "id": "SVEJ1",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+          "application": "Humidity Multisensor",
+          "deviceSelector": "T:2",
+          "valueSelector": "J:$.CurrentRelativeHumidity",
+          "unitSelector": "C:%"
+        }
+      }
+    },
+    {
+      "topics": [ "shellyplusht/+/status/temperature:0" ],
+      "handler": "SVEJ",
+      "id": "SVEJ2",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+          "application": "Temperature Shelly Plus HT",
+          "deviceSelector": "T:1",
+          "valueSelector": "J:$.tC",
+          "unitSelector": "C:°C"
+        }
+      }
+    },
+    {
+      "topics": [ "shellyplusht/+/status/humidity:0" ],
+      "handler": "SVEJ",
+      "id": "SVE4",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+          "application": "Humidity Shelly Plus HT",
+          "deviceSelector": "T:1",
+          "valueSelector": "J:$.rh",
+          "unitSelector": "C:%"
+        }
+      }
+    }
+  ],
+  "archiver": {
+    "dir": "/archive"
+  }
+}

deployment/instances/udi/soil/config.json

@@ -0,0 +1,22 @@
+{
+  "mqtt": {
+    "broker": "ssl://eu1.cloud.thethings.network:8883",
+    "username": "de-hottis-app01@ttn",
+    "password": "ENV",
+    "tlsEnable": "true"
+  },
+  "topicMappings": [
+    {
+      "topics": [ "v3/#" ],
+      "handler": "TTN",
+      "id": "TTN0",
+      "config": {
+        "attributes": {
+        }
+      }
+    }
+  ],
+  "archiver": {
+    "dir": "/archive"
+  }
+}

deployment/load-config.sh

@@ -1,47 +0,0 @@
-#!/bin/bash
-
-FILE=$1
-if [ "$FILE" = "" ]; then
-  echo "give config file to load as first argument"
-  exit 1
-fi
-SECRET_NAME=$2
-if [ "$SECRET_NAME" = "" ]; then
-  echo "give secret name to create/modify as second argument"
-  exit 1
-fi
-NAMESPACE=$3
-if [ "$NAMESPACE" = "" ]; then
-  echo "give namespace as third argument"
-  exit 1
-fi
-
-kubectl create secret generic $SECRET_NAME \
-  --from-literal=UDI_CONF="`cat $FILE`" \
-  -n $NAMESPACE \
-  --dry-run=client \
-  -o yaml \
-  --save-config | \
-kubectl apply -f -
-
-. ~/Workspace/MyKubernetesEnv/ENVDB
-DATABASE=udi
-LOGIN=udi
-PASSWORD=`openssl rand -base64 24` 
-psql <<EOF
-ALTER USER $LOGIN WITH PASSWORD '$PASSWORD';
-GRANT ALL PRIVILEGES ON DATABASE $DATABASE TO $LOGIN;
-COMMIT;
-EOF
-
-kubectl create secret generic udi-db-cred \
-  --dry-run=client \
-  -o yaml \
-  --save-config \
-  --from-literal=PGUSER="$LOGIN" \
-  --from-literal=PGHOST="timescaledb.database.svc.cluster.local" \
-  --from-literal=PGPASSWORD="$PASSWORD" \
-  --from-literal=PGSSLMODE="require" \
-  --from-literal=PGDATABASE="$DATABASE" | \
-kubectl apply -f - -n $NAMESPACE
-

deployment/secrets.enc

@@ -0,0 +1,34 @@
+U2FsdGVkX1+v9CRp5LutINfnhA50Nwr9JvSomaQcLKASxmGuqlww0UvBr95u3Ton
+VFWFT+AlVKftqMXoXcWqcSDZ9dSc2iyBKJglxP1gfY6r+Rc5/FMMYt+blYexv8s3
+hXmfnzXBFJJsaFAIHaENJacmU2iI89A534WbCrt0yiq3ul4CznuhxCGCFcfUMZPh
+/6a7wrZ9tIZmt87Z7lxjC765anBoCuoX3iGHw7AtSsBh38SnTARnhcuxEu3aSXJD
+s+kGRzEmmFDD2vxW95MXkXkDceuzru4oPgWKiTvlvuZS5hER8JVLVZqGYVhsk/+Y
+hZmHPn5wVGBbm/3/mPDm0rTnIVOhdXUqnEPGxz0m7h/0i6QvaKo9ylWscxIMoKp4
+IUKLqEPkULG8OnKzzkCZ3yfyHqUJNFq0KZA0FQXkCifkURGKo9EFz9rYrepGsL7C
+A7TYBL3f5jHQAOmCob2tBp46sh2E4n8RhglAX6JynCdszd65U+DdwERloNlekg+I
++x2Hx94ROYlvYELK1bdITATpijzCsHinfNJ3PHDz85SYh9JAA3ljdOUazoQaT8+n
+R674SnVX7CR793KCgV0wulyHOkAvjkC26mCt4NaEsGW499vYSyzfpneJSx2WY/I2
+MrMNW7IUGjHDZEdSQisGbs3WZd9sZdZFC858QTLKsXVy90lH41ih6SysxNXMSP6G
+Hz7IHGAk2GhzuEj2NALE98hQ9I76T6POOhmREpTlJEkyHGujsW8N4+n5tOx1S2pR
+4Sm8Sl0zQYMHgJcGq/e+HTkCuQ6cuO0yAn2T1oJjwe2OA5OpReljGp6b/DjCY/P6
+/J7hL8EpHa2OTUccSYWNNTfB8oz5oyOnvQTKc3GY6uNJYFfzNkLz9ZsxTi/t6Q3P
+WyxIVCR8dE6z3xPgZxq34K1W44TGsLI/r3DDgTMIhaXr1fcKO69AcAAXFehMqy9L
+Rygg8jQkOxTonSwHQA6MB8oVXfMjsTxKX7NnDHj6dndNRuUMCf5ulmfFCUg7jqxH
+rUL3XOb71mgUWRaI1NY4wOf6wM7NwsGDz+Uyp8LWC7XrOXJsWv6vPNQl/5W4nIRA
+jPYHRf1kC0ZmylB+R5CXh+kf2248iiBPS4fNdEWdj+95hCJ8eSaxu7ROtUt6ZtXO
+aN/0Y9UcU8isjXe7PJilqN8O0KCHHUk4lFlROD/FOkhiPfwoVBfJbObv+TWMB0ir
+pIxeklB1QQrxcxT/ozBi00Bp0Um03nJWU18WRY3lheO+ulEmb9HDvy/iedh2gGVG
+F7JVS68k+5Y8xi+MtcKY3Gm/bh89HjwNgJY7eH+SItpNGoZybbjJNawsd4HnQVS4
+nrtma0RYl3+y48Yf9pWPggKuZ0l7Glji4+dXasDVz7MKOhU9WotIFjmhyLhLkSa+
+tQvEqunAYK8lAXCB6Pgjb7V0AAroLSRR5FGcWaFWXkVYJPBqTopz16W1BgyTH2Bv
+EfYCGSJJXKwuo3jRgHWn/NM3WA25hespcDieKZKU6t50RHiExgySdI4daAM4Kj7X
+OOAIvH1mwOrPGAOpQGOBEaQbUfKcVl3rF0uHAvSy6UeEOFd9NUnb0FSCFcz79SNn
+Q0PkkvII3m9qvRFsKlf6OR6mm5IKARs/juzZxu8zHahJC3NbJ7NcTPMu9LGGrV3t
+vIifvLAsSMhqY2BC+FCVbEIErJa5UhOE6xFOUOi4RDpi4/vfBdCWRoijHQs6HSrk
+c/ihUIOxK580QqHjT4tY9UnwqEY/UlA44SInrSy+ZretWlzls0fLaPbhRv9VNi+v
+sJjAVcwE+56/lPN95599dB3/2PmdzC00BZBhZDQ9v5g/NzMJ/42PO4OWflL9R7Ix
+DVkV8kIuHlU8r+bMavAKDkURGaB/BnsZHLG2754WomcX7WElJYRK7CzlCKvTBGc7
+WvG7QdfUvQj+m0zZHVxf9K1ZK5nw31yM1Y2fOp91oNpgqYv01URpJYs9Ef25L0Nu
+Rj0Unpohunw2LQLp3k/zbSlTny/HmndcOflB4TNvV4nICW+hiU6eH7f/Xy3E+M2d
+2yOBnTfEJlJcLO/Q8ImH7Ex7b30Q+3elkSUlHCU0zTVVYmh3yQIouDvPAjy6Hy7v
+52Mwww/oSasugEQaUpkEFNfs3N0O/s1ngaErqAZtv+5Ab2FwftYTJ+q4lQY8yqRl

migration/.gitignore

@@ -0,0 +1 @@
+.venv

migration/migrate-level.py

@@ -0,0 +1,29 @@
+import psycopg2
+from loguru import logger
+
+try:
+    srcConn = psycopg2.connect(database="level_monitoring_berresheim")
+    srcConn.autocommit = False
+
+    with srcConn.cursor() as srcCur:
+        srcCur.execute("select time, application_name, raw_level, level, status, battery from measurement_t")
+        for srcObj in srcCur:
+            timestamp = srcObj[0]
+            deviceName = srcObj[1]
+            rawLevel = srcObj[2]
+            level = srcObj[3]
+            status = srcObj[4]
+            battery = srcObj[5]
+
+            logger.info(f"{timestamp=}, {deviceName=}, {rawLevel=}, {level=}, {status=}, {battery=}")
+
+            destTime = timestamp
+            destApplication = "de-hottis-level-monitoring"
+            destDevice = "eui-a84041a2c18341d6"
+            destAttributes = '{"ApplicationId":"de-hottis-level-monitoring", "DeviceType":"dragino-ldds75", "Hint": "Migrated"}'
+            destValues = '{"Battery":{"unit":"V","label":"Battery","value":' + str(battery) + ',"variable":"Voltage"}, "Distance":{"unit":mm","label":"Distance","variable":"Level","value":' + str(rawLevel) + '}, "CorrectedDistance":{"unit":"mm", "label":"CorrectedDistance", "variable":"Level","value":' + str(level) + '}}'
+            logger.info(f"{destTime=}, {destApplication=}, {destDevice=}, {destAttributes=}, {destValues=}")
+finally:
+    if srcConn:
+        srcConn.close()
+

migration/requirements.txt

@@ -0,0 +1,2 @@
+loguru==0.7.2
+psycopg2==2.9.9

queries/hottis.sql

@@ -37,3 +37,11 @@ create or replace view gas_v as
     from measurements
     where application = 'Gas' and
           attributes->>'Status' = 'Ok';
+
+create or replace view temperature_v as
+  select time,
+         cast(values->'Value'->>'value' as float) as temperature,
+         device
+    from measurements
+    where application in ('Temperature Multisensor', 'Temperature Shelly Plus HT');
+

queries/pg.sql

@@ -0,0 +1,20 @@
+create or replace view power_v as
+  select time,
+         cast(values->'ActivePowerL1'->>'value' as float) as power_l1,
+         cast(values->'ActivePowerL2'->>'value' as float) as power_l2,
+         cast(values->'ActivePowerL3'->>'value' as float) as power_l3,
+         device
+    from measurements
+    where application = 'com-passavant-geiger-poc' and
+          attributes->>'FPort' = '1';
+
+create or replace view power_factor_v as
+  select time,
+         cast(values->'PowerfactorL1'->>'value' as float) as factor_l1,
+         cast(values->'PowerfactorL2'->>'value' as float) as factor_l2,
+         cast(values->'PowerfactorL3'->>'value' as float) as factor_l3,
+         device
+    from measurements
+    where application = 'com-passavant-geiger-poc' and
+          attributes->>'FPort' = '1';
+

src/udi/ENVDB.udiload

@@ -0,0 +1,10 @@
+if [ "$1" = "" ]; then
+  echo "set namespace as argument"
+fi
+N=$1
+PGHOST=`kubectl get services traefik -n system -o jsonpath="{.status.loadBalancer.ingress[0].ip}"`
+PGPASSWORD=`kubectl get secrets udi-db-cred -n $N -o jsonpath="{.data.PGPASSWORD}" | base64 --decode`
+PGUSER=`kubectl get secrets udi-db-cred -n $N -o jsonpath="{.data.PGUSER}" | base64 --decode`
+PGSSLMODE=`kubectl get secrets udi-db-cred -n $N -o jsonpath="{.data.PGSSLMODE}" | base64 --decode`
+PGDATABASE=`kubectl get secrets udi-db-cred -n $N -o jsonpath="{.data.PGDATABASE}" | base64 --decode`
+export PGUSER PGHOST PGPASSWORD PGSSLMODE PGDATABASE

src/udi/ENVDB.uditest

@@ -0,0 +1,6 @@
+PGUSER="uditest"
+PGHOST=`kubectl get services traefik -n system -o jsonpath="{.status.loadBalancer.ingress[0].ip}"`
+PGPASSWORD=`kubectl get secrets uditest-db-cred -n udi-test -o jsonpath="{.data.PGPASSWORD}" | base64 --decode`
+PGSSLMODE=require
+PGDATABASE="uditest"
+export PGUSER PGHOST PGPASSWORD PGSSLMODE PGDATABASE

src/udi/config-test-ttn.json

@@ -0,0 +1,21 @@
+{
+  "mqtt": {
+    "broker": "ssl://eu1.cloud.thethings.network:8883",
+    "username": "de-hottis-lora-test1@ttn",
+    "tlsEnable": "true"
+  },
+  "topicMappings": [
+    {
+      "topics": [ "v3/#" ],
+      "handler": "TTN",
+      "id": "TTN0",
+      "config": {
+        "attributes": {
+        }
+      }
+    }
+  ],
+  "archiver": {
+    "dir": "./tmp/udi"
+  }
+}

src/udi/config-test.json

@@ -0,0 +1,133 @@
+{
+  "mqtt": {
+    "broker": "mqtt://172.23.1.102:1883",
+    "tlsEnable": "false"
+  },
+  "topicMappings": [
+    {
+      "topics": [ "IoT/PV/Values" ],
+      "handler": "PV",
+      "id": "PV",
+      "config": {
+        "attributes": {
+        }
+      }
+    },
+    {
+      "topics": [ "IoT/MBGW3/Measurement" ],
+      "handler": "MBGW3",
+      "id": "MBGW3",
+      "config": {
+        "attributes": {
+        }
+      }
+    },
+    {
+      "topics": [ "dt1/ai/periodic/1" ],
+      "handler": "DT1T",
+      "id": "DT1T.0",
+      "config": {
+        "attributes": {
+          "Application": "Temperature Wago",
+          "Device": "Freezer",
+          "HardLow": "-273",
+          "SoftLow": "-50",
+          "SoftHigh": "20",
+          "HardHigh": "100"
+        }
+      }
+    },
+    {
+      "topics": [ "dt1/ai/periodic/3" ],
+      "handler": "DT1T",
+      "id": "DT1T.1",
+      "config": {
+        "attributes": {
+          "Application": "Temperature Wago",
+          "Device": "Outdoor",
+          "HardLow": "-273",
+          "SoftLow": "-60",
+          "SoftHigh": "60",
+          "HardHigh": "100"
+        }
+      }
+    },
+    {
+      "topics": [ "IoT/OneWireGW/Bus 1/#" ],
+      "handler": "SVER",
+      "id": "SVER0",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+          "application": "Temperature Heating",
+          "payloadRegex": "(\\d+(\\.\\d+)?)\\s*([^0-9\\s]\\S*)",
+          "deviceFrom": "topic", 
+          "devicePart": "3",
+          "valueFrom": "payload",
+          "valuePart": "1",
+          "unitFrom": "payload",
+          "unitPart": "3"
+        }
+      }
+    },
+    {
+      "topics": [ "NR/Multisensor/+/Temperatur" ],
+      "handler": "SVEJ",
+      "id": "SVEJ0",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+          "application": "Temperature Multisensor",
+          "deviceSelector": "T:2",
+          "valueSelector": "J:$.CurrentTemperature",
+          "unitSelector": "C:°C"
+        }
+      }
+    },
+    {
+      "topics": [ "NR/Multisensor/+/Feuchte" ],
+      "handler": "SVEJ",
+      "id": "SVEJ1",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+          "application": "Humidity Multisensor",
+          "deviceSelector": "T:2",
+          "valueSelector": "J:$.CurrentRelativeHumidity",
+          "unitSelector": "C:%"
+        }
+      }
+    },
+    {
+      "topics": [ "shellyplusht/+/status/temperature:0" ],
+      "handler": "SVEJ",
+      "id": "SVEJ2",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+          "application": "Temperature Shelly Plus HT",
+          "deviceSelector": "T:1",
+          "valueSelector": "J:$.tC",
+          "unitSelector": "C:°C"
+        }
+      }
+    },
+    {
+      "topics": [ "shellyplusht/+/status/humidity:0" ],
+      "handler": "SVEJ",
+      "id": "SVE4",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+          "application": "Humidity Shelly Plus HT",
+          "deviceSelector": "T:1",
+          "valueSelector": "J:$.rh",
+          "unitSelector": "C:%"
+        }
+      }
+    }
+  ],
+  "archiver": {
+    "dir": "./tmp/udi"
+  }
+}

src/udi/config/config.go

@@ -5,7 +5,6 @@ import "log"
 import "os"
 
 type HandlerConfigT struct {
-  DatabaseConnStr string `json:"databaseConnStr"`
   Attributes map[string]string `json:"attributes"`
 }
 
@@ -13,7 +12,7 @@ type ConfigT struct {
   Mqtt struct {
     Broker string `json:"broker"`
     Username string `json:"username"`
-    Password string `json:"password"`
+    Password string
     TlsEnable string `json:"tlsEnable"`
   } `json:"mqtt"`
   TopicMappings []struct {
@@ -35,5 +34,7 @@ func LoadConfiguration() {
   if err != nil {
     log.Fatalf("Unable to parse configuration: %s", err)
   }
+
+  Config.Mqtt.Password = os.Getenv("MQTT_PASSWORD")
 }
 

src/udi/config/example.json

@@ -1,26 +0,0 @@
-{
-	"mqtt": {
-		"broker": "172.23.1.102:1883",
-		"username": "",
-		"password": "",
-		"tlsEnable": "false"
-	},
-	"topicMappings": [
-		{
-			"topics": ["IoT/MBGW3/Measurement"],
-			"handler": "IoT"
-		}
-	],
-	"handlers": [
-		{
-			"name": "IoT",
-			"databaseConnStr": "",
-			"attributes": {
-			}
-		}
-	],
-	"archiver": {
-		"dir": "/mnt/udi/archive"
-	}
-}
-

src/udi/counter/counter.go

@@ -0,0 +1,94 @@
+package counter
+
+import (
+  "log"
+  "time"
+  "encoding/json"
+)
+
+type statsTuple_t struct {
+  Successful int `json:"good"`
+  Failed int `json:"bad"`
+}
+
+type stats_t struct {
+  Received statsTuple_t `json:"received"`
+  Archived statsTuple_t `json:"archived"`
+  Dispatched statsTuple_t `json:"dispatched"`
+  Handled map[string]statsTuple_t `json:"handled"`
+  Stored statsTuple_t `json:"stored"`
+}
+
+var stats stats_t
+
+func S(id string) {
+  switch id {
+  case "Received": 
+    stats.Received.Successful = stats.Received.Successful + 1
+  case "Archived": 
+    stats.Archived.Successful += 1
+  case "Dispatched":
+    stats.Dispatched.Successful += 1
+  case "Stored":
+    stats.Stored.Successful += 1
+  default:
+    log.Printf("Unknown stats id %s", id)
+  }
+}
+
+func F(id string) {
+  switch id {
+  case "Received": 
+    stats.Received.Failed += 1
+  case "Archived": 
+    stats.Archived.Failed += 1
+  case "Dispatched":
+    stats.Dispatched.Failed += 1
+  case "Stored":
+    stats.Stored.Failed += 1
+  default:
+    log.Printf("Unknown stats id %s", id)
+  }
+}
+
+func SH(id string) {
+  if _, ok := stats.Handled[id]; ok {
+    tuple := stats.Handled[id]
+    tuple.Successful += 1
+    stats.Handled[id] = tuple
+  } else {
+    stats.Handled[id] = statsTuple_t { Successful:1, Failed:0, }
+  }
+}
+
+func FH(id string) {
+  if _, ok := stats.Handled[id]; ok {
+    tuple := stats.Handled[id]
+    tuple.Failed += 1
+    stats.Handled[id] = tuple
+  } else {
+    stats.Handled[id] = statsTuple_t { Successful:0, Failed:1, }
+  }
+}
+
+func InitCounter() {
+  stats = stats_t {
+    Received: statsTuple_t {Successful:0,Failed:0,},
+    Archived: statsTuple_t {Successful:0,Failed:0,},
+    Dispatched: statsTuple_t {Successful:0,Failed:0,},
+    Stored: statsTuple_t {Successful:0,Failed:0,},
+    Handled: make(map[string]statsTuple_t),
+  }
+
+  go func() { 
+    for {
+      sj, err := json.Marshal(stats)
+      if err != nil {
+        log.Printf("Unable to marshal stats object: %s", err)
+      }
+      log.Println(string(sj))
+      time.Sleep(time.Second * 60)
+    }
+  }()
+}
+

src/udi/database/abstract_database.go

@@ -35,8 +35,8 @@ type DeviceType struct {
 
 type Device struct {
   gorm.Model
-  Label string `gorm:"not null"`
-  ApplicationID int `gorm:"not null"`
+  Label string `gorm:"not null;uniqueIndex:idx_label_application_id"`
+  ApplicationID int `gorm:"not null;uniqueIndex:idx_label_application_id"`
   Application Application
   DeviceTypeID int `gorm:"not null"`
   DeviceType DeviceType

src/udi/database/database.go

@@ -4,6 +4,8 @@ package database
 import (
   "log"
   //"time"
+  "fmt"
+  "udi/counter"
   "gorm.io/driver/postgres"
   "gorm.io/gorm"
 )
@@ -13,33 +15,60 @@ type DatabaseHandle struct {
   dbh *gorm.DB 
 }
 
-func NewDatabaseHandle(dsn string) *DatabaseHandle {
+func NewDatabaseHandle() *DatabaseHandle {
   var db DatabaseHandle
-  conn, err := gorm.Open(postgres.Open(dsn))
+  // inject the whole database configuration via the well-known PG* env variables
+  conn, err := gorm.Open(postgres.Open(""))
   if err != nil {
     log.Printf("Unable to open database connection: %s", err)
     db.initialized = false
   } else {
     db.dbh = conn
     db.initialized = true
-    log.Println("Database connection opened")
+    //log.Println("Database connection opened")
   }
   return &db
 }
 
-func (dbh *DatabaseHandle) StoreMeasurement(measurement *Measurement) {
-  if ! dbh.initialized {
+func (self *DatabaseHandle) StoreMeasurement(measurement *Measurement) {
+  if ! self.initialized {
     log.Printf("Database connection not initialized, can not store, measurement %s lost", measurement)
+    counter.F("Stored")
     return
   }
 
-  result := dbh.dbh.Create(measurement)
+  result := self.dbh.Create(measurement)
   if result.Error != nil {
     log.Printf("Unable to insert, measurement %s lost, error: %s", measurement, result.Error)
+    counter.F("Stored")
     return
   }
 
-  log.Println("Successfully stored measurement")
+  //log.Println("Successfully stored measurement")
+  counter.S("Stored")
+}
+
+func (self *DatabaseHandle) GetDeviceByLabelAndApplication(applicationLabel string, deviceLabel string) (*Device, error) {
+  if ! self.initialized {
+    err := fmt.Errorf("Database connection not initialized")
+    return nil, err
+  }
+
+  var device Device
+  result := self.dbh.
+    Preload("Application").
+    Preload("DeviceType").
+    Joins("JOIN applications ON devices.application_id = applications.id").
+    Where("devices.label = ? AND applications.label = ?", deviceLabel, applicationLabel).
+    First(&device)
+
+  if result.Error != nil {
+    err := fmt.Errorf("Query failed: %s", result.Error)
+    return nil, err
+  }
+
+  return &device, nil
 }
 
 
+

src/udi/dispatcher/dispatcher.go

@@ -7,6 +7,7 @@ import "fmt"
 import "net/url"
 import "udi/mqtt"
 import "udi/config"
+import "udi/counter"
 import "udi/handlers/handler"
 import "udi/handlers/ttn"
 import "udi/handlers/iot"
@@ -14,13 +15,14 @@ import "udi/handlers/pv"
 import "udi/handlers/mbgw3"
 import "udi/handlers/sver"
 import "udi/handlers/svej"
+import "udi/handlers/dt1t"
 
 
 var handlerMap map[string]handler.Handler = make(map[string]handler.Handler)
 var archiverChannel chan handler.MessageT = make(chan handler.MessageT, 100)
 
 func InitDispatcher() {
-  log.Printf("Initializing dispatcher")
+  log.Printf("Dispatcher initializing")
   go archiver()
 
   for _, mapping := range config.Config.TopicMappings {
@@ -29,38 +31,41 @@ func InitDispatcher() {
     var factory interface{}
     switch mapping.Handler {
     case "TTN":
-      factory = ttn.NewTTNHandler
+      factory = ttn.New
     case "IoT":
-      factory = iot.NewIoTHandler
+      factory = iot.New
     case "PV":
-      factory = pv.NewPvHandler
+      factory = pv.New
     case "MBGW3":
-      factory = mbgw3.NewMbgw3Handler
+      factory = mbgw3.New
     case "SVER":
-      factory = sver.NewSverHandler
+      factory = sver.New
     case "SVEJ":
-      factory = svej.NewSvejHandler
+      factory = svej.New
+    case "DT1T":
+      factory = dt1t.New
     default:
       factory = nil
       log.Printf("No handler %s found, ignore mapping", mapping.Handler)
     }
 
-    fn, ok := factory.(func(config.HandlerConfigT) handler.Handler)
+    fn, ok := factory.(func(string, config.HandlerConfigT) handler.Handler)
     if ! ok {
       log.Println("Typ Assertion failed")
       break
     }
-    handler := fn(mapping.Config)
+    handler := fn(mapping.Id, mapping.Config)
     handlerMap[mapping.Id] = handler
   }
 
-  log.Printf("handlerMap: %s", handlerMap)
+  //log.Printf("handlerMap: %s", handlerMap)
 }
 
 func storeMessage(filename string, item handler.MessageT) {
   file, err := os.OpenFile(filename, os.O_APPEND | os.O_CREATE | os.O_WRONLY, 0644)
   if err != nil {
     log.Printf("Unable to open archiving file %s, message is not archived: %s", filename, err)
+    counter.F("Archived")
     return
   }
   defer file.Close()
@@ -68,9 +73,11 @@ func storeMessage(filename string, item handler.MessageT) {
   _, err = file.WriteString(string(archivingString) + "\n")
   if err != nil {
     log.Printf("Unable to write message, message is not archived: %s", err)
+    counter.F("Archived")
     return
   }
-  log.Println("Successfully archived message")
+  //log.Println("Successfully archived message")
+  counter.S("Archived")
 }
 
 func archiver() {
@@ -86,9 +93,10 @@ func archiver() {
         err := os.MkdirAll(currentArchivingDir, 0755)
         if err != nil {
           log.Printf("Unable to create archiving dir %s: %s", currentArchivingDir, err)
+          counter.F("Archived")
         }
         lastArchivingDir = currentArchivingDir
-        log.Printf("Archiving dir %s created", currentArchivingDir)
+        //log.Printf("Archiving dir %s created", currentArchivingDir)
       }
       archivingFilename := fmt.Sprintf("%s/%s", currentArchivingDir, url.PathEscape(message.Topic))
       storeMessage(archivingFilename, message)
@@ -100,7 +108,7 @@ func InputDispatcher() {
   for {
     select {
     case mqttMessage := <- mqtt.InputChannel:
-      log.Printf("Message arrived in inputDispatcher, topic: %s\n", mqttMessage.Topic)
+      //log.Printf("Message arrived in inputDispatcher, topic: %s\n", mqttMessage.Topic)
       message := handler.MessageT { time.Now(), mqttMessage.Topic, string(mqttMessage.Payload) }
       archiverChannel <- message
       handleMessage(message)
@@ -114,16 +122,19 @@ func handleMessage(message handler.MessageT) {
     for _, subscribedTopic := range mapping.Topics {
       // log.Printf("Testing %s in %s", message.Topic, subscribedTopic)
       if mqtt.TopicMatchesSubscription(message.Topic, subscribedTopic) {
-        log.Printf("Handle message in handler %s", mapping.Id)
+        //log.Printf("Handle message in handler %s", mapping.Id)
         handler, exists := handlerMap[mapping.Id]
         if exists {
           handler.Handle(message)
+          counter.S("Dispatched")
           return
         } else {
           log.Printf("Handler %s not found, message %s is lost", mapping.Id, message)
+          counter.F("Dispatched")
         }
       }
     }
   }
   log.Printf("No match for topic %s, message %s is lost", message.Topic, message)
+  counter.F("Dispatched")
 }

src/udi/handlers/dt1t/dt1t.go

@@ -0,0 +1,80 @@
+package dt1t
+
+import (
+  "log"
+  "fmt"
+  "time"
+  "strconv"
+  "udi/handlers/handler"
+  "udi/database"
+  "udi/config"
+)
+
+
+type Dt1tHandler struct {
+  handler.CommonHandler
+  ready bool
+  label string
+  dbh *database.DatabaseHandle
+  application string
+  device string
+
+}
+
+func New(id string, config config.HandlerConfigT) handler.Handler {
+  t := &Dt1tHandler {
+  }
+
+  if config.Attributes["Application"] == "" {
+    log.Println("Error: application not configured")
+    return t
+  }
+  t.application = config.Attributes["Application"]
+  if config.Attributes["Device"] == "" {
+    log.Println("Error: device not configured")
+    return t
+  }
+  t.device = config.Attributes["Device"]
+  t.Id = id
+  
+  t.dbh = database.NewDatabaseHandle()
+  t.ready = true
+  return t
+}
+
+func (self *Dt1tHandler) Handle(message handler.MessageT) {
+  if ! self.ready {
+    self.Lost("Handler is not marked as ready", nil, message)
+    return
+  }
+  // log.Printf("Handler DT1T %d processing %s -> %s", self.id, message.Topic, message.Payload)
+
+  temperature, err := strconv.Atoi(message.Payload)
+  if err != nil {
+    self.Lost("Invalid raw value", err, message)
+    return
+  }
+  if temperature & 0x8000 != 0{
+    temperature = ((temperature - 1) ^ 0xffff) * -1
+  }
+  temperatureF := float32(temperature) / 10.0
+
+  var measurement database.Measurement
+  measurement.Time = time.Now()
+  measurement.Application = self.application
+  measurement.Device = self.device
+  
+  var variable database.VariableType
+  variable.Label = "Temperature"
+  variable.Variable = ""
+  variable.Unit = "°C"
+  variable.Value = fmt.Sprintf("%f", temperatureF)
+  measurement.Values = make(map[string]database.VariableType)
+  measurement.Values["Value"] = variable
+
+  // log.Printf("Prepared measurement item: %s", measurement)
+  self.dbh.StoreMeasurement(&measurement)
+  self.S()
+}
+
+

src/udi/handlers/handler/handler.go

@@ -1,6 +1,10 @@
 package handler
 
-import "time"
+import (
+  "time"
+  "log"
+  "udi/counter"
+)
 
 type MessageT struct {
   Timestamp time.Time
@@ -11,5 +15,32 @@ type MessageT struct {
 type Handler interface {
   GetId() string
   Handle(MessageT)
+  Lost(msg string, err error, message MessageT)
+  S()
+  F()
+}
+
+type CommonHandler struct {
+  Id string
+}
+
+func (self *CommonHandler) S() {
+  counter.SH(self.Id)
+}
+func (self *CommonHandler) F() {
+  counter.FH(self.Id)
+}
+
+func (self *CommonHandler) GetId() string {
+  return self.Id
+}
+
+func (self *CommonHandler) Lost(msg string, err error, message MessageT) {
+  if err != nil {
+    log.Printf("Error: %s, message %s is lost", msg, message)
+  } else {
+    log.Printf("Error: %s (%s), message %s is lost", msg, err, message)
+  }
+  self.F()
 }
 

src/udi/handlers/iot/iot.go

@@ -1,29 +1,23 @@
 package iot
 
 import "log"
-import "fmt"
+import "udi/config"
 import "udi/handlers/handler"
 
-var idSeq int = 0
 
 type IoTHandler struct {
-  id int
+  handler.CommonHandler
 }
 
-func NewIoTHandler() handler.Handler {
+func New(id string, config config.HandlerConfigT) handler.Handler {
   t := &IoTHandler {
-    id: idSeq,
   }
-  idSeq += 1
+  t.Id = id
   return t
 }
 
-func (self *IoTHandler) GetId() string {
-  return fmt.Sprintf("IoT%d", self.id)
-}
-
 func (self *IoTHandler) Handle(message handler.MessageT) {
-  log.Printf("Handler IoT %d processing %s -> %s", self.id, message.Topic, message.Payload)
+  log.Printf("Handler IoT %d processing %s -> %s", self.Id, message.Topic, message.Payload)
 }
 
 

src/udi/handlers/mbgw3/mbgw3.go

@@ -1,11 +1,8 @@
 package mbgw3
 
 import (
-  "log"
-  //"reflect"
   "time"
   "strconv"
-  "fmt"
   "encoding/json"
   "udi/config"
   "udi/handlers/handler"
@@ -13,10 +10,8 @@ import (
 )
 
 
-var idSeq int = 0
-
 type Mbgw3Handler struct {
-  id int
+  handler.CommonHandler
   dbh *database.DatabaseHandle
 }
 
@@ -31,26 +26,21 @@ type Observation struct {
 }
 
 
-func NewMbgw3Handler(config config.HandlerConfigT) handler.Handler {
+func New(id string, config config.HandlerConfigT) handler.Handler {
   t := &Mbgw3Handler {
-    id: idSeq,
   }
-  idSeq += 1
-  t.dbh = database.NewDatabaseHandle(config.DatabaseConnStr)
+  t.Id = id
+  t.dbh = database.NewDatabaseHandle()
   return t
 }
 
-func (self *Mbgw3Handler) GetId() string {
-  return fmt.Sprintf("MBGW3%d", self.id)
-}
-
 func (self *Mbgw3Handler) Handle(message handler.MessageT) {
-//  log.Printf("Handler MBGW3 %d processing %s -> %s", self.id, message.Topic, message.Payload)
+  //log.Printf("Handler MBGW3 %d processing %s -> %s", self.Id, message.Topic, message.Payload)
 
   var observation Observation
   err := json.Unmarshal([]byte(message.Payload), &observation)
   if err != nil {
-    log.Printf("Unable to parse payload into Observation struct, message %s -> %s is lost, error ", message.Topic, message.Payload, err)
+    self.Lost("Unable to parse payload into Observation struct", err, message)
     return
   }
 
@@ -95,9 +85,10 @@ func (self *Mbgw3Handler) Handle(message handler.MessageT) {
     }
   }
 
-  // log.Printf("Prepared measurement item: %s", measurement)
+  //log.Printf("Prepared measurement item: %s", measurement)
 
   self.dbh.StoreMeasurement(&measurement)
+  self.S()
 }
 
 

src/udi/handlers/pv/pv.go

@@ -1,10 +1,8 @@
 package pv
 
 import (
-  "log"
   "reflect"
   "time"
-  "fmt"
   "encoding/json"
   "udi/config"
   "udi/handlers/handler"
@@ -12,10 +10,8 @@ import (
 )
 
 
-var idSeq int = 0
-
 type PvHandler struct {
-  id int
+  handler.CommonHandler
   dbh *database.DatabaseHandle
 }
 
@@ -40,18 +36,14 @@ type PvValue struct {
 }
 
 
-func NewPvHandler(config config.HandlerConfigT) handler.Handler {
+func New(id string, config config.HandlerConfigT) handler.Handler {
   t := &PvHandler {
-    id: idSeq,
   }
-  idSeq += 1
-  t.dbh = database.NewDatabaseHandle(config.DatabaseConnStr)
+  t.Id = id
+  t.dbh = database.NewDatabaseHandle()
   return t
 }
 
-func (self *PvHandler) GetId() string {
-  return fmt.Sprintf("PV%d", self.id)
-}
 
 func (self *PvHandler) Handle(message handler.MessageT) {
   //log.Printf("Handler PV %d processing %s -> %s", self.id, message.Topic, message.Payload)
@@ -59,7 +51,7 @@ func (self *PvHandler) Handle(message handler.MessageT) {
   var pvValue PvValue
   err := json.Unmarshal([]byte(message.Payload), &pvValue)
   if err != nil {
-    log.Printf("Unable to parse payload into pvValue struct, message %s -> %s is lost, error: %s", message.Topic, message.Payload, err)
+    self.Lost("Unable to parse payload into pvValue struct", err, message)
     return
   }
 
@@ -85,6 +77,7 @@ func (self *PvHandler) Handle(message handler.MessageT) {
   }
 
   self.dbh.StoreMeasurement(&measurement)
+  self.S()
 }
 
 

src/udi/handlers/svej/svej.go

@@ -13,10 +13,8 @@ import (
   "udi/database"
 )
 
-var idSeq int = 0
-
 type SingleValueExtractorJsonpathHandler struct {
-  id int
+  handler.CommonHandler
   ready bool
   application string
   deviceSelector string
@@ -36,12 +34,10 @@ C:ConstantValue
 */
 
 
-func NewSvejHandler(config config.HandlerConfigT) handler.Handler {
+func New(id string, config config.HandlerConfigT) handler.Handler {
   t := &SingleValueExtractorJsonpathHandler {
-    id: idSeq,
     ready: false,
   }
-  idSeq += 1
 
   if config.Attributes["application"] == "" {
     log.Println("Error: application not configured")
@@ -77,19 +73,12 @@ func NewSvejHandler(config config.HandlerConfigT) handler.Handler {
     t.unitJsonpath = jp
   }
 
+  t.Id = id
   t.ready = true
-  t.dbh = database.NewDatabaseHandle(config.DatabaseConnStr)
+  t.dbh = database.NewDatabaseHandle()
   return t
 }
 
-func (self *SingleValueExtractorJsonpathHandler) GetId() string {
-  return fmt.Sprintf("SVE%d", self.id)
-}
-
-func lost(msg string, message handler.MessageT) {
-  log.Printf("Error: %s, message %s is lost", msg, message)
-}
-
 func extractionHelper(subTopics []string, jPayload interface{}, selector string, jp *jsonpath.Compiled) (string, error) {
   var res string
   switch selector[:2] {
@@ -119,10 +108,10 @@ func extractionHelper(subTopics []string, jPayload interface{}, selector string,
 
 func (self *SingleValueExtractorJsonpathHandler) Handle(message handler.MessageT) {
   if ! self.ready {
-    log.Println("Handler is not marked as ready, message %s is lost", message)
+    self.Lost("Handler is not marked as ready", nil, message)
     return
   }
-  log.Printf("Handler SingleValueExtractorJsonpath %d processing %s -> %s", self.id, message.Topic, message.Payload)
+  //log.Printf("Handler SingleValueExtractorJsonpath %d processing %s -> %s", self.Id, message.Topic, message.Payload)
 
   var measurement database.Measurement
   measurement.Time = time.Now()
@@ -133,23 +122,23 @@ func (self *SingleValueExtractorJsonpathHandler) Handle(message handler.MessageT
   var jPayload interface{}
   err := json.Unmarshal([]byte(message.Payload), &jPayload)
   if err != nil {
-    lost(fmt.Sprintf("Unable to unmarshal payload: %s", err), message)
+    self.Lost("Unable to unmarshal payload", err, message)
     return
   }
 
   device, err1 := extractionHelper(subTopics, jPayload, self.deviceSelector, self.deviceJsonpath)
   if err1 != nil {
-    lost(fmt.Sprintf("Device extraction failed with %s", err1), message)
+    self.Lost("Device extraction failed", err1, message)
     return
   }
   value, err2 := extractionHelper(subTopics, jPayload, self.valueSelector, self.valueJsonpath)
   if err2 != nil {
-    lost(fmt.Sprintf("Value extraction failed with %s", err2), message)
+    self.Lost("Value extraction failed", err2, message)
     return
   }
   unit, err3 := extractionHelper(subTopics, jPayload, self.unitSelector, self.unitJsonpath)
   if err3 != nil {
-    lost(fmt.Sprintf("Unit extraction failed with %s", err3), message)
+    self.Lost("Unit extraction failed", err3, message)
     return
   }
 
@@ -163,7 +152,8 @@ func (self *SingleValueExtractorJsonpathHandler) Handle(message handler.MessageT
   measurement.Values = make(map[string]database.VariableType)
   measurement.Values["Value"] = variable
 
-  log.Printf("Prepared measurement item: %s", measurement)
+  //log.Printf("Prepared measurement item: %s", measurement)
   self.dbh.StoreMeasurement(&measurement)
+  self.S()
 }
 

src/udi/handlers/sver/sver.go

@@ -1,21 +1,19 @@
 package sver
 
 import (
-  "log"
   "time"
   "strconv"
   "strings"
   "regexp"
-  "fmt"
+  "log"
   "udi/config"
   "udi/handlers/handler"
   "udi/database"
 )
 
-var idSeq int = 0
 
 type SingleValueExtractorRegexHandler struct {
-  id int
+  handler.CommonHandler
   ready bool
   config localConfig
   payloadRegex *regexp.Regexp
@@ -40,12 +38,10 @@ type localConfig struct {
 }
 
 
-func NewSverHandler(config config.HandlerConfigT) handler.Handler {
+func New(id string, config config.HandlerConfigT) handler.Handler {
   t := &SingleValueExtractorRegexHandler {
-    id: idSeq,
     ready: false,
   }
-  idSeq += 1
 
   var localConfig localConfig
   if config.Attributes["application"] == "" {
@@ -77,13 +73,13 @@ func NewSverHandler(config config.HandlerConfigT) handler.Handler {
   // empty device is valid
   localConfig.device = config.Attributes["device"]
 
-  if config.Attributes["valueFrom"] != TOPIC_SEL && config.Attributes["valueFrom"] != PAYLOAD_SEL && config.Attributes["valueFrom"] != PAYLOAD_FULL_SEL {
+  if config.Attributes["valueFrom"] != PAYLOAD_SEL && config.Attributes["valueFrom"] != PAYLOAD_FULL_SEL {
     log.Printf("Error: invalid value %s for valueFrom", config.Attributes["valueFrom"])
     return t
   }
   localConfig.valueFrom = config.Attributes["valueFrom"]
 
-  if config.Attributes["valueFrom"] != PAYLOAD_FULL_SEL {
+  if config.Attributes["valueFrom"] == PAYLOAD_SEL {
     valuePart, err2 := strconv.Atoi(config.Attributes["valuePart"])
     if err2 != nil {
       log.Printf("Error: unable to convert valuePart to number: %s", err2)
@@ -92,13 +88,13 @@ func NewSverHandler(config config.HandlerConfigT) handler.Handler {
     localConfig.valuePart = valuePart
   }
 
-  if config.Attributes["unitFrom"] != TOPIC_SEL && config.Attributes["unitFrom"] != PAYLOAD_SEL && config.Attributes["unitFrom"] != CONSTANT_SEL {
+  if config.Attributes["unitFrom"] != PAYLOAD_SEL && config.Attributes["unitFrom"] != CONSTANT_SEL {
     log.Printf("Error: invalid value %s for unitFrom", config.Attributes["unitFrom"])
     return t
   }
   localConfig.unitFrom = config.Attributes["unitFrom"]
 
-  if config.Attributes["unitFrom"] != CONSTANT_SEL {
+  if config.Attributes["unitFrom"] == PAYLOAD_SEL {
     unitPart, err3 := strconv.Atoi(config.Attributes["unitPart"])
     if err3 != nil {
       log.Printf("Error: unable to convert unitPart to number: %s", err3)
@@ -112,22 +108,15 @@ func NewSverHandler(config config.HandlerConfigT) handler.Handler {
 
   t.config = localConfig
 
+  t.Id = id
   t.ready = true
-  t.dbh = database.NewDatabaseHandle(config.DatabaseConnStr)
+  t.dbh = database.NewDatabaseHandle()
   return t
 }
 
-func (self *SingleValueExtractorRegexHandler) GetId() string {
-  return fmt.Sprintf("SVE%d", self.id)
-}
-
-func lost(msg string, message handler.MessageT) {
-  log.Printf("Error: %s, message %s is lost", msg, message)
-}
-
 func (self *SingleValueExtractorRegexHandler) Handle(message handler.MessageT) {
   if ! self.ready {
-    log.Println("Handler is not marked as ready, message %s is lost", message)
+    self.Lost("Handler is not marked as ready", nil, message)
     return
   }
   //log.Printf("Handler SingleValueExtractor %d processing %s -> %s", self.id, message.Topic, message.Payload)
@@ -148,17 +137,17 @@ func (self *SingleValueExtractorRegexHandler) Handle(message handler.MessageT) {
   switch self.config.deviceFrom {
   case TOPIC_SEL:
     if self.config.devicePart >= len(subTopics) {
-      lost("devicePart out of range", message)
+      self.Lost("devicePart out of range", nil, message)
       return
     }
     measurement.Device = subTopics[self.config.devicePart]
   case PAYLOAD_SEL:
     if self.payloadRegex == nil {
-      lost("no payloadRegex defined, devicePart can't be used", message)
+      self.Lost("no payloadRegex defined, devicePart can't be used", nil, message)
       return
     }
     if self.config.devicePart >= len(payloadMatches) {
-      lost("devicePart out of range", message)
+      self.Lost("devicePart out of range", nil, message)
       return
     }
     measurement.Device = payloadMatches[self.config.devicePart]
@@ -172,19 +161,13 @@ func (self *SingleValueExtractorRegexHandler) Handle(message handler.MessageT) {
   variable.Variable = ""
 
   switch self.config.valueFrom {
-  case TOPIC_SEL:
-    if self.config.valuePart >= len(subTopics) {
-      lost("valuePart out of range", message)
-      return
-    }
-    variable.Value = subTopics[self.config.valuePart]
   case PAYLOAD_SEL:
     if self.payloadRegex == nil {
-      lost("no payloadRegex defined, valuePart can't be used", message)
+      self.Lost("no payloadRegex defined, valuePart can't be used", nil, message)
       return
     }
     if self.config.valuePart >= len(payloadMatches) {
-      lost("valuePart out of range", message)
+      self.Lost("valuePart out of range", nil, message)
       return
     }
     variable.Value = payloadMatches[self.config.valuePart]
@@ -193,19 +176,13 @@ func (self *SingleValueExtractorRegexHandler) Handle(message handler.MessageT) {
   }
 
   switch self.config.unitFrom {
-  case TOPIC_SEL:
-    if self.config.unitPart >= len(subTopics) {
-      lost("unitPart out of range", message)
-      return
-    }
-    variable.Unit = subTopics[self.config.unitPart]
   case PAYLOAD_SEL:
     if self.payloadRegex == nil {
-      lost("no payloadRegex defined, unitPart can't be used", message)
+      self.Lost("no payloadRegex defined, unitPart can't be used", nil, message)
       return
     }
     if self.config.unitPart >= len(payloadMatches) {
-      lost("unitPart out of range", message)
+      self.Lost("unitPart out of range", nil, message)
       return
     }
     variable.Unit = payloadMatches[self.config.unitPart]
@@ -217,5 +194,6 @@ func (self *SingleValueExtractorRegexHandler) Handle(message handler.MessageT) {
 
   //log.Printf("Prepared measurement item: %s", measurement)
   self.dbh.StoreMeasurement(&measurement)
+  self.S()
 }
 

src/udi/handlers/ttn/models/draginoLdds75/draginoLdds75.go

@@ -0,0 +1,80 @@
+package draginoLdds75
+
+import (
+  "fmt"
+  "log"
+  "strings"
+  "strconv"
+  "encoding/json"
+  "udi/database"
+)
+
+/*
+    "decoded_payload": {
+      "Bat": 3.299,
+      "Distance": "352 mm",
+      "Interrupt_flag": 0,
+      "Sensor_flag": 1,
+      "TempC_DS18B20": "0.00"
+    },
+*/
+type message struct {
+  Bat float32 `json:"Bat"`
+  Distance string `json:"Distance"`
+  Interrupt_flag int `json:"Interrupt_flag"`
+  Sensor_flag int `json:"Sensor_flag"`
+  TempC_DS18B20 string `json:"TempC_DS18B20"`
+}
+
+func Parse(fPort int, decodedPayload []byte, variables *map[string]database.VariableType, device *database.Device) error {
+  if fPort != 2 {
+    return fmt.Errorf("Unexpected fPort %d", fPort)
+  }
+  var message message
+  err := json.Unmarshal(decodedPayload, &message)
+  if err != nil {
+    return fmt.Errorf("Unable to parse payload, fPort %d, error %s", fPort, err)
+  }
+  distanceParts := strings.Split(message.Distance, " ")
+  if len(distanceParts) != 2 && distanceParts[1] != "mm" {
+    return fmt.Errorf("Invalid format for distance value: %s", message.Distance)
+  }
+  distance, err2 := strconv.Atoi(distanceParts[0])
+  if err2 != nil {
+    return fmt.Errorf("Distance value is no number: %s -> %s", message.Distance, err2)
+  }
+  (*variables)["Battery"] = database.VariableType {
+    Label: "Battery",
+    Variable: "Voltage",
+    Unit: "V",
+    Value: message.Bat,
+  }
+  (*variables)["Distance"] = database.VariableType {
+    Label: "Distance",
+    Variable: "Level",
+    Unit: "mm",
+    Value: distance,
+  }
+  groundLevelI, exists := device.Attributes["GroundLevel"]
+  groundLevelS, ok := groundLevelI.(string)
+  groundLevel, err3 := strconv.Atoi(groundLevelS)
+  if exists && err3 == nil && ok {
+    log.Println("add corrected distance")
+    correctedDistance := groundLevel - distance
+    (*variables)["CorrectedDistance"] = database.VariableType {
+      Label: "CorrectedDistance",
+      Variable: "Level",
+      Unit: "mm",
+      Value: correctedDistance,
+    }
+  } else {
+    log.Printf("no ground level: %s %s %s", exists, err3, ok)
+    log.Printf("Device: %s", device)
+    log.Printf("Attributes: %s", device.Attributes)
+  }
+
+  return nil
+}
+
+
+

src/udi/handlers/ttn/models/draginoLdds75/example-ldds75.json

@@ -0,0 +1,83 @@
+{
+  "end_device_ids": {
+    "device_id": "eui-a84041a14185f67f",
+    "application_ids": {
+      "application_id": "de-hottis-level-monitoring"
+    },
+    "dev_eui": "A84041A14185F67F",
+    "join_eui": "A840410000000101",
+    "dev_addr": "260B3987"
+  },
+  "correlation_ids": [
+    "gs:uplink:01HHP8EVC1N78FGTNBTNZYGCVY"
+  ],
+  "received_at": "2023-12-15T08:11:04.142793605Z",
+  "uplink_message": {
+    "session_key_id": "AYZ/cI2YeDLCZr4urRDzCw==",
+    "f_port": 2,
+    "f_cnt": 27758,
+    "frm_payload": "DOMBYAAAAAE=",
+    "decoded_payload": {
+      "Bat": 3.299,
+      "Distance": "352 mm",
+      "Interrupt_flag": 0,
+      "Sensor_flag": 1,
+      "TempC_DS18B20": "0.00"
+    },
+    "rx_metadata": [
+      {
+        "gateway_ids": {
+          "gateway_id": "eui-00005813d35e3021",
+          "eui": "00005813D35E3021"
+        },
+        "timestamp": 3654294763,
+        "rssi": -85,
+        "channel_rssi": -85,
+        "snr": 8.8,
+        "location": {
+          "latitude": 52.17065267448476,
+          "longitude": 7.629437184774199,
+          "source": "SOURCE_REGISTRY"
+        },
+        "uplink_token": "CiIKIAoUZXVpLTAwMDA1ODEzZDM1ZTMwMjESCAAAWBPTXjAhEOu5wM4NGgwIl5TwqwYQ5Z/7vgMg+OvDp63eBA==",
+        "channel_index": 2,
+        "received_at": "2023-12-15T08:11:03.937349093Z"
+      }
+    ],
+    "settings": {
+      "data_rate": {
+        "lora": {
+          "bandwidth": 125000,
+          "spreading_factor": 7,
+          "coding_rate": "4/5"
+        }
+      },
+      "frequency": "867500000",
+      "timestamp": 3654294763
+    },
+    "received_at": "2023-12-15T08:11:03.938112475Z",
+    "consumed_airtime": "0.056576s",
+    "locations": {
+      "user": {
+        "latitude": 52.1710648323742,
+        "longitude": 7.62751003482794,
+        "altitude": 37,
+        "source": "SOURCE_REGISTRY"
+      }
+    },
+    "version_ids": {
+      "brand_id": "dragino",
+      "model_id": "ldds75",
+      "hardware_version": "_unknown_hw_version_",
+      "firmware_version": "1.1.4",
+      "band_id": "EU_863_870"
+    },
+    "network_ids": {
+      "net_id": "000013",
+      "ns_id": "EC656E0000000181",
+      "tenant_id": "ttn",
+      "cluster_id": "eu1",
+      "cluster_address": "eu1.cloud.thethings.network"
+    }
+  }
+}

src/udi/handlers/ttn/models/draginoLmds200/draginoLmds200.go

@@ -0,0 +1,81 @@
+package draginoLmds200
+
+import (
+  "fmt"
+  "strconv"
+  "encoding/json"
+  "udi/database"
+)
+
+/*
+    "decoded_payload": {
+      "Bat": 3.082,
+      "DALARM_count": 0,
+      "Distance_alarm": 0,
+      "Interrupt_alarm": 0,
+      "dis1": 105,
+      "dis2": 201
+    },
+*/
+type message struct {
+  Bat float32 `json:"Bat"`
+  DALARM_count int `json:"DALARM_count"`
+  Distance_alarm int `json:"Distance_alarm"`
+  Interrupt_alarm int `json:"Interrupt_alarm"`
+  Dis1 int `json:"dis1"`
+  Dis2 int `json:"dis2"`
+}
+
+func Parse(fPort int, decodedPayload []byte, variables *map[string]database.VariableType, device *database.Device) error {
+  if fPort != 2 {
+    return fmt.Errorf("Unexpected fPort %d", fPort)
+  }
+  var message message
+  err := json.Unmarshal(decodedPayload, &message)
+  if err != nil {
+    return fmt.Errorf("Unable to parse payload, fPort %d, error %s", fPort, err)
+  }
+  (*variables)["Battery"] = database.VariableType {
+    Label: "Battery",
+    Variable: "Voltage",
+    Unit: "V",
+    Value: message.Bat,
+  }
+  distance1 := message.Dis1 * 10
+  (*variables)["Distance1"] = database.VariableType {
+    Label: "Distance1",
+    Variable: "Level",
+    Unit: "mm",
+    Value: distance1,
+  }
+  distance2 := message.Dis2 * 10
+  (*variables)["Distance2"] = database.VariableType {
+    Label: "Distance2",
+    Variable: "Level",
+    Unit: "mm",
+    Value: distance2,
+  }
+  groundLevelI, exists := device.Attributes["GroundLevel"]
+  groundLevelS, ok := groundLevelI.(string)
+  groundLevel, err3 := strconv.Atoi(groundLevelS)
+  if exists && err3 == nil && ok {
+    correctedDistance1 := groundLevel - distance1
+    (*variables)["CorrectedDistance1"] = database.VariableType {
+      Label: "CorrectedDistance1",
+      Variable: "Level",
+      Unit: "mm",
+      Value: correctedDistance1,
+    }
+    correctedDistance2 := groundLevel - distance2
+    (*variables)["CorrectedDistance2"] = database.VariableType {
+      Label: "CorrectedDistance2",
+      Variable: "Level",
+      Unit: "mm",
+      Value: correctedDistance2,
+    }
+  }
+  return nil
+}
+
+
+

src/udi/handlers/ttn/models/draginoLmds200/lmds200-example.json

@@ -0,0 +1,84 @@
+{
+  "end_device_ids": {
+    "device_id": "eui-a840419641867eb5",
+    "application_ids": {
+      "application_id": "de-hottis-level-monitoring"
+    },
+    "dev_eui": "A840419641867EB5",
+    "join_eui": "A840410000000101",
+    "dev_addr": "260B91F9"
+  },
+  "correlation_ids": [
+    "gs:uplink:01HH1R112BNDQQ52N9FVV0TKPW"
+  ],
+  "received_at": "2023-12-07T08:59:05.369293395Z",
+  "uplink_message": {
+    "session_key_id": "AYa9JUhNJp00t+hKqkQUog==",
+    "f_port": 2,
+    "f_cnt": 25665,
+    "frm_payload": "DAoAaQDJAA==",
+    "decoded_payload": {
+      "Bat": 3.082,
+      "DALARM_count": 0,
+      "Distance_alarm": 0,
+      "Interrupt_alarm": 0,
+      "dis1": 105,
+      "dis2": 201
+    },
+    "rx_metadata": [
+      {
+        "gateway_ids": {
+          "gateway_id": "eui-00005813d35e3021",
+          "eui": "00005813D35E3021"
+        },
+        "timestamp": 1141271036,
+        "rssi": -100,
+        "channel_rssi": -100,
+        "snr": 7.3,
+        "location": {
+          "latitude": 52.17065267448476,
+          "longitude": 7.629437184774199,
+          "source": "SOURCE_REGISTRY"
+        },
+        "uplink_token": "CiIKIAoUZXVpLTAwMDA1ODEzZDM1ZTMwMjESCAAAWBPTXjAhEPzTmaAEGgsI2ZLGqwYQnfLnTSDggLjIm5IF",
+        "channel_index": 6,
+        "received_at": "2023-12-07T08:59:05.163182877Z"
+      }
+    ],
+    "settings": {
+      "data_rate": {
+        "lora": {
+          "bandwidth": 125000,
+          "spreading_factor": 7,
+          "coding_rate": "4/5"
+        }
+      },
+      "frequency": "868300000",
+      "timestamp": 1141271036
+    },
+    "received_at": "2023-12-07T08:59:05.163964824Z",
+    "consumed_airtime": "0.056576s",
+    "locations": {
+      "user": {
+        "latitude": 52.1707216912195,
+        "longitude": 7.63066603211241,
+        "altitude": 39,
+        "source": "SOURCE_REGISTRY"
+      }
+    },
+    "version_ids": {
+      "brand_id": "dragino",
+      "model_id": "lmds200",
+      "hardware_version": "_unknown_hw_version_",
+      "firmware_version": "1.0",
+      "band_id": "EU_863_870"
+    },
+    "network_ids": {
+      "net_id": "000013",
+      "ns_id": "EC656E0000000181",
+      "tenant_id": "ttn",
+      "cluster_id": "eu1",
+      "cluster_address": "eu1.cloud.thethings.network"
+    }
+  }
+}

src/udi/handlers/ttn/models/draginoLse01/draginoLse01.go

@@ -0,0 +1,63 @@
+package draginoLse01
+
+import (
+  "fmt"
+  "encoding/json"
+  "udi/database"
+)
+
+/*
+{
+  "Bat":3.211,
+  "TempC_DS18B20":"0.0",
+  "conduct_SOIL":32,
+  "temp_SOIL":"7.56",
+  "water_SOIL":"25.92"
+}
+*/
+
+type message struct {
+  Bat float32 `json:"Bat"`
+  TempC_DS18B20 string `json:"TempC_DS18B20"`
+  Conduct_SOIL int `json:"conduct_SOIL"`
+  Temp_SOIL string `json:"temp_SOIL"`
+  Water_SOIL string `json:"water_SOIL"`
+}
+
+func Parse(fPort int, decodedPayload []byte, variables *map[string]database.VariableType, device *database.Device) error {
+  if fPort != 2 {
+    return fmt.Errorf("Unexpected fPort %d", fPort)
+  }
+  var message message
+  err := json.Unmarshal(decodedPayload, &message)
+  if err != nil {
+    return fmt.Errorf("Unable to parse payload, fPort %d, error %s", fPort, err)
+  }
+  (*variables)["Battery"] = database.VariableType {
+    Label: "Battery",
+    Variable: "Voltage",
+    Unit: "V",
+    Value: message.Bat,
+  }
+  (*variables)["Conductance"] = database.VariableType {
+    Label: "Conductance",
+    Variable: "Conductance",
+    Unit: "uS/cm",
+    Value: message.Conduct_SOIL,
+  }
+  (*variables)["Temperature"] = database.VariableType {
+    Label: "Temperature",
+    Variable: "Temperature",
+    Unit: "°C",
+    Value: message.Temp_SOIL,
+  }
+  (*variables)["Water"] = database.VariableType {
+    Label: "Water",
+    Variable: "Water",
+    Unit: "%",
+    Value: message.Water_SOIL,
+  }
+
+  return nil
+}
+

src/udi/handlers/ttn/models/emuProfIILoRaCfg1/emu-fprot1-example.json

@@ -0,0 +1,124 @@
+{
+  "end_device_ids": {
+    "device_id": "eui-102ceffffe01089c",
+    "application_ids": {
+      "application_id": "com-passavant-geiger-poc"
+    },
+    "dev_eui": "102CEFFFFE01089C",
+    "join_eui": "102CEF0000000000",
+    "dev_addr": "260B0E1A"
+  },
+  "correlation_ids": [
+    "gs:uplink:01HHF7YF14Y7HQBF9D8N8D20ZM"
+  ],
+  "received_at": "2023-12-12T14:47:26.197129491Z",
+  "uplink_message": {
+    "session_key_id": "AYxJcJyrJgr7XiIUdO3EBA==",
+    "f_port": 1,
+    "f_cnt": 11738,
+    "frm_payload": "7HF4ZQsAAAAADAAAAAANAAAAAA4AAAAAFwAYABkA8ADc",
+    "decoded_payload": {
+      "Active Power L1": {
+        "cfgphase": 1,
+        "unit": "W",
+        "value": 0
+      },
+      "Active Power L123": {
+        "unit": "W",
+        "value": 0
+      },
+      "Active Power L2": {
+        "cfgphase": 2,
+        "unit": "W",
+        "value": 0
+      },
+      "Active Power L3": {
+        "cfgphase": 3,
+        "unit": "W",
+        "value": 0
+      },
+      "Powerfactor L1": {
+        "cfgphase": 1,
+        "unit": "Cos",
+        "value": 0,
+        "value_raw": 0
+      },
+      "Powerfactor L2": {
+        "cfgphase": 2,
+        "unit": "Cos",
+        "value": 0,
+        "value_raw": 0
+      },
+      "Powerfactor L3": {
+        "cfgphase": 3,
+        "unit": "Cos",
+        "value": 0,
+        "value_raw": 0
+      },
+      "errorcode": {
+        "CTRatioChange": false,
+        "ImpulseRatioChange": false,
+        "ImpulseWidthChange": false,
+        "LogbookFull": false,
+        "PowerFail": false,
+        "TimeChanged": false,
+        "VTRatioChange": false,
+        "value": 0
+      },
+      "medium": {
+        "desc": "Electricity",
+        "type": 1
+      },
+      "timeStamp": 1702392300
+    },
+    "rx_metadata": [
+      {
+        "gateway_ids": {
+          "gateway_id": "eui-b827ebfffe8b01dd",
+          "eui": "B827EBFFFE8B01DD"
+        },
+        "time": "2023-12-12T14:47:25.951668977Z",
+        "timestamp": 3710351237,
+        "rssi": -89,
+        "channel_rssi": -89,
+        "snr": 14,
+        "location": {
+          "latitude": 51.404164272478724,
+          "longitude": 7.060088589208832,
+          "source": "SOURCE_REGISTRY"
+        },
+        "uplink_token": "CiIKIAoUZXVpLWI4MjdlYmZmZmU4YjAxZGQSCLgn6//+iwHdEIXvnekNGgwI/eThqwYQpoGI1wMgiL+rkf5r",
+        "received_at": "2023-12-12T14:47:25.969479115Z"
+      }
+    ],
+    "settings": {
+      "data_rate": {
+        "lora": {
+          "bandwidth": 125000,
+          "spreading_factor": 7,
+          "coding_rate": "4/5"
+        }
+      },
+      "frequency": "867100000",
+      "timestamp": 3710351237,
+      "time": "2023-12-12T14:47:25.951668977Z"
+    },
+    "received_at": "2023-12-12T14:47:25.988957776Z",
+    "confirmed": true,
+    "consumed_airtime": "0.092416s",
+    "version_ids": {
+      "brand_id": "emu",
+      "model_id": "emu-prof-ii",
+      "hardware_version": "1.0",
+      "firmware_version": "1.0",
+      "band_id": "EU_863_870"
+    },
+    "network_ids": {
+      "net_id": "000013",
+      "ns_id": "EC656E0000000181",
+      "tenant_id": "ttn",
+      "cluster_id": "eu1",
+      "cluster_address": "eu1.cloud.thethings.network"
+    }
+  }
+}

src/udi/handlers/ttn/models/emuProfIILoRaCfg1/emu-fprot2-example.json

@@ -0,0 +1,93 @@
+{
+  "end_device_ids": {
+    "device_id": "eui-102ceffffe01089c",
+    "application_ids": {
+      "application_id": "com-passavant-geiger-poc"
+    },
+    "dev_eui": "102CEFFFFE01089C",
+    "join_eui": "102CEF0000000000",
+    "dev_addr": "260B0E1A"
+  },
+  "correlation_ids": [
+    "gs:uplink:01HH53T0RPG7QFT6N267J3Q2PA"
+  ],
+  "received_at": "2023-12-08T16:22:41.900339885Z",
+  "uplink_message": {
+    "session_key_id": "AYxJcJyrJgr7XiIUdO3EBA==",
+    "f_port": 2,
+    "f_cnt": 374,
+    "frm_payload": "BEFzZSQAAAAAAAAAACYAAAAAAAAAACgAAAAAAAAAACoAAAAAAAAAAMk=",
+    "decoded_payload": {
+      "Active Energy Export T1 64bit": {
+        "unit": "Wh",
+        "value": 0
+      },
+      "Active Energy Import T1 64bit": {
+        "unit": "Wh",
+        "value": 0
+      },
+      "Reactive Energy Export T1 64bit": {
+        "unit": "varh",
+        "value": 0
+      },
+      "Reactive Energy Import T1 64bit": {
+        "unit": "varh",
+        "value": 0
+      },
+      "medium": {
+        "desc": "Electricity",
+        "type": 1
+      },
+      "timeStamp": 1702052100
+    },
+    "rx_metadata": [
+      {
+        "gateway_ids": {
+          "gateway_id": "eui-b827ebfffe8b01dd",
+          "eui": "B827EBFFFE8B01DD"
+        },
+        "time": "2023-12-08T16:22:40.969499111Z",
+        "timestamp": 383972090,
+        "rssi": -93,
+        "channel_rssi": -93,
+        "snr": 12.75,
+        "location": {
+          "latitude": 51.404164272478724,
+          "longitude": 7.060088589208832,
+          "source": "SOURCE_REGISTRY"
+        },
+        "uplink_token": "CiIKIAoUZXVpLWI4MjdlYmZmZmU4YjAxZGQSCLgn6//+iwHdEPrli7cBGgwI0YXNqwYQoteqxwIgkMGUtJb+pwI=",
+        "received_at": "2023-12-08T16:22:41.668146811Z"
+      }
+    ],
+    "settings": {
+      "data_rate": {
+        "lora": {
+          "bandwidth": 125000,
+          "spreading_factor": 7,
+          "coding_rate": "4/5"
+        }
+      },
+      "frequency": "867700000",
+      "timestamp": 383972090,
+      "time": "2023-12-08T16:22:40.969499111Z"
+    },
+    "received_at": "2023-12-08T16:22:41.687510109Z",
+    "confirmed": true,
+    "consumed_airtime": "0.102656s",
+    "version_ids": {
+      "brand_id": "emu",
+      "model_id": "emu-prof-ii",
+      "hardware_version": "1.0",
+      "firmware_version": "1.0",
+      "band_id": "EU_863_870"
+    },
+    "network_ids": {
+      "net_id": "000013",
+      "ns_id": "EC656E0000000181",
+      "tenant_id": "ttn",
+      "cluster_id": "eu1",
+      "cluster_address": "eu1.cloud.thethings.network"
+    }
+  }
+}

src/udi/handlers/ttn/models/emuProfIILoRaCfg1/emuProfIILoRaCfg1.go

@@ -0,0 +1,257 @@
+package emuProfIILoRaCfg1
+
+// provisioning device with
+// f_port=1: 01 00 0A 0B 0C 0D 0E 17 18 19 F0 4C
+//           01 00 08 0B 0C 0D 0E 17 18 19 F0 BE
+// f_port=2: 01 00 0A 24 26 28 2A 7A
+//           01 00 08 24 26 28 2A BE
+
+
+import (
+  "fmt"
+  //"log"
+  "encoding/json"
+  "udi/database"
+)
+
+
+/*
+    {
+      "Active Energy Export T1 64bit": {
+        "unit": "Wh",
+        "value": 0
+      },
+      "Active Energy Import T1 64bit": {
+        "unit": "Wh",
+        "value": 0
+      },
+      "Reactive Energy Export T1 64bit": {
+        "unit": "varh",
+        "value": 0
+      },
+      "Reactive Energy Import T1 64bit": {
+        "unit": "varh",
+        "value": 0
+      },
+      "medium": {
+        "desc": "Electricity",
+        "type": 1
+      },
+      "timeStamp": 1702052100
+    }
+*/
+type emuMessage2 struct {
+  ActiveEnergyExport struct {
+    Value int `json:"value"`
+    Unit string `json:"unit"`
+  } `json:"Active Energy Export T1 64bit"`
+  ReactiveEnergyExport struct {
+    Value int `json:"value"`
+    Unit string `json:"unit"`
+  } `json:"Reactive Energy Export T1 64bit"`
+  ActiveEnergyImport struct {
+    Value int `json:"value"`
+    Unit string `json:"unit"`
+  } `json:"Active Energy Import T1 64bit"`
+  ReactiveEnergyImport struct {
+    Value int `json:"value"`
+    Unit string `json:"unit"`
+  } `json:"Reactive Energy Import T1 64bit"`
+  Medium struct {
+    Desc string `json:"desc"`
+    Type int `json:"type"`
+  } `json:"medium"`
+  Timestamp int `json:"timestamp"`
+}
+
+/*
+    {
+      "Active Power L1": {
+        "cfgphase": 1,
+        "unit": "W",
+        "value": 0
+      },
+      "Active Power L123": {
+        "unit": "W",
+        "value": 0
+      },
+      "Active Power L2": {
+        "cfgphase": 2,
+        "unit": "W",
+        "value": 0
+      },
+      "Active Power L3": {
+        "cfgphase": 3,
+        "unit": "W",
+        "value": 0
+      },
+      "Powerfactor L1": {
+        "cfgphase": 1,
+        "unit": "Cos",
+        "value": 0,
+        "value_raw": 0
+      },
+      "Powerfactor L2": {
+        "cfgphase": 2,
+        "unit": "Cos",
+        "value": 0,
+        "value_raw": 0
+      },
+      "Powerfactor L3": {
+        "cfgphase": 3,
+        "unit": "Cos",
+        "value": 0,
+        "value_raw": 0
+      },
+      "errorcode": {
+        "CTRatioChange": false,
+        "ImpulseRatioChange": false,
+        "ImpulseWidthChange": false,
+        "LogbookFull": false,
+        "PowerFail": false,
+        "TimeChanged": false,
+        "VTRatioChange": false,
+        "value": 0
+      },
+      "medium": {
+        "desc": "Electricity",
+        "type": 1
+      },
+      "timeStamp": 1702392300
+    }
+*/
+type emuMessage1 struct {
+  ActivePowerL1 struct {
+    CfgPhase int `json:"cfgphase"`
+    Unit string `json:"unit"`
+    Value int `json:"value"` 
+  } `json:"Active Power L1"`
+  ActivePowerL2 struct {
+    CfgPhase int `json:"cfgphase"`
+    Unit string `json:"unit"`
+    Value int `json:"value"` 
+  } `json:"Active Power L2"`
+  ActivePowerL3 struct {
+    CfgPhase int `json:"cfgphase"`
+    Unit string `json:"unit"`
+    Value int `json:"value"` 
+  } `json:"Active Power L3"`
+  ActivePowerL123 struct {
+    Unit string `json:"unit"`
+    Value int `json:"value"` 
+  } `json:"Active Power L123"`
+  PowerfactorL1 struct {
+    CfgPhase int `json:"cfgphase"`
+    Unit string `json:"unit"`
+    Value float32 `json:"value"` 
+    ValueRaw float32 `json:"value_raw"` 
+  } `json:"Powerfactor L1"`
+  PowerfactorL2 struct {
+    CfgPhase int `json:"cfgphase"`
+    Unit string `json:"unit"`
+    Value float32 `json:"value"` 
+    ValueRaw float32 `json:"value_raw"` 
+  } `json:"Powerfactor L2"`
+  PowerfactorL3 struct {
+    CfgPhase int `json:"cfgphase"`
+    Unit string `json:"unit"`
+    Value float32 `json:"value"` 
+    ValueRaw float32 `json:"value_raw"` 
+  } `json:"Powerfactor L3"`
+  ErrorCode struct {
+    CTRatioChange bool `json:"CTRatioChange"`
+    ImpulseRatioChange bool `json:"ImpulseRatioChange"`
+    ImpulseWidthChange bool `json:"ImpulseWidthChange"`
+    LogbookFull bool `json:"LogbookFull"`
+    PowerFail bool `json:"PowerFail"`
+    TimeChanged bool `json:"TimeChanged"`
+    VTRatioChange bool `json:"VTRatioChange"`
+    Value int `json:"value"`
+  } `json:"errorcode"`
+  Medium struct {
+    Desc string `json:"desc"`
+    Type int `json:"type"`
+  } `json:"medium"`
+  Timestamp int `json:"timestamp"`
+}
+
+
+
+func Parse(fPort int, decodedPayload []byte, variables *map[string]database.VariableType, _ *database.Device) error {
+  //log.Printf("Parse input: %d, %s", fPort, decodedPayload)
+  switch fPort {
+  case 1:
+    var emuMessage1 emuMessage1
+    err := json.Unmarshal(decodedPayload, &emuMessage1)
+    if err != nil {
+      return fmt.Errorf("Unable to parse payload, fPort %d, error %s", fPort, err)
+    }
+    (*variables)["ActivePowerL1"] = database.VariableType { 
+      Variable: "ActivePowerL1", 
+      Unit: emuMessage1.ActivePowerL1.Unit,
+      Value: emuMessage1.ActivePowerL1.Value,
+    }
+    (*variables)["ActivePowerL2"] = database.VariableType { 
+      Variable: "ActivePowerL2", 
+      Unit: emuMessage1.ActivePowerL2.Unit,
+      Value: emuMessage1.ActivePowerL2.Value,
+    }
+    (*variables)["ActivePowerL3"] = database.VariableType { 
+      Variable: "ActivePowerL3", 
+      Unit: emuMessage1.ActivePowerL3.Unit,
+      Value: emuMessage1.ActivePowerL3.Value,
+    }
+    (*variables)["ActivePowerL123"] = database.VariableType { 
+      Variable: "ActivePowerL123", 
+      Unit: emuMessage1.ActivePowerL123.Unit,
+      Value: emuMessage1.ActivePowerL123.Value,
+    }
+    (*variables)["PowerfactorL1"] = database.VariableType { 
+      Variable: "PowerfactorL1", 
+      Unit: emuMessage1.PowerfactorL1.Unit,
+      Value: emuMessage1.PowerfactorL1.Value,
+    }
+    (*variables)["PowerfactorL2"] = database.VariableType { 
+      Variable: "PowerfactorL2", 
+      Unit: emuMessage1.PowerfactorL2.Unit,
+      Value: emuMessage1.PowerfactorL2.Value,
+    }
+    (*variables)["PowerfactorL3"] = database.VariableType { 
+      Variable: "PowerfactorL3", 
+      Unit: emuMessage1.PowerfactorL3.Unit,
+      Value: emuMessage1.PowerfactorL3.Value,
+    }
+
+    return nil
+  case 2:
+    var emuMessage2 emuMessage2
+    err := json.Unmarshal(decodedPayload, &emuMessage2)
+    if err != nil {
+      return fmt.Errorf("Unable to parse payload, fPort %d, error %s", fPort, err)
+    }
+    (*variables)["ActiveEnergyExport"] = database.VariableType { 
+      Variable: "ActiveEnergyExport", 
+      Unit: emuMessage2.ActiveEnergyExport.Unit,
+      Value: emuMessage2.ActiveEnergyExport.Value,
+    }
+    (*variables)["ActiveEnergyImport"] = database.VariableType { 
+      Variable: "ActiveEnergyImport", 
+      Unit: emuMessage2.ActiveEnergyImport.Unit,
+      Value: emuMessage2.ActiveEnergyImport.Value,
+    }
+    (*variables)["ReactiveEnergyExport"] = database.VariableType { 
+      Variable: "ReactiveEnergyExport", 
+      Unit: emuMessage2.ReactiveEnergyExport.Unit,
+      Value: emuMessage2.ReactiveEnergyExport.Value,
+    }
+    (*variables)["ReactiveEnergyImport"] = database.VariableType { 
+      Variable: "ReactiveEnergyImport", 
+      Unit: emuMessage2.ReactiveEnergyImport.Unit,
+      Value: emuMessage2.ReactiveEnergyImport.Value,
+    }
+    return nil
+  default:
+    return fmt.Errorf("Unexpected fPort %d", fPort)
+  }
+}
+

src/udi/handlers/ttn/ttn.go

@@ -1,29 +1,162 @@
 package ttn
 
-import "log"
-import "fmt"
-import "udi/handlers/handler"
+import (
+  "fmt"
+  "log"
+  "time"
+  "encoding/json"
+  "udi/config"
+  "udi/handlers/handler"
+  "udi/handlers/ttn/models/emuProfIILoRaCfg1"
+  "udi/handlers/ttn/models/draginoLdds75"
+  "udi/handlers/ttn/models/draginoLmds200"
+  "udi/handlers/ttn/models/draginoLse01"
+  "udi/database"
+)
 
-var idSeq int = 0
 
 type TTNHandler struct {
-  id int
+  handler.CommonHandler
+  dbh *database.DatabaseHandle
 }
 
-func NewTTNHandler() handler.Handler {
+type DecodedPayloaderHolder struct {
+  Payload []byte
+}
+
+type uplinkMessage struct {
+  EndDeviceIds struct {
+    DeviceId string `json:"device_id"`
+    ApplicationIds struct {
+      ApplicationId string `json:"application_id"`
+    } `json:"application_ids"`
+    DevEui string `json:"dev_eui"`
+    JoinEui string `json:"join_eui"`
+    DevAddr string `json:"dev_addr"`
+  } `json:"end_device_ids"`
+  ReceivedAt string `json:"received_at"`
+  UplinkMessage struct {
+    FCnt int `json:"f_cnt"`
+    FPort int `json:"f_port"`
+    FrmPayload string `json:"frm_payload"`
+    DecodedPayload DecodedPayloaderHolder `json:"decoded_payload"`
+    RxMetadata []struct {
+      GatewayIds struct {
+        GatewayId string `json:"gateway_id"`
+        Eui string `json:"eui"`
+      } `json:"gateway_ids"`
+      Time string `json:"time"`
+      Rssi int `json:"rssi"`
+      ChannelRssi int `json:"channel_rssi"`
+      Snr float32 `json:"snr"`
+      ChannelIndex int `json:"channel_index"`
+    } `json:"rx_metadata"`
+    ConsumedAirtime string `json:"consumed_airtime"`
+  } `json:"uplink_message"`
+}
+
+type gatewayAttributes struct {
+  GatewayId string `json:"gateway_id"`
+  Rssi int `json:"rssi"`
+  Snr float32 `json:"snr"`
+}
+
+type attributes struct {
+  DeviceId string `json:"device_id"`
+  ApplicationId string `json:"application_id"`
+  FCnt int `json:"f_cnt"`
+  FPort int `json:"f_port"`
+  FrmPayload string `json:"frm_payload"`
+  Gateways []gatewayAttributes `json:"gateways"`
+  ConsumedAirtime string `json:"consumed_airtime"`
+}
+
+func (self *DecodedPayloaderHolder) UnmarshalJSON(data []byte) error {
+  self.Payload = data
+  return nil
+}
+
+func New(id string, config config.HandlerConfigT) handler.Handler {
   t := &TTNHandler {
-    id: idSeq,
   }
-  idSeq += 1
+  t.Id = id
+  t.dbh = database.NewDatabaseHandle()
   return t
 }
 
-func (self *TTNHandler) GetId() string {
-  return fmt.Sprintf("TTN%d", self.id)
-}
-
 func (self *TTNHandler) Handle(message handler.MessageT) {
-  log.Printf("Handler TTN %d processing %s -> %s", self.id, message.Topic, message.Payload)
+  log.Printf("Handler TTN %d processing %s -> %s", self.Id, message.Topic, message.Payload)
+
+  var measurement database.Measurement
+  measurement.Time = time.Now()
+
+  var uplinkMessage uplinkMessage
+  err := json.Unmarshal([]byte(message.Payload), &uplinkMessage)
+  if err != nil {
+    self.Lost("Error when unmarshaling message", err, message)
+    return
+  }
+  log.Printf("Parsed message: %s", uplinkMessage)
+
+  var attributes attributes
+  attributes.DeviceId = uplinkMessage.EndDeviceIds.DeviceId
+  attributes.ApplicationId = uplinkMessage.EndDeviceIds.ApplicationIds.ApplicationId
+  attributes.FCnt = uplinkMessage.UplinkMessage.FCnt
+  attributes.FPort = uplinkMessage.UplinkMessage.FPort
+  attributes.FrmPayload = uplinkMessage.UplinkMessage.FrmPayload
+  attributes.ConsumedAirtime = uplinkMessage.UplinkMessage.ConsumedAirtime
+  for _, rxm := range uplinkMessage.UplinkMessage.RxMetadata {
+    log.Printf("RXM: %s", rxm)
+    g := gatewayAttributes { GatewayId: rxm.GatewayIds.GatewayId, Rssi: rxm.Rssi, Snr: rxm.Snr }
+    attributes.Gateways = append(attributes.Gateways, g)
+  }
+  log.Printf("Attributes: %s", attributes)
+  measurement.Attributes = map[string]interface{} {
+    "DeviceId": attributes.DeviceId,
+    "ApplicationId": attributes.ApplicationId,
+    "FCnt": attributes.FCnt,
+    "FPort": attributes.FPort,
+    "FrmPayload": attributes.FrmPayload,
+    "Gateways": attributes.Gateways,
+    "ConsumedAirtime": attributes.ConsumedAirtime,
+  }
+
+  log.Printf("ApplicationId: %s, DeviceId: %s", attributes.ApplicationId, attributes.DeviceId)
+  device, err2 := self.dbh.GetDeviceByLabelAndApplication(attributes.ApplicationId, attributes.DeviceId)
+  if err2 != nil {
+    self.Lost("Error when loading device", err2, message)
+    return
+  }
+  measurement.Application = attributes.ApplicationId
+  measurement.Device = attributes.DeviceId
+  measurement.Attributes["DeviceType"] = device.DeviceType.ModelIdentifier
+
+  log.Printf("DeviceLabel: %s, DeviceType: %s", device.Label, device.DeviceType.ModelIdentifier)
+
+  var parser func(int, []byte, *map[string]database.VariableType, *database.Device) error
+  switch device.DeviceType.ModelIdentifier {
+  case "emu-prof-ii-lora-cfg1":
+    parser = emuProfIILoRaCfg1.Parse
+  case "dragino-ldds75":
+    parser = draginoLdds75.Parse
+  case "dragino-lmds200":
+    parser = draginoLmds200.Parse
+  case "dragino-lse01":
+    parser = draginoLse01.Parse
+  default:
+    self.Lost(fmt.Sprintf("No parser found for %s", device.DeviceType.ModelIdentifier), nil, message)
+    return
+  }
+
+  measurement.Values = make(map[string]database.VariableType)
+  err3 := parser(uplinkMessage.UplinkMessage.FPort, uplinkMessage.UplinkMessage.DecodedPayload.Payload, &(measurement.Values), device)
+  if err3 != nil {
+    self.Lost("Model parser failed", err3, message)
+    return
+  }
+  log.Printf("Prepared measurement item: %s", measurement)
+  self.dbh.StoreMeasurement(&measurement)
+  self.S()
 }
 
 

src/udi/main.go

@@ -5,6 +5,7 @@ import "os"
 import "os/signal"
 import "udi/mqtt"
 import "udi/config"
+import "udi/counter"
 import "udi/dispatcher"
 
 
@@ -23,6 +24,8 @@ func main() {
   mqtt.StartMqttClient()
   defer mqtt.StopMqttClient()
 
+  counter.InitCounter()
+
   log.Println("UDI running")
 
   c := make(chan os.Signal, 1)

src/udi/mqtt/mqtt.go

@@ -7,6 +7,7 @@ import MQTT "github.com/eclipse/paho.mqtt.golang"
 import "github.com/google/uuid"
 import "crypto/tls"
 import "udi/config"
+import "udi/counter"
 
 type Message struct {
   Topic string
@@ -26,10 +27,12 @@ func onMessageReceived(client MQTT.Client, message MQTT.Message) {
   }
   select {
   case InputChannel <- m:
+    counter.S("Received")
     {}
     //log.Println("Message sent to channel")
   default:
     log.Println("Channel full, message lost")
+    counter.F("Received")
   }
 }
 
@@ -54,7 +57,7 @@ func onConnect(client MQTT.Client) {
     if token := client.Subscribe(topic, 0, onMessageReceived); token.Wait() && token.Error() != nil {
       log.Fatalf("Unable to subscribe to topic %s, error %s", topic, token.Error())
     }
-    log.Printf("Successfully subscribed to topic %s", topic)
+    log.Printf("Topic %s subscribed", topic)
   }
 }
 
@@ -101,19 +104,19 @@ func StartMqttClient() {
 
   enableTls := config.Config.Mqtt.TlsEnable
   if enableTls == "true" {
-    log.Println("Enabling TLS connection")
+    //log.Println("Enabling TLS connection")
     tlsConfig := &tls.Config { 
       InsecureSkipVerify: true, 
     }
     opts.SetTLSConfig(tlsConfig)
 	}
 
-  log.Println("Trying to connect to broker")
+  log.Println("Broker connecting")
   mqttClient = MQTT.NewClient(opts)
   if token := mqttClient.Connect(); token.Wait() && token.Error() != nil {
     log.Fatalf("Unable to connect to broker %s, error %s", broker, token.Error())
   }
-  log.Printf("Successfully connected to broker %s", broker)
+  //log.Printf("Successfully connected to broker %s", broker)
 
   go outputDispatcher(mqttClient)