float fix 1

.gitignore

@@ -1,5 +1,8 @@
 src/udi/udi
+src/udi/main
 src/udi/migrate_schema
 tmp/
 ENVDB
 ENVDB.cluster
+deployment/secrets.txt
+deployment/secrets

.woodpecker.yml

@@ -1,29 +1,39 @@
+when:
+  - event: tag
+
+
 steps:
   build:
+    image: golang:1.22.5-alpine3.20
+    commands:
+      - GOPATH=/woodpecker/go
+      - cd src/udi
+      - go mod tidy
+      - go build -a -installsuffix nocgo -o udi main.go
+      - cp udi ../..
+
+  dockerize:
     image: plugins/kaniko
     settings:
-      repo:
+      repo: ${FORGE_NAME}/${CI_REPO}
-        from_secret: image_name
       registry: 
-        from_secret: container_registry
+        from_secret: local_registry
-      tags: latest,${CI_COMMIT_SHA},${CI_COMMIT_TAG}
+      tags: latest,${CI_COMMIT_TAG}
       username: 
-        from_secret: container_registry_username
+        from_secret: local_username
       password: 
-        from_secret: container_registry_password
+        from_secret: local_password
       dockerfile: Dockerfile
-    when:
-      - event: [push, tag]
 
   deploy:
-    image: portainer/kubectl-shell:latest
+    image: quay.io/wollud1969/k8s-admin-helper:0.4.1
-    secrets:
+    environment:
-      - source: kube_config
+      KUBE_CONFIG_CONTENT:
-        target: KUBE_CONFIG_CONTENT
+        from_secret: kube_config
+      GPG_PASSPHRASE:
+        from_secret: gpg_passphrase
     commands:
       - export IMAGE_TAG=$CI_COMMIT_TAG
       - printf "$KUBE_CONFIG_CONTENT" > /tmp/kubeconfig
       - export KUBECONFIG=/tmp/kubeconfig
       - ./deployment/deploy.sh
-    when:
-      - event: tag

Dockerfile

@@ -1,16 +1,8 @@
-FROM golang:1.21-alpine as builder
-
-RUN mkdir -p /go/src
-COPY ./src/ /go/src
-WORKDIR /go/src/udi
-RUN go build -a -installsuffix nocgo -o udi main.go
-
-
 FROM scratch
 
 ENV UDI_CONF ""
 
-COPY --from=builder /go/src/udi ./
+COPY udi ./
+
 ENTRYPOINT ["./udi"]
 
-

attic/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
 }

attic/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)

attic/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
+}
+

attic/ttn/models/draginoLsn50/draginoLsn50.go

@@ -0,0 +1,74 @@
+package draginoLsn50
+
+import (
+  "fmt"
+  "encoding/json"
+  "udi/database"
+)
+
+/*
+      "decoded_payload": {
+        "ALARM_status": "FALSE",
+        "BatV": 3.659,
+        "Temp_Black": 3276.7,
+        "Temp_Red": 22.6,
+        "Temp_White": 3276.7,
+        "Work_mode": "DS18B20"
+      },
+*/
+type message struct {
+  ALARM_status string `json:"ALARM_status"`
+  Bat float32 `json:"BatV"`
+  Work_mode string `json:"Work_mode"`
+  Temp_Black float32 `json:"Temp_Black"`
+  Temp_Red float32 `json:"Temp_Red"`
+  Temp_White float32 `json:"Temp_White"`
+}
+
+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)
+  }
+  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)["Alarm"] = database.VariableType {
+    Label: "Alarm",
+    Variable: "Alarm",
+    Unit: "",
+    Value: message.ALARM_status,
+  }
+  (*variables)["Temp_Red"] = database.VariableType {
+    Label: "Temp_Red",
+    Variable: "Temperature",
+    Unit: "°C",
+    Value: message.Temp_Red,
+  }
+  (*variables)["Temp_Black"] = database.VariableType {
+    Label: "Temp_Black",
+    Variable: "Temperature",
+    Unit: "°C",
+    Value: message.Temp_Black,
+  }
+  (*variables)["Temp_White"] = database.VariableType {
+    Label: "Temp_White",
+    Variable: "Temperature",
+    Unit: "°C",
+    Value: message.Temp_White,
+  }
+
+  (*attributes)["Status"] = "Ok"
+
+  return nil
+}
+
+
+

attic/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:

attic/ttn/models/hottisGy21/hottisGY21.go

@@ -0,0 +1,55 @@
+package hottisGy21
+
+import (
+  //"log"
+  "fmt"
+  "bytes"
+  "encoding/base64"
+  "encoding/binary"
+  "udi/database"
+)
+
+type hottisGy21Values struct {
+  Connected uint8
+  Status uint8
+  RawHumidity uint16
+  RawTemperature uint16
+}
+
+
+func Parse(fPort int, _ []byte, frmPayload string, variables *map[string]database.VariableType, attributes *map[string]interface{}, _ *database.Device) error {
+  if fPort != 2 {
+    return fmt.Errorf("Unexpected fPort %d", fPort)
+  }
+
+  b, err := base64.StdEncoding.DecodeString(frmPayload)
+  if err != nil {
+    return fmt.Errorf("Unable to base64-decode payload: %v", err)
+  }
+
+  var values hottisGy21Values
+  err = binary.Read(bytes.NewReader(b), binary.LittleEndian, &values)
+  if err != nil {
+    return fmt.Errorf("Unable to cast into struct: %v", err)
+  }
+  var temperature float32 = -46.85 + 175.72 * (float32(values.RawTemperature) / 65536.0)
+  var humidity float32 = -6 + 125 * (float32(values.RawHumidity) / 65536.0);
+
+  // log.Printf("CO2: %f, Temp: %f, Hum: %f, Status: %d", co2concentration, temperature, humidity, values.Status)
+
+  (*variables)["Humidity"] = database.VariableType {
+    Label: "Humidity",
+    Variable: "Humidity",
+    Unit: "%",
+    Value: humidity,
+  }
+  (*variables)["Temperature"] = database.VariableType {
+    Label: "Temperature",
+    Variable: "Temperature",
+    Unit: "°C",
+    Value: temperature,
+  }
+
+  (*attributes)["Status"] = values.Status
+  return nil
+}

attic/ttn/models/hottisScd30/hottisScd30.go

@@ -0,0 +1,69 @@
+package hottisScd30
+
+import (
+  //"log"
+  "fmt"
+  "bytes"
+  "encoding/base64"
+  "encoding/binary"
+  "udi/database"
+)
+
+type hottisScd30Values struct {
+  Status uint8
+  CO2Conc int32
+  Temp int32
+  Hum int32
+  Bri int32
+}
+
+
+func Parse(fPort int, _ []byte, frmPayload string, variables *map[string]database.VariableType, attributes *map[string]interface{}, _ *database.Device) error {
+  if fPort != 2 {
+    return fmt.Errorf("Unexpected fPort %d", fPort)
+  }
+
+  b, err := base64.StdEncoding.DecodeString(frmPayload)
+  if err != nil {
+    return fmt.Errorf("Unable to base64-decode payload: %v", err)
+  }
+
+  var values hottisScd30Values
+  err = binary.Read(bytes.NewReader(b), binary.LittleEndian, &values)
+  if err != nil {
+    return fmt.Errorf("Unable to cast into struct: %v", err)
+  }
+  var co2concentration float32 = float32(values.CO2Conc) / 100;
+  var temperature float32 = float32(values.Temp) / 100;
+  var humidity float32 = float32(values.Hum) / 100;
+
+  // log.Printf("CO2: %f, Temp: %f, Hum: %f, Status: %d", co2concentration, temperature, humidity, values.Status)
+
+  (*variables)["CO2concentration"] = database.VariableType {
+    Label: "CO2concentration",
+    Variable: "Concentration",
+    Unit: "ppm",
+    Value: co2concentration,
+  }
+  (*variables)["Temperature"] = database.VariableType {
+    Label: "Temperature",
+    Variable: "Temperature",
+    Unit: "°C",
+    Value: temperature,
+  }
+  (*variables)["Humidity"] = database.VariableType {
+    Label: "Humidity",
+    Variable: "Humidity",
+    Unit: "%",
+    Value: humidity,
+  }
+  (*variables)["Brightness"] = database.VariableType {
+    Label: "Brightness",
+    Variable: "Brightness",
+    Unit: "",
+    Value: values.Bri,
+  }
+
+  (*attributes)["Status"] = values.Status
+  return nil
+}

attic/ttn/models/hottisThreeWayThermometer/hottisThreeWayThermometer.go

@@ -0,0 +1,93 @@
+package hottisThreeWayThermometer
+
+import (
+  "log"
+  "fmt"
+  "bytes"
+  "strconv"
+  "encoding/base64"
+  "encoding/binary"
+  "udi/database"
+)
+
+type hottisThreeWayThermometerValues struct {
+  Status uint8
+  Battery uint16
+  SensorAddress1 uint64
+  Value1 int32
+  SensorAddress2 uint64
+  Value2 int32
+  SensorAddress3 uint64
+  Value3 int32 
+}
+
+func getSensorName(sensorsMap *map[string]interface{}, sensorAddress uint64) string {
+  key := strconv.FormatUint(sensorAddress, 10)
+	if sensorName, exists := (*sensorsMap)[key].(string); exists {
+		return sensorName
+	}
+	return "Sensor" + key
+}
+
+func Parse(fPort int, _ []byte, frmPayload string, variables *map[string]database.VariableType, attributes *map[string]interface{}, device *database.Device) error {
+  deviceAttrs := (*device).Attributes
+  sensorsMap := deviceAttrs["Sensors"].(map[string]interface{})
+
+  if fPort != 2 {
+    return fmt.Errorf("Unexpected fPort %d", fPort)
+  }
+
+  b, err := base64.StdEncoding.DecodeString(frmPayload)
+  if err != nil {
+    return fmt.Errorf("Unable to base64-decode payload: %v", err)
+  }
+
+  var values hottisThreeWayThermometerValues
+  err = binary.Read(bytes.NewReader(b), binary.LittleEndian, &values)
+  if err != nil {
+    return fmt.Errorf("Unable to cast into struct: %v", err)
+  }
+
+  var battery float32 = float32(values.Battery) / 1000.0
+  var value1 float32 = float32(values.Value1) / 128.0
+  var value2 float32 = float32(values.Value2) / 128.0
+  var value3 float32 = float32(values.Value3) / 128.0
+
+  //log.Printf("Status: %d, Battery: %d", values.Status, values.Battery);
+  log.Printf("Status: %d", values.Status);
+  log.Printf("Battery: %d, %f", values.Battery, battery);
+  log.Printf("Sensor1: Addr: %d, Value: %f", values.SensorAddress1, value1);
+  log.Printf("Sensor2: Addr: %d, Value: %f", values.SensorAddress2, value2);
+  log.Printf("Sensor3: Addr: %d, Value: %f", values.SensorAddress3, value3);
+
+  (*variables)["Battery"] = database.VariableType {
+    Label: "Battery",
+    Variable: "Voltage",
+    Unit: "V",
+    Value: battery,
+  }
+  (*variables)["Temperature1"] = database.VariableType {
+    Label: getSensorName(&sensorsMap, values.SensorAddress1),
+    Variable: "Temperature",
+    Unit: "°C",
+    Value: value1,
+  }
+  (*variables)["Temperature2"] = database.VariableType {
+    Label: getSensorName(&sensorsMap, values.SensorAddress2),
+    Variable: "Temperature",
+    Unit: "°C",
+    Value: value2,
+  }
+  (*variables)["Temperature3"] = database.VariableType {
+    Label: getSensorName(&sensorsMap, values.SensorAddress3),
+    Variable: "Temperature",
+    Unit: "°C",
+    Value: value3,
+  }
+
+  (*attributes)["Status"] = values.Status
+  (*attributes)["SensorAddress1"] = values.SensorAddress1
+  (*attributes)["SensorAddress2"] = values.SensorAddress2
+  (*attributes)["SensorAddress3"] = values.SensorAddress3
+  return nil
+}

attic/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
+}

attic/ttn/ttn.go

@@ -1,8 +1,8 @@
 package ttn
 
 import (
-  "log"
   "fmt"
+  "log"
   "time"
   "encoding/json"
   "udi/config"
@@ -10,13 +10,18 @@ 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/draginoLsn50"
+  "udi/handlers/ttn/models/rawPayloadPrinter"
+  "udi/handlers/ttn/models/hottisScd30"
+  "udi/handlers/ttn/models/hottisGy21"
+  "udi/handlers/ttn/models/hottisThreeWayThermometer"
   "udi/database"
 )
 
-var idSeq int = 0
 
 type TTNHandler struct {
-  id int
+  handler.CommonHandler
   dbh *database.DatabaseHandle
 }
 
@@ -76,25 +81,17 @@ 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()
+  log.Printf("Handler TTN %d initialized", id)
   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 +99,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 +112,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 +130,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 +139,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 +147,37 @@ func (self *TTNHandler) Handle(message handler.MessageT) {
     parser = draginoLdds75.Parse
   case "dragino-lmds200":
     parser = draginoLmds200.Parse
+  case "dragino-lse01":
+    parser = draginoLse01.Parse
+  case "dragino-lsn50":
+    parser = draginoLsn50.Parse
+  case "raw-payload-printer":
+    parser = rawPayloadPrinter.Parse
+  case "hottis-scd30":
+    parser = hottisScd30.Parse
+  case "hottis-gy21":
+    parser = hottisGy21.Parse
+  case "hottis-threeway-thermometer":
+    parser = hottisThreeWayThermometer.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()
 }
 
 

deployment/deploy-yml.tmpl

@@ -17,7 +17,7 @@ metadata:
   labels:
     app: udi
   annotations:
-    secret.reloader.stakater.com/reload: "%PRE%-udi-conf,%PRE%-udi-db-cred,%PRE%-mqtt-password"
+    secret.reloader.stakater.com/reload: "%PRE%-udi-conf,%PRE%-udi-db-cred,%PRE%-mqtt-password,%PRE%-udi-influxdb-cred"
 spec:
   replicas: 1
   selector:
@@ -36,6 +36,8 @@ spec:
                 name: %PRE%-udi-db-cred
             - secretRef:
                 name: %PRE%-mqtt-password
+            - secretRef:
+                name: %PRE%-udi-influxdb-cred
             - configMapRef:
                 name: %PRE%-udi-conf
           volumeMounts:

deployment/deploy.sh

@@ -4,8 +4,14 @@ if [ "$IMAGE_TAG" == "" ]; then
   echo "Make sure IMAGE_TAG is set"
   exit 1
 fi
+if [ "$GPG_PASSPHRASE" == "" ]; then
+  echo "Make sure GPG_PASSPHRASE is set"
+  exit 1
+fi
 
-IMAGE_NAME=gitea.hottis.de/wn/udi
+
+
+IMAGE_NAME=$FORGE_NAME/$CI_REPO
 
 CONFIG_FILE=config.json
 
@@ -13,6 +19,14 @@ 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
+
+gpg --decrypt --yes --batch --passphrase "$GPG_PASSPHRASE" --homedir /tmp/.gnupg -o /tmp/secrets secrets.asc
+. /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`
@@ -29,37 +43,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=$VARIABLE_PREFIX"_MQTT_PASSWORD"
-    MQTT_PASSWORD_VARIABLE=`echo $MQTT_PASSWORD_VARIABLE | tr - _`
     MQTT_PASSWORD="${!MQTT_PASSWORD_VARIABLE}"
-    echo "MQTT_PASSWORD_VARIABLE: $MQTT_PASSWORD_VARIABLE"
+    # echo "MQTT_PASSWORD_VARIABLE: $MQTT_PASSWORD_VARIABLE"
-    echo "MQTT_PASSWORD: $MQTT_PASSWORD"
+    # 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"
+    LOGIN_VARIABLE=$VARIABLE_PREFIX"_PGUSER"
-    NEW_UDI_DB_PASSWORD=`tr -dc 'a-zA-Z0-9' < /dev/urandom | head -c 32`
+    NEW_UDI_DB_LOGIN="${!LOGIN_VARIABLE}"
-    NEW_UDI_DB_DATABASE="udi""-""$NAMESPACE""-""$INSTANCE"
+    PASSWORD_VARIABLE=$VARIABLE_PREFIX"_PGPASSWORD"
-    NEW_UDI_DB_HOST=timescaledb.database.svc.cluster.local
+    NEW_UDI_DB_PASSWORD="${!PASSWORD_VARIABLE}"
-
+    DATABASE_VARIABLE=$VARIABLE_PREFIX"_PGDATABASE"
-    DATABASE_MASTER_POD=`kubectl get pods -n database -l app=StackGresCluster -l role=master -o jsonpath='{.items[0].metadata.name}'`
+    NEW_UDI_DB_DATABASE="${!DATABASE_VARIABLE}"
-    kubectl exec -i $DATABASE_MASTER_POD -c postgres-util -n database -- psql <<EOF
+    NEW_UDI_DB_HOST=database.database1.svc.cluster.local
-    BEGIN;
+    INFLUXDB_URL=$VARIABLE_PREFIX"_INFLUXDB_URL"
-    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 \
@@ -72,6 +80,13 @@ EOF
       --from-literal=PGSSLMODE="require" | \
       kubectl apply -f - -n $NAMESPACE
 
+    kubectl create secret generic $INSTANCE-udi-influxdb-cred \
+      --dry-run=client \
+      -o yaml \
+      --save-config \
+      --from-literal=INFLUXDB_URL="${!INFLUXDB_URL}" | \
+      kubectl apply -f - -n $NAMESPACE
+
     # set configuration as configMap
     kubectl create configmap $INSTANCE-udi-conf \
       --from-literal=UDI_CONF="`cat $CONFIG_FILE`" \

deployment/disabled-instances/udi-saerbeck/default/config.json

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

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

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

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

@@ -1,23 +1,22 @@
 {
   "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": {
         }
       }
     }
   ],
   "archiver": {
-    "dir": "./tmp/udi"
+    "dir": "/archive"
   }
 }

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

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

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

@@ -0,0 +1,226 @@
+{
+  "mqtt": {
+    "broker": "mqtt://mosquitto-broker-mqtt-anon-cluster.mosquitto.svc.cluster.local:1883",
+    "tlsEnable": "false"
+  },
+  "topicMappings": [
+    {
+      "topics": [ "snmp" ],
+      "handler": "PREP",
+      "id": "SNMP",
+      "config": {
+        "attributes": {
+        }
+      }
+    },
+    {
+      "topics": [ "tsm" ],
+      "handler": "PREP",
+      "id": "TSM",
+      "config": {
+        "attributes": {
+        }
+      }
+    },
+    {
+      "topics": [ "dt1/ai/periodic/1" ],
+      "handler": "DT1T",
+      "id": "DT1T.0",
+      "config": {
+        "attributes": {
+          "Application": "Temperature Wago",
+          "Device": "Freezer",
+          "HardLow": "-273",
+          "SoftLow": "-50",
+          "SoftHigh": "20",
+          "HardHigh": "100"
+        }
+      }
+    },
+    {
+      "topics": [ "dt1/ai/periodic/3" ],
+      "handler": "DT1T",
+      "id": "DT1T.1",
+      "config": {
+        "attributes": {
+          "Application": "Temperature Wago",
+          "Device": "Outdoor",
+          "HardLow": "-273",
+          "SoftLow": "-60",
+          "SoftHigh": "60",
+          "HardHigh": "100"
+        }
+      }
+    },
+    {
+      "topics": [ "IoT/PV/Values" ],
+      "handler": "PV",
+      "id": "PV",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+        }
+      }
+    },
+    {
+      "topics": [ "IoT/Car/Values" ],
+      "handler": "Car",
+      "id": "Car",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+        }
+      }
+    },
+    {
+      "topics": [ "locative/event/#" ],
+      "handler": "Locative",
+      "id": "Locative",
+      "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",
+          "valueType": "float",
+          "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": [ "zigbee2mqtt/+" ],
+      "handler": "Z2M",
+      "id": "Z2M",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+        }
+      }
+    },
+    {
+      "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": "SVEJ3",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+          "application": "Humidity Shelly Plus HT",
+          "deviceSelector": "T:1",
+          "valueSelector": "J:$.rh",
+          "unitSelector": "C:%"
+        }
+      }
+    },
+    {
+      "topics": [ "shellies/sensor/+/status/temperature:0" ],
+      "handler": "SVEJ",
+      "id": "SVEJ4",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+          "application": "Shellies Sensor Temperature",
+          "deviceSelector": "T:2",
+          "valueSelector": "J:$.tC",
+          "unitSelector": "C:°C"
+        }
+      }
+    },
+    {
+      "topics": [ "shellies/sensor/+/status/humidity:0" ],
+      "handler": "SVEJ",
+      "id": "SVEJ5",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+          "application": "Shellies Sensor Humidity",
+          "deviceSelector": "T:2",
+          "valueSelector": "J:$.rh",
+          "unitSelector": "C:%"
+        }
+      }
+    },
+    {
+      "topics": [ "shellies/sensor/+/status/devicepower:0" ],
+      "handler": "SVEJ",
+      "id": "SVEJ6",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+          "application": "Shellies Sensor Power",
+          "deviceSelector": "T:2",
+          "valueSelector": "J:$.battery.percent",
+          "unitSelector": "C:%"
+        }
+      }
+    }
+  ],
+  "archiver": {
+    "dir": "/archive"
+  }
+}

deployment/instances/udi/default/config.json

@@ -1,105 +0,0 @@
-{
-  "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"
-  }
-}

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.asc

@@ -0,0 +1,8 @@
+-----BEGIN PGP MESSAGE-----
+
+jA0ECQMIYUoTHR96Qfb90psBoxuk38UXPXTWPCmdW690bi2+w34S4NLHZvHfe3Ra
+nck319+PXvr0agfHGZ733hhTQv4sa8I2o6ICrgFqtKGfHmgnqL5kYNP9+NuV/IsF
+x3dxwjEejsZ5GYn/zk+CQceItQ8nyyJc2ms1KwTu2r4hMzuHmnVtvKxNCzPrw2N5
+SJIRhh41eequFkzELQqqXXu10raBFsttOemVhA==
+=TySu
+-----END PGP MESSAGE-----

queries/berresheim.sql

@@ -0,0 +1,8 @@
+create or replace view level_v as
+  select time, 
+         cast(values->'CorrectedDistance'->>'value' as float) as level,
+         cast(values->'Battery'->>'value' as float) as battery,
+         attributes->>'Status' as status,
+         device
+    from measurements
+    where application = 'de-hottis-level-monitoring';

queries/hottis.sql

@@ -43,5 +43,140 @@ create or replace view temperature_v as
          cast(values->'Value'->>'value' as float) as temperature,
          device
     from measurements
-    where application in ('Temperature Multisensor', 'Temperature Shelly Plus HT');
+    where application in ('Temperature Multisensor', 'Temperature Shelly Plus HT')
+  union
+  select time,
+         cast(values->'Temperature'->>'value' as float) as temperature,
+         device
+    from measurements
+    where application = 'Zigbee2MQTT Hottis Eupenstr.' and
+          attributes->>'DeviceModel' in ('WSDCGQ11LM', 'WSDCGQ01LM');
 
+create or replace view voltage_v as
+  select time,
+         cast(values->'Voltage'->>'value' as float) / 1000 as voltage,
+         device
+    from measurements
+    where application = 'Zigbee2MQTT Hottis Eupenstr.' and
+          attributes->>'DeviceModel' in ('WSDCGQ11LM', 'WSDCGQ01LM');
+
+create or replace view temperature2_v as
+  select time,
+         cast(values->'Value'->>'value' as float) as temperature,
+         device
+    from measurements
+    where application = 'Temperature Wago';
+
+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')
+  union
+  select time,
+         cast(values->'Humidity'->>'value' as float) as temperature,
+         device
+    from measurements
+    where application = 'Zigbee2MQTT Hottis Eupenstr.' and
+          attributes->>'DeviceModel' in ('WSDCGQ11LM', 'WSDCGQ01LM');
+
+create or replace view soil_v as
+  select time,
+         cast(values->'Water'->>'value' as float) as water,
+         cast(values->'Conductance'->>'value' as float) as conductance,
+         cast(values->'Temperature'->>'value' as float) as temperature,
+         device
+    from measurements
+    where application = 'de-hottis-app01' and attributes->>'DeviceType' = 'dragino-lse01';
+
+create or replace view co2_v as
+  select time,
+         cast(m.values->'CO2concentration'->>'value' as float) as co2concentration,
+         cast(m.values->'Humidity'->>'value' as float) as humidity,
+         cast(m.values->'Temperature'->>'value' as float) as temperature,
+         cast(m.values->'Brightness'->>'value' as int) as brightness,
+         m.device as device,
+         d.attributes->>'Label' as label 
+    from measurements m, devices d
+    where m.application = 'de-hottis-app01' and 
+          m.attributes->>'DeviceType' = 'hottis-scd30' and
+          m.device = d.label;
+
+create or replace view locative_v as
+  select time,
+         device as person,
+         values->'Location'->>'value' as location,
+         values->'Trigger'->>'value' as direction
+    from measurements
+    where application = 'Locative';
+
+create or replace view router_v as
+  select time,
+         device,
+         cast(values->'wan-in'->>'value'  as int) as wanInOctetsPerSeconds,
+         cast(values->'wan-out'->>'value'  as int) as wanOutOctetsPerSeconds
+    from measurements
+    where application = 'SNMP' and device = '172.16.3.1';
+
+create or replace view lora_sht21_v as
+  select time,
+         cast(values->'Humidity'->>'value' as float) as humidity,
+         cast(values->'Temperature'->>'value' as float) as temperature,
+         m.device as device,
+         d.attributes->>'Label' as label
+    from measurements m, devices d
+    where m.application = 'de-hottis-app01' and 
+          m.attributes->>'DeviceType' = 'hottis-gy21' and
+          m.device = d.label;
+
+create or replace view ntp_server_snmp_v as
+  select time,
+         device,
+         cast(values->'load1'->>'value'  as float) as laLoad1,
+         cast(values->'lan-in'->>'value'  as int) as lanInOctetsPerSeconds,
+         cast(values->'lan-out'->>'value'  as int) as lanOutOctetsPerSeconds
+    from measurements
+    where application = 'SNMP' and device = '172.16.13.10';
+
+create or replace view ntp_server_variables_v as
+  select time,
+         device,
+         cast(values->'rootdisp'->>'value'  as float) as rootdisp
+    from measurements
+    where application = 'TSM' and device = '172.16.13.10';
+
+
+--  Status            string  `unit:""   json:"status"`
+--  Timestamp         string  `unit:""   json:"timestamp"`
+--  VoltageL1         float32 `unit:"V"  json:"voltageL1"`
+--  VoltageL2         float32 `unit:"V"  json:"voltageL2"`
+--  VoltageL3         float32 `unit:"V"  json:"voltageL3"`
+--  CurrentL1         float32 `unit:"A"  json:"currentL1"`
+--  CurrentL2         float32 `unit:"A"  json:"currentL2"`
+--  CurrentL3         float32 `unit:"A"  json:"currentL3"`
+--  PowerL1           float32 `unit:"W"  json:"powerL1"`
+--  PowerL2           float32 `unit:"W"  json:"powerL2"`
+--  PowerL3           float32 `unit:"W"  json:"powerL3"`
+--  TotalImportEnergy float32 `unit:"Wh" json:"totalImportEnergy"`
+--  TotalExportEnergy float32 `unit:"Wh" json:"totalExportEnergy"`
+--  Cnt               int     `unit:""   json:"cnt"`
+
+
+create or replace view car_values_v as
+  select time, 
+         cast(values->'VoltageL1'->>'value' as float) as voltage_l1,
+         cast(values->'VoltageL2'->>'value' as float) as voltage_l2,
+         cast(values->'VoltageL3'->>'value' as float) as voltage_l3,
+         cast(values->'CurrentL1'->>'value' as float) as current_l1,
+         cast(values->'CurrentL2'->>'value' as float) as current_l2,
+         cast(values->'CurrentL3'->>'value' as float) as current_l3,
+         cast(values->'PowerL1'->>'value' as float) as power_l1,
+         cast(values->'PowerL2'->>'value' as float) as power_l2,
+         cast(values->'PowerL3'->>'value' as float) as power_l3,
+         cast(values->'TotalImportEnergy'->>'value' as float) as total_import_energy,
+         cast(values->'TotalExportEnergy'->>'value' as float) as total_export_energy,
+         values->'Status'->>'value' as status,
+         device
+    from measurements
+    where application = 'Car';

queries/old-daily-temperature-query.sql

@@ -0,0 +1,11 @@
+select
+        extract('day' from time)::varchar || '.' || extract('month' from time)::varchar || '.' || extract('year' from time)::varchar as day,
+        avg(temperature)::numeric(10,0) as temperature
+    from room_climate_measurement_t
+    where
+        category = 'Outdoor' and
+        location = 'Outdoor' and
+        extract('hour' from time) = 12 and
+        time::date = now()::date
+    group by day
+

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"}

queries/pg.sql

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

queries/saerbeck.sql

@@ -0,0 +1,44 @@
+create or replace view badesee_temperature_v as
+  select time, 
+         cast(values->'Temp_Red'->>'value' as float) as Temp1,
+         cast(values->'Temp_White'->>'value' as float) as Temp2,
+         cast(values->'Temp_Black'->>'value' as float) as Temp3,
+         cast(values->'Battery'->>'value' as float) as battery,
+         attributes->>'Status' as status,
+         device
+    from measurements
+    where application = 'de-hottis-saerbeck-monitoring' and
+          device = 'eui-a84041318187ec13';
+
+create or replace view cubecell_threeway_temperature2_v as
+  select time,
+         cast(values->'Temperature2'->>'value' as float) as value,
+         values->'Temperature2'->>'label' as label
+    from measurements
+    where application = 'de-hottis-saerbeck-monitoring' and
+    device = 'eui-70b3d57ed0068fa4';
+
+create or replace view cubecell_threeway_temperature1_v as
+  select time,
+         cast(values->'Temperature1'->>'value' as float) as value,
+         values->'Temperature1'->>'label' as label
+    from measurements
+    where application = 'de-hottis-saerbeck-monitoring' and
+    device = 'eui-70b3d57ed0068fa4';
+
+create or replace view cubecell_threeway_temperature3_v as
+  select time,
+         cast(values->'Temperature3'->>'value' as float) as value,
+         values->'Temperature3'->>'label' as label
+    from measurements
+    where application = 'de-hottis-saerbeck-monitoring' and
+    device = 'eui-70b3d57ed0068fa4';
+
+create or replace view cubecell_threeway_battery_v as
+  select time,
+         cast(values->'Battery'->>'value' as float) as value,
+         values->'Battery'->>'label' as label
+    from measurements
+    where application = 'de-hottis-saerbeck-monitoring' and
+    device = 'eui-70b3d57ed0068fa4';
+

src/udi/ENVDB.udiload

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

src/udi/ENVDB.uditest

@@ -1,6 +1,6 @@
-PGUSER="udi"
+PGUSER="udi-hottis"
 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 default-udi-db-cred -n udi -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-2.json

@@ -4,6 +4,54 @@
     "tlsEnable": "false"
   },
   "topicMappings": [
+    {
+      "topics": [ "snmp" ],
+      "handler": "PREP",
+      "id": "SNMP",
+      "config": {
+        "attributes": {
+        }
+      }
+    },
+    {
+      "topics": [ "tsm" ],
+      "handler": "PREP",
+      "id": "TSM",
+      "config": {
+        "attributes": {
+        }
+      }
+    },
+    {
+      "topics": [ "dt1/ai/periodic/1" ],
+      "handler": "DT1T",
+      "id": "DT1T.0",
+      "config": {
+        "attributes": {
+          "Application": "Temperature Wago",
+          "Device": "Freezer",
+          "HardLow": "-273",
+          "SoftLow": "-50",
+          "SoftHigh": "20",
+          "HardHigh": "100"
+        }
+      }
+    },
+    {
+      "topics": [ "dt1/ai/periodic/3" ],
+      "handler": "DT1T",
+      "id": "DT1T.1",
+      "config": {
+        "attributes": {
+          "Application": "Temperature Wago",
+          "Device": "Outdoor",
+          "HardLow": "-273",
+          "SoftLow": "-60",
+          "SoftHigh": "60",
+          "HardHigh": "100"
+        }
+      }
+    },
     {
       "topics": [ "IoT/PV/Values" ],
       "handler": "PV",
@@ -14,6 +62,16 @@
         }
       }
     },
+    {
+      "topics": [ "IoT/Car/Values" ],
+      "handler": "Car",
+      "id": "Car",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+        }
+      }
+    },
     {
       "topics": [ "IoT/MBGW3/Measurement" ],
       "handler": "MBGW3",
@@ -43,30 +101,12 @@
       }
     },
     {
-      "topics": [ "NR/Multisensor/+/Temperatur" ],
+      "topics": [ "zigbee2mqtt/+" ],
-      "handler": "SVEJ",
+      "handler": "Z2M",
-      "id": "SVEJ0",
+      "id": "Z2M",
       "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:%"
         }
       }
     },

src/udi/config-test-ttn.json

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

src/udi/config-test.json

@@ -0,0 +1,31 @@
+{
+  "mqtt": {
+    "broker": "mqtt://172.23.1.102:1883",
+    "tlsEnable": "false"
+  },
+  "topicMappings": [
+    {
+      "topics": [ "IoT/PV/Values" ],
+      "handler": "PV",
+      "id": "PV",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+        }
+      }
+    },
+    {
+      "topics": [ "IoT/Car/Values" ],
+      "handler": "Car",
+      "id": "Car",
+      "config": {
+        "databaseConnStr": "",
+        "attributes": {
+        }
+      }
+    }
+  ],
+  "archiver": {
+    "dir": "./tmp/udi"
+  }
+}

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

@@ -1,46 +1,38 @@
 package database
 
 import "time"
-import "gorm.io/gorm"
-
 
 type VariableType struct {
-  Label string `json:"label"`
+	Label    string      `json:"label"`
-  Variable string `json:"variable"`
+	Variable string      `json:"variable"`
-  Unit string `json:"unit"`
+	Unit     string      `json:"unit"`
-  Value interface{} `json:"value,omitempty"`
+	Value    interface{} `json:"value,omitempty"`
+	Status   string      `json:"status,omitempty"`
 }
 
 type Measurement struct {
-  Time time.Time `gorm:"not null;primary_key"`
+	Time        time.Time
-  Application string `gorm:"not null"`
+	Application string
-  Device string
+	Device      string
-  Attributes map[string]interface{} `gorm:"serializer:json;type:jsonb"`
+	Attributes  map[string]interface{}
-  Values map[string]VariableType `gorm:"serializer:json;type:jsonb"`
+	Values      map[string]VariableType
+}
+
+// Simplified structures for backward compatibility
+type DeviceType struct {
+	Label           string
+	ModelIdentifier string
+	Attributes      map[string]interface{}
 }
 
 type Application struct {
-  gorm.Model
+	Label      string
-  Label string `gorm:"not null"`
+	Attributes map[string]interface{}
-  Attributes map[string]interface{} `gorm:"serializer:json;type:jsonb"`
-}
-
-
-type DeviceType struct {
-  gorm.Model
-  Label string `gorm:"not null"`
-  ModelIdentifier string
-  Attributes map[string]interface{} `gorm:"serializer:json;type:jsonb"`
 }
 
 type Device struct {
-  gorm.Model
+	Label       string
-  Label string `gorm:"not null;uniqueIndex:idx_label_application_id"`
+	Application Application
-  ApplicationID int `gorm:"not null;uniqueIndex:idx_label_application_id"`
+	DeviceType  DeviceType
-  Application Application
+	Attributes  map[string]interface{}
-  DeviceTypeID int `gorm:"not null"`
-  DeviceType DeviceType
-  Attributes map[string]interface{} `gorm:"serializer:json;type:jsonb"`
 }
-
-

src/udi/database/database.go

@@ -1,70 +1,155 @@
 package database
 
-
 import (
-  "log"
+	"fmt"
-  //"time"
+	"log"
-  "fmt"
+	"os"
-  "gorm.io/driver/postgres"
+	"udi/counter"
-  "gorm.io/gorm"
+
+	influxdb "github.com/influxdata/influxdb1-client/v2"
 )
 
 type DatabaseHandle struct {
-  initialized bool
+	initialized bool
-  dbh *gorm.DB 
+	client      influxdb.Client
+	database    string
 }
 
 func NewDatabaseHandle() *DatabaseHandle {
-  var db DatabaseHandle
+	var db DatabaseHandle
-  // inject the whole database configuration via the well-known PG* env variables
+
-  conn, err := gorm.Open(postgres.Open(""))
+	// Read configuration from environment variables
-  if err != nil {
+	influxURL := os.Getenv("INFLUXDB_URL")
-    log.Printf("Unable to open database connection: %s", err)
+	if influxURL == "" {
-    db.initialized = false
+		influxURL = "http://localhost:8086"
-  } else {
+	}
-    db.dbh = conn
+
-    db.initialized = true
+	influxDB := os.Getenv("INFLUXDB_DATABASE")
-    log.Println("Database connection opened")
+	if influxDB == "" {
-  }
+		influxDB = "udi"
-  return &db
+	}
+
+	username := os.Getenv("INFLUXDB_USER")
+	password := os.Getenv("INFLUXDB_PASSWORD")
+
+	// Create InfluxDB client
+	client, err := influxdb.NewHTTPClient(influxdb.HTTPConfig{
+		Addr:     influxURL,
+		Username: username,
+		Password: password,
+	})
+
+	if err != nil {
+		log.Printf("Unable to create InfluxDB client (config: URL: %s, Username: %s, Password: %s): %s", influxDB, username, password, err)
+		db.initialized = false
+		return &db
+	}
+
+	// Test connection
+	_, _, err = client.Ping(0)
+	if err != nil {
+		log.Printf("Unable to ping InfluxDB: %s", err)
+		db.initialized = false
+		client.Close()
+		return &db
+	}
+
+	db.client = client
+	db.database = influxDB
+	db.initialized = true
+	log.Printf("InfluxDB connection opened (URL: %s, Database: %s)", influxURL, influxDB)
+
+	return &db
 }
 
 func (self *DatabaseHandle) StoreMeasurement(measurement *Measurement) {
-  if ! self.initialized {
+	if !self.initialized {
-    log.Printf("Database connection not initialized, can not store, measurement %s lost", measurement)
+		log.Printf("Database connection not initialized, can not store, measurement %v lost", measurement)
-    return
+		counter.F("Stored")
-  }
+		return
+	}
 
-  result := self.dbh.Create(measurement)
+	// Create batch points
-  if result.Error != nil {
+	bp, err := influxdb.NewBatchPoints(influxdb.BatchPointsConfig{
-    log.Printf("Unable to insert, measurement %s lost, error: %s", measurement, result.Error)
+		Database:  self.database,
-    return
+		Precision: "s",
-  }
+	})
+	if err != nil {
+		log.Printf("Unable to create batch points: %s", err)
+		counter.F("Stored")
+		return
+	}
 
-  log.Println("Successfully stored measurement")
+	// Build tags
+	tags := make(map[string]string)
+	if measurement.Device != "" {
+		tags["device"] = measurement.Device
+	}
+
+	// Build fields from Values
+	fields := make(map[string]interface{})
+	for key, varType := range measurement.Values {
+		// Store the value with the variable name as field key
+		fields[key] = varType.Value
+
+		// Optionally store metadata as separate fields
+		if varType.Unit != "" {
+			fields[key+"_unit"] = varType.Unit
+		}
+		// This is already the column name, so we can skip it
+		//if varType.Variable != "" {
+		//	fields[key+"_variable"] = varType.Variable
+		//}
+		if varType.Status != "" {
+			fields[key+"_status"] = varType.Status
+		}
+	}
+	// Add attributes as fields
+	for key, value := range measurement.Attributes {
+		if strValue, ok := value.(string); ok {
+			fields[key] = strValue
+		} else {
+			fields[key] = fmt.Sprintf("%v", value)
+		}
+	}
+
+	// Ensure we have at least one field
+	if len(fields) == 0 {
+		log.Printf("No fields to store in measurement, skipping")
+		counter.F("Stored")
+		return
+	}
+
+	// Create point
+	pt, err := influxdb.NewPoint(
+		measurement.Application,
+		tags,
+		fields,
+		measurement.Time,
+	)
+	if err != nil {
+		log.Printf("Unable to create point: %s", err)
+		counter.F("Stored")
+		return
+	}
+
+	bp.AddPoint(pt)
+
+	// Write batch
+	err = self.client.Write(bp)
+	if err != nil {
+		log.Printf("Unable to write to InfluxDB, measurement lost, error: %s", err)
+		counter.F("Stored")
+		return
+	}
+
+	log.Println("Successfully stored measurement")
+	counter.S("Stored")
 }
 
-func (self *DatabaseHandle) GetDeviceByLabelAndApplication(applicationLabel string, deviceLabel string) (*Device, error) {
+func (self *DatabaseHandle) Close() {
-  if ! self.initialized {
+	if self.initialized && self.client != nil {
-    err := fmt.Errorf("Database connection not initialized")
+		self.client.Close()
-    return nil, err
+		log.Println("InfluxDB connection closed")
-  }
+	}
-
-  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/database/migrate_database.go

@@ -1,55 +0,0 @@
-package database
-
-import (
-  "log"
-  //"time"
-  "gorm.io/driver/postgres"
-  "gorm.io/gorm"
-)
-
-func Migrate() {
-  dsn := ""
-  db, err := gorm.Open(postgres.Open(dsn))
-  if err != nil {
-    log.Fatalf("Unable to open database connection: %s", err)
-  }
-
-  db.AutoMigrate(&Application{})
-  log.Println("Application created")
-
-  db.AutoMigrate(&DeviceType{})
-  log.Println("DeviceType created")
-
-  db.AutoMigrate(&Device{})
-  log.Println("Device created")
-
-  db.AutoMigrate(&Measurement{})
-  log.Println("Measurement created")
-
-  log.Println("Remember to call create_hypertable on measurements, sowhat I can't do that for you.")
-
-  /*
-  m := Measurement { 
-    Time: time.Now(), 
-    Application: "app", 
-    Attributes: nil,
-    Values: []SensorType {
-      { Variable: "Temperature", Unit: "Degree Celsius", Value: 1.0 },
-      { Variable: "Temperature", Unit: "Degree Celsius", Value: 3.0 },
-    },
-  }
-  db.Create(&m)
-
-  m = Measurement { 
-    Time: time.Now(), 
-    Application: "app", 
-    Attributes: nil,
-    Values: []SensorType {
-      { Variable: "Temperature", Unit: "Degree Celsius", Value: 10.0 },
-      { Variable: "Temperature", Unit: "Degree Celsius", Value: 30.0 },
-    },
-  }
-  db.Create(&m)
-  */
-}
-

src/udi/database/schema.sql

@@ -1,41 +0,0 @@
-create extension if not exists timescaledb;
-
-create table application_t (
-  id serial not null primary key,
-  label text not null unique,
-  attributes jsonb
-);
-
-create table sensor_type_t (
-  id serial not null primary key,
-  label text not null unique,
-  variable text not null,
-  unit text not null,
-  converter text not null,
-  attributes jsonb
-);
-
-create table sensor_t (
-  id serial not null primary key,
-  label text not null,
-  application int references application_t(id),
-  sensor_type int references sensor_type_t(id),
-  attributes jsonb,
-  unique (label, application)
-);
-
-create table measurement_t (
-  time timestamp without time zone not null,
-  application text not null,
-  sensor_type text not null,
-  sensor text not null,
-  variable text not null,
-  unit text not null,
-  value double precision not null,
-  attributes jsonb
-);
-
-select create_hypertable('measurement_t', 'time');
-
-
-

src/udi/dispatcher/dispatcher.go

@@ -1,129 +1,154 @@
 package dispatcher
 
-import "log"
+import (
-import "time"
+	"fmt"
-import "os"
+	"log"
-import "fmt"
+	"net/url"
-import "net/url"
+	"os"
-import "udi/mqtt"
+	"time"
-import "udi/config"
+	"udi/config"
-import "udi/handlers/handler"
+	"udi/counter"
-import "udi/handlers/ttn"
+	"udi/handlers/car"
-import "udi/handlers/iot"
+	"udi/handlers/dt1t"
-import "udi/handlers/pv"
+	"udi/handlers/handler"
-import "udi/handlers/mbgw3"
+	"udi/handlers/iot"
-import "udi/handlers/sver"
+	"udi/handlers/locative"
-import "udi/handlers/svej"
+	"udi/handlers/mbgw3"
+	"udi/handlers/prepared"
+	"udi/handlers/pv"
+	"udi/handlers/svej"
+	"udi/handlers/sver"
 
+	// "udi/handlers/ttn"
+	"udi/handlers/z2m"
+	"udi/mqtt"
+)
 
 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()
+	go archiver()
 
-  for _, mapping := range config.Config.TopicMappings {
+	for _, mapping := range config.Config.TopicMappings {
-    // log.Printf("Trying to initialize %s", mapping)
+		// log.Printf("Trying to initialize %s", mapping)
 
-    var factory interface{}
+		var factory interface{}
-    switch mapping.Handler {
+		switch mapping.Handler {
-    case "TTN":
+		// case "TTN":
-      factory = ttn.NewTTNHandler
+		// 	factory = ttn.New
-    case "IoT":
+		case "IoT":
-      factory = iot.NewIoTHandler
+			factory = iot.New
-    case "PV":
+		case "PV":
-      factory = pv.NewPvHandler
+			factory = pv.New
-    case "MBGW3":
+		case "MBGW3":
-      factory = mbgw3.NewMbgw3Handler
+			factory = mbgw3.New
-    case "SVER":
+		case "SVER":
-      factory = sver.NewSverHandler
+			factory = sver.New
-    case "SVEJ":
+		case "SVEJ":
-      factory = svej.NewSvejHandler
+			factory = svej.New
-    default:
+		case "DT1T":
-      factory = nil
+			factory = dt1t.New
-      log.Printf("No handler %s found, ignore mapping", mapping.Handler)
+		case "Locative":
-    }
+			factory = locative.New
+		case "PREP":
+			factory = prepared.New
+		case "Z2M":
+			factory = z2m.New
+		case "Car":
+			factory = car.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 {
+		if !ok {
-      log.Println("Typ Assertion failed")
+			log.Println("Typ Assertion failed")
-      break
+			break
-    }
+		}
-    handler := fn(mapping.Config)
+		handler := fn(mapping.Id, mapping.Config)
-    handlerMap[mapping.Id] = handler
+		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)
+	file, err := os.OpenFile(filename, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
-  if err != nil {
+	if err != nil {
-    log.Printf("Unable to open archiving file %s, message is not archived: %s", filename, err)
+		log.Printf("Unable to open archiving file %s, message is not archived: %s", filename, err)
-    return
+		counter.F("Archived")
-  }
+		return
-  defer file.Close()
+	}
-  archivingString := fmt.Sprintf("%s - %s - %s\n", item.Timestamp.Format("2006-01-02 15:04:05"), item.Topic, item.Payload)
+	defer file.Close()
-  _, err = file.WriteString(string(archivingString) + "\n")
+	archivingString := fmt.Sprintf("%s - %s - %s\n", item.Timestamp.Format("2006-01-02 15:04:05"), item.Topic, item.Payload)
-  if err != nil {
+	_, err = file.WriteString(string(archivingString) + "\n")
-    log.Printf("Unable to write message, message is not archived: %s", err)
+	if err != nil {
-    return
+		log.Printf("Unable to write message, message is not archived: %s", err)
-  }
+		counter.F("Archived")
-  log.Println("Successfully archived message")
+		return
+	}
+	//log.Println("Successfully archived message")
+	counter.S("Archived")
 }
 
 func archiver() {
-  archivingRootDir := config.Config.Archiver.Dir
+	archivingRootDir := config.Config.Archiver.Dir
-  var lastArchivingDir string
+	var lastArchivingDir string
 
-  for {
+	for {
-    select {
+		select {
-    case message := <- archiverChannel:
+		case message := <-archiverChannel:
-      currentDateStr := message.Timestamp.Format("2006/01/02/15")
+			currentDateStr := message.Timestamp.Format("2006/01/02/15")
-      currentArchivingDir := archivingRootDir + "/" + currentDateStr
+			currentArchivingDir := archivingRootDir + "/" + currentDateStr
-      if currentArchivingDir != lastArchivingDir {
+			if currentArchivingDir != lastArchivingDir {
-        err := os.MkdirAll(currentArchivingDir, 0755)
+				err := os.MkdirAll(currentArchivingDir, 0755)
-        if err != nil {
+				if err != nil {
-          log.Printf("Unable to create archiving dir %s: %s", currentArchivingDir, err)
+					log.Printf("Unable to create archiving dir %s: %s", currentArchivingDir, err)
-        }
+					counter.F("Archived")
-        lastArchivingDir = currentArchivingDir
+				}
-        log.Printf("Archiving dir %s created", currentArchivingDir)
+				lastArchivingDir = currentArchivingDir
-      }
+				//log.Printf("Archiving dir %s created", currentArchivingDir)
-      archivingFilename := fmt.Sprintf("%s/%s", currentArchivingDir, url.PathEscape(message.Topic))
+			}
-      storeMessage(archivingFilename, message)
+			archivingFilename := fmt.Sprintf("%s/%s", currentArchivingDir, url.PathEscape(message.Topic))
-    }
+			storeMessage(archivingFilename, message)
-  }
+		}
+	}
 }
 
 func InputDispatcher() {
-  for {
+	for {
-    select {
+		select {
-    case mqttMessage := <- mqtt.InputChannel:
+		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) }
+			message := handler.MessageT{time.Now(), mqttMessage.Topic, string(mqttMessage.Payload)}
-      archiverChannel <- message
+			archiverChannel <- message
-      handleMessage(message)
+			handleMessage(message)
-    }
+		}
-  }
+	}
 }
 
 func handleMessage(message handler.MessageT) {
-  for _, mapping := range config.Config.TopicMappings {
+	for _, mapping := range config.Config.TopicMappings {
-    // log.Printf("Testing %s -> %s", mapping.Topics, mapping.Handler)
+		// log.Printf("Testing %s -> %s", mapping.Topics, mapping.Handler)
-    for _, subscribedTopic := range mapping.Topics {
+		for _, subscribedTopic := range mapping.Topics {
-      // log.Printf("Testing %s in %s", message.Topic, subscribedTopic)
+			// log.Printf("Testing %s in %s", message.Topic, subscribedTopic)
-      if mqtt.TopicMatchesSubscription(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]
+				handler, exists := handlerMap[mapping.Id]
-        if exists {
+				if exists {
-          handler.Handle(message)
+					handler.Handle(message)
-          return
+					counter.S("Dispatched")
-        } else {
+					return
-          log.Printf("Handler %s not found, message %s is lost", mapping.Id, message)
+				} 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)
+		}
+	}
+	log.Printf("No match for topic %s, message %s is lost", message.Topic, message)
+	counter.F("Dispatched")
 }

src/udi/go.mod

@@ -1,24 +1,16 @@
 module udi
 
-go 1.21.3
+go 1.22.3
 
 require (
-	github.com/eclipse/paho.mqtt.golang v1.4.3
+	github.com/eclipse/paho.mqtt.golang v1.5.0
-	github.com/google/uuid v1.4.0
+	github.com/google/uuid v1.6.0
+	github.com/influxdata/influxdb1-client v0.0.0-20220302092344-a9ab5670611c
 	github.com/oliveagle/jsonpath v0.0.0-20180606110733-2e52cf6e6852
-	gorm.io/driver/postgres v1.5.4
-	gorm.io/gorm v1.25.5
 )
 
 require (
-	github.com/gorilla/websocket v1.5.0 // indirect
+	github.com/gorilla/websocket v1.5.3 // indirect
-	github.com/jackc/pgpassfile v1.0.0 // indirect
+	golang.org/x/net v0.34.0 // indirect
-	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
+	golang.org/x/sync v0.10.0 // indirect
-	github.com/jackc/pgx/v5 v5.4.3 // indirect
-	github.com/jinzhu/inflection v1.0.0 // indirect
-	github.com/jinzhu/now v1.1.5 // indirect
-	golang.org/x/crypto v0.14.0 // indirect
-	golang.org/x/net v0.10.0 // indirect
-	golang.org/x/sync v0.1.0 // indirect
-	golang.org/x/text v0.13.0 // indirect
 )

src/udi/go.sum

@@ -1,44 +1,14 @@
-github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/eclipse/paho.mqtt.golang v1.5.0 h1:EH+bUVJNgttidWFkLLVKaQPGmkTUfQQqjOsyvMGvD6o=
-github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/eclipse/paho.mqtt.golang v1.5.0/go.mod h1:du/2qNQVqJf/Sqs4MEL77kR8QTqANF7XU7Fk0aOTAgk=
-github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
-github.com/eclipse/paho.mqtt.golang v1.4.3 h1:2kwcUGn8seMUfWndX0hGbvH8r7crgcJguQNCyp70xik=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/eclipse/paho.mqtt.golang v1.4.3/go.mod h1:CSYvoAlsMkhYOXh/oKyxa8EcBci6dVkLCbo5tTC1RIE=
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
-github.com/google/uuid v1.4.0 h1:MtMxsa51/r9yyhkyLsVeVt0B+BGQZzpQiTQ4eHZ8bc4=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
-github.com/google/uuid v1.4.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/influxdata/influxdb1-client v0.0.0-20220302092344-a9ab5670611c h1:qSHzRbhzK8RdXOsAdfDgO49TtqC1oZ+acxPrkfTxcCs=
-github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
+github.com/influxdata/influxdb1-client v0.0.0-20220302092344-a9ab5670611c/go.mod h1:qj24IKcXYK6Iy9ceXlo3Tc+vtHo9lIhSX5JddghvEPo=
-github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
-github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
-github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
-github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk=
-github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
-github.com/jackc/pgx/v5 v5.4.3 h1:cxFyXhxlvAifxnkKKdlxv8XqUf59tDlYjnV5YYfsJJY=
-github.com/jackc/pgx/v5 v5.4.3/go.mod h1:Ig06C2Vu0t5qXC60W8sqIthScaEnFvojjj9dSljmHRA=
-github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E=
-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=
+golang.org/x/net v0.34.0 h1:Mb7Mrk043xzHgnRM88suvJFwzVrRfHEHJEl5/71CKw0=
-github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+golang.org/x/net v0.34.0/go.mod h1:di0qlW3YNM5oh6GqDGQr92MyTozJPmybPK4Ev/Gm31k=
-github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
-github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
-github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
-github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
-golang.org/x/crypto v0.14.0 h1:wBqGXzWJW6m1XrIKlAH0Hs1JJ7+9KBwnIO8v66Q9cHc=
-golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4=
-golang.org/x/net v0.10.0 h1:X2//UzNDwYmtCLn7To6G58Wr6f5ahEAQgKNzv9Y951M=
-golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
-golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
-golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/text v0.13.0 h1:ablQoSUd0tRdKxZewP80B+BaqeKJuVhuRxj/dkrun3k=
-golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
-gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-gorm.io/driver/postgres v1.5.4 h1:Iyrp9Meh3GmbSuyIAGyjkN+n9K+GHX9b9MqsTL4EJCo=
-gorm.io/driver/postgres v1.5.4/go.mod h1:Bgo89+h0CRcdA33Y6frlaHHVuTdOf87pmyzwW9C/BH0=
-gorm.io/gorm v1.25.5 h1:zR9lOiiYf09VNh5Q1gphfyia1JpiClIWG9hQaxB/mls=
-gorm.io/gorm v1.25.5/go.mod h1:hbnx/Oo0ChWMn1BIhpy1oYozzpM15i4YPuHDmfYtwg8=

src/udi/handlers/car/car.go

@@ -0,0 +1,94 @@
+package car
+
+import (
+	"encoding/json"
+	"log"
+	"reflect"
+	"time"
+	"udi/config"
+	"udi/database"
+	"udi/handlers/handler"
+)
+
+type CarHandler struct {
+	handler.CommonHandler
+	dbh *database.DatabaseHandle
+}
+
+/*
+{
+  "status": "Ok",
+  "timestamp": "2025-12-15T13:11:15.648243",
+  "voltageL1": 228.68,
+  "voltageL2": 227.69,
+  "voltageL3": 228.53,
+  "currentL1": 0.0,
+  "currentL2": 0.0,
+  "currentL3": 0.0,
+  "powerL1": 0.0,
+  "powerL2": 0.0,
+  "powerL3": 0.0,
+  "totalImportEnergy": 0.0,
+  "totalExportEnergy": 0.0,
+  "cnt": 399300}
+*/
+
+type CarValue struct {
+  Status            string  `unit:""   json:"status"`
+  Timestamp         string  `unit:""   json:"timestamp"`
+  VoltageL1         float32 `unit:"V"  json:"voltageL1"`
+  VoltageL2         float32 `unit:"V"  json:"voltageL2"`
+  VoltageL3         float32 `unit:"V"  json:"voltageL3"`
+  CurrentL1         float32 `unit:"A"  json:"currentL1"`
+  CurrentL2         float32 `unit:"A"  json:"currentL2"`
+  CurrentL3         float32 `unit:"A"  json:"currentL3"`
+  PowerL1           float32 `unit:"W"  json:"powerL1"`
+  PowerL2           float32 `unit:"W"  json:"powerL2"`
+  PowerL3           float32 `unit:"W"  json:"powerL3"`
+  TotalImportEnergy float32 `unit:"Wh" json:"totalImportEnergy"`
+  TotalExportEnergy float32 `unit:"Wh" json:"totalExportEnergy"`
+  Cnt               int     `unit:""   json:"cnt"`
+}
+
+func New(id string, config config.HandlerConfigT) handler.Handler {
+	t := &CarHandler{}
+	t.Id = id
+	t.dbh = database.NewDatabaseHandle()
+	log.Printf("Handler Car %d initialized", id)
+	return t
+}
+
+func (self *CarHandler) Handle(message handler.MessageT) {
+	//log.Printf("Handler Car %d processing %s -> %s", self.id, message.Topic, message.Payload)
+
+	var carValue CarValue
+	err := json.Unmarshal([]byte(message.Payload), &carValue)
+	if err != nil {
+		self.Lost("Unable to parse payload into carValue struct", err, message)
+		return
+	}
+
+	variables := make(map[string]database.VariableType)
+	carValueStructValue := reflect.ValueOf(carValue)
+	for i := 0; i < carValueStructValue.NumField(); i++ {
+		field := carValueStructValue.Type().Field(i)
+		fieldValue := carValueStructValue.Field(i)
+		v := database.VariableType{
+			Label:    "",
+			Variable: field.Name,
+			Unit:     field.Tag.Get("unit"),
+			Value:    fieldValue.Interface(),
+		}
+		variables[field.Name] = v
+	}
+
+	measurement := database.Measurement{
+		Time:        time.Now(),
+		Application: "Car",
+		Device:      "Powermeter",
+		Values:      variables,
+	}
+
+	self.dbh.StoreMeasurement(&measurement)
+	self.S()
+}

src/udi/handlers/dt1t/dt1t.go

@@ -0,0 +1,81 @@
+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()
+  log.Printf("Handler DT1T %d initialized", id)
+  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/locative/locative.go

@@ -0,0 +1,74 @@
+package locative
+
+import (
+  "reflect"
+  "time"
+  "log"
+  "encoding/json"
+  "udi/config"
+  "udi/handlers/handler"
+  "udi/database"
+)
+
+
+type LocativeHandler struct {
+  handler.CommonHandler
+  dbh *database.DatabaseHandle
+}
+
+type locativeEvent struct {
+  Trigger string `json:"trigger"`
+  Device string `json:"device"`
+  Location string `json:"location"`
+  Latitude string `json:"latitude"`
+  Longitude string `json:"longitude"`
+  Person string `json:"person"`
+  Timestamp string `json:"timestamp"`
+}
+
+func New(id string, config config.HandlerConfigT) handler.Handler {
+  t := &LocativeHandler {
+  }
+  t.Id = id
+  t.dbh = database.NewDatabaseHandle()
+  log.Printf("Handler Locative %d initialized", id)
+  return t
+}
+
+
+func (self *LocativeHandler) Handle(message handler.MessageT) {
+  log.Printf("Handler Locative %d processing %s -> %s", self.Id, message.Topic, message.Payload)
+
+  var locativeEvent locativeEvent
+  err := json.Unmarshal([]byte(message.Payload), &locativeEvent)
+  if err != nil {
+    self.Lost("Unable to parse payload into locativeEvent struct", err, message)
+    return
+  }
+
+  variables := make(map[string]database.VariableType)
+  locativeEventStructValue := reflect.ValueOf(locativeEvent)
+  for i := 0; i < locativeEventStructValue.NumField(); i++ {
+    field := locativeEventStructValue.Type().Field(i)
+    fieldValue := locativeEventStructValue.Field(i)
+    v := database.VariableType {
+      Label: "",
+      Variable: field.Name,
+      Unit: "",
+      Value: fieldValue.Interface(),
+    }
+    variables[field.Name] = v
+  }
+
+  measurement := database.Measurement {
+    Time: time.Now(),
+    Application: "Locative",
+    Device: locativeEvent.Person,
+    Values: variables,
+  }
+
+  self.dbh.StoreMeasurement(&measurement)
+  self.S()
+}
+
+

src/udi/handlers/mbgw3/mbgw3.go

@@ -1,11 +1,9 @@
 package mbgw3
 
 import (
-  "log"
-  //"reflect"
   "time"
+  "log"
   "strconv"
-  "fmt"
   "encoding/json"
   "udi/config"
   "udi/handlers/handler"
@@ -13,10 +11,8 @@ import (
 )
 
 
-var idSeq int = 0
-
 type Mbgw3Handler struct {
-  id int
+  handler.CommonHandler
   dbh *database.DatabaseHandle
 }
 
@@ -31,26 +27,22 @@ 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()
+  log.Printf("Handler MBGW3 %d initialized", id)
   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
   }
 
@@ -82,12 +74,22 @@ func (self *Mbgw3Handler) Handle(message handler.MessageT) {
 
   measurement.Values = make(map[string]database.VariableType)
   unitMap := map[string]string { "Energy": "Wh", "Power": "W", "Voltage": "V", "Current": "A", "Volume": "m3" }
+  keyCount := make(map[string]int)
+  
   for k, v := range observation.Values {
     unit, exists := unitMap[k]
     if ! exists {
       unit = "Unmapped Unit"
     }
-    measurement.Values[k] = database.VariableType {
+    
+    // Check if key already exists and create unique key if needed
+    keyCount[k]++
+    uniqueKey := k
+    if keyCount[k] > 1 {
+      uniqueKey = k + strconv.Itoa(keyCount[k])
+    }
+    
+    measurement.Values[uniqueKey] = database.VariableType {
       Label: "",
       Variable: k,
       Unit: unit,
@@ -95,9 +97,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/prepared/prepared.go

@@ -0,0 +1,79 @@
+package prepared
+
+import (
+  "time"
+  "log"
+  "encoding/json"
+  "udi/config"
+  "udi/handlers/handler"
+  "udi/database"
+)
+
+
+type PreparedHandler struct {
+  handler.CommonHandler
+  dbh *database.DatabaseHandle
+}
+
+type endpoint_t struct {
+  Label string `json:"label"`
+  Variable string `json:"variable"`
+  Value string `json:"value"`
+  Unit string `json:"unit"`
+  Status string `json:"status"`
+}
+
+type observation_t struct {
+  Device string `json:"device"`
+  Label string `json:"label"`
+  Variables map[string]endpoint_t `json:"variables"`
+}
+
+
+func New(id string, config config.HandlerConfigT) handler.Handler {
+  t := &PreparedHandler {
+  }
+  t.Id = id
+  t.dbh = database.NewDatabaseHandle()
+  log.Printf("Handler Prepared %d initialized", id)
+  return t
+}
+
+func (self *PreparedHandler) Handle(message handler.MessageT) {
+  log.Printf("Handler Prepared %d processing %s -> %s", self.Id, message.Topic, message.Payload)
+
+  var observation observation_t
+  err := json.Unmarshal([]byte(message.Payload), &observation)
+  if err != nil {
+    self.Lost("Unable to parse payload into Observation struct", err, message)
+    return
+  }
+
+  var measurement database.Measurement
+  measurement.Time = time.Now()
+
+  measurement.Application = self.Id
+  measurement.Device = observation.Device
+
+  measurement.Attributes = map[string]interface{} {
+    "Label": observation.Label,
+  }
+
+  measurement.Values = make(map[string]database.VariableType)
+  for k, v := range observation.Variables {
+    measurement.Values[k] = database.VariableType {
+      Label: v.Label,
+      Variable: v.Variable,
+      Unit: v.Unit,
+      Value: v.Value,
+      Status: v.Status,
+    }
+  }
+
+  log.Printf("Prepared measurement item: %s", measurement)
+
+  self.dbh.StoreMeasurement(&measurement)
+  self.S()
+}
+
+

src/udi/handlers/pv/pv.go

@@ -1,10 +1,9 @@
 package pv
 
 import (
-  "log"
   "reflect"
   "time"
-  "fmt"
+  "log"
   "encoding/json"
   "udi/config"
   "udi/handlers/handler"
@@ -12,10 +11,8 @@ import (
 )
 
 
-var idSeq int = 0
-
 type PvHandler struct {
-  id int
+  handler.CommonHandler
   dbh *database.DatabaseHandle
 }
 
@@ -40,18 +37,15 @@ 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()
+  log.Printf("Handler PV %d initialized", id)
   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 +53,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 +79,7 @@ func (self *PvHandler) Handle(message handler.MessageT) {
   }
 
   self.dbh.StoreMeasurement(&measurement)
+  self.S()
 }
 
 

src/udi/handlers/svej/svej.go

@@ -1,31 +1,30 @@
 package svej
 
 import (
-  "log"
+	"encoding/json"
-  "time"
+	"fmt"
-  "strconv"
+	"log"
-  "strings"
+	"strconv"
-  "fmt"
+	"strings"
-  "github.com/oliveagle/jsonpath"
+	"time"
-  "encoding/json"
+	"udi/config"
-  "udi/config"
+	"udi/database"
-  "udi/handlers/handler"
+	"udi/handlers/handler"
-  "udi/database"
+
+	"github.com/oliveagle/jsonpath"
 )
 
-var idSeq int = 0
-
 type SingleValueExtractorJsonpathHandler struct {
-  id int
+	handler.CommonHandler
-  ready bool
+	ready          bool
-  application string
+	application    string
-  deviceSelector string
+	deviceSelector string
-  valueSelector string
+	valueSelector  string
-  unitSelector string
+	unitSelector   string
-  deviceJsonpath *jsonpath.Compiled
+	deviceJsonpath *jsonpath.Compiled
-  valueJsonpath *jsonpath.Compiled
+	valueJsonpath  *jsonpath.Compiled
-  unitJsonpath *jsonpath.Compiled
+	unitJsonpath   *jsonpath.Compiled
-  dbh *database.DatabaseHandle
+	dbh            *database.DatabaseHandle
 }
 
 /*
@@ -35,135 +34,131 @@ T:TopicPartIndex
 C:ConstantValue
 */
 
+func New(id string, config config.HandlerConfigT) handler.Handler {
+	t := &SingleValueExtractorJsonpathHandler{
+		ready: false,
+	}
 
-func NewSvejHandler(config config.HandlerConfigT) handler.Handler {
+	if config.Attributes["application"] == "" {
-  t := &SingleValueExtractorJsonpathHandler {
+		log.Println("Error: application not configured")
-    id: idSeq,
+		return t
-    ready: false,
+	}
-  }
+	t.application = config.Attributes["application"]
-  idSeq += 1
 
-  if config.Attributes["application"] == "" {
+	t.deviceSelector = config.Attributes["deviceSelector"]
-    log.Println("Error: application not configured")
+	if t.deviceSelector[:2] == "J:" {
-    return t
+		jp, err := jsonpath.Compile(t.deviceSelector[2:])
-  }
+		if err != nil {
-  t.application = config.Attributes["application"]
+			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.deviceSelector = config.Attributes["deviceSelector"]
+	t.Id = id
-  if t.deviceSelector[:2] == "J:" {
+	t.ready = true
-    jp, err := jsonpath.Compile(t.deviceSelector[2:])
+	t.dbh = database.NewDatabaseHandle()
-    if err != nil {
+	log.Printf("Handler SVEJ %s initialized", id)
-      log.Printf("Unable to compile deviceJsonpath: %s, %s", t.deviceSelector[2:], err)
+	return t
-      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()
-  return t
 }
 
-func (self *SingleValueExtractorJsonpathHandler) GetId() string {
+func (self *SingleValueExtractorJsonpathHandler) ExtractionHelper(subTopics []string, jPayload interface{}, selector string, jp *jsonpath.Compiled) (interface{}, error) {
-  return fmt.Sprintf("SVE%d", self.id)
+	var res interface{}
+	switch selector[:2] {
+	case "J:":
+		// extract using jsonpath from payload
+		r, e := jp.Lookup(jPayload)
+		if e != nil {
+			return "", fmt.Errorf("jp.Lookup failed with %s", e)
+		}
+		res = r
+	case "T:":
+		// T: extract from topic
+		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:":
+		// use constant value
+		res = selector[2:]
+	default:
+		return "", fmt.Errorf("Invalid selector: %s", selector[:2])
+	}
+	return res, nil
 }
 
-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 {
+	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
+		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
+	var measurement database.Measurement
-  measurement.Time = time.Now()
+	measurement.Time = time.Now()
-  measurement.Application = self.application
+	measurement.Application = self.application
 
-  subTopics := strings.Split(message.Topic, "/")
+	subTopics := strings.Split(message.Topic, "/")
-  //log.Printf("Subtopics: %s", strings.Join(subTopics, ", "))
+	log.Printf("Subtopics: %s", strings.Join(subTopics, ", "))
-  var jPayload interface{}
+	var jPayload interface{}
-  err := json.Unmarshal([]byte(message.Payload), &jPayload)
+	err := json.Unmarshal([]byte(message.Payload), &jPayload)
-  if err != nil {
+	if err != nil {
-    lost(fmt.Sprintf("Unable to unmarshal payload: %s", err), message)
+		self.Lost("Unable to unmarshal payload", err, message)
-    return
+		return
-  }
+	}
 
-  device, err1 := extractionHelper(subTopics, jPayload, self.deviceSelector, self.deviceJsonpath)
+	device, err1 := self.ExtractionHelper(subTopics, jPayload, self.deviceSelector, self.deviceJsonpath)
-  if err1 != nil {
+	if err1 != nil {
-    lost(fmt.Sprintf("Device extraction failed with %s", err1), message)
+		self.Lost("Device extraction failed", err1, message)
-    return
+		return
-  }
+	}
-  value, err2 := extractionHelper(subTopics, jPayload, self.valueSelector, self.valueJsonpath)
+	log.Printf("device: %s", device)
-  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
+	value, err2 := self.ExtractionHelper(subTopics, jPayload, self.valueSelector, self.valueJsonpath)
+	if err2 != nil {
+		self.Lost("Value extraction failed", err2, message)
+		return
+	}
 
-  var variable database.VariableType
+	unit, err3 := self.ExtractionHelper(subTopics, jPayload, self.unitSelector, self.unitJsonpath)
-  variable.Label = ""
+	if err3 != nil {
-  variable.Variable = ""
+		self.Lost("Unit extraction failed", err3, message)
-  variable.Unit = unit
+		return
-  variable.Value = value
+	}
-  measurement.Values = make(map[string]database.VariableType)
-  measurement.Values["Value"] = variable
 
-  log.Printf("Prepared measurement item: %s", measurement)
+	measurement.Device = device.(string)
-  self.dbh.StoreMeasurement(&measurement)
+
+	var variable database.VariableType
+	variable.Label = ""
+	variable.Variable = ""
+	variable.Unit = unit.(string)
+	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)
+	self.S()
 }
-

src/udi/handlers/sver/sver.go

@@ -1,29 +1,22 @@
 package sver
 
 import (
-  "log"
+	"log"
-  "time"
+	"regexp"
-  "strconv"
+	"strconv"
-  "strings"
+	"strings"
-  "regexp"
+	"time"
-  "fmt"
+	"udi/config"
-  "reflect"
+	"udi/database"
-  "encoding/json"
+	"udi/handlers/handler"
-  "github.com/oliveagle/jsonpath"
-  "udi/config"
-  "udi/handlers/handler"
-  "udi/database"
 )
 
-var idSeq int = 0
-
 type SingleValueExtractorRegexHandler struct {
-  id int
+	handler.CommonHandler
-  ready bool
+	ready        bool
-  config localConfig
+	config       localConfig
-  payloadRegex *regexp.Regexp
+	payloadRegex *regexp.Regexp
-  payloadJsonpath *jsonpath.Compiled
+	dbh          *database.DatabaseHandle
-  dbh *database.DatabaseHandle
 }
 
 const TOPIC_SEL = "topic"
@@ -32,204 +25,191 @@ const PAYLOAD_FULL_SEL = "payload-full"
 const CONSTANT_SEL = "constant"
 
 type localConfig struct {
-  application string
+	application string
-  deviceFrom string
+	deviceFrom  string
-  devicePart int
+	devicePart  int
-  device string
+	device      string
-  valueFrom string
+	valueFrom   string
-  valuePart int
+	valuePart   int
-  unitFrom string
+	valueType   string
-  unitPart int
+	unitFrom    string
-  unit string
+	unitPart    int
+	unit        string
 }
 
+func New(id string, config config.HandlerConfigT) handler.Handler {
+	t := &SingleValueExtractorRegexHandler{
+		ready: false,
+	}
 
-func NewSverHandler(config config.HandlerConfigT) handler.Handler {
+	var localConfig localConfig
-  t := &SingleValueExtractorRegexHandler {
+	if config.Attributes["application"] == "" {
-    id: idSeq,
+		log.Println("Error: application not configured")
-    ready: false,
+		return t
-  }
+	}
-  idSeq += 1
+	localConfig.application = config.Attributes["application"]
 
-  var localConfig localConfig
+	payloadRegex := config.Attributes["payloadRegex"]
-  if config.Attributes["application"] == "" {
+	if payloadRegex != "" {
-    log.Println("Error: application not configured")
+		t.payloadRegex = regexp.MustCompile(payloadRegex)
-    return t
+	} else {
-  }
+		t.payloadRegex = nil
-  localConfig.application = config.Attributes["application"]
+	}
 
-  payloadRegex := config.Attributes["payloadRegex"]
+	if config.Attributes["deviceFrom"] != TOPIC_SEL && config.Attributes["deviceFrom"] != PAYLOAD_SEL && config.Attributes["deviceFrom"] != CONSTANT_SEL {
-  if payloadRegex != "" {
+		log.Printf("Error: invalid value %s for deviceFrom", config.Attributes["deviceFrom"])
-    t.payloadRegex = regexp.MustCompile(payloadRegex)
+		return t
-  } else {
+	}
-    t.payloadRegex = nil
+	localConfig.deviceFrom = config.Attributes["deviceFrom"]
-  }
-  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 {
+	devicePart, err1 := strconv.Atoi(config.Attributes["devicePart"])
-    log.Printf("Error: invalid value %s for deviceFrom", config.Attributes["deviceFrom"])
+	if err1 != nil {
-    return t
+		log.Printf("Error: unable to convert devicePart to number: %s", err1)
-  }
+		return t
-  localConfig.deviceFrom = config.Attributes["deviceFrom"]
+	}
+	localConfig.devicePart = devicePart
 
-  devicePart, err1 := strconv.Atoi(config.Attributes["devicePart"])
+	// empty device is valid
-  if err1 != nil {
+	localConfig.device = config.Attributes["device"]
-    log.Printf("Error: unable to convert devicePart to number: %s", err1)
-    return t
-  }
-  localConfig.devicePart = devicePart
 
-  // empty device is valid
+	if config.Attributes["valueFrom"] != PAYLOAD_SEL && config.Attributes["valueFrom"] != PAYLOAD_FULL_SEL {
-  localConfig.device = config.Attributes["device"]
+		log.Printf("Error: invalid value %s for valueFrom", config.Attributes["valueFrom"])
+		return t
+	}
+	localConfig.valueFrom = config.Attributes["valueFrom"]
 
-  if config.Attributes["valueFrom"] != PAYLOAD_SEL && config.Attributes["valueFrom"] != PAYLOAD_FULL_SEL {
+	if config.Attributes["valueFrom"] == PAYLOAD_SEL {
-    log.Printf("Error: invalid value %s for valueFrom", config.Attributes["valueFrom"])
+		valuePart, err2 := strconv.Atoi(config.Attributes["valuePart"])
-    return t
+		if err2 != nil {
-  }
+			log.Printf("Error: unable to convert valuePart to number: %s", err2)
-  localConfig.valueFrom = config.Attributes["valueFrom"]
+			return t
+		}
+		localConfig.valuePart = valuePart
+	}
 
-  if config.Attributes["valueFrom"] == PAYLOAD_SEL {
+	if config.Attributes["valueType"] != "float" && config.Attributes["valueType"] != "string" {
-    valuePart, err2 := strconv.Atoi(config.Attributes["valuePart"])
+		log.Printf("Error: invalid value %s for valueType", config.Attributes["valueType"])
-    if err2 != nil {
+		return t
-      log.Printf("Error: unable to convert valuePart to number: %s", err2)
+	}
-      return t
+	localConfig.valueType = config.Attributes["valueType"]
-    }
-    localConfig.valuePart = valuePart
-  }
 
-  if 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"])
+		log.Printf("Error: invalid value %s for unitFrom", config.Attributes["unitFrom"])
-    return t
+		return t
-  }
+	}
-  localConfig.unitFrom = config.Attributes["unitFrom"]
+	localConfig.unitFrom = config.Attributes["unitFrom"]
 
-  if config.Attributes["unitFrom"] == PAYLOAD_SEL {
+	if config.Attributes["unitFrom"] == PAYLOAD_SEL {
-    unitPart, err3 := strconv.Atoi(config.Attributes["unitPart"])
+		unitPart, err3 := strconv.Atoi(config.Attributes["unitPart"])
-    if err3 != nil {
+		if err3 != nil {
-      log.Printf("Error: unable to convert unitPart to number: %s", err3)
+			log.Printf("Error: unable to convert unitPart to number: %s", err3)
-      return t
+			return t
-    }
+		}
-    localConfig.unitPart = unitPart
+		localConfig.unitPart = unitPart
-  }
+	}
 
-  // empty unit is valid
+	// empty unit is valid
-  localConfig.unit = config.Attributes["unit"]
+	localConfig.unit = config.Attributes["unit"]
 
-  t.config = localConfig
+	t.config = localConfig
 
-  t.ready = true
+	t.Id = id
-  t.dbh = database.NewDatabaseHandle()
+	t.ready = true
-  return t
+	t.dbh = database.NewDatabaseHandle()
-}
+	log.Printf("Handler SVER %d initialized", id)
-
+	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 {
+	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
+		return
-  }
+	}
-  //log.Printf("Handler SingleValueExtractor %d processing %s -> %s", self.id, message.Topic, message.Payload)
+	//log.Printf("Handler SingleValueExtractor %d processing %s -> %s", self.id, message.Topic, message.Payload)
 
-  var measurement database.Measurement
+	var measurement database.Measurement
-  measurement.Time = time.Now()
+	measurement.Time = time.Now()
-  measurement.Application = self.config.application
+	measurement.Application = self.config.application
 
-  subTopics := strings.Split(message.Topic, "/")
+	subTopics := strings.Split(message.Topic, "/")
-  //log.Printf("Subtopics: %s", strings.Join(subTopics, ", "))
+	//log.Printf("Subtopics: %s", strings.Join(subTopics, ", "))
 
-  var payloadMatches []string
+	var payloadMatches []string
-  if self.payloadRegex != nil {
+	if self.payloadRegex != nil {
-    payloadMatches = self.payloadRegex.FindStringSubmatch(message.Payload)
+		payloadMatches = self.payloadRegex.FindStringSubmatch(message.Payload)
-    //log.Printf("Matches: %s", strings.Join(payloadMatches, ", "))
+		//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 {
+	switch self.config.deviceFrom {
-  case TOPIC_SEL:
+	case TOPIC_SEL:
-    if self.config.devicePart >= len(subTopics) {
+		if self.config.devicePart >= len(subTopics) {
-      lost("devicePart out of range", message)
+			self.Lost("devicePart out of range", nil, message)
-      return
+			return
-    }
+		}
-    measurement.Device = subTopics[self.config.devicePart]
+		measurement.Device = subTopics[self.config.devicePart]
-  case PAYLOAD_SEL:
+	case PAYLOAD_SEL:
-    if self.payloadRegex == nil && self.payloadJsonpath == nil {
+		if self.payloadRegex == nil {
-      lost("no payloadRegex or payloadJsonpath defined, devicePart can't be used", message)
+			self.Lost("no payloadRegex defined, devicePart can't be used", nil, message)
-      return
+			return
-    }
+		}
-    if self.config.devicePart >= len(payloadMatches) {
+		if self.config.devicePart >= len(payloadMatches) {
-      lost("devicePart out of range", message)
+			self.Lost("devicePart out of range", nil, message)
-      return
+			return
-    }
+		}
-    measurement.Device = payloadMatches[self.config.devicePart]
+		measurement.Device = payloadMatches[self.config.devicePart]
-  case CONSTANT_SEL:
+	case CONSTANT_SEL:
-    measurement.Device = self.config.device
+		measurement.Device = self.config.device
-  }
+	}
 
-  measurement.Values = make(map[string]database.VariableType)
+	measurement.Values = make(map[string]database.VariableType)
-  var variable database.VariableType
+	var variable database.VariableType
-  variable.Label = ""
+	variable.Label = ""
-  variable.Variable = ""
+	variable.Variable = ""
 
-  switch self.config.valueFrom {
+	var value string
-  case PAYLOAD_SEL:
+	switch self.config.valueFrom {
-    if self.payloadRegex == nil && self.payloadJsonpath == nil {
+	case PAYLOAD_SEL:
-      lost("no payloadRegex or payloadJsonpath defined, valuePart can't be used", message)
+		if self.payloadRegex == nil {
-      return
+			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)
+		if self.config.valuePart >= len(payloadMatches) {
-      return
+			self.Lost("valuePart out of range", nil, message)
-    }
+			return
-    variable.Value = payloadMatches[self.config.valuePart]
+		}
-  case PAYLOAD_FULL_SEL:
+		value = payloadMatches[self.config.valuePart]
-    variable.Value = message.Payload
+	case PAYLOAD_FULL_SEL:
-  }
+		value = message.Payload
+	}
+	if self.config.valueType == "float" {
+		fValue, err := strconv.ParseFloat(value, 64)
+		if err != nil {
+			self.Lost("Unable to convert value to float", err, message)
+			return
+		}
+		variable.Value = fValue
+	} else {
+		variable.Value = value
+	}
 
-  switch self.config.unitFrom {
+	switch self.config.unitFrom {
-  case PAYLOAD_SEL:
+	case PAYLOAD_SEL:
-    if self.payloadRegex == nil && self.payloadJsonpath == nil {
+		if self.payloadRegex == nil {
-      lost("no payloadRegex or payloadJsonpath defined, unitPart can't be used", message)
+			self.Lost("no payloadRegex defined, unitPart can't be used", nil, message)
-      return
+			return
-    }
+		}
-    if self.config.unitPart >= len(payloadMatches) {
+		if self.config.unitPart >= len(payloadMatches) {
-      lost("unitPart out of range", message)
+			self.Lost("unitPart out of range", nil, message)
-      return
+			return
-    }
+		}
-    variable.Unit = payloadMatches[self.config.unitPart]
+		variable.Unit = payloadMatches[self.config.unitPart]
-  case CONSTANT_SEL:
+	case CONSTANT_SEL:
-    variable.Unit = self.config.unit
+		variable.Unit = self.config.unit
-  }
+	}
 
-  measurement.Values["Value"] = variable
+	measurement.Values["Value"] = variable
 
-  //log.Printf("Prepared measurement item: %s", measurement)
+	//log.Printf("Prepared measurement item: %s", measurement)
-  self.dbh.StoreMeasurement(&measurement)
+	self.dbh.StoreMeasurement(&measurement)
+	self.S()
 }
-

src/udi/handlers/z2m/z2m.go

@@ -0,0 +1,80 @@
+package z2m
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"strings"
+	"time"
+	"udi/config"
+	"udi/database"
+	"udi/handlers/handler"
+)
+
+type Z2MHandler struct {
+	handler.CommonHandler
+	dbh *database.DatabaseHandle
+}
+
+func New(id string, config config.HandlerConfigT) handler.Handler {
+	t := &Z2MHandler{}
+	t.Id = id
+	t.dbh = database.NewDatabaseHandle()
+	log.Printf("Handler Z2M %d initialized", id)
+	return t
+}
+
+func (self *Z2MHandler) Handle(message handler.MessageT) {
+	log.Printf("Handler Z2M %d processing %s -> %s", self.Id, message.Topic, message.Payload)
+
+	var measurement database.Measurement
+	measurement.Time = time.Now()
+
+	subTopics := strings.Split(message.Topic, "/")
+	deviceId := subTopics[1]
+	log.Printf("DeviceId: %s", deviceId)
+
+	measurement.Device = deviceId
+
+	// Parse JSON direkt in eine map
+	var jsonData map[string]interface{}
+	err := json.Unmarshal([]byte(message.Payload), &jsonData)
+	if err != nil {
+		self.Lost("Failed to parse JSON payload", err, message)
+		return
+	}
+
+	measurement.Attributes = make(map[string]interface{})
+	measurement.Values = make(map[string]database.VariableType)
+
+	// Extract device info for application naming
+	if deviceData, ok := jsonData["device"]; ok {
+		if deviceMap, ok := deviceData.(map[string]any); ok {
+			manufacturerId, hasManufacturer := deviceMap["manufacturerID"]
+			model, hasModel := deviceMap["model"]
+
+			if !hasManufacturer || !hasModel {
+				self.Lost("Missing manufacturerID or model in device data", fmt.Errorf("manufacturerID: %v, model: %v", hasManufacturer, hasModel), handler.MessageT{})
+				return
+			}
+
+			measurement.Application = "z2m_" + fmt.Sprintf("%v", manufacturerId) + "_" + model.(string)
+		}
+		delete(jsonData, "device")
+	}
+
+	// Konvertiere die restlichen Elemente in VariableType-Map
+	for key, value := range jsonData {
+		measurement.Values[key] = database.VariableType{
+			Label:    key,
+			Variable: "",
+			Unit:     "",
+			Value:    value,
+		}
+	}
+
+	measurement.Attributes["Status"] = "ok"
+	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/migrate_schema.go

@@ -1,18 +0,0 @@
-package main
-
-import "log"
-import "udi/database"
-
-
-
-func main() {
-  log.SetPrefix("UDI Migrate Schema: ")
-  log.SetFlags(log.Ldate | log.Ltime | log.Lshortfile)
-
-  log.Println("Starting")
-
-  database.Migrate()
-
-  log.Println("Done")
-}
-

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)