svej

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)
@@ -35,8 +36,10 @@ func InitDispatcher() {
       factory = pv.NewPvHandler
     case "MBGW3":
       factory = mbgw3.NewMbgw3Handler
-    case "SVE":
-      factory = sve.NewSveHandler
+    case "SVER":
+      factory = sver.NewSverHandler
+    case "SVEJ":
+      factory = svej.NewSvejHandler
     default:
       factory = nil
       log.Printf("No handler %s found, ignore mapping", mapping.Handler)

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"
@@ -7,8 +7,6 @@ import (
   "strings"
   "regexp"
   "fmt"
-  "encoding/json"
-  "github.com/oliveagle/jsonpath"
   "udi/config"
   "udi/handlers/handler"
   "udi/database"
@@ -16,18 +14,16 @@ import (
 
 var idSeq int = 0
 
-type SingleValueExtractorHandler struct {
+type SingleValueExtractorRegexHandler struct {
   id int
   ready bool
   config localConfig
   payloadRegex *regexp.Regexp
-  payloadJsonpath *jsonpath.Compiled
   dbh *database.DatabaseHandle
 }
 
 const TOPIC_SEL = "topic"
-const PAYLOAD_REGEX_SEL = "payload-regex"
-const PAYLOAD_JSONPATH_SEL = "payload-jsonpath"
+const PAYLOAD_SEL = "payload"
 const PAYLOAD_FULL_SEL = "payload-full"
 const CONSTANT_SEL = "constant"
 
@@ -44,8 +40,8 @@ type localConfig struct {
 }
 
 
-func NewSveHandler(config config.HandlerConfigT) handler.Handler {
-  t := &SingleValueExtractorHandler {
+func NewSverHandler(config config.HandlerConfigT) handler.Handler {
+  t := &SingleValueExtractorRegexHandler {
     id: idSeq,
     ready: false,
   }
@@ -64,20 +60,8 @@ func NewSveHandler(config config.HandlerConfigT) handler.Handler {
   } else {
     t.payloadRegex = nil
   }
-  payloadJsonpath := config.Attributes["payloadJsonpath"]
-  if payloadJsonpath != "" {
-    j, err := jsonpath.Compile(payloadJsonpath)
-    if err != nil {
-      log.Printf("Unable to compile jsonpath %s", payloadJsonpath)
-      t.payloadJsonpath = nil
-    } else {
-      t.payloadJsonpath = j
-    }
-  } else {
-    t.payloadJsonpath = nil
-  }
 
-  if config.Attributes["deviceFrom"] != TOPIC_SEL && config.Attributes["deviceFrom"] != PAYLOAD_REGEX_SEL && config.Attributes["deviceFrom"] != CONSTANT_SEL {
+  if config.Attributes["deviceFrom"] != TOPIC_SEL && config.Attributes["deviceFrom"] != PAYLOAD_SEL && config.Attributes["deviceFrom"] != CONSTANT_SEL {
     log.Printf("Error: invalid value %s for deviceFrom", config.Attributes["deviceFrom"])
     return t
   }
@@ -93,13 +77,13 @@ func NewSveHandler(config config.HandlerConfigT) handler.Handler {
   // empty device is valid
   localConfig.device = config.Attributes["device"]
 
-  if config.Attributes["valueFrom"] != PAYLOAD_REGEX_SEL && config.Attributes["valueFrom"] != PAYLOAD_JSONPATH_SEL && config.Attributes["valueFrom"] != PAYLOAD_FULL_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_REGEX_SEL {
+  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)
@@ -108,7 +92,7 @@ func NewSveHandler(config config.HandlerConfigT) handler.Handler {
     localConfig.valuePart = valuePart
   }
 
-  if config.Attributes["unitFrom"] != TOPIC_SEL && config.Attributes["unitFrom"] != PAYLOAD_REGEX_SEL && config.Attributes["unitFrom"] != CONSTANT_SEL {
+  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"])
     return t
   }
@@ -133,7 +117,7 @@ func NewSveHandler(config config.HandlerConfigT) handler.Handler {
   return t
 }
 
-func (self *SingleValueExtractorHandler) GetId() string {
+func (self *SingleValueExtractorRegexHandler) GetId() string {
   return fmt.Sprintf("SVE%d", self.id)
 }
 
@@ -141,7 +125,7 @@ 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
@@ -168,12 +152,12 @@ func (self *SingleValueExtractorHandler) Handle(message handler.MessageT) {
       return
     }
     measurement.Device = subTopics[self.config.devicePart]
-  case PAYLOAD_REGEX_SEL:
+  case PAYLOAD_SEL:
     if self.payloadRegex == nil {
       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
     }
@@ -188,7 +172,13 @@ func (self *SingleValueExtractorHandler) Handle(message handler.MessageT) {
   variable.Variable = ""
 
   switch self.config.valueFrom {
-  case PAYLOAD_REGEX_SEL:
+  case TOPIC_SEL:
+    if self.config.valuePart >= len(subTopics) {
+      lost("valuePart out of range", message)
+      return
+    }
+    variable.Value = subTopics[self.config.valuePart]
+  case PAYLOAD_SEL:
     if self.payloadRegex == nil {
       lost("no payloadRegex defined, valuePart can't be used", message)
       return
@@ -198,19 +188,6 @@ func (self *SingleValueExtractorHandler) Handle(message handler.MessageT) {
       return
     }
     variable.Value = payloadMatches[self.config.valuePart]
-  case PAYLOAD_JSONPATH_SEL:
-    if self.payloadJsonpath == nil {
-      lost("no payloadJsonpath defined, valuePart can't be used", message)
-      return
-    }
-    var jsonData interface{}
-    json.Unmarshal([]byte(message.Payload), &jsonData)
-    result, err := self.payloadJsonpath.Lookup(jsonData)
-    if err != nil {
-      lost(fmt.Sprintf("jsonpath error: %s", err), message)
-      return
-    }
-    variable.Value = result
   case PAYLOAD_FULL_SEL:
     variable.Value = message.Payload
   }
@@ -222,12 +199,12 @@ func (self *SingleValueExtractorHandler) Handle(message handler.MessageT) {
       return
     }
     variable.Unit = subTopics[self.config.unitPart]
-  case PAYLOAD_REGEX_SEL:
+  case PAYLOAD_SEL:
     if self.payloadRegex == nil {
       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
     }