From 8085f8937e8cf12af4889ee8976e488fdbfc5f65 Mon Sep 17 00:00:00 2001 From: Wolfgang Hottgenroth Date: Wed, 6 Dec 2023 12:30:53 +0100 Subject: [PATCH 1/2] rename sve to sver, fixes in sver --- src/udi/handlers/{sve/sve.go => sver/sver.go} | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) rename src/udi/handlers/{sve/sve.go => sver/sver.go} (97%) diff --git a/src/udi/handlers/sve/sve.go b/src/udi/handlers/sver/sver.go similarity index 97% rename from src/udi/handlers/sve/sve.go rename to src/udi/handlers/sver/sver.go index fe64147..bc841f8 100644 --- a/src/udi/handlers/sve/sve.go +++ b/src/udi/handlers/sver/sver.go @@ -157,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 } @@ -183,7 +183,7 @@ 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 } @@ -204,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 } From d7c30ef0eb2d142878246218609e299112009261 Mon Sep 17 00:00:00 2001 From: Wolfgang Hottgenroth Date: Wed, 6 Dec 2023 14:32:50 +0100 Subject: [PATCH 2/2] svej --- src/udi/dispatcher/dispatcher.go | 9 +- src/udi/go.mod | 1 + src/udi/go.sum | 2 + src/udi/handlers/svej/svej.go | 169 +++++++++++++++++++++++++++++++ src/udi/handlers/sver/sver.go | 12 +-- 5 files changed, 184 insertions(+), 9 deletions(-) create mode 100644 src/udi/handlers/svej/svej.go diff --git a/src/udi/dispatcher/dispatcher.go b/src/udi/dispatcher/dispatcher.go index c4ac749..2220c1b 100644 --- a/src/udi/dispatcher/dispatcher.go +++ b/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) diff --git a/src/udi/go.mod b/src/udi/go.mod index c6b2572..fe51759 100644 --- a/src/udi/go.mod +++ b/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 ) diff --git a/src/udi/go.sum b/src/udi/go.sum index 0a0dc0d..67dea6a 100644 --- a/src/udi/go.sum +++ b/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= diff --git a/src/udi/handlers/svej/svej.go b/src/udi/handlers/svej/svej.go new file mode 100644 index 0000000..6923d36 --- /dev/null +++ b/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) +} + diff --git a/src/udi/handlers/sver/sver.go b/src/udi/handlers/sver/sver.go index bc841f8..0eb39c4 100644 --- a/src/udi/handlers/sver/sver.go +++ b/src/udi/handlers/sver/sver.go @@ -1,4 +1,4 @@ -package sve +package sver import ( "log" @@ -14,7 +14,7 @@ import ( var idSeq int = 0 -type SingleValueExtractorHandler struct { +type SingleValueExtractorRegexHandler struct { id int ready bool config localConfig @@ -40,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, } @@ -117,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) } @@ -125,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