Compare commits

...

5 Commits

28 changed files with 385 additions and 417 deletions

145
src/udi/config-test-2.json Normal file
View File

@@ -0,0 +1,145 @@
{
"mqtt": {
"broker": "mqtt://172.23.1.102:1883",
"tlsEnable": "false"
},
"topicMappings": [
{
"topics": [ "snmp" ],
"handler": "PREP",
"id": "SNMP",
"config": {
"attributes": {
}
}
},
{
"topics": [ "tsm" ],
"handler": "PREP",
"id": "TSM",
"config": {
"attributes": {
}
}
},
{
"topics": [ "dt1/ai/periodic/1" ],
"handler": "DT1T",
"id": "DT1T.0",
"config": {
"attributes": {
"Application": "Temperature Wago",
"Device": "Freezer",
"HardLow": "-273",
"SoftLow": "-50",
"SoftHigh": "20",
"HardHigh": "100"
}
}
},
{
"topics": [ "dt1/ai/periodic/3" ],
"handler": "DT1T",
"id": "DT1T.1",
"config": {
"attributes": {
"Application": "Temperature Wago",
"Device": "Outdoor",
"HardLow": "-273",
"SoftLow": "-60",
"SoftHigh": "60",
"HardHigh": "100"
}
}
},
{
"topics": [ "IoT/PV/Values" ],
"handler": "PV",
"id": "PV",
"config": {
"databaseConnStr": "",
"attributes": {
}
}
},
{
"topics": [ "IoT/Car/Values" ],
"handler": "Car",
"id": "Car",
"config": {
"databaseConnStr": "",
"attributes": {
}
}
},
{
"topics": [ "IoT/MBGW3/Measurement" ],
"handler": "MBGW3",
"id": "MBGW3",
"config": {
"databaseConnStr": "",
"attributes": {
}
}
},
{
"topics": [ "IoT/OneWireGW/Bus 1/#" ],
"handler": "SVER",
"id": "SVER0",
"config": {
"databaseConnStr": "",
"attributes": {
"application": "Temperature Heating",
"payloadRegex": "(\\d+(\\.\\d+)?)\\s*([^0-9\\s]\\S*)",
"deviceFrom": "topic",
"devicePart": "3",
"valueFrom": "payload",
"valuePart": "1",
"unitFrom": "payload",
"unitPart": "3"
}
}
},
{
"topics": [ "zigbee2mqtt/+" ],
"handler": "Z2M",
"id": "Z2M",
"config": {
"databaseConnStr": "",
"attributes": {
}
}
},
{
"topics": [ "shellyplusht/+/status/temperature:0" ],
"handler": "SVEJ",
"id": "SVEJ2",
"config": {
"databaseConnStr": "",
"attributes": {
"application": "Temperature Shelly Plus HT",
"deviceSelector": "T:1",
"valueSelector": "J:$.tC",
"unitSelector": "C:°C"
}
}
},
{
"topics": [ "shellyplusht/+/status/humidity:0" ],
"handler": "SVEJ",
"id": "SVE4",
"config": {
"databaseConnStr": "",
"attributes": {
"application": "Humidity Shelly Plus HT",
"deviceSelector": "T:1",
"valueSelector": "J:$.rh",
"unitSelector": "C:%"
}
}
}
],
"archiver": {
"dir": "./tmp/udi"
}
}

View File

@@ -5,23 +5,19 @@
},
"topicMappings": [
{
"topics": [ "NR/Multisensor/+/Temperatur" ],
"handler": "SVEJ",
"id": "SVEJ0",
"topics": [ "IoT/PV/Values" ],
"handler": "PV",
"id": "PV",
"config": {
"databaseConnStr": "",
"attributes": {
"application": "Temperature Multisensor",
"deviceSelector": "T:2",
"valueSelector": "J:$.CurrentTemperature",
"unitSelector": "C:°C"
}
}
},
{
"topics": [ "zigbee2mqtt/+" ],
"handler": "Z2M",
"id": "Z2M",
"topics": [ "IoT/Car/Values" ],
"handler": "Car",
"id": "Car",
"config": {
"databaseConnStr": "",
"attributes": {

View File

@@ -1,47 +1,38 @@
package database
import "time"
import "gorm.io/gorm"
type VariableType struct {
Label string `json:"label"`
Variable string `json:"variable"`
Unit string `json:"unit"`
Value interface{} `json:"value,omitempty"`
Status string `json:"status,omitempty"`
Label string `json:"label"`
Variable string `json:"variable"`
Unit string `json:"unit"`
Value interface{} `json:"value,omitempty"`
Status string `json:"status,omitempty"`
}
type Measurement struct {
Time time.Time `gorm:"not null;primary_key"`
Application string `gorm:"not null"`
Device string
Attributes map[string]interface{} `gorm:"serializer:json;type:jsonb"`
Values map[string]VariableType `gorm:"serializer:json;type:jsonb"`
Time time.Time
Application string
Device string
Attributes map[string]interface{}
Values map[string]VariableType
}
// Simplified structures for backward compatibility
type DeviceType struct {
Label string
ModelIdentifier string
Attributes map[string]interface{}
}
type Application struct {
gorm.Model
Label string `gorm:"not null"`
Attributes map[string]interface{} `gorm:"serializer:json;type:jsonb"`
}
type DeviceType struct {
gorm.Model
Label string `gorm:"not null"`
ModelIdentifier string
Attributes map[string]interface{} `gorm:"serializer:json;type:jsonb"`
Label string
Attributes map[string]interface{}
}
type Device struct {
gorm.Model
Label string `gorm:"not null;uniqueIndex:idx_label_application_id"`
ApplicationID int `gorm:"not null;uniqueIndex:idx_label_application_id"`
Application Application
DeviceTypeID int `gorm:"not null"`
DeviceType DeviceType
Attributes map[string]interface{} `gorm:"serializer:json;type:jsonb"`
Label string
Application Application
DeviceType DeviceType
Attributes map[string]interface{}
}

View File

@@ -1,96 +1,155 @@
package database
import (
"log"
//"time"
"fmt"
"udi/counter"
"gorm.io/driver/postgres"
"gorm.io/gorm"
"fmt"
"log"
"os"
"udi/counter"
influxdb "github.com/influxdata/influxdb1-client/v2"
)
type DatabaseHandle struct {
initialized bool
dbh *gorm.DB
initialized bool
client influxdb.Client
database string
}
func NewDatabaseHandle() *DatabaseHandle {
var db DatabaseHandle
// inject the whole database configuration via the well-known PG* env variables
conn, err := gorm.Open(postgres.Open(""))
if err != nil {
log.Printf("Unable to open database connection: %s", err)
db.initialized = false
} else {
db.dbh = conn
db.initialized = true
log.Println("Database connection opened")
}
return &db
var db DatabaseHandle
// Read configuration from environment variables
influxURL := os.Getenv("INFLUXDB_URL")
if influxURL == "" {
influxURL = "http://localhost:8086"
}
influxDB := os.Getenv("INFLUXDB_DATABASE")
if influxDB == "" {
influxDB = "udi"
}
username := os.Getenv("INFLUXDB_USER")
password := os.Getenv("INFLUXDB_PASSWORD")
// Create InfluxDB client
client, err := influxdb.NewHTTPClient(influxdb.HTTPConfig{
Addr: influxURL,
Username: username,
Password: password,
})
if err != nil {
log.Printf("Unable to create InfluxDB client: %s", err)
db.initialized = false
return &db
}
// Test connection
_, _, err = client.Ping(0)
if err != nil {
log.Printf("Unable to ping InfluxDB: %s", err)
db.initialized = false
client.Close()
return &db
}
db.client = client
db.database = influxDB
db.initialized = true
log.Printf("InfluxDB connection opened (URL: %s, Database: %s)", influxURL, influxDB)
return &db
}
func (self *DatabaseHandle) StoreMeasurement(measurement *Measurement) {
if ! self.initialized {
log.Printf("Database connection not initialized, can not store, measurement %s lost", measurement)
counter.F("Stored")
return
}
if !self.initialized {
log.Printf("Database connection not initialized, can not store, measurement %v lost", measurement)
counter.F("Stored")
return
}
result := self.dbh.Create(measurement)
if result.Error != nil {
log.Printf("Unable to insert, measurement %s lost, error: %s", measurement, result.Error)
counter.F("Stored")
return
}
// Create batch points
bp, err := influxdb.NewBatchPoints(influxdb.BatchPointsConfig{
Database: self.database,
Precision: "s",
})
if err != nil {
log.Printf("Unable to create batch points: %s", err)
counter.F("Stored")
return
}
log.Println("Successfully stored measurement")
counter.S("Stored")
// Build tags
tags := make(map[string]string)
if measurement.Device != "" {
tags["device"] = measurement.Device
}
// Build fields from Values
fields := make(map[string]interface{})
for key, varType := range measurement.Values {
// Store the value with the variable name as field key
fields[key] = varType.Value
// Optionally store metadata as separate fields
if varType.Unit != "" {
fields[key+"_unit"] = varType.Unit
}
// This is already the column name, so we can skip it
//if varType.Variable != "" {
// fields[key+"_variable"] = varType.Variable
//}
if varType.Status != "" {
fields[key+"_status"] = varType.Status
}
}
// Add attributes as fields
for key, value := range measurement.Attributes {
if strValue, ok := value.(string); ok {
fields[key] = strValue
} else {
fields[key] = fmt.Sprintf("%v", value)
}
}
// Ensure we have at least one field
if len(fields) == 0 {
log.Printf("No fields to store in measurement, skipping")
counter.F("Stored")
return
}
// Create point
pt, err := influxdb.NewPoint(
measurement.Application,
tags,
fields,
measurement.Time,
)
if err != nil {
log.Printf("Unable to create point: %s", err)
counter.F("Stored")
return
}
bp.AddPoint(pt)
// Write batch
err = self.client.Write(bp)
if err != nil {
log.Printf("Unable to write to InfluxDB, measurement lost, error: %s", err)
counter.F("Stored")
return
}
log.Println("Successfully stored measurement")
counter.S("Stored")
}
func (self *DatabaseHandle) GetDeviceByLabelAndApplication(applicationLabel string, deviceLabel string) (*Device, error) {
if ! self.initialized {
err := fmt.Errorf("Database connection not initialized")
return nil, err
}
var device Device
result := self.dbh.
Preload("Application").
Preload("DeviceType").
Joins("JOIN applications ON devices.application_id = applications.id").
Where("devices.label = ? AND applications.label = ?", deviceLabel, applicationLabel).
First(&device)
if result.Error != nil {
err := fmt.Errorf("Query failed: %s", result.Error)
return nil, err
}
return &device, nil
func (self *DatabaseHandle) Close() {
if self.initialized && self.client != nil {
self.client.Close()
log.Println("InfluxDB connection closed")
}
}
func (self *DatabaseHandle) GetDeviceByLabel(deviceLabel string) (*Device, error) {
if ! self.initialized {
err := fmt.Errorf("Database connection not initialized")
return nil, err
}
var device Device
result := self.dbh.
Preload("Application").
Preload("DeviceType").
Where("devices.label = ?", deviceLabel).
First(&device)
if result.Error != nil {
err := fmt.Errorf("Query failed: %s", result.Error)
return nil, err
}
return &device, nil
}

View File

@@ -1,55 +0,0 @@
package database
import (
"log"
//"time"
"gorm.io/driver/postgres"
"gorm.io/gorm"
)
func Migrate() {
dsn := ""
db, err := gorm.Open(postgres.Open(dsn))
if err != nil {
log.Fatalf("Unable to open database connection: %s", err)
}
db.AutoMigrate(&Application{})
log.Println("Application created")
db.AutoMigrate(&DeviceType{})
log.Println("DeviceType created")
db.AutoMigrate(&Device{})
log.Println("Device created")
db.AutoMigrate(&Measurement{})
log.Println("Measurement created")
log.Println("Remember to call create_hypertable on measurements, sowhat I can't do that for you.")
/*
m := Measurement {
Time: time.Now(),
Application: "app",
Attributes: nil,
Values: []SensorType {
{ Variable: "Temperature", Unit: "Degree Celsius", Value: 1.0 },
{ Variable: "Temperature", Unit: "Degree Celsius", Value: 3.0 },
},
}
db.Create(&m)
m = Measurement {
Time: time.Now(),
Application: "app",
Attributes: nil,
Values: []SensorType {
{ Variable: "Temperature", Unit: "Degree Celsius", Value: 10.0 },
{ Variable: "Temperature", Unit: "Degree Celsius", Value: 30.0 },
},
}
db.Create(&m)
*/
}

View File

@@ -1,41 +0,0 @@
create extension if not exists timescaledb;
create table application_t (
id serial not null primary key,
label text not null unique,
attributes jsonb
);
create table sensor_type_t (
id serial not null primary key,
label text not null unique,
variable text not null,
unit text not null,
converter text not null,
attributes jsonb
);
create table sensor_t (
id serial not null primary key,
label text not null,
application int references application_t(id),
sensor_type int references sensor_type_t(id),
attributes jsonb,
unique (label, application)
);
create table measurement_t (
time timestamp without time zone not null,
application text not null,
sensor_type text not null,
sensor text not null,
variable text not null,
unit text not null,
value double precision not null,
attributes jsonb
);
select create_hypertable('measurement_t', 'time');

View File

@@ -18,7 +18,8 @@ import (
"udi/handlers/pv"
"udi/handlers/svej"
"udi/handlers/sver"
"udi/handlers/ttn"
// "udi/handlers/ttn"
"udi/handlers/z2m"
"udi/mqtt"
)
@@ -35,8 +36,8 @@ func InitDispatcher() {
var factory interface{}
switch mapping.Handler {
case "TTN":
factory = ttn.New
// case "TTN":
// factory = ttn.New
case "IoT":
factory = iot.New
case "PV":

View File

@@ -5,21 +5,12 @@ go 1.22.3
require (
github.com/eclipse/paho.mqtt.golang v1.5.0
github.com/google/uuid v1.6.0
github.com/influxdata/influxdb1-client v0.0.0-20220302092344-a9ab5670611c
github.com/oliveagle/jsonpath v0.0.0-20180606110733-2e52cf6e6852
gorm.io/driver/postgres v1.5.11
gorm.io/gorm v1.25.12
)
require (
github.com/gorilla/websocket v1.5.3 // indirect
github.com/jackc/pgpassfile v1.0.0 // indirect
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
github.com/jackc/pgx/v5 v5.7.2 // indirect
github.com/jackc/puddle/v2 v2.2.1 // indirect
github.com/jinzhu/inflection v1.0.0 // indirect
github.com/jinzhu/now v1.1.5 // indirect
golang.org/x/crypto v0.32.0 // indirect
golang.org/x/net v0.34.0 // indirect
golang.org/x/sync v0.10.0 // indirect
golang.org/x/text v0.21.0 // indirect
)

View File

@@ -1,46 +1,14 @@
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/eclipse/paho.mqtt.golang v1.4.3 h1:2kwcUGn8seMUfWndX0hGbvH8r7crgcJguQNCyp70xik=
github.com/eclipse/paho.mqtt.golang v1.4.3/go.mod h1:CSYvoAlsMkhYOXh/oKyxa8EcBci6dVkLCbo5tTC1RIE=
github.com/google/uuid v1.4.0 h1:MtMxsa51/r9yyhkyLsVeVt0B+BGQZzpQiTQ4eHZ8bc4=
github.com/google/uuid v1.4.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk=
github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
github.com/jackc/pgx/v5 v5.5.5 h1:amBjrZVmksIdNjxGW/IiIMzxMKZFelXbUoPNb+8sjQw=
github.com/jackc/pgx/v5 v5.5.5/go.mod h1:ez9gk+OAat140fv9ErkZDYFWmXLfV+++K0uAOiwgm1A=
github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk=
github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E=
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/eclipse/paho.mqtt.golang v1.5.0 h1:EH+bUVJNgttidWFkLLVKaQPGmkTUfQQqjOsyvMGvD6o=
github.com/eclipse/paho.mqtt.golang v1.5.0/go.mod h1:du/2qNQVqJf/Sqs4MEL77kR8QTqANF7XU7Fk0aOTAgk=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
github.com/influxdata/influxdb1-client v0.0.0-20220302092344-a9ab5670611c h1:qSHzRbhzK8RdXOsAdfDgO49TtqC1oZ+acxPrkfTxcCs=
github.com/influxdata/influxdb1-client v0.0.0-20220302092344-a9ab5670611c/go.mod h1:qj24IKcXYK6Iy9ceXlo3Tc+vtHo9lIhSX5JddghvEPo=
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=
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
golang.org/x/crypto v0.32.0 h1:euUpcYgM8WcP71gNpTqQCn6rC2t6ULUPiOzfWaXVVfc=
golang.org/x/crypto v0.32.0/go.mod h1:ZnnJkOaASj8g0AjIduWNlq2NRxL0PlBrbKVyZ6V/Ugc=
golang.org/x/net v0.34.0 h1:Mb7Mrk043xzHgnRM88suvJFwzVrRfHEHJEl5/71CKw0=
golang.org/x/net v0.34.0/go.mod h1:di0qlW3YNM5oh6GqDGQr92MyTozJPmybPK4Ev/Gm31k=
golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gorm.io/driver/postgres v1.5.9 h1:DkegyItji119OlcaLjqN11kHoUgZ/j13E0jkJZgD6A8=
gorm.io/driver/postgres v1.5.9/go.mod h1:DX3GReXH+3FPWGrrgffdvCk3DQ1dwDPdmbenSkweRGI=
gorm.io/gorm v1.25.11 h1:/Wfyg1B/je1hnDx3sMkX+gAlxrlZpn6X0BXRlwXlvHg=
gorm.io/gorm v1.25.11/go.mod h1:xh7N7RHfYlNc5EmcI/El95gXusucDrQnHXe0+CgWcLQ=

View File

@@ -1,15 +0,0 @@
package gs361ah04
type Observation struct {
LinkQuality uint8 `unit:"" json:"linkquality"`
Battery uint8 `unit:"%" json:"battery"`
AwayMode string `unit:"" json:"away_mode"`
ChildLock string `unit:"" json:"child_lock"`
CurrentHeatingSetpoint float32 `unit:"°C" json:"current_heating_setpoint"`
LocalTemperature float32 `unit:"°C" json:"local_temperature"`
Preset string `unit:"" json:"preset"`
SystemMode string `unit:"" json:"system_mode"`
ValveDetection string `unit:"" json:"valve_detection"`
WindowDetection string `unit:"" json:"window_detection"`
}

View File

@@ -1,11 +0,0 @@
package wsdcgq01lm
type Observation struct {
LinkQuality uint8 `unit:"" json:"linkquality"`
Battery uint8 `unit:"%" json:"battery"`
Humidity float32 `unit:"%H" json:"humidity"`
Pressure float32 `unit:"mbar" json:"pressure"`
Temperature float32 `unit:"°C" json:"temperature"`
Voltage uint16 `unit:"mV" json:"voltage"`
}

View File

@@ -1,10 +0,0 @@
package wsdcgq11lm
type Observation struct {
LinkQuality uint8 `unit:"" json:"linkquality"`
Battery uint8 `unit:"%" json:"battery"`
Humidity float32 `unit:"%H" json:"humidity"`
Temperature float32 `unit:"°C" json:"temperature"`
Voltage uint16 `unit:"mV" json:"voltage"`
}

View File

@@ -1,113 +1,80 @@
package z2m
import (
"fmt"
"log"
"time"
"strings"
"reflect"
"encoding/json"
"udi/config"
"udi/handlers/handler"
"udi/database"
"udi/handlers/z2m/models/wsdcgq11lm"
"udi/handlers/z2m/models/wsdcgq01lm"
"udi/handlers/z2m/models/gs361ah04"
"encoding/json"
"fmt"
"log"
"strings"
"time"
"udi/config"
"udi/database"
"udi/handlers/handler"
)
type Z2MHandler struct {
handler.CommonHandler
dbh *database.DatabaseHandle
}
func parse(T any, payload string, variables *map[string]database.VariableType) error {
observationType := reflect.TypeOf(T)
observation := reflect.New(observationType).Interface()
err := json.Unmarshal([]byte(payload), observation)
if err != nil {
return fmt.Errorf("Unable to parse payload into Observation struct: %v, %s", err, payload)
}
observationValue := reflect.ValueOf(observation).Elem()
for i := 0; i < observationType.NumField(); i++ {
field := observationType.Field(i)
name := field.Name
unit := field.Tag.Get("unit")
value := observationValue.Field(i).Interface()
(*variables)[name] = database.VariableType {
Label: name,
Variable: "y",
Unit: unit,
Value: value,
}
}
return nil
handler.CommonHandler
dbh *database.DatabaseHandle
}
func New(id string, config config.HandlerConfigT) handler.Handler {
t := &Z2MHandler {
}
t.Id = id
t.dbh = database.NewDatabaseHandle()
log.Printf("Handler Z2M %d initialized", id)
return t
t := &Z2MHandler{}
t.Id = id
t.dbh = database.NewDatabaseHandle()
log.Printf("Handler Z2M %d initialized", id)
return t
}
func (self *Z2MHandler) Handle(message handler.MessageT) {
log.Printf("Handler Z2M %d processing %s -> %s", self.Id, message.Topic, message.Payload)
log.Printf("Handler Z2M %d processing %s -> %s", self.Id, message.Topic, message.Payload)
var measurement database.Measurement
measurement.Time = time.Now()
var measurement database.Measurement
measurement.Time = time.Now()
subTopics := strings.Split(message.Topic, "/")
deviceId := subTopics[1]
log.Printf("DeviceId: %s", deviceId)
subTopics := strings.Split(message.Topic, "/")
deviceId := subTopics[1]
log.Printf("DeviceId: %s", deviceId)
device, err1 := self.dbh.GetDeviceByLabel(deviceId)
if err1 != nil {
self.Lost("Error when loading device", err1, message)
return
}
log.Printf("Device: %s", device)
measurement.Device = deviceId
measurement.Application = device.Application.Label
measurement.Device = device.Attributes["Label"].(string)
// Parse JSON direkt in eine map
var jsonData map[string]interface{}
err := json.Unmarshal([]byte(message.Payload), &jsonData)
if err != nil {
self.Lost("Failed to parse JSON payload", err, message)
return
}
var T any
switch device.DeviceType.ModelIdentifier {
case "WSDCGQ11LM":
T = wsdcgq11lm.Observation{}
case "WSDCGQ01LM":
T = wsdcgq01lm.Observation{}
case "GS361A-H04":
T = gs361ah04.Observation{}
default:
self.Lost(fmt.Sprintf("No parser found for %s", device.DeviceType.ModelIdentifier), nil, message)
return
}
measurement.Attributes = make(map[string]interface{})
measurement.Values = make(map[string]database.VariableType)
measurement.Values = make(map[string]database.VariableType)
measurement.Attributes = make(map[string]interface{})
err3 := parse(T,
message.Payload,
&(measurement.Values))
if err3 != nil {
self.Lost("Model parser failed", err3, message)
return
}
// Extract device info for application naming
if deviceData, ok := jsonData["device"]; ok {
if deviceMap, ok := deviceData.(map[string]any); ok {
manufacturerId, hasManufacturer := deviceMap["manufacturerID"]
model, hasModel := deviceMap["model"]
measurement.Attributes["Status"] = "ok"
measurement.Attributes["DeviceId"] = deviceId
measurement.Attributes["DeviceModel"] = device.DeviceType.ModelIdentifier
log.Printf("Prepared measurement item: %s", measurement)
self.dbh.StoreMeasurement(&measurement)
self.S()
if !hasManufacturer || !hasModel {
self.Lost("Missing manufacturerID or model in device data", fmt.Errorf("manufacturerID: %v, model: %v", hasManufacturer, hasModel), handler.MessageT{})
return
}
measurement.Application = "z2m_" + fmt.Sprintf("%v", manufacturerId) + "_" + model.(string)
}
delete(jsonData, "device")
}
// Konvertiere die restlichen Elemente in VariableType-Map
for key, value := range jsonData {
measurement.Values[key] = database.VariableType{
Label: key,
Variable: "",
Unit: "",
Value: value,
}
}
measurement.Attributes["Status"] = "ok"
log.Printf("Prepared measurement item: %s", measurement)
self.dbh.StoreMeasurement(&measurement)
self.S()
}

View File

@@ -1,18 +0,0 @@
package main
import "log"
import "udi/database"
func main() {
log.SetPrefix("UDI Migrate Schema: ")
log.SetFlags(log.Ldate | log.Ltime | log.Lshortfile)
log.Println("Starting")
database.Migrate()
log.Println("Done")
}