svej

.gitignore

@@ -3,3 +3,4 @@ src/udi/udi
 src/udi/migrate_schema
 tmp/
 ENVDB
+ENVDB.cluster

deployment/deploy-yml.tmpl

@@ -18,6 +18,8 @@ metadata:
   namespace: udi
   labels:
     app: udi
+  annotations:
+    secret.reloader.stakater.com/reload: "udi-conf,udi-db-cred"
 spec:
   replicas: 1
   selector:
@@ -37,6 +39,9 @@ spec:
                 secretKeyRef:
                   name: udi-conf
                   key: UDI_CONF
+          envFrom:
+            - secretRef: 
+                name: udi-db-cred
           volumeMounts:
             - mountPath: /archive
               name: udi-archive

deployment/load-config.sh

@@ -23,3 +23,25 @@ kubectl create secret generic $SECRET_NAME \
   -o yaml \
   --save-config | \
 kubectl apply -f -
+
+. ~/Workspace/MyKubernetesEnv/ENVDB
+DATABASE=udi
+LOGIN=udi
+PASSWORD=`openssl rand -base64 24` 
+psql <<EOF
+ALTER USER $LOGIN WITH PASSWORD '$PASSWORD';
+GRANT ALL PRIVILEGES ON DATABASE $DATABASE TO $LOGIN;
+COMMIT;
+EOF
+
+kubectl create secret generic udi-db-cred \
+  --dry-run=client \
+  -o yaml \
+  --save-config \
+  --from-literal=PGUSER="$LOGIN" \
+  --from-literal=PGHOST="timescaledb.database.svc.cluster.local" \
+  --from-literal=PGPASSWORD="$PASSWORD" \
+  --from-literal=PGSSLMODE="require" \
+  --from-literal=PGDATABASE="$DATABASE" | \
+kubectl apply -f - -n $NAMESPACE
+

queries/hottis.sql

@@ -23,3 +23,17 @@ create or replace view power_v as
     where application = 'Power' and
           attributes->>'Status' = 'Ok';
 
+create or replace view temperature_heating_v as
+  select time,
+         cast(values->'Value'->>'value' as float) as temperature,
+         device
+    from measurements
+    where application = 'Temperature Heating';
+
+create or replace view gas_v as
+  select time,
+         cast(values->'Volume'->>'value' as float) as volume,
+         device
+    from measurements
+    where application = 'Gas' and
+          attributes->>'Status' = 'Ok';

src/udi/config/config.go

@@ -17,10 +17,11 @@ type ConfigT struct {
     TlsEnable string `json:"tlsEnable"`
   } `json:"mqtt"`
   TopicMappings []struct {
+    Id string `json:"id"`
     Topics []string `json:"topics"`
     Handler string `json:"handler"`
+    Config HandlerConfigT `json:"config"`
   } `json:"topicMappings"`
-  Handlers map[string]HandlerConfigT `json:"handlers"`
   Archiver struct {
     Dir string `json:"dir"`
   }

src/udi/dispatcher/dispatcher.go

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

src/udi/go.mod

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

src/udi/go.sum

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

src/udi/handlers/handler/handler.go

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

src/udi/handlers/iot/iot.go

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

src/udi/handlers/mbgw3/mbgw3.go

@@ -5,6 +5,7 @@ import (
   //"reflect"
   "time"
   "strconv"
+  "fmt"
   "encoding/json"
   "udi/config"
   "udi/handlers/handler"
@@ -30,7 +31,7 @@ type Observation struct {
 }
 
 
-func NewMbgw3Handler(config config.HandlerConfigT) *Mbgw3Handler {
+func NewMbgw3Handler(config config.HandlerConfigT) handler.Handler {
   t := &Mbgw3Handler {
     id: idSeq,
   }
@@ -39,6 +40,10 @@ func NewMbgw3Handler(config config.HandlerConfigT) *Mbgw3Handler {
   return t
 }
 
+func (self *Mbgw3Handler) GetId() string {
+  return fmt.Sprintf("MBGW3%d", self.id)
+}
+
 func (self *Mbgw3Handler) Handle(message handler.MessageT) {
 //  log.Printf("Handler MBGW3 %d processing %s -> %s", self.id, message.Topic, message.Payload)
 

src/udi/handlers/pv/pv.go

@@ -4,6 +4,7 @@ import (
   "log"
   "reflect"
   "time"
+  "fmt"
   "encoding/json"
   "udi/config"
   "udi/handlers/handler"
@@ -39,7 +40,7 @@ type PvValue struct {
 }
 
 
-func NewPvHandler(config config.HandlerConfigT) *PvHandler {
+func NewPvHandler(config config.HandlerConfigT) handler.Handler {
   t := &PvHandler {
     id: idSeq,
   }
@@ -48,6 +49,10 @@ func NewPvHandler(config config.HandlerConfigT) *PvHandler {
   return t
 }
 
+func (self *PvHandler) GetId() string {
+  return fmt.Sprintf("PV%d", self.id)
+}
+
 func (self *PvHandler) Handle(message handler.MessageT) {
   //log.Printf("Handler PV %d processing %s -> %s", self.id, message.Topic, message.Payload)
 

src/udi/handlers/svej/svej.go

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

src/udi/handlers/sver/sver.go

@@ -1,4 +1,4 @@
-package sve
+package sver
 
 import (
   "log"
@@ -6,6 +6,7 @@ import (
   "strconv"
   "strings"
   "regexp"
+  "fmt"
   "udi/config"
   "udi/handlers/handler"
   "udi/database"
@@ -13,7 +14,7 @@ import (
 
 var idSeq int = 0
 
-type SingleValueExtractorHandler struct {
+type SingleValueExtractorRegexHandler struct {
   id int
   ready bool
   config localConfig
@@ -23,6 +24,7 @@ type SingleValueExtractorHandler struct {
 
 const TOPIC_SEL = "topic"
 const PAYLOAD_SEL = "payload"
+const PAYLOAD_FULL_SEL = "payload-full"
 const CONSTANT_SEL = "constant"
 
 type localConfig struct {
@@ -38,8 +40,8 @@ type localConfig struct {
 }
 
 
-func NewSveHandler(config config.HandlerConfigT) *SingleValueExtractorHandler {
-  t := &SingleValueExtractorHandler {
+func NewSverHandler(config config.HandlerConfigT) handler.Handler {
+  t := &SingleValueExtractorRegexHandler {
     id: idSeq,
     ready: false,
   }
@@ -75,18 +77,20 @@ func NewSveHandler(config config.HandlerConfigT) *SingleValueExtractorHandler {
   // empty device is valid
   localConfig.device = config.Attributes["device"]
 
-  if config.Attributes["valueFrom"] != TOPIC_SEL && config.Attributes["valueFrom"] != PAYLOAD_SEL {
+  if config.Attributes["valueFrom"] != TOPIC_SEL && config.Attributes["valueFrom"] != PAYLOAD_SEL && config.Attributes["valueFrom"] != PAYLOAD_FULL_SEL {
     log.Printf("Error: invalid value %s for valueFrom", config.Attributes["valueFrom"])
     return t
   }
   localConfig.valueFrom = config.Attributes["valueFrom"]
 
+  if config.Attributes["valueFrom"] != PAYLOAD_FULL_SEL {
     valuePart, err2 := strconv.Atoi(config.Attributes["valuePart"])
     if err2 != nil {
       log.Printf("Error: unable to convert valuePart to number: %s", err2)
       return t
     }
     localConfig.valuePart = valuePart
+  }
 
   if config.Attributes["unitFrom"] != TOPIC_SEL && config.Attributes["unitFrom"] != PAYLOAD_SEL && config.Attributes["unitFrom"] != CONSTANT_SEL {
     log.Printf("Error: invalid value %s for unitFrom", config.Attributes["unitFrom"])
@@ -94,12 +98,14 @@ func NewSveHandler(config config.HandlerConfigT) *SingleValueExtractorHandler {
   }
   localConfig.unitFrom = config.Attributes["unitFrom"]
 
+  if config.Attributes["unitFrom"] != CONSTANT_SEL {
     unitPart, err3 := strconv.Atoi(config.Attributes["unitPart"])
     if err3 != nil {
       log.Printf("Error: unable to convert unitPart to number: %s", err3)
       return t
     }
     localConfig.unitPart = unitPart
+  }
 
   // empty unit is valid
   localConfig.unit = config.Attributes["unit"]
@@ -111,11 +117,15 @@ func NewSveHandler(config config.HandlerConfigT) *SingleValueExtractorHandler {
   return t
 }
 
+func (self *SingleValueExtractorRegexHandler) GetId() string {
+  return fmt.Sprintf("SVE%d", self.id)
+}
+
 func lost(msg string, message handler.MessageT) {
   log.Printf("Error: %s, message %s is lost", msg, message)
 }
 
-func (self *SingleValueExtractorHandler) Handle(message handler.MessageT) {
+func (self *SingleValueExtractorRegexHandler) Handle(message handler.MessageT) {
   if ! self.ready {
     log.Println("Handler is not marked as ready, message %s is lost", message)
     return
@@ -147,7 +157,7 @@ func (self *SingleValueExtractorHandler) Handle(message handler.MessageT) {
       lost("no payloadRegex defined, devicePart can't be used", message)
       return
     }
-    if self.config.devicePart >= len(subTopics) {
+    if self.config.devicePart >= len(payloadMatches) {
       lost("devicePart out of range", message)
       return
     }
@@ -173,11 +183,13 @@ func (self *SingleValueExtractorHandler) Handle(message handler.MessageT) {
       lost("no payloadRegex defined, valuePart can't be used", message)
       return
     }
-    if self.config.valuePart >= len(subTopics) {
+    if self.config.valuePart >= len(payloadMatches) {
       lost("valuePart out of range", message)
       return
     }
     variable.Value = payloadMatches[self.config.valuePart]
+  case PAYLOAD_FULL_SEL:
+    variable.Value = message.Payload
   }
 
   switch self.config.unitFrom {
@@ -192,7 +204,7 @@ func (self *SingleValueExtractorHandler) Handle(message handler.MessageT) {
       lost("no payloadRegex defined, unitPart can't be used", message)
       return
     }
-    if self.config.unitPart >= len(subTopics) {
+    if self.config.unitPart >= len(payloadMatches) {
       lost("unitPart out of range", message)
       return
     }

src/udi/handlers/ttn/ttn.go

@@ -1,6 +1,7 @@
 package ttn
 
 import "log"
+import "fmt"
 import "udi/handlers/handler"
 
 var idSeq int = 0
@@ -9,7 +10,7 @@ type TTNHandler struct {
   id int
 }
 
-func NewTTNHandler() *TTNHandler {
+func NewTTNHandler() handler.Handler {
   t := &TTNHandler {
     id: idSeq,
   }
@@ -17,6 +18,10 @@ func NewTTNHandler() *TTNHandler {
   return t
 }
 
+func (self *TTNHandler) GetId() string {
+  return fmt.Sprintf("TTN%d", self.id)
+}
+
 func (self *TTNHandler) Handle(message handler.MessageT) {
   log.Printf("Handler TTN %d processing %s -> %s", self.id, message.Topic, message.Payload)
 }