wn/universal-data-ingest
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

small j for float
Some checks failed
ci/woodpecker/tag/woodpecker Pipeline failed
Details

Browse Source
This commit is contained in:
Wolfgang Hottgenroth
2026-03-06 20:25:04 +01:00
parent 8c5626942f
commit b940f715c0
2 changed files with 140 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
deployment/instances/udi-influx/default/config.json
View File

@@ -185,7 +185,7 @@
"attributes": { "attributes": {
"application": "Shellies Sensor Temperature", "application": "Shellies Sensor Temperature",
"deviceSelector": "T:2", "deviceSelector": "T:2",
"valueSelector": "J:$.tC", "valueSelector": "j:$.tC",
"unitSelector": "C:°C" "unitSelector": "C:°C"
} }
} }
@@ -199,7 +199,7 @@
"attributes": { "attributes": {
"application": "Shellies Sensor Humidity", "application": "Shellies Sensor Humidity",
"deviceSelector": "T:2", "deviceSelector": "T:2",
"valueSelector": "J:$.rh", "valueSelector": "j:$.rh",
"unitSelector": "C:%" "unitSelector": "C:%"
} }
} }
@@ -213,7 +213,7 @@
"attributes": { "attributes": {
"application": "Shellies Sensor Power", "application": "Shellies Sensor Power",
"deviceSelector": "T:2", "deviceSelector": "T:2",
"valueSelector": "J:$.battery.percent", "valueSelector": "j:$.battery.percent",
"unitSelector": "C:%" "unitSelector": "C:%"
} }
} }

272
src/udi/handlers/svej/svej.go
View File

@@ -1,29 +1,30 @@
package svej package svej
import ( 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"
) )
type SingleValueExtractorJsonpathHandler struct { type SingleValueExtractorJsonpathHandler struct {
handler.CommonHandler 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
} }
/* /*
@@ -33,134 +34,135 @@ T:TopicPartIndex
C:ConstantValue C:ConstantValue
*/ */
func New(id string, config config.HandlerConfigT) handler.Handler { func New(id string, config config.HandlerConfigT) handler.Handler {
t := &SingleValueExtractorJsonpathHandler { t := &SingleValueExtractorJsonpathHandler{
ready: false, ready: false,
} }
if config.Attributes["application"] == "" { if config.Attributes["application"] == "" {
log.Println("Error: application not configured") log.Println("Error: application not configured")
return t return t
} }
t.application = config.Attributes["application"] 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.Id = id t.deviceSelector = config.Attributes["deviceSelector"]
t.ready = true if t.deviceSelector[:2] == "J:" || t.deviceSelector[:2] == "j:" {
t.dbh = database.NewDatabaseHandle() jp, err := jsonpath.Compile(t.deviceSelector[2:])
log.Printf("Handler SVEJ %d initialized", id) if err != nil {
return t 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:" || 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:" || 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.Id = id
t.ready = true
t.dbh = database.NewDatabaseHandle()
log.Printf("Handler SVEJ %s initialized", id)
return t
} }
func (self *SingleValueExtractorJsonpathHandler) ExtractionHelper(subTopics []string, jPayload interface{}, selector string, jp *jsonpath.Compiled) (string, error) { func (self *SingleValueExtractorJsonpathHandler) ExtractionHelper(subTopics []string, jPayload interface{}, selector string, jp *jsonpath.Compiled) (interface{}, error) {
var res string var res interface{}
switch selector[:2] { switch selector[:2] {
case "J:": case "J:", "j:":
// extract using jsonpath from payload // extract using jsonpath from payload
r, e := jp.Lookup(jPayload) r, e := jp.Lookup(jPayload)
if e != nil { if e != nil {
return "", fmt.Errorf("jp.Lookup failed with %s", e) return "", fmt.Errorf("jp.Lookup failed with %s", e)
} }
res = fmt.Sprint(r) if selector[:2] == "j:" {
case "T:": res = r.(float64)
// T: extract from topic } else {
i, e := strconv.Atoi(selector[2:]) res = fmt.Sprint(r)
if e != nil { }
return "", fmt.Errorf("Atoi failed with %s", e) case "T:":
} // T: extract from topic
if i >= len(subTopics) { i, e := strconv.Atoi(selector[2:])
return "", fmt.Errorf("not enough subtopics") if e != nil {
} return "", fmt.Errorf("Atoi failed with %s", e)
res = subTopics[i] }
case "C:": if i >= len(subTopics) {
// use constant value return "", fmt.Errorf("not enough subtopics")
res = selector[2:] }
default: res = subTopics[i]
return "", fmt.Errorf("Invalid selector: %s", selector[:2]) case "C:":
} // use constant value
return res, nil res = selector[2:]
default:
return "", fmt.Errorf("Invalid selector: %s", selector[:2])
}
return res, nil
} }
func (self *SingleValueExtractorJsonpathHandler) Handle(message handler.MessageT) { func (self *SingleValueExtractorJsonpathHandler) Handle(message handler.MessageT) {
if ! self.ready { if !self.ready {
self.Lost("Handler is not marked as ready", nil, 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, "/")
log.Printf("Subtopics: %s", strings.Join(subTopics, ", "))
var jPayload interface{}
err := json.Unmarshal([]byte(message.Payload), &jPayload)
if err != nil {
self.Lost("Unable to unmarshal payload", err, message)
return
}
device, err1 := self.ExtractionHelper(subTopics, jPayload, self.deviceSelector, self.deviceJsonpath) subTopics := strings.Split(message.Topic, "/")
if err1 != nil { log.Printf("Subtopics: %s", strings.Join(subTopics, ", "))
self.Lost("Device extraction failed", err1, message) var jPayload interface{}
return err := json.Unmarshal([]byte(message.Payload), &jPayload)
} if err != nil {
log.Printf("device: %s", device) self.Lost("Unable to unmarshal payload", err, message)
return
}
value, err2 := self.ExtractionHelper(subTopics, jPayload, self.valueSelector, self.valueJsonpath) device, err1 := self.ExtractionHelper(subTopics, jPayload, self.deviceSelector, self.deviceJsonpath)
if err2 != nil { if err1 != nil {
self.Lost("Value extraction failed", err2, message) self.Lost("Device extraction failed", err1, message)
return return
} }
log.Printf("device: %s", device)
unit, err3 := self.ExtractionHelper(subTopics, jPayload, self.unitSelector, self.unitJsonpath) value, err2 := self.ExtractionHelper(subTopics, jPayload, self.valueSelector, self.valueJsonpath)
if err3 != nil { if err2 != nil {
self.Lost("Unit extraction failed", err3, message) self.Lost("Value extraction failed", err2, message)
return return
} }
measurement.Device = device unit, err3 := self.ExtractionHelper(subTopics, jPayload, self.unitSelector, self.unitJsonpath)
if err3 != nil {
self.Lost("Unit extraction failed", err3, message)
return
}
var variable database.VariableType measurement.Device = device
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) var variable database.VariableType
self.dbh.StoreMeasurement(&measurement) variable.Label = ""
self.S() 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)
self.S()
} }