diff --git a/deployment/instances/udi-influx/default/config.json b/deployment/instances/udi-influx/default/config.json index d774402..0c25906 100644 --- a/deployment/instances/udi-influx/default/config.json +++ b/deployment/instances/udi-influx/default/config.json @@ -185,7 +185,7 @@ "attributes": { "application": "Shellies Sensor Temperature", "deviceSelector": "T:2", - "valueSelector": "J:$.tC", + "valueSelector": "j:$.tC", "unitSelector": "C:°C" } } @@ -199,7 +199,7 @@ "attributes": { "application": "Shellies Sensor Humidity", "deviceSelector": "T:2", - "valueSelector": "J:$.rh", + "valueSelector": "j:$.rh", "unitSelector": "C:%" } } @@ -213,7 +213,7 @@ "attributes": { "application": "Shellies Sensor Power", "deviceSelector": "T:2", - "valueSelector": "J:$.battery.percent", + "valueSelector": "j:$.battery.percent", "unitSelector": "C:%" } } diff --git a/src/udi/handlers/svej/svej.go b/src/udi/handlers/svej/svej.go index 4229e0b..6d22485 100644 --- a/src/udi/handlers/svej/svej.go +++ b/src/udi/handlers/svej/svej.go @@ -1,29 +1,30 @@ package svej import ( - "log" - "time" - "strconv" - "strings" - "fmt" - "github.com/oliveagle/jsonpath" - "encoding/json" - "udi/config" - "udi/handlers/handler" - "udi/database" + "encoding/json" + "fmt" + "log" + "strconv" + "strings" + "time" + "udi/config" + "udi/database" + "udi/handlers/handler" + + "github.com/oliveagle/jsonpath" ) type SingleValueExtractorJsonpathHandler struct { - handler.CommonHandler - ready bool - application string - deviceSelector string - valueSelector string - unitSelector string - deviceJsonpath *jsonpath.Compiled - valueJsonpath *jsonpath.Compiled - unitJsonpath *jsonpath.Compiled - dbh *database.DatabaseHandle + handler.CommonHandler + ready bool + application string + deviceSelector string + valueSelector string + unitSelector string + deviceJsonpath *jsonpath.Compiled + valueJsonpath *jsonpath.Compiled + unitJsonpath *jsonpath.Compiled + dbh *database.DatabaseHandle } /* @@ -33,134 +34,135 @@ T:TopicPartIndex C:ConstantValue */ - func New(id string, config config.HandlerConfigT) handler.Handler { - t := &SingleValueExtractorJsonpathHandler { - ready: false, - } + t := &SingleValueExtractorJsonpathHandler{ + ready: false, + } - 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 - } + if config.Attributes["application"] == "" { + log.Println("Error: application not configured") + return t + } + t.application = config.Attributes["application"] - t.Id = id - t.ready = true - t.dbh = database.NewDatabaseHandle() - log.Printf("Handler SVEJ %d initialized", id) - return t + t.deviceSelector = config.Attributes["deviceSelector"] + if t.deviceSelector[:2] == "J:" || 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:" || 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) { - var res string - 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 = fmt.Sprint(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 (self *SingleValueExtractorJsonpathHandler) ExtractionHelper(subTopics []string, jPayload interface{}, selector string, jp *jsonpath.Compiled) (interface{}, error) { + var res interface{} + switch selector[:2] { + case "J:", "j:": + // extract using jsonpath from payload + r, e := jp.Lookup(jPayload) + if e != nil { + return "", fmt.Errorf("jp.Lookup failed with %s", e) + } + if selector[:2] == "j:" { + res = r.(float64) + } else { + res = fmt.Sprint(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 (self *SingleValueExtractorJsonpathHandler) Handle(message handler.MessageT) { - if ! self.ready { - self.Lost("Handler is not marked as ready", nil, message) - return - } - log.Printf("Handler SingleValueExtractorJsonpath %d processing %s -> %s", self.Id, message.Topic, message.Payload) + if !self.ready { + self.Lost("Handler is not marked as ready", nil, 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 { - self.Lost("Unable to unmarshal payload", err, message) - return - } + var measurement database.Measurement + measurement.Time = time.Now() + measurement.Application = self.application - device, err1 := self.ExtractionHelper(subTopics, jPayload, self.deviceSelector, self.deviceJsonpath) - if err1 != nil { - self.Lost("Device extraction failed", err1, message) - return - } - log.Printf("device: %s", device) + 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 + } - value, err2 := self.ExtractionHelper(subTopics, jPayload, self.valueSelector, self.valueJsonpath) - if err2 != nil { - self.Lost("Value extraction failed", err2, message) - return - } + device, err1 := self.ExtractionHelper(subTopics, jPayload, self.deviceSelector, self.deviceJsonpath) + if err1 != nil { + self.Lost("Device extraction failed", err1, message) + return + } + log.Printf("device: %s", device) - unit, err3 := self.ExtractionHelper(subTopics, jPayload, self.unitSelector, self.unitJsonpath) - if err3 != nil { - self.Lost("Unit extraction failed", err3, message) - return - } + value, err2 := self.ExtractionHelper(subTopics, jPayload, self.valueSelector, self.valueJsonpath) + if err2 != nil { + self.Lost("Value extraction failed", err2, message) + 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 - variable.Label = "" - variable.Variable = "" - variable.Unit = unit - variable.Value = value - measurement.Values = make(map[string]database.VariableType) - measurement.Values["Value"] = variable + measurement.Device = device - log.Printf("Prepared measurement item: %s", measurement) - self.dbh.StoreMeasurement(&measurement) - self.S() + 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) + self.S() } -