add status in ttn handlers

deployment/decrypt-secrets.sh

@@ -11,7 +11,7 @@ if [ "$MD5_CHECKSUM" = "" ]; then
 fi
 
 SECRETS_CIPHERTEXT_FILE=secrets.enc
-SECRETS_PLAINTEXT_FILE=secrets
+SECRETS_PLAINTEXT_FILE=/tmp/secrets
 TMP_FILE=`mktemp`
 POD_NAME_SUFFIX=`date +%s`
 
@@ -37,6 +37,7 @@ if [ "$MD5_CHECKSUM" != "$CALCULATED_CHECKSUM" ]; then
   exit 1
 fi
 
+# cat $TMP_FILE
 mv $TMP_FILE $SECRETS_PLAINTEXT_FILE
 
 

deployment/deploy.sh

@@ -5,9 +5,6 @@ if [ "$IMAGE_TAG" == "" ]; then
   exit 1
 fi
 
-./deployment/decrypt-secrets.sh || exit 1 
-. secrets
-rm secrets
 
 IMAGE_NAME=gitea.hottis.de/wn/udi
 
@@ -17,6 +14,11 @@ 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`
@@ -33,38 +35,31 @@ for NAMESPACE_DIR in `find $INSTANCES_DIR -type d -mindepth 1 -maxdepth 1`; do
     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="$NAMESPACE""_""$INSTANCE""_MQTT_PASSWORD"
-    MQTT_PASSWORD_VARIABLE=`echo $MQTT_PASSWORD_VARIABLE | tr - _`
+    MQTT_PASSWORD_VARIABLE=$VARIABLE_PREFIX"_MQTT_PASSWORD"
     MQTT_PASSWORD="${!MQTT_PASSWORD_VARIABLE}"
-    echo "MQTT_PASSWORD_VARIABLE: $MQTT_PASSWORD_VARIABLE"
-    echo "MQTT_PASSWORD: $MQTT_PASSWORD"
+    # 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
+      kubectl apply -f - -n $NAMESPACE
 
-    # set database configuration as secret
-    ## prepare configuration to access database to set udi database password
-    PGUSER=`kubectl get secret -n database timescaledb -o jsonpath="{.data.superuser-username}" | base64 -d`
-    PGHOST=`kubectl get services traefik -n system -o jsonpath="{.status.loadBalancer.ingress[0].ip}"`
-    PGPASSWORD=`kubectl get secret -n database timescaledb -o jsonpath="{.data.superuser-password}" | base64 -d`
-    PGSSLMODE=require
 
-    NEW_UDI_DB_LOGIN="udi""-""$NAMESPACE""-""$INSTANCE"
-    NEW_UDI_DB_PASSWORD=`tr -dc 'a-zA-Z0-9' < /dev/urandom | head -c 32`
-    NEW_UDI_DB_DATABASE="udi""-""$NAMESPACE""-""$INSTANCE"
+    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
 
-    DATABASE_MASTER_POD=`kubectl get pods -n database -l app=StackGresCluster -l role=master -o jsonpath='{.items[0].metadata.name}'`
-    kubectl exec -i $DATABASE_MASTER_POD -c postgres-util -n database -- psql <<EOF
-    BEGIN;
-    ALTER USER "$NEW_UDI_DB_LOGIN" WITH PASSWORD '$NEW_UDI_DB_PASSWORD';
-    COMMIT;
-EOF
-
     kubectl create secret generic $INSTANCE-udi-db-cred \
       --dry-run=client \
       -o yaml \

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

@@ -11,13 +11,12 @@
       "handler": "TTN",
       "id": "TTN0",
       "config": {
-        "databaseConnStr": "",
         "attributes": {
         }
       }
     }
   ],
   "archiver": {
-    "dir": "./tmp/udi"
+    "dir": "/archive"
   }
 }

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

@@ -1,17 +1,16 @@
 {
   "mqtt": {
     "broker": "ssl://eu1.cloud.thethings.network:8883",
-    "username": "com-passavant-geiger-poc@ttn",
+    "username": "de-hottis-lora-test1@ttn",
     "password": "ENV",
     "tlsEnable": "true"
   },
   "topicMappings": [
     {
-      "topics": [ "v3/com-passavant-geiger-poc@ttn/devices/#" ],
+      "topics": [ "v3/#" ],
       "handler": "TTN",
       "id": "TTN0",
       "config": {
-        "databaseConnStr": "",
         "attributes": {
         }
       }

deployment/instances/udi/default/config.json

@@ -4,6 +4,36 @@
     "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",

deployment/instances/udi/soil/config.json

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

deployment/load-config.sh

@@ -1,55 +0,0 @@
-#!/bin/bash
-
-FILE=$1
-if [ "$FILE" = "" ]; then
-  echo "give config file to load as first argument"
-  exit 1
-fi
-MQTT_PASSWORD=$2
-if [ "$MQTT_PASSWORD" = "" ]; then
-  echo "give mqtt password as second argument"
-  exit 1
-fi
-NAMESPACE=$3
-if [ "$NAMESPACE" = "" ]; then
-  echo "give namespace as third argument"
-  exit 1
-fi
-
-kubectl create secret generic udi-conf \
-  --from-literal=UDI_CONF="`cat $FILE`" \
-  -n $NAMESPACE \
-  --dry-run=client \
-  -o yaml \
-  --save-config | \
-kubectl apply -f -
-
-kubectl create secret generic mqtt-password \
-  --from-literal=MQTT_PASSWORD="$MQTT_PASSWORD" \
-  -n $NAMESPACE \
-  --dry-run=client \
-  -o yaml \
-  --save-config | \
-kubectl apply -f -
-
-. ~/Workspace/MyKubernetesEnv/ENVDB
-DATABASE="udi-$NAMESPACE"
-LOGIN="udi-$NAMESPACE"
-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

@@ -1,4 +1,32 @@
-U2FsdGVkX1+235sIaS3YkXthSjtLu/5ky8o0KGw4E0Bh2avnKV6Qg9XiKe5JnJOk
-IQcWgB9rwqg1oNFD1diaotk5AEGvejJawiUcsvHywx7U0XqGt7vhNdf3tp/Mjc0z
-BzbHykKfwnFzX3PACw78HJb+zk10DyDgEQ09o7wE6CZVCx5MXdbcZzrJ1a7a3edQ
-+FKkrwK5L/byPJk7lOmdOxC+Kq+uVGWRToUniABbYYaBDvtpXytan8BVZcKSjQQ/
+U2FsdGVkX1+DXC4uFXaRWr40xvTwUDMfmx3gZmixNJWP4djN5e5JZYmq2uWB/kQr
+0eCD4UM9cRnwyqCJudsOJnB8pT6XQgl/ZkSZavSOxG7r0uh90IqOe25nxWH3iiza
+oPWW0qR9KXB4qNQEAHkoww+dz7B2zFaDSQPgzm2oV9SWXfjhu0nDPcBO2e3gzSvU
+vCuwLnmG/4oacBgAeJHyys2NmW1e2ZnjbFOT+hMBtGPwwEIQ/mbq7IWrfiREUJ7B
+U1LoN4NPnkTtbFf63AkuQ6Lq0mkH6a6ZoVfkyg7kS9VIhznoDcZb29S/N6cGgoRG
+KFu1VbyjoCXnskZ/a9rw4e3Epau9qoeupmALlmcogOK3J0g1EhltmFMsfMnTDIkf
+Pj3t4+WZn0xozX89gLwNYYATALQfL+mAFRHpwx924Mh12tRzgSHyq+BHbcI5yjKA
+eNyP+gUT6nmrSGhlwBXsUzILsOCxxdhNsSd0h3S3huhuS+RivnXpGJQnR5vXmc8d
+iQE28Rx9YyrOM8+lpIL5mz7O7cSjEzwuIWLsnvs2nowYl8erWu7e+R9e2e8ulAgt
+UokWp2tOMFfvPU6bHJwDcre42Ozv8QefPQb9E8hcZe6U3ibpw715sV4YvSchxLCF
+cGzwgRSceKd+LOuoXOTfcUAvMzkOcV5/VRbX/ZbESt5ntw+g2AZqp3T+7iAnPmtx
+q2RN94Yes/8yKkvAV+wO6qBv7mCn79ZPYdf8C+eWVQv/uJ9BRmJMxy0FfUFz+kY0
++QL72T1BJwBRWkWmAmOuQSvQ6q+MtzekwuXfQgmcdtdWZAUxcnsjJBKAFRthEgU9
+/N2UkHezMx5hn2wa7K+Z8v29zmlwQaRJGVW0xzpzKlYrneAqm65o+aIEtWMiIUwz
+zuCPN7tZcjxzOmwnLXupOclw2fE76vFzxc8g6pRH5bbpmNynB6SszCM8Gya2LUOV
+OK8B7fThcx4XimWRE8jrBQHCaYEEqgXaLxUSrMV7s9yLbKRQI24YTWcIMY8pmp5t
+/Viq6qqtiWZ7BwrtTR8KFzNEJcz5hC7LDaP1hJR4MEbRv8VPsyM3mxOWflmmR6jI
+x8UBv585SLRiqNMFqJ+BryrBbiwtbm15z1jUyQTSGM8J+dULbLgDp1O1fQphXJKV
+7dbkV5CXPoglOK/jy3bj4zSfG0Fr068aBkXvkBlJ2X088Xxoj59mep+ZI0SJj4L1
+vK6EecRxUHUQUltarj9A2LaHw5iqG5QD5px3O1wN4xqiuh95FuwmvIa2DNlclgca
+4xzWXUFwoJFoKLsaX8dkxC3Zs4YybSEpRntU8+ElQeAQoDB5gPgjvKoep+JHCSWZ
+w+ZhNT+F5+tbRToPwyWn9k9lRBaDcDWQAOQtcoGxzZ0I4j0CWnC0uzfTJ48skhKk
+xBo5sujuyZ3m0I/icdVY0hAt9Ok+3hB4hrvfMAT13zK0u+a917d6HKfeh95BNXAN
+CzSUtC/J0VU1tk2cF71pJS6T1oTxm/+ptwPMclOiKqzgkxoZzITd08JLe6d50HCI
+fw4LL7Z65HOE4kxHySxtCHWBw0d+44C+H9+g6SKWDNCUpv+xbc9VAMM6/rl8vW2G
+bJHkSqxvx2mi3X3Ti4BLXNb7IWYjJVirTNtzYXbqgoPDotviuyoyB3v9bAGbg+gQ
+KpzQJR+j+ODHITG9wJs14WtL/Ll3TO6Tz3XGfmgLiPs7N5oReNdQYrW5TadzttoQ
++WDKYoFXLXyHOT43BHRu+6V16Mpj/khdR7DFoj5AKbS3IoSuMiniowf2sztvtnsY
+j7jwL3zaNv7qSf0p4TYo1HhXSggunaDRqBebpNVKAbHTU0ygiSizZAKIXb97/Gbp
+rop1vSH0GNZWcV653vNFCKoSecVPwAA7LRQcW1RpyzE/NBdRLmh+rbONeh9FlJA4
+JWpMK7RNA0JsaTy7Ti9/I7cYxUpAxP/6oHaH+P16bpoppyx6toH0Q94uXTU/Nlpl
+PpipMYgTHN8SPwSBWUzIYQ==

migration/.gitignore

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

migration/migrate-level.py

@@ -0,0 +1,38 @@
+import psycopg2
+from loguru import logger
+
+try:
+    srcConn = psycopg2.connect(database="level_monitoring_berresheim")
+    srcConn.autocommit = False
+    destConn = psycopg2.connect(database="udi-berresheim")
+    destConn.autocommit = False
+
+    with srcConn.cursor() as srcCur, destConn.cursor() as destCur:
+        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=}")
+
+            destCur.execute("insert into measurements (time, application, device, attributes, values) values(%s, %s, %s, %s, %s)",
+                            (destTime, destApplication, destDevice, destAttributes, destValues))
+            destConn.commit()
+finally:
+    if srcConn:
+        srcConn.close()
+    if destConn:
+        destConn.close()
+
+

migration/requirements.txt

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

queries/hottis.sql

@@ -45,3 +45,10 @@ create or replace view temperature_v as
     from measurements
     where application in ('Temperature Multisensor', 'Temperature Shelly Plus HT');
 
+create or replace view humidity_v as
+  select time,
+         cast(values->'Value'->>'value' as float) as humidity,
+         device
+    from measurements
+    where application in ('Humidity Multisensor');
+

queries/old-pv-yield-query.sql

@@ -0,0 +1,73 @@
+-- query
+
+with 
+    first_day_in_year as (
+        select 
+                date_trunc('day', min(time)) as day
+            from pv_power_measurement_t
+            where 
+                time between date_trunc('year', time) and now()
+    ),
+    first_value_in_year as (
+        select 
+                time_bucket('1 day', time) as interval,
+                first(exportenergyactive, time) as energy
+            from pv_power_measurement_t
+            where 
+                time between (select day from first_day_in_year) and (select day from first_day_in_year) + interval '1 day' and
+                status = 'Ok'
+            group by interval
+    ),
+    first_day_in_month as (
+        select 
+                date_trunc('day', min(time)) as day
+            from pv_power_measurement_t
+            where 
+                time between date_trunc('month', now()) and now()
+    ),
+    first_value_in_month as (
+        select 
+                time_bucket('1 day', time) as interval,
+                first(exportenergyactive, time) as energy
+            from pv_power_measurement_t
+            where 
+                time between (select day from first_day_in_month) and (select day from first_day_in_month) + interval '1 day' and
+                status = 'Ok'
+            group by interval
+    ),
+    first_value_in_day as (
+        select
+                time_bucket('1 day', time) as interval,
+                first(exportenergyactive, time) as energy
+            from pv_power_measurement_t
+            where time >= date_trunc('day', now())
+            group by interval
+    ),
+    last_value as (
+        select  
+                time_bucket('1 day', time) as interval,
+                last(exportenergyactive, time) as energy
+            from pv_power_measurement_t
+            where 
+                time between date_trunc('day', now()) and date_trunc('day', now()) + interval '1 day' and
+                status = 'Ok'
+            group by interval
+    )
+    select
+        extract(year from (select day from first_day_in_year))::text as period_value,
+        'Year' as period_name,
+        round(((select energy from last_value) - (select energy from first_value_in_year))::numeric, 2) as yield
+    union
+    select
+        to_char((select day from first_day_in_month), 'Month') as period_value,
+        'Month' as period_name,
+        round(((select energy from last_value) - (select energy from first_value_in_month))::numeric, 2) as yield
+    union
+    select
+        now()::date::text as period_value,
+        'Day' as period_name,
+        round(((select energy from last_value) - (select energy from first_value_in_day))::numeric, 2) as yield;
+
+-- output format
+-- wn@atuin:~/Workspace/go-workspace/src/universal-data-ingest  [main ≡ +0 ~1 -0 !]$ mosquitto_sub -h 172.23.1.102 -v -t IoT/PV/Yields
+-- IoT/PV/Yields {"Month":"1.43","Year":"285.39","Day":"0.00"}

src/udi/ENVDB.uditest

@@ -1,6 +1,6 @@
-PGUSER="udi"
+PGUSER="uditest"
 PGHOST=`kubectl get services traefik -n system -o jsonpath="{.status.loadBalancer.ingress[0].ip}"`
-PGPASSWORD=`kubectl get secrets udi-db-cred -n udi -o jsonpath="{.data.PGPASSWORD}" | base64 --decode`
+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-iot2.json

@@ -1,21 +0,0 @@
-{
-  "mqtt": {
-    "broker": "mqtt://172.23.1.102:1883",
-    "tlsEnable": "false"
-  },
-  "topicMappings": [
-    {
-      "topics": [ "ttn/#" ],
-      "handler": "TTN",
-      "id": "TTN0",
-      "config": {
-        "databaseConnStr": "",
-        "attributes": {
-        }
-      }
-    }
-  ],
-  "archiver": {
-    "dir": "./tmp/udi"
-  }
-}

src/udi/config-test-ttn.json

@@ -1,8 +1,7 @@
 {
   "mqtt": {
     "broker": "ssl://eu1.cloud.thethings.network:8883",
-    "username": "de-hottis-level-monitoring@ttn",
-    "passwordEnvVar": "MQTT_PASSWORD",
+    "username": "de-hottis-lora-test1@ttn",
     "tlsEnable": "true"
   },
   "topicMappings": [
@@ -11,7 +10,6 @@
       "handler": "TTN",
       "id": "TTN0",
       "config": {
-        "databaseConnStr": "",
         "attributes": {
         }
       }

src/udi/config-test.json

@@ -9,7 +9,6 @@
       "handler": "PV",
       "id": "PV",
       "config": {
-        "databaseConnStr": "",
         "attributes": {
         }
       }
@@ -19,11 +18,40 @@
       "handler": "MBGW3",
       "id": "MBGW3",
       "config": {
-        "databaseConnStr": "",
         "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",

src/udi/config-ttn.json

@@ -1,29 +0,0 @@
-{
-  "mqtt": {
-    "broker": "mqtt://172.23.1.102:1883",
-    "tlsEnable": "false"
-  },
-  "topicMappings": [
-    {
-      "topics": [ "v3/de-hottis-level-monitoring@ttn/devices/+/up" ],
-      "handler": "TTN"
-    }
-  ],
-  "handlers": [
-    {
-      "name": "TTN",
-      "databaseConnStr": "",
-      "attributes": {
-      }
-    },
-		{
-			"name": "IoT",
-			"databaseConnStr": "",
-			"attributes": {
-			}
-		}
-  ],
-  "archiver": {
-    "dir": "./tmp/udi"
-  }
-}

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/database.go

@@ -5,6 +5,7 @@ import (
   "log"
   //"time"
   "fmt"
+  "udi/counter"
   "gorm.io/driver/postgres"
   "gorm.io/gorm"
 )
@@ -24,7 +25,7 @@ func NewDatabaseHandle() *DatabaseHandle {
   } else {
     db.dbh = conn
     db.initialized = true
-    log.Println("Database connection opened")
+    //log.Println("Database connection opened")
   }
   return &db
 }
@@ -32,16 +33,19 @@ func NewDatabaseHandle() *DatabaseHandle {
 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 := 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) {

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.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.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()
   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,28 +1,22 @@
 package sver
 
 import (
-  "log"
   "time"
   "strconv"
   "strings"
   "regexp"
-  "fmt"
-  "reflect"
-  "encoding/json"
-  "github.com/oliveagle/jsonpath"
+  "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
-  payloadJsonpath *jsonpath.Compiled
   dbh *database.DatabaseHandle
 }
 
@@ -44,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"] == "" {
@@ -64,18 +56,6 @@ func NewSverHandler(config config.HandlerConfigT) handler.Handler {
   } else {
     t.payloadRegex = nil
   }
-  payloadJsonpath := config.Attributes["payloadJsonpath"]
-  if payloadJsonpath != "" {
-    j, err := jsonpath.Compile(payloadJsonpath)
-    if err != nil {
-      log.Printf("Unable to compile jsonpath %s", payloadJsonpath)
-      t.payloadJsonpath = nil
-    } else {
-      t.payloadJsonpath = j
-    }
-  } else {
-    t.payloadJsonpath = nil
-  }
 
   if config.Attributes["deviceFrom"] != TOPIC_SEL && config.Attributes["deviceFrom"] != PAYLOAD_SEL && config.Attributes["deviceFrom"] != CONSTANT_SEL {
     log.Printf("Error: invalid value %s for deviceFrom", config.Attributes["deviceFrom"])
@@ -128,22 +108,15 @@ func NewSverHandler(config config.HandlerConfigT) handler.Handler {
 
   t.config = localConfig
 
+  t.Id = id
   t.ready = true
   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)
@@ -160,31 +133,21 @@ func (self *SingleValueExtractorRegexHandler) Handle(message handler.MessageT) {
     payloadMatches = self.payloadRegex.FindStringSubmatch(message.Payload)
     //log.Printf("Matches: %s", strings.Join(payloadMatches, ", "))
   }
-  if self.payloadJsonpath != nil {
-    var jsonData interface{}
-    json.Unmarshal([]byte(message.Payload), &jsonData)
-    p, err := self.payloadJsonpath.Lookup(jsonData)
-    if err != nil {
-      lost(fmt.Sprintf("jsonpath error: %s", err), message)
-      return
-    }
-    log.Printf("XXXX: %s", reflect.TypeOf(p))
-  }
 
   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 && self.payloadJsonpath == nil {
-      lost("no payloadRegex or payloadJsonpath defined, devicePart can't be used", message)
+    if self.payloadRegex == nil {
+      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]
@@ -199,12 +162,12 @@ func (self *SingleValueExtractorRegexHandler) Handle(message handler.MessageT) {
 
   switch self.config.valueFrom {
   case PAYLOAD_SEL:
-    if self.payloadRegex == nil && self.payloadJsonpath == nil {
-      lost("no payloadRegex or payloadJsonpath defined, valuePart can't be used", message)
+    if self.payloadRegex == nil {
+      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]
@@ -214,12 +177,12 @@ func (self *SingleValueExtractorRegexHandler) Handle(message handler.MessageT) {
 
   switch self.config.unitFrom {
   case PAYLOAD_SEL:
-    if self.payloadRegex == nil && self.payloadJsonpath == nil {
-      lost("no payloadRegex or payloadJsonpath defined, unitPart can't be used", message)
+    if self.payloadRegex == nil {
+      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]
@@ -231,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

@@ -2,7 +2,7 @@ package draginoLdds75
 
 import (
   "fmt"
-  "log"
+  // "log"
   "strings"
   "strconv"
   "encoding/json"
@@ -26,7 +26,7 @@ type message struct {
   TempC_DS18B20 string `json:"TempC_DS18B20"`
 }
 
-func Parse(fPort int, decodedPayload []byte, variables *map[string]database.VariableType, device *database.Device) error {
+func Parse(fPort int, decodedPayload []byte, _ string, variables *map[string]database.VariableType, attributes *map[string]interface{}, device *database.Device) error {
   if fPort != 2 {
     return fmt.Errorf("Unexpected fPort %d", fPort)
   }
@@ -55,11 +55,20 @@ func Parse(fPort int, decodedPayload []byte, variables *map[string]database.Vari
     Unit: "mm",
     Value: distance,
   }
+
+  if distance == 20 {
+    (*attributes)["Status"] = "invalid value"
+  } else if distance == 0 {
+    (*attributes)["Status"] = "no sensor detected"
+  } else {
+    (*attributes)["Status"] = "Ok"
+  }
+
   groundLevelI, exists := device.Attributes["GroundLevel"]
   groundLevelS, ok := groundLevelI.(string)
   groundLevel, err3 := strconv.Atoi(groundLevelS)
   if exists && err3 == nil && ok {
-    log.Println("add corrected distance")
+    //log.Println("add corrected distance")
     correctedDistance := groundLevel - distance
     (*variables)["CorrectedDistance"] = database.VariableType {
       Label: "CorrectedDistance",
@@ -67,11 +76,11 @@ func Parse(fPort int, decodedPayload []byte, variables *map[string]database.Vari
       Unit: "mm",
       Value: correctedDistance,
     }
-  } else {
+  } /* 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/draginoLmds200/draginoLmds200.go

@@ -26,7 +26,7 @@ type message struct {
   Dis2 int `json:"dis2"`
 }
 
-func Parse(fPort int, decodedPayload []byte, variables *map[string]database.VariableType, device *database.Device) error {
+func Parse(fPort int, decodedPayload []byte, _ string, variables *map[string]database.VariableType, attributes *map[string]interface{}, device *database.Device) error {
   if fPort != 2 {
     return fmt.Errorf("Unexpected fPort %d", fPort)
   }
@@ -55,6 +55,15 @@ func Parse(fPort int, decodedPayload []byte, variables *map[string]database.Vari
     Unit: "mm",
     Value: distance2,
   }
+
+  if distance1 == 2 {
+    (*attributes)["Status"] = "invalid value"
+  } else if distance1 == 1 {
+    (*attributes)["Status"] = "no sensor detected"
+  } else {
+    (*attributes)["Status"] = "Ok"
+  }
+
   groundLevelI, exists := device.Attributes["GroundLevel"]
   groundLevelS, ok := groundLevelI.(string)
   groundLevel, err3 := strconv.Atoi(groundLevelS)

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, _ string, variables *map[string]database.VariableType, _ *map[string]interface{}, _ *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/emuProfIILoRaCfg1.go

@@ -177,7 +177,7 @@ type emuMessage1 struct {
 
 
 
-func Parse(fPort int, decodedPayload []byte, variables *map[string]database.VariableType, _ *database.Device) error {
+func Parse(fPort int, decodedPayload []byte, _ string, variables *map[string]database.VariableType, _ *map[string]interface{}, _ *database.Device) error {
   //log.Printf("Parse input: %d, %s", fPort, decodedPayload)
   switch fPort {
   case 1:

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

@@ -0,0 +1,25 @@
+package rawPayloadPrinter
+
+import (
+  "log"
+  "fmt"
+  "encoding/base64"
+  "encoding/hex"
+  "udi/database"
+)
+
+
+func Parse(fPort int, _ []byte, frmPayload string, variables *map[string]database.VariableType, _ *map[string]interface{}, _ *database.Device) error {
+  if fPort != 2 {
+    return fmt.Errorf("Unexpected fPort %d", fPort)
+  }
+
+  bytes, err := base64.StdEncoding.DecodeString(frmPayload)
+  if err != nil {
+    return fmt.Errorf("Unable to base64-decode payload: %v", err)
+  }
+  hexString := hex.EncodeToString(bytes)
+
+  log.Printf("Payload: %s", hexString)
+  return nil
+}

src/udi/handlers/ttn/ttn.go

@@ -1,8 +1,8 @@
 package ttn
 
 import (
-  "log"
   "fmt"
+  //"log"
   "time"
   "encoding/json"
   "udi/config"
@@ -10,13 +10,14 @@ import (
   "udi/handlers/ttn/models/emuProfIILoRaCfg1"
   "udi/handlers/ttn/models/draginoLdds75"
   "udi/handlers/ttn/models/draginoLmds200"
+  "udi/handlers/ttn/models/draginoLse01"
+  "udi/handlers/ttn/models/rawPayloadPrinter"
   "udi/database"
 )
 
-var idSeq int = 0
 
 type TTNHandler struct {
-  id int
+  handler.CommonHandler
   dbh *database.DatabaseHandle
 }
 
@@ -76,25 +77,16 @@ func (self *DecodedPayloaderHolder) UnmarshalJSON(data []byte) error {
   return nil
 }
 
-func NewTTNHandler(config config.HandlerConfigT) handler.Handler {
+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 lost(msg string, message handler.MessageT) {
-  log.Printf("Error: %s, message %s is lost", msg, message)
-}
-
 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()
@@ -102,7 +94,7 @@ func (self *TTNHandler) Handle(message handler.MessageT) {
   var uplinkMessage uplinkMessage
   err := json.Unmarshal([]byte(message.Payload), &uplinkMessage)
   if err != nil {
-    lost(fmt.Sprintf("Error when unmarshaling message: %s, ", err), message)
+    self.Lost("Error when unmarshaling message", err, message)
     return
   }
   //log.Printf("Parsed message: %s", uplinkMessage)
@@ -115,7 +107,7 @@ func (self *TTNHandler) Handle(message handler.MessageT) {
   attributes.FrmPayload = uplinkMessage.UplinkMessage.FrmPayload
   attributes.ConsumedAirtime = uplinkMessage.UplinkMessage.ConsumedAirtime
   for _, rxm := range uplinkMessage.UplinkMessage.RxMetadata {
-    log.Printf("RXM: %s", rxm)
+    //log.Printf("RXM: %s", rxm)
     g := gatewayAttributes { GatewayId: rxm.GatewayIds.GatewayId, Rssi: rxm.Rssi, Snr: rxm.Snr }
     attributes.Gateways = append(attributes.Gateways, g)
   }
@@ -133,7 +125,7 @@ func (self *TTNHandler) Handle(message handler.MessageT) {
   //log.Printf("ApplicationId: %s, DeviceId: %s", attributes.ApplicationId, attributes.DeviceId)
   device, err2 := self.dbh.GetDeviceByLabelAndApplication(attributes.ApplicationId, attributes.DeviceId)
   if err2 != nil {
-    lost(fmt.Sprintf("Error when loading device: %s, ", err2), message)
+    self.Lost("Error when loading device", err2, message)
     return
   }
   measurement.Application = attributes.ApplicationId
@@ -142,7 +134,7 @@ func (self *TTNHandler) Handle(message handler.MessageT) {
 
   //log.Printf("DeviceLabel: %s, DeviceType: %s", device.Label, device.DeviceType.ModelIdentifier)
 
-  var parser func(int, []byte, *map[string]database.VariableType, *database.Device) error
+  var parser func(int, []byte, string, *map[string]database.VariableType, *map[string]interface{}, *database.Device) error
   switch device.DeviceType.ModelIdentifier {
   case "emu-prof-ii-lora-cfg1":
     parser = emuProfIILoRaCfg1.Parse
@@ -150,19 +142,29 @@ func (self *TTNHandler) Handle(message handler.MessageT) {
     parser = draginoLdds75.Parse
   case "dragino-lmds200":
     parser = draginoLmds200.Parse
+  case "dragino-lse01":
+    parser = draginoLse01.Parse
+  case "raw-payload-printer":
+    parser = rawPayloadPrinter.Parse
   default:
-    lost(fmt.Sprintf("No parser found for %s", device.DeviceType.ModelIdentifier), message)
+    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)
+  err3 := parser(uplinkMessage.UplinkMessage.FPort, 
+                 uplinkMessage.UplinkMessage.DecodedPayload.Payload, 
+                 uplinkMessage.UplinkMessage.FrmPayload,
+                 &(measurement.Values), 
+                 &(measurement.Attributes),
+                 device)
   if err3 != nil {
-    lost(fmt.Sprintf("Model parser failed: %s", err3), message)
+    self.Lost("Model parser failed", err3, message)
     return
   }
-  log.Printf("Prepared measurement item: %s", measurement)
+  //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)