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

float fix 3
All checks were successful
ci/woodpecker/tag/woodpecker Pipeline was successful
Details

Browse Source
This commit is contained in:
Wolfgang Hottgenroth
2026-03-06 21:35:32 +01:00
parent f63c22912a
commit 42ff8b51ed
1 changed files with 48 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
src/udi/handlers/prepared/prepared.go
View File

@@ -1,79 +1,74 @@
package prepared package prepared
import ( import (
"time" "encoding/json"
"log" "log"
"encoding/json" "time"
"udi/config" "udi/config"
"udi/handlers/handler" "udi/database"
"udi/database" "udi/handlers/handler"
) )
type PreparedHandler struct { type PreparedHandler struct {
handler.CommonHandler handler.CommonHandler
dbh *database.DatabaseHandle dbh *database.DatabaseHandle
} }
type endpoint_t struct { type endpoint_t struct {
Label string `json:"label"` Label string `json:"label"`
Variable string `json:"variable"` Variable string `json:"variable"`
Value string `json:"value"` Value interface{} `json:"value"`
Unit string `json:"unit"` Unit string `json:"unit"`
Status string `json:"status"` Status string `json:"status"`
} }
type observation_t struct { type observation_t struct {
Device string `json:"device"` Device string `json:"device"`
Label string `json:"label"` Label string `json:"label"`
Variables map[string]endpoint_t `json:"variables"` Variables map[string]endpoint_t `json:"variables"`
} }
func New(id string, config config.HandlerConfigT) handler.Handler { func New(id string, config config.HandlerConfigT) handler.Handler {
t := &PreparedHandler { t := &PreparedHandler{}
} t.Id = id
t.Id = id t.dbh = database.NewDatabaseHandle()
t.dbh = database.NewDatabaseHandle() log.Printf("Handler Prepared %d initialized", id)
log.Printf("Handler Prepared %d initialized", id) return t
return t
} }
func (self *PreparedHandler) Handle(message handler.MessageT) { func (self *PreparedHandler) Handle(message handler.MessageT) {
log.Printf("Handler Prepared %d processing %s -> %s", self.Id, message.Topic, message.Payload) log.Printf("Handler Prepared %d processing %s -> %s", self.Id, message.Topic, message.Payload)
var observation observation_t var observation observation_t
err := json.Unmarshal([]byte(message.Payload), &observation) err := json.Unmarshal([]byte(message.Payload), &observation)
if err != nil { if err != nil {
self.Lost("Unable to parse payload into Observation struct", err, message) self.Lost("Unable to parse payload into Observation struct", err, message)
return return
} }
var measurement database.Measurement var measurement database.Measurement
measurement.Time = time.Now() measurement.Time = time.Now()
measurement.Application = self.Id measurement.Application = self.Id
measurement.Device = observation.Device measurement.Device = observation.Device
measurement.Attributes = map[string]interface{} { measurement.Attributes = map[string]interface{}{
"Label": observation.Label, "Label": observation.Label,
} }
measurement.Values = make(map[string]database.VariableType) measurement.Values = make(map[string]database.VariableType)
for k, v := range observation.Variables { for k, v := range observation.Variables {
measurement.Values[k] = database.VariableType { measurement.Values[k] = database.VariableType{
Label: v.Label, Label: v.Label,
Variable: v.Variable, Variable: v.Variable,
Unit: v.Unit, Unit: v.Unit,
Value: v.Value, Value: v.Value,
Status: v.Status, Status: v.Status,
} }
} }
log.Printf("Prepared measurement item: %s", measurement) log.Printf("Prepared measurement item: %s", measurement)
self.dbh.StoreMeasurement(&measurement) self.dbh.StoreMeasurement(&measurement)
self.S() self.S()
} }