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

drop some obsolete instances

Browse Source
This commit is contained in:
Wolfgang Hottgenroth
2026-02-04 13:01:42 +01:00
parent a78c6952f0
commit 01acf835a8
3 changed files with 68 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
deployment/instances/udi-pg/default/config.json
View File

@@ -1,21 +0,0 @@
{
"mqtt": {
"broker": "ssl://eu1.cloud.thethings.network:8883",
"username": "com-passavant-geiger-poc@ttn",
"tlsEnable": "true"
},
"topicMappings": [
{
"topics": [ "v3/com-passavant-geiger-poc@ttn/devices/#" ],
"handler": "TTN",
"id": "TTN0",
"config": {
"attributes": {
}
}
}
],
"archiver": {
"dir": "/archive"
}
}

22
deployment/instances/udi/soil/config.json
View File

@@ -1,22 +0,0 @@
{
"mqtt": {
"broker": "ssl://eu1.cloud.thethings.network:8883",
"username": "de-hottis-app01@ttn",
"password": "ENV",
"tlsEnable": "true"
},
"topicMappings": [
{
"topics": [ "v3/#" ],
"handler": "TTN",
"id": "TTN0",
"config": {
"attributes": {
}
}
}
],
"archiver": {
"dir": "/archive"
}
}

151
src/udi/handlers/mbgw3/mbgw3.go
View File

@@ -1,106 +1,91 @@
package mbgw3 package mbgw3
import ( import (
"time" "encoding/json"
"log" "log"
"strconv" "strconv"
"encoding/json" "time"
"udi/config" "udi/config"
"udi/handlers/handler" "udi/database"
"udi/database" "udi/handlers/handler"
) )
type Mbgw3Handler struct { type Mbgw3Handler struct {
handler.CommonHandler handler.CommonHandler
dbh *database.DatabaseHandle dbh *database.DatabaseHandle
} }
type Observation struct { type Observation struct {
Status string `json:"Status"` Status string `json:"Status"`
RequestId string `json:"RequestId"` RequestId string `json:"RequestId"`
Device string `json:"Device"` Device string `json:"Device"`
Errors string `json:"Errors"` Errors string `json:"Errors"`
ErrorRatio string `json:"ErrorRatio"` ErrorRatio string `json:"ErrorRatio"`
Requests string `json:"Requests"` Requests string `json:"Requests"`
Values map[string]string Values map[string]string
} }
func New(id string, config config.HandlerConfigT) handler.Handler { func New(id string, config config.HandlerConfigT) handler.Handler {
t := &Mbgw3Handler { t := &Mbgw3Handler{}
} t.Id = id
t.Id = id t.dbh = database.NewDatabaseHandle()
t.dbh = database.NewDatabaseHandle() log.Printf("Handler MBGW3 %d initialized", id)
log.Printf("Handler MBGW3 %d initialized", id) return t
return t
} }
func (self *Mbgw3Handler) Handle(message handler.MessageT) { func (self *Mbgw3Handler) Handle(message handler.MessageT) {
//log.Printf("Handler MBGW3 %d processing %s -> %s", self.Id, message.Topic, message.Payload) //log.Printf("Handler MBGW3 %d processing %s -> %s", self.Id, message.Topic, message.Payload)
var observation Observation var observation Observation
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()
if observation.Device == "Gas" { if observation.Device == "Gas" {
measurement.Application = "Gas" measurement.Application = "Gas"
measurement.Device = "Gasmeter" measurement.Device = "Gasmeter"
} else { } else {
measurement.Application = "Power" measurement.Application = "Power"
measurement.Device = observation.Device measurement.Device = observation.Device
} }
measurement.Attributes = make(map[string]interface{}) measurement.Attributes = make(map[string]interface{})
if v, err := strconv.Atoi(observation.RequestId); err == nil { if v, err := strconv.Atoi(observation.RequestId); err == nil {
measurement.Attributes["RequestId"] = v measurement.Attributes["RequestId"] = v
} }
if v, err := strconv.ParseFloat(observation.ErrorRatio, 32); err == nil { if v, err := strconv.ParseFloat(observation.ErrorRatio, 32); err == nil {
measurement.Attributes["ErrorRatio"] = v measurement.Attributes["ErrorRatio"] = v
} }
if v, err := strconv.Atoi(observation.Errors); err == nil { if v, err := strconv.Atoi(observation.Errors); err == nil {
measurement.Attributes["Errors"] = v measurement.Attributes["Errors"] = v
} }
if v, err := strconv.Atoi(observation.Requests); err == nil { if v, err := strconv.Atoi(observation.Requests); err == nil {
measurement.Attributes["Requests"] = v measurement.Attributes["Requests"] = v
} }
measurement.Attributes["Status"] = observation.Status measurement.Attributes["Status"] = observation.Status
measurement.Values = make(map[string]database.VariableType) measurement.Values = make(map[string]database.VariableType)
unitMap := map[string]string { "Energy": "Wh", "Power": "W", "Voltage": "V", "Current": "A", "Volume": "m3" } unitMap := map[string]string{"Energy": "Wh", "Power": "W", "Voltage": "V", "Current": "A", "Volume": "m3"}
keyCount := make(map[string]int) for k, v := range observation.Values {
unit, exists := unitMap[k]
if !exists {
unit = "Unmapped Unit"
}
measurement.Values[k] = database.VariableType{
Label: "",
Variable: k,
Unit: unit,
Value: v,
}
}
for k, v := range observation.Values { //log.Printf("Prepared measurement item: %s", measurement)
unit, exists := unitMap[k]
if ! exists {
unit = "Unmapped Unit"
}
// Check if key already exists and create unique key if needed self.dbh.StoreMeasurement(&measurement)
keyCount[k]++ self.S()
uniqueKey := k
if keyCount[k] > 1 {
uniqueKey = k + strconv.Itoa(keyCount[k])
}
measurement.Values[uniqueKey] = database.VariableType {
Label: "",
Variable: k,
Unit: unit,
Value: v,
}
}
//log.Printf("Prepared measurement item: %s", measurement)
self.dbh.StoreMeasurement(&measurement)
self.S()
} }