fix deploy

.gitignore

@@ -1,5 +1,5 @@
-config-*.json
 src/udi/udi
 src/udi/migrate_schema
 tmp/
 ENVDB
+ENVDB.cluster

.woodpecker.yml

@@ -25,6 +25,8 @@ steps:
     commands:
       - 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 - 
+      - for N in "udi udi-pg"; do
+          cat $CI_WORKSPACE/deployment/deploy-yml.tmpl | sed -e 's,%IMAGE%,'$IMAGE_NAME':'$CI_COMMIT_TAG',' | kubectl apply -n $N -f - ;
+        done
     when:
       - event: tag

deployment/deploy-yml.tmpl

@@ -2,7 +2,6 @@ apiVersion: v1
 kind: PersistentVolumeClaim
 metadata:
   name: udi-archive
-  namespace: udi
 spec:
   accessModes:
     - ReadWriteOnce
@@ -15,9 +14,10 @@ apiVersion: apps/v1
 kind: Deployment
 metadata:
   name: udi
-  namespace: udi
   labels:
     app: udi
+  annotations:
+    secret.reloader.stakater.com/reload: "udi-conf,udi-db-cred,mqtt-password"
 spec:
   replicas: 1
   selector:
@@ -31,12 +31,13 @@ spec:
       containers:
         - name: udi
           image: %IMAGE%
-          env:
-            - name: UDI_CONF
-              valueFrom:
-                secretKeyRef:
-                  name: udi-conf
-                  key: UDI_CONF
+          envFrom:
+            - secretRef: 
+                name: udi-db-cred
+            - secretRef:
+                name: mqtt-password
+            - secretRef:
+                name: udi-conf
           volumeMounts:
             - mountPath: /archive
               name: udi-archive

deployment/load-config.sh

@@ -5,9 +5,9 @@ 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"
+MQTT_PASSWORD=$2
+if [ "$MQTT_PASSWORD" = "" ]; then
+  echo "give mqtt password as second argument"
   exit 1
 fi
 NAMESPACE=$3
@@ -16,10 +16,40 @@ if [ "$NAMESPACE" = "" ]; then
   exit 1
 fi
 
-kubectl create secret generic $SECRET_NAME \
+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
+

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');
+

src/udi/ENVDB.udi

@@ -0,0 +1,6 @@
+PGUSER="udi"
+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`
+PGSSLMODE=require
+PGDATABASE="uditest"
+export PGUSER PGHOST PGPASSWORD PGSSLMODE PGDATABASE

src/udi/config-iot-cluster.json

@@ -0,0 +1,105 @@
+{
+  "mqtt": {
+    "broker": "mqtt://emqx01-anonymous-cluster-internal.broker.svc.cluster.local:1883",
+    "tlsEnable": "false"
+  },
+  "topicMappings": [
+    {
+      "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"
+  }
+}

src/udi/config-iot.json

@@ -0,0 +1,105 @@
+{
+  "mqtt": {
+    "broker": "mqtt://172.23.1.102:1883",
+    "tlsEnable": "false"
+  },
+  "topicMappings": [
+    {
+      "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": "./tmp/udi"
+  }
+}

src/udi/config-iot2.json

@@ -0,0 +1,21 @@
+{
+  "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-pg-cluster.json

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

src/udi/config-pg.json

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

src/udi/config-ttn.json

@@ -0,0 +1,29 @@
+{
+  "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/config/config.go

@@ -17,10 +17,11 @@ type ConfigT struct {
     TlsEnable string `json:"tlsEnable"`
   } `json:"mqtt"`
   TopicMappings []struct {
+    Id string `json:"id"`
     Topics []string `json:"topics"`
     Handler string `json:"handler"`
+    Config HandlerConfigT `json:"config"`
   } `json:"topicMappings"`
-  Handlers map[string]HandlerConfigT `json:"handlers"`
   Archiver struct {
     Dir string `json:"dir"`
   }
@@ -34,5 +35,9 @@ func LoadConfiguration() {
   if err != nil {
     log.Fatalf("Unable to parse configuration: %s", err)
   }
+
+  if Config.Mqtt.Password == "ENV" {
+    Config.Mqtt.Password = os.Getenv("MQTT_PASSWORD")
+  }
 }
 

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,7 @@ package database
 import (
   "log"
   //"time"
+  "fmt"
   "gorm.io/driver/postgres"
   "gorm.io/gorm"
 )
@@ -27,13 +28,13 @@ func NewDatabaseHandle(dsn string) *DatabaseHandle {
   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)
     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)
     return
@@ -42,4 +43,27 @@ func (dbh *DatabaseHandle) StoreMeasurement(measurement *Measurement) {
   log.Println("Successfully stored measurement")
 }
 
+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

@@ -12,7 +12,8 @@ import "udi/handlers/ttn"
 import "udi/handlers/iot"
 import "udi/handlers/pv"
 import "udi/handlers/mbgw3"
-import "udi/handlers/sve"
+import "udi/handlers/sver"
+import "udi/handlers/svej"
 
 
 var handlerMap map[string]handler.Handler = make(map[string]handler.Handler)
@@ -21,11 +22,39 @@ var archiverChannel chan handler.MessageT = make(chan handler.MessageT, 100)
 func InitDispatcher() {
   log.Printf("Initializing dispatcher")
   go archiver()
-  handlerMap["TTN"] = ttn.NewTTNHandler()
-  handlerMap["IoT"] = iot.NewIoTHandler()
-  handlerMap["PV"] = pv.NewPvHandler(config.Config.Handlers["PV"])
-  handlerMap["MBGW3"] = mbgw3.NewMbgw3Handler(config.Config.Handlers["MBGW3"])
-  handlerMap["SVE"] = sve.NewSveHandler(config.Config.Handlers["SVE"])
+
+  for _, mapping := range config.Config.TopicMappings {
+    // log.Printf("Trying to initialize %s", mapping)
+
+    var factory interface{}
+    switch mapping.Handler {
+    case "TTN":
+      factory = ttn.NewTTNHandler
+    case "IoT":
+      factory = iot.NewIoTHandler
+    case "PV":
+      factory = pv.NewPvHandler
+    case "MBGW3":
+      factory = mbgw3.NewMbgw3Handler
+    case "SVER":
+      factory = sver.NewSverHandler
+    case "SVEJ":
+      factory = svej.NewSvejHandler
+    default:
+      factory = nil
+      log.Printf("No handler %s found, ignore mapping", mapping.Handler)
+    }
+
+    fn, ok := factory.(func(config.HandlerConfigT) handler.Handler)
+    if ! ok {
+      log.Println("Typ Assertion failed")
+      break
+    }
+    handler := fn(mapping.Config)
+    handlerMap[mapping.Id] = handler
+  }
+
+  log.Printf("handlerMap: %s", handlerMap)
 }
 
 func storeMessage(filename string, item handler.MessageT) {
@@ -85,13 +114,13 @@ 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.Handler)
-        handler, exists := handlerMap[mapping.Handler]
+        log.Printf("Handle message in handler %s", mapping.Id)
+        handler, exists := handlerMap[mapping.Id]
         if exists {
           handler.Handle(message)
           return
         } else {
-          log.Printf("Handler %s not found, message %s is lost", mapping.Handler, message)
+          log.Printf("Handler %s not found, message %s is lost", mapping.Id, message)
         }
       }
     }

src/udi/go.mod

@@ -5,6 +5,7 @@ go 1.21.3
 require (
 	github.com/eclipse/paho.mqtt.golang v1.4.3
 	github.com/google/uuid v1.4.0
+	github.com/oliveagle/jsonpath v0.0.0-20180606110733-2e52cf6e6852
 	gorm.io/driver/postgres v1.5.4
 	gorm.io/gorm v1.25.5
 )

src/udi/go.sum

@@ -17,6 +17,8 @@ github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD
 github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc=
 github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ=
 github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8=
+github.com/oliveagle/jsonpath v0.0.0-20180606110733-2e52cf6e6852 h1:Yl0tPBa8QPjGmesFh1D0rDy+q1Twx6FyU7VWHi8wZbI=
+github.com/oliveagle/jsonpath v0.0.0-20180606110733-2e52cf6e6852/go.mod h1:eqOVx5Vwu4gd2mmMZvVZsgIqNSaW3xxRThUJ0k/TPk4=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=

src/udi/handlers/handler/handler.go

@@ -9,6 +9,7 @@ type MessageT struct {
 }
 
 type Handler interface {
+  GetId() string
   Handle(MessageT)
 }
 

src/udi/handlers/iot/iot.go

@@ -1,6 +1,7 @@
 package iot
 
 import "log"
+import "fmt"
 import "udi/handlers/handler"
 
 var idSeq int = 0
@@ -9,7 +10,7 @@ type IoTHandler struct {
   id int
 }
 
-func NewIoTHandler() *IoTHandler {
+func NewIoTHandler() handler.Handler {
   t := &IoTHandler {
     id: idSeq,
   }
@@ -17,6 +18,10 @@ func NewIoTHandler() *IoTHandler {
   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)
 }

src/udi/handlers/mbgw3/mbgw3.go

@@ -5,6 +5,7 @@ import (
   //"reflect"
   "time"
   "strconv"
+  "fmt"
   "encoding/json"
   "udi/config"
   "udi/handlers/handler"
@@ -30,7 +31,7 @@ type Observation struct {
 }
 
 
-func NewMbgw3Handler(config config.HandlerConfigT) *Mbgw3Handler {
+func NewMbgw3Handler(config config.HandlerConfigT) handler.Handler {
   t := &Mbgw3Handler {
     id: idSeq,
   }
@@ -39,6 +40,10 @@ func NewMbgw3Handler(config config.HandlerConfigT) *Mbgw3Handler {
   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)
 

src/udi/handlers/pv/pv.go

@@ -4,6 +4,7 @@ import (
   "log"
   "reflect"
   "time"
+  "fmt"
   "encoding/json"
   "udi/config"
   "udi/handlers/handler"
@@ -39,7 +40,7 @@ type PvValue struct {
 }
 
 
-func NewPvHandler(config config.HandlerConfigT) *PvHandler {
+func NewPvHandler(config config.HandlerConfigT) handler.Handler {
   t := &PvHandler {
     id: idSeq,
   }
@@ -48,6 +49,10 @@ func NewPvHandler(config config.HandlerConfigT) *PvHandler {
   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)
 

src/udi/handlers/svej/svej.go

@@ -0,0 +1,169 @@
+package svej
+
+import (
+  "log"
+  "time"
+  "strconv"
+  "strings"
+  "fmt"
+  "github.com/oliveagle/jsonpath"
+  "encoding/json"
+  "udi/config"
+  "udi/handlers/handler"
+  "udi/database"
+)
+
+var idSeq int = 0
+
+type SingleValueExtractorJsonpathHandler struct {
+  id int
+  ready bool
+  application string
+  deviceSelector string
+  valueSelector string
+  unitSelector string
+  deviceJsonpath *jsonpath.Compiled
+  valueJsonpath *jsonpath.Compiled
+  unitJsonpath *jsonpath.Compiled
+  dbh *database.DatabaseHandle
+}
+
+/*
+Valid values for selectors:
+J:JsonpathExpression
+T:TopicPartIndex
+C:ConstantValue
+*/
+
+
+func NewSvejHandler(config config.HandlerConfigT) handler.Handler {
+  t := &SingleValueExtractorJsonpathHandler {
+    id: idSeq,
+    ready: false,
+  }
+  idSeq += 1
+
+  if config.Attributes["application"] == "" {
+    log.Println("Error: application not configured")
+    return t
+  }
+  t.application = config.Attributes["application"]
+  
+  t.deviceSelector = config.Attributes["deviceSelector"]
+  if t.deviceSelector[:2] == "J:" {
+    jp, err := jsonpath.Compile(t.deviceSelector[2:])
+    if err != nil {
+      log.Printf("Unable to compile deviceJsonpath: %s, %s", t.deviceSelector[2:], err)
+      return t
+    }
+    t.deviceJsonpath = jp
+  }
+  t.valueSelector = config.Attributes["valueSelector"]
+  if t.valueSelector[:2] == "J:" {
+    jp, err := jsonpath.Compile(t.valueSelector[2:])
+    if err != nil {
+      log.Printf("Unable to compile valueJsonpath: %s, %s", t.valueSelector[2:], err)
+      return t
+    }
+    t.valueJsonpath = jp
+  }
+  t.unitSelector = config.Attributes["unitSelector"]
+  if t.unitSelector[:2] == "J:" {
+    jp, err := jsonpath.Compile(t.unitSelector[2:])
+    if err != nil {
+      log.Printf("Unable to compile unitJsonpath: %s, %s", t.unitSelector[2:], err)
+      return t
+    }
+    t.unitJsonpath = jp
+  }
+
+  t.ready = true
+  t.dbh = database.NewDatabaseHandle(config.DatabaseConnStr)
+  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] {
+  case "J:":
+    r, e := jp.Lookup(jPayload)
+    if e != nil {
+      return "", fmt.Errorf("jp.Lookup failed with %s", e)
+    }
+    res = fmt.Sprint(r)
+  case "T:":
+    i, e := strconv.Atoi(selector[2:])
+    if e != nil {
+      return "", fmt.Errorf("Atoi failed with %s", e)
+    }
+    if i >= len(subTopics) {
+      return "", fmt.Errorf("not enough subtopics")
+    }
+    res = subTopics[i]
+  case "C:":
+    res = selector[2:]
+  default:
+    return "", fmt.Errorf("Invalid selector: %s", selector[:2])
+  }
+  return res, nil
+}
+
+
+func (self *SingleValueExtractorJsonpathHandler) Handle(message handler.MessageT) {
+  if ! self.ready {
+    log.Println("Handler is not marked as ready, message %s is lost", message)
+    return
+  }
+  log.Printf("Handler SingleValueExtractorJsonpath %d processing %s -> %s", self.id, message.Topic, message.Payload)
+
+  var measurement database.Measurement
+  measurement.Time = time.Now()
+  measurement.Application = self.application
+  
+  subTopics := strings.Split(message.Topic, "/")
+  //log.Printf("Subtopics: %s", strings.Join(subTopics, ", "))
+  var jPayload interface{}
+  err := json.Unmarshal([]byte(message.Payload), &jPayload)
+  if err != nil {
+    lost(fmt.Sprintf("Unable to unmarshal payload: %s", 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)
+    return
+  }
+  value, err2 := extractionHelper(subTopics, jPayload, self.valueSelector, self.valueJsonpath)
+  if err2 != nil {
+    lost(fmt.Sprintf("Value extraction failed with %s", 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)
+    return
+  }
+
+  measurement.Device = device
+
+  var variable database.VariableType
+  variable.Label = ""
+  variable.Variable = ""
+  variable.Unit = unit
+  variable.Value = value
+  measurement.Values = make(map[string]database.VariableType)
+  measurement.Values["Value"] = variable
+
+  log.Printf("Prepared measurement item: %s", measurement)
+  self.dbh.StoreMeasurement(&measurement)
+}
+

src/udi/handlers/sver/sver.go

@@ -1,4 +1,4 @@
-package sve
+package sver
 
 import (
   "log"
@@ -6,6 +6,10 @@ import (
   "strconv"
   "strings"
   "regexp"
+  "fmt"
+  "reflect"
+  "encoding/json"
+  "github.com/oliveagle/jsonpath"
   "udi/config"
   "udi/handlers/handler"
   "udi/database"
@@ -13,16 +17,18 @@ import (
 
 var idSeq int = 0
 
-type SingleValueExtractorHandler struct {
+type SingleValueExtractorRegexHandler struct {
   id int
   ready bool
   config localConfig
   payloadRegex *regexp.Regexp
+  payloadJsonpath *jsonpath.Compiled
   dbh *database.DatabaseHandle
 }
 
 const TOPIC_SEL = "topic"
 const PAYLOAD_SEL = "payload"
+const PAYLOAD_FULL_SEL = "payload-full"
 const CONSTANT_SEL = "constant"
 
 type localConfig struct {
@@ -38,13 +44,13 @@ type localConfig struct {
 }
 
 
-func NewSveHandler(config config.HandlerConfigT) *SingleValueExtractorHandler {
-  t := &SingleValueExtractorHandler {
+func NewSverHandler(config config.HandlerConfigT) handler.Handler {
+  t := &SingleValueExtractorRegexHandler {
     id: idSeq,
     ready: false,
   }
   idSeq += 1
-  
+
   var localConfig localConfig
   if config.Attributes["application"] == "" {
     log.Println("Error: application not configured")
@@ -58,6 +64,18 @@ func NewSveHandler(config config.HandlerConfigT) *SingleValueExtractorHandler {
   } 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"])
@@ -75,31 +93,35 @@ func NewSveHandler(config config.HandlerConfigT) *SingleValueExtractorHandler {
   // empty device is valid
   localConfig.device = config.Attributes["device"]
 
-  if config.Attributes["valueFrom"] != TOPIC_SEL && config.Attributes["valueFrom"] != PAYLOAD_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"]
 
-  valuePart, err2 := strconv.Atoi(config.Attributes["valuePart"])
-  if err2 != nil {
-    log.Printf("Error: unable to convert valuePart to number: %s", err2)
-    return t
+  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)
+      return t
+    }
+    localConfig.valuePart = valuePart
   }
-  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"]
 
-  unitPart, err3 := strconv.Atoi(config.Attributes["unitPart"])
-  if err3 != nil {
-    log.Printf("Error: unable to convert unitPart to number: %s", err3)
-    return t
+  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)
+      return t
+    }
+    localConfig.unitPart = unitPart
   }
-  localConfig.unitPart = unitPart
 
   // empty unit is valid
   localConfig.unit = config.Attributes["unit"]
@@ -111,11 +133,15 @@ func NewSveHandler(config config.HandlerConfigT) *SingleValueExtractorHandler {
   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 *SingleValueExtractorHandler) Handle(message handler.MessageT) {
+func (self *SingleValueExtractorRegexHandler) Handle(message handler.MessageT) {
   if ! self.ready {
     log.Println("Handler is not marked as ready, message %s is lost", message)
     return
@@ -134,6 +160,16 @@ func (self *SingleValueExtractorHandler) 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:
@@ -143,11 +179,11 @@ func (self *SingleValueExtractorHandler) Handle(message handler.MessageT) {
     }
     measurement.Device = subTopics[self.config.devicePart]
   case PAYLOAD_SEL:
-    if self.payloadRegex == nil {
-      lost("no payloadRegex defined, devicePart can't be used", message)
+    if self.payloadRegex == nil && self.payloadJsonpath == nil {
+      lost("no payloadRegex or payloadJsonpath defined, devicePart can't be used", message)
       return
     }
-    if self.config.devicePart >= len(subTopics) {
+    if self.config.devicePart >= len(payloadMatches) {
       lost("devicePart out of range", message)
       return
     }
@@ -162,37 +198,27 @@ func (self *SingleValueExtractorHandler) 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)
+    if self.payloadRegex == nil && self.payloadJsonpath == nil {
+      lost("no payloadRegex or payloadJsonpath defined, valuePart can't be used", message)
       return
     }
-    if self.config.valuePart >= len(subTopics) {
+    if self.config.valuePart >= len(payloadMatches) {
       lost("valuePart out of range", message)
       return
     }
     variable.Value = payloadMatches[self.config.valuePart]
+  case PAYLOAD_FULL_SEL:
+    variable.Value = message.Payload
   }
 
   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)
+    if self.payloadRegex == nil && self.payloadJsonpath == nil {
+      lost("no payloadRegex or payloadJsonpath defined, unitPart can't be used", message)
       return
     }
-    if self.config.unitPart >= len(subTopics) {
+    if self.config.unitPart >= len(payloadMatches) {
       lost("unitPart out of range", message)
       return
     }

src/udi/handlers/ttn/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/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,259 @@
+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) (map[string]database.VariableType, 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 nil, fmt.Errorf("Unable to parse payload, fPort %d, error %s", fPort, err)
+    }
+    variables := make(map[string]database.VariableType)
+    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 variables, nil
+  case 2:
+    var emuMessage2 emuMessage2
+    err := json.Unmarshal(decodedPayload, &emuMessage2)
+    if err != nil {
+      return nil, fmt.Errorf("Unable to parse payload, fPort %d, error %s", fPort, err)
+    }
+    variables := make(map[string]database.VariableType)
+    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 variables, nil
+  default:
+    return nil, fmt.Errorf("Unexpected fPort %d", fPort)
+  }
+}
+

src/udi/handlers/ttn/ttn.go

@@ -1,24 +1,159 @@
 package ttn
 
 import "log"
+import "fmt"
+import "time"
+import "encoding/json"
+import "udi/config"
 import "udi/handlers/handler"
+import "udi/handlers/ttn/models/emuProfIILoRaCfg1"
+import "udi/database"
 
 var idSeq int = 0
 
 type TTNHandler struct {
   id int
+  dbh *database.DatabaseHandle
 }
 
-func NewTTNHandler() *TTNHandler {
+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 NewTTNHandler(config config.HandlerConfigT) handler.Handler {
   t := &TTNHandler {
     id: idSeq,
   }
   idSeq += 1
+  t.dbh = database.NewDatabaseHandle(config.DatabaseConnStr)
   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()
+
+  var uplinkMessage uplinkMessage
+  err := json.Unmarshal([]byte(message.Payload), &uplinkMessage)
+  if err != nil {
+    lost(fmt.Sprintf("Error when unmarshaling message: %s, ", 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 {
+    lost(fmt.Sprintf("Error when loading device: %s, ", err2), message)
+    return
+  }
+  measurement.Application = attributes.ApplicationId
+  measurement.Device = attributes.DeviceId
+
+  //log.Printf("DeviceLabel: %s, DeviceType: %s", device.Label, device.DeviceType.ModelIdentifier)
+
+  var parser func(int, []byte) (map[string]database.VariableType, error)
+  switch device.DeviceType.ModelIdentifier {
+  case "emu-prof-ii-lora-cfg1":
+    parser = emuProfIILoRaCfg1.Parse
+  default:
+    lost(fmt.Sprintf("No parser found for %s", device.DeviceType.ModelIdentifier), message)
+    return
+  }
+
+  variables, err3 := parser(uplinkMessage.UplinkMessage.FPort, uplinkMessage.UplinkMessage.DecodedPayload.Payload)
+  if err3 != nil {
+    lost(fmt.Sprintf("Model parser failed: %s", err3), message)
+    return
+  }
+  measurement.Values = variables
+  log.Printf("Prepared measurement item: %s", measurement)
+  self.dbh.StoreMeasurement(&measurement)
 }