Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
01acf835a8
|
@@ -12,6 +12,40 @@ steps:
|
||||
- go build -a -installsuffix nocgo -o udi main.go
|
||||
- cp udi ../..
|
||||
|
||||
scan:
|
||||
image: quay.io/wollud1969/woodpecker-helper:0.5.1
|
||||
environment:
|
||||
TRIVY_TOKEN:
|
||||
from_secret: trivy_token
|
||||
TRIVY_URL:
|
||||
from_secret: trivy_url
|
||||
DTRACK_API_KEY:
|
||||
from_secret: dtrack_api_key
|
||||
DTRACK_API_URL:
|
||||
from_secret: dtrack_api_url
|
||||
commands:
|
||||
- export GOPATH=/woodpecker/go # the export is required, otherwise trivy will not consider the variable
|
||||
- HOME=/home/`id -nu`
|
||||
- TAG="${CI_COMMIT_TAG:-$CI_COMMIT_SHA}"
|
||||
- |
|
||||
trivy fs \
|
||||
--server $TRIVY_URL \
|
||||
--token $TRIVY_TOKEN \
|
||||
--format cyclonedx \
|
||||
--scanners license \
|
||||
--output /tmp/sbom.xml \
|
||||
.
|
||||
- cat /tmp/sbom.xml
|
||||
- |
|
||||
curl -X "POST" \
|
||||
-H "Content-Type: multipart/form-data" \
|
||||
-H "X-Api-Key: $DTRACK_API_KEY" \
|
||||
-F "autoCreate=true" \
|
||||
-F "projectName=$CI_REPO" \
|
||||
-F "projectVersion=$TAG" \
|
||||
-F "bom=@/tmp/sbom.xml"\
|
||||
"$DTRACK_API_URL/api/v1/bom"
|
||||
|
||||
dockerize:
|
||||
image: plugins/kaniko
|
||||
settings:
|
||||
|
||||
44
deployment/decrypt-secrets.sh
Executable file
44
deployment/decrypt-secrets.sh
Executable file
@@ -0,0 +1,44 @@
|
||||
#!/bin/bash
|
||||
|
||||
if [ "$ENCRYPTION_KEY" = "" ]; then
|
||||
echo "ENCRYPTION_KEY not set"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
if [ "$MD5_CHECKSUM" = "" ]; then
|
||||
echo "No checksum given"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
|
||||
SECRETS_CIPHERTEXT_FILE=secrets.enc
|
||||
SECRETS_PLAINTEXT_FILE=/tmp/secrets
|
||||
TMP_FILE=`mktemp`
|
||||
POD_NAME_SUFFIX=`date +%s`
|
||||
|
||||
cat $SECRETS_CIPHERTEXT_FILE | \
|
||||
kubectl run openssl-$POD_NAME_SUFFIX \
|
||||
--rm \
|
||||
--image bitnami/debian-base-buildpack:latest \
|
||||
--env KEY=$ENCRYPTION_KEY \
|
||||
-i \
|
||||
-q \
|
||||
-- \
|
||||
/bin/sh -c "openssl enc -aes-256-cbc -salt -pass env:KEY -a -d" > \
|
||||
$TMP_FILE
|
||||
|
||||
if [ `uname` = "Darwin" ]; then
|
||||
CALCULATED_CHECKSUM=`cat $TMP_FILE | md5`
|
||||
elif [ `uname` = "Linux" ]; then
|
||||
CALCULATED_CHECKSUM=`cat $TMP_FILE | md5sum - | awk '{print $1}'`
|
||||
fi
|
||||
|
||||
if [ "$MD5_CHECKSUM" != "$CALCULATED_CHECKSUM" ]; then
|
||||
echo "Invalid checksum"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
#cat $TMP_FILE
|
||||
mv $TMP_FILE $SECRETS_PLAINTEXT_FILE
|
||||
|
||||
|
||||
@@ -17,7 +17,7 @@ metadata:
|
||||
labels:
|
||||
app: udi
|
||||
annotations:
|
||||
secret.reloader.stakater.com/reload: "%PRE%-udi-conf,%PRE%-udi-db-cred,%PRE%-mqtt-password,%PRE%-udi-influxdb-cred"
|
||||
secret.reloader.stakater.com/reload: "%PRE%-udi-conf,%PRE%-udi-db-cred,%PRE%-mqtt-password"
|
||||
spec:
|
||||
replicas: 1
|
||||
selector:
|
||||
@@ -36,8 +36,6 @@ spec:
|
||||
name: %PRE%-udi-db-cred
|
||||
- secretRef:
|
||||
name: %PRE%-mqtt-password
|
||||
- secretRef:
|
||||
name: %PRE%-udi-influxdb-cred
|
||||
- configMapRef:
|
||||
name: %PRE%-udi-conf
|
||||
volumeMounts:
|
||||
|
||||
@@ -67,7 +67,6 @@ for NAMESPACE_DIR in `find $INSTANCES_DIR -type d -mindepth 1 -maxdepth 1`; do
|
||||
DATABASE_VARIABLE=$VARIABLE_PREFIX"_PGDATABASE"
|
||||
NEW_UDI_DB_DATABASE="${!DATABASE_VARIABLE}"
|
||||
NEW_UDI_DB_HOST=database.database1.svc.cluster.local
|
||||
INFLUXDB_URL=$VARIABLE_PREFIX"_INFLUXDB_URL"
|
||||
|
||||
kubectl create secret generic $INSTANCE-udi-db-cred \
|
||||
--dry-run=client \
|
||||
@@ -80,13 +79,6 @@ for NAMESPACE_DIR in `find $INSTANCES_DIR -type d -mindepth 1 -maxdepth 1`; do
|
||||
--from-literal=PGSSLMODE="require" | \
|
||||
kubectl apply -f - -n $NAMESPACE
|
||||
|
||||
kubectl create secret generic $INSTANCE-udi-influxdb-cred \
|
||||
--dry-run=client \
|
||||
-o yaml \
|
||||
--save-config \
|
||||
--from-literal=INFLUXDB_URL="${!INFLUXDB_URL}" | \
|
||||
kubectl apply -f - -n $NAMESPACE
|
||||
|
||||
# set configuration as configMap
|
||||
kubectl create configmap $INSTANCE-udi-conf \
|
||||
--from-literal=UDI_CONF="`cat $CONFIG_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"
|
||||
}
|
||||
}
|
||||
@@ -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"
|
||||
}
|
||||
}
|
||||
27
deployment/encrypt-secrets.sh
Executable file
27
deployment/encrypt-secrets.sh
Executable file
@@ -0,0 +1,27 @@
|
||||
#!/bin/bash
|
||||
|
||||
ENCRYPTION_KEY=`openssl rand -hex 32`
|
||||
echo $ENCRYPTION_KEY
|
||||
|
||||
SECRETS_PLAINTEXT_FILE=secrets.txt
|
||||
SECRETS_CIPHERTEXT_FILE=secrets.enc
|
||||
|
||||
if [ `uname` = "Darwin" ]; then
|
||||
cat $SECRETS_PLAINTEXT_FILE | md5
|
||||
elif [ `uname` = "Linux" ]; then
|
||||
cat $SECRETS_PLAINTEXT_FILE | md5sum - | awk '{print $1}'
|
||||
fi
|
||||
|
||||
POD_NAME_SUFFIX=`date +%s`
|
||||
|
||||
cat $SECRETS_PLAINTEXT_FILE | \
|
||||
kubectl run openssl-$POD_NAME_SUFFIX \
|
||||
--rm \
|
||||
--image bitnami/debian-base-buildpack:latest \
|
||||
--env KEY=$ENCRYPTION_KEY \
|
||||
-i \
|
||||
-q \
|
||||
-- \
|
||||
/bin/sh -c "openssl enc -aes-256-cbc -salt -pass env:KEY -a" > \
|
||||
$SECRETS_CIPHERTEXT_FILE
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"mqtt": {
|
||||
"broker": "mqtt://mosquitto-broker-mqtt-anon-cluster.mosquitto.svc.cluster.local:1883",
|
||||
"broker": "mqtt://emqx01-anonymous-cluster-internal.broker.svc.cluster.local:1883",
|
||||
"tlsEnable": "false"
|
||||
},
|
||||
"topicMappings": [
|
||||
@@ -105,7 +105,6 @@
|
||||
"devicePart": "3",
|
||||
"valueFrom": "payload",
|
||||
"valuePart": "1",
|
||||
"valueType": "float",
|
||||
"unitFrom": "payload",
|
||||
"unitPart": "3"
|
||||
}
|
||||
@@ -166,7 +165,7 @@
|
||||
{
|
||||
"topics": [ "shellyplusht/+/status/humidity:0" ],
|
||||
"handler": "SVEJ",
|
||||
"id": "SVEJ3",
|
||||
"id": "SVE4",
|
||||
"config": {
|
||||
"databaseConnStr": "",
|
||||
"attributes": {
|
||||
@@ -176,48 +175,6 @@
|
||||
"unitSelector": "C:%"
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"topics": [ "shellies/sensor/+/status/temperature:0" ],
|
||||
"handler": "SVEJ",
|
||||
"id": "SVEJ4",
|
||||
"config": {
|
||||
"databaseConnStr": "",
|
||||
"attributes": {
|
||||
"application": "Shellies Sensor Temperature",
|
||||
"deviceSelector": "T:2",
|
||||
"valueSelector": "J:$.tC",
|
||||
"unitSelector": "C:°C"
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"topics": [ "shellies/sensor/+/status/humidity:0" ],
|
||||
"handler": "SVEJ",
|
||||
"id": "SVEJ5",
|
||||
"config": {
|
||||
"databaseConnStr": "",
|
||||
"attributes": {
|
||||
"application": "Shellies Sensor Humidity",
|
||||
"deviceSelector": "T:2",
|
||||
"valueSelector": "J:$.rh",
|
||||
"unitSelector": "C:%"
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"topics": [ "shellies/sensor/+/status/devicepower:0" ],
|
||||
"handler": "SVEJ",
|
||||
"id": "SVEJ6",
|
||||
"config": {
|
||||
"databaseConnStr": "",
|
||||
"attributes": {
|
||||
"application": "Shellies Sensor Power",
|
||||
"deviceSelector": "T:2",
|
||||
"valueSelector": "J:$.battery.percent",
|
||||
"unitSelector": "C:%"
|
||||
}
|
||||
}
|
||||
}
|
||||
],
|
||||
"archiver": {
|
||||
@@ -1,8 +1,20 @@
|
||||
-----BEGIN PGP MESSAGE-----
|
||||
|
||||
jA0ECQMIYUoTHR96Qfb90psBoxuk38UXPXTWPCmdW690bi2+w34S4NLHZvHfe3Ra
|
||||
nck319+PXvr0agfHGZ733hhTQv4sa8I2o6ICrgFqtKGfHmgnqL5kYNP9+NuV/IsF
|
||||
x3dxwjEejsZ5GYn/zk+CQceItQ8nyyJc2ms1KwTu2r4hMzuHmnVtvKxNCzPrw2N5
|
||||
SJIRhh41eequFkzELQqqXXu10raBFsttOemVhA==
|
||||
=TySu
|
||||
jA0ECQMIetkwUzO5q5z+0ukB9IS9Z6rRbXzGHS4dOfwLMbyx2auDu7PYaAgHLHkI
|
||||
25aul+Q2RMUc1WtWgtyIPyVrq84ctjx4AmnXU3Gbsqte6QJIk0RX8w886gTI+NeF
|
||||
RJahlDNeL9LLltQInkFPDXgLC0u/8a+az0aVZUnfAHVhcYkkgQ0JZasc0VQjLqHV
|
||||
dQ3sHmGj1sUC712AujE1f+T5J9CImBdK4r4sZeFq0c3km7hIpOJNkg8LX+0fj9Un
|
||||
/LPPg0Qjq15dnb44UB6g+gsCUCJqTylGAMH4HLvsgRLbs4T4Za0O9lUKZ7UzuvlZ
|
||||
MnbJLIyJt2zxBVQ6NeuchvTmb8Ppkrk2qrO2mXC4hB0YyGtmT3Dy178pNdy7mYR6
|
||||
zLPx7wO+x4MOJOY8uzclHLucAfBhLTgTFCrje+268e99bXuFhAXOA2zqYSyhBI5M
|
||||
XnHUr14XsTACy8CwZRoIpvy9Roe0N3pJq/kdkx/zJ3ieBUWazR5HYFtBeixcELHa
|
||||
EIH9vZCqGi4NfatsZGYyGofVQhV9TnV5yyAruBDsdb0JIYIz1rznzW4megwR6O0K
|
||||
I7W4cTtzT6XdUZQGyWHZO+FSEQ3zoFxLPWeVWj4MJSj0qvd/G2Z07tAw8MJueGRK
|
||||
DsQJOwhrZVH79zuQE+zF+TR3K7A9O3pq0C618gXzXYnVIRoxYqAGo5RyPGlswxRS
|
||||
3sAf8spYQ3iSvo2whpRJ1c7ytmGKPbA3+4YXmJLemCCxiJK3yNJya6xg/hVTeJlL
|
||||
w76+IPGx+VwfURNt+JDf3Xfq2pnppDjW59a69FMraiWCovSl6cUGHr9X+mgQ+C39
|
||||
7OUhKLHQP4s+Yeak5St8/MKKG57bC1k4j07sh9pCXiAPPST3gVhSDmAGPGUDdISd
|
||||
ieOw/ofwzATB+qgEyhPWcRcR/RU0fYwz2q0AxdI6brLgxgGPO6gvQiulY0+MKtuZ
|
||||
kYUnBJiOQ4PBqnPDhtUjGvjT8YAbadWj+pH44bCGd7Q0Kg==
|
||||
=KCiQ
|
||||
-----END PGP MESSAGE-----
|
||||
|
||||
@@ -1,145 +0,0 @@
|
||||
{
|
||||
"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"
|
||||
}
|
||||
}
|
||||
@@ -5,19 +5,23 @@
|
||||
},
|
||||
"topicMappings": [
|
||||
{
|
||||
"topics": [ "IoT/PV/Values" ],
|
||||
"handler": "PV",
|
||||
"id": "PV",
|
||||
"topics": [ "NR/Multisensor/+/Temperatur" ],
|
||||
"handler": "SVEJ",
|
||||
"id": "SVEJ0",
|
||||
"config": {
|
||||
"databaseConnStr": "",
|
||||
"attributes": {
|
||||
"application": "Temperature Multisensor",
|
||||
"deviceSelector": "T:2",
|
||||
"valueSelector": "J:$.CurrentTemperature",
|
||||
"unitSelector": "C:°C"
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"topics": [ "IoT/Car/Values" ],
|
||||
"handler": "Car",
|
||||
"id": "Car",
|
||||
"topics": [ "zigbee2mqtt/+" ],
|
||||
"handler": "Z2M",
|
||||
"id": "Z2M",
|
||||
"config": {
|
||||
"databaseConnStr": "",
|
||||
"attributes": {
|
||||
|
||||
@@ -1,38 +1,47 @@
|
||||
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
|
||||
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{}
|
||||
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"`
|
||||
}
|
||||
|
||||
type Application struct {
|
||||
Label string
|
||||
Attributes map[string]interface{}
|
||||
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"`
|
||||
}
|
||||
|
||||
type Device struct {
|
||||
Label string
|
||||
Application Application
|
||||
DeviceType DeviceType
|
||||
Attributes map[string]interface{}
|
||||
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"`
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -1,155 +1,96 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"udi/counter"
|
||||
|
||||
influxdb "github.com/influxdata/influxdb1-client/v2"
|
||||
import (
|
||||
"log"
|
||||
//"time"
|
||||
"fmt"
|
||||
"udi/counter"
|
||||
"gorm.io/driver/postgres"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
type DatabaseHandle struct {
|
||||
initialized bool
|
||||
client influxdb.Client
|
||||
database string
|
||||
initialized bool
|
||||
dbh *gorm.DB
|
||||
}
|
||||
|
||||
func NewDatabaseHandle() *DatabaseHandle {
|
||||
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 (config: URL: %s, Username: %s, Password: %s): %s", influxDB, username, password, 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
|
||||
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
|
||||
}
|
||||
|
||||
func (self *DatabaseHandle) StoreMeasurement(measurement *Measurement) {
|
||||
if !self.initialized {
|
||||
log.Printf("Database connection not initialized, can not store, measurement %v lost", measurement)
|
||||
counter.F("Stored")
|
||||
return
|
||||
}
|
||||
if ! self.initialized {
|
||||
log.Printf("Database connection not initialized, can not store, measurement %s lost", measurement)
|
||||
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
|
||||
}
|
||||
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
|
||||
}
|
||||
|
||||
// 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")
|
||||
log.Println("Successfully stored measurement")
|
||||
counter.S("Stored")
|
||||
}
|
||||
|
||||
func (self *DatabaseHandle) Close() {
|
||||
if self.initialized && self.client != nil {
|
||||
self.client.Close()
|
||||
log.Println("InfluxDB connection closed")
|
||||
}
|
||||
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) 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
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
55
src/udi/database/migrate_database.go
Normal file
55
src/udi/database/migrate_database.go
Normal file
@@ -0,0 +1,55 @@
|
||||
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)
|
||||
*/
|
||||
}
|
||||
|
||||
41
src/udi/database/schema.sql
Normal file
41
src/udi/database/schema.sql
Normal file
@@ -0,0 +1,41 @@
|
||||
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');
|
||||
|
||||
|
||||
|
||||
@@ -18,8 +18,7 @@ import (
|
||||
"udi/handlers/pv"
|
||||
"udi/handlers/svej"
|
||||
"udi/handlers/sver"
|
||||
|
||||
// "udi/handlers/ttn"
|
||||
"udi/handlers/ttn"
|
||||
"udi/handlers/z2m"
|
||||
"udi/mqtt"
|
||||
)
|
||||
@@ -36,8 +35,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":
|
||||
|
||||
@@ -5,12 +5,21 @@ 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
|
||||
)
|
||||
|
||||
@@ -1,14 +1,46 @@
|
||||
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/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/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=
|
||||
|
||||
@@ -1,106 +1,91 @@
|
||||
package mbgw3
|
||||
|
||||
import (
|
||||
"time"
|
||||
"log"
|
||||
"strconv"
|
||||
"encoding/json"
|
||||
"udi/config"
|
||||
"udi/handlers/handler"
|
||||
"udi/database"
|
||||
"encoding/json"
|
||||
"log"
|
||||
"strconv"
|
||||
"time"
|
||||
"udi/config"
|
||||
"udi/database"
|
||||
"udi/handlers/handler"
|
||||
)
|
||||
|
||||
|
||||
type Mbgw3Handler struct {
|
||||
handler.CommonHandler
|
||||
dbh *database.DatabaseHandle
|
||||
handler.CommonHandler
|
||||
dbh *database.DatabaseHandle
|
||||
}
|
||||
|
||||
type Observation struct {
|
||||
Status string `json:"Status"`
|
||||
RequestId string `json:"RequestId"`
|
||||
Device string `json:"Device"`
|
||||
Errors string `json:"Errors"`
|
||||
ErrorRatio string `json:"ErrorRatio"`
|
||||
Requests string `json:"Requests"`
|
||||
Values map[string]string
|
||||
Status string `json:"Status"`
|
||||
RequestId string `json:"RequestId"`
|
||||
Device string `json:"Device"`
|
||||
Errors string `json:"Errors"`
|
||||
ErrorRatio string `json:"ErrorRatio"`
|
||||
Requests string `json:"Requests"`
|
||||
Values map[string]string
|
||||
}
|
||||
|
||||
|
||||
func New(id string, config config.HandlerConfigT) handler.Handler {
|
||||
t := &Mbgw3Handler {
|
||||
}
|
||||
t.Id = id
|
||||
t.dbh = database.NewDatabaseHandle()
|
||||
log.Printf("Handler MBGW3 %d initialized", id)
|
||||
return t
|
||||
t := &Mbgw3Handler{}
|
||||
t.Id = id
|
||||
t.dbh = database.NewDatabaseHandle()
|
||||
log.Printf("Handler MBGW3 %d initialized", id)
|
||||
return t
|
||||
}
|
||||
|
||||
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
|
||||
err := json.Unmarshal([]byte(message.Payload), &observation)
|
||||
if err != nil {
|
||||
self.Lost("Unable to parse payload into Observation struct", err, message)
|
||||
return
|
||||
}
|
||||
var observation Observation
|
||||
err := json.Unmarshal([]byte(message.Payload), &observation)
|
||||
if err != nil {
|
||||
self.Lost("Unable to parse payload into Observation struct", err, message)
|
||||
return
|
||||
}
|
||||
|
||||
var measurement database.Measurement
|
||||
measurement.Time = time.Now()
|
||||
var measurement database.Measurement
|
||||
measurement.Time = time.Now()
|
||||
|
||||
if observation.Device == "Gas" {
|
||||
measurement.Application = "Gas"
|
||||
measurement.Device = "Gasmeter"
|
||||
} else {
|
||||
measurement.Application = "Power"
|
||||
measurement.Device = observation.Device
|
||||
}
|
||||
if observation.Device == "Gas" {
|
||||
measurement.Application = "Gas"
|
||||
measurement.Device = "Gasmeter"
|
||||
} else {
|
||||
measurement.Application = "Power"
|
||||
measurement.Device = observation.Device
|
||||
}
|
||||
|
||||
measurement.Attributes = make(map[string]interface{})
|
||||
if v, err := strconv.Atoi(observation.RequestId); err == nil {
|
||||
measurement.Attributes["RequestId"] = v
|
||||
}
|
||||
if v, err := strconv.ParseFloat(observation.ErrorRatio, 32); err == nil {
|
||||
measurement.Attributes["ErrorRatio"] = v
|
||||
}
|
||||
if v, err := strconv.Atoi(observation.Errors); err == nil {
|
||||
measurement.Attributes["Errors"] = v
|
||||
}
|
||||
if v, err := strconv.Atoi(observation.Requests); err == nil {
|
||||
measurement.Attributes["Requests"] = v
|
||||
}
|
||||
measurement.Attributes["Status"] = observation.Status
|
||||
measurement.Attributes = make(map[string]interface{})
|
||||
if v, err := strconv.Atoi(observation.RequestId); err == nil {
|
||||
measurement.Attributes["RequestId"] = v
|
||||
}
|
||||
if v, err := strconv.ParseFloat(observation.ErrorRatio, 32); err == nil {
|
||||
measurement.Attributes["ErrorRatio"] = v
|
||||
}
|
||||
if v, err := strconv.Atoi(observation.Errors); err == nil {
|
||||
measurement.Attributes["Errors"] = v
|
||||
}
|
||||
if v, err := strconv.Atoi(observation.Requests); err == nil {
|
||||
measurement.Attributes["Requests"] = v
|
||||
}
|
||||
measurement.Attributes["Status"] = observation.Status
|
||||
|
||||
measurement.Values = make(map[string]database.VariableType)
|
||||
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"
|
||||
}
|
||||
|
||||
// Check if key already exists and create unique key if needed
|
||||
keyCount[k]++
|
||||
uniqueKey := k
|
||||
if keyCount[k] > 1 {
|
||||
uniqueKey = k + strconv.Itoa(keyCount[k])
|
||||
}
|
||||
|
||||
measurement.Values[uniqueKey] = database.VariableType {
|
||||
Label: "",
|
||||
Variable: k,
|
||||
Unit: unit,
|
||||
Value: v,
|
||||
}
|
||||
}
|
||||
measurement.Values = make(map[string]database.VariableType)
|
||||
unitMap := map[string]string{"Energy": "Wh", "Power": "W", "Voltage": "V", "Current": "A", "Volume": "m3"}
|
||||
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,
|
||||
}
|
||||
}
|
||||
|
||||
//log.Printf("Prepared measurement item: %s", measurement)
|
||||
//log.Printf("Prepared measurement item: %s", measurement)
|
||||
|
||||
self.dbh.StoreMeasurement(&measurement)
|
||||
self.S()
|
||||
self.dbh.StoreMeasurement(&measurement)
|
||||
self.S()
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -1,30 +1,29 @@
|
||||
package svej
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
"udi/config"
|
||||
"udi/database"
|
||||
"udi/handlers/handler"
|
||||
|
||||
"github.com/oliveagle/jsonpath"
|
||||
"log"
|
||||
"time"
|
||||
"strconv"
|
||||
"strings"
|
||||
"fmt"
|
||||
"github.com/oliveagle/jsonpath"
|
||||
"encoding/json"
|
||||
"udi/config"
|
||||
"udi/handlers/handler"
|
||||
"udi/database"
|
||||
)
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
/*
|
||||
@@ -34,131 +33,134 @@ 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"]
|
||||
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
|
||||
}
|
||||
|
||||
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.ready = true
|
||||
t.dbh = database.NewDatabaseHandle()
|
||||
log.Printf("Handler SVEJ %s initialized", id)
|
||||
return t
|
||||
t.Id = id
|
||||
t.ready = true
|
||||
t.dbh = database.NewDatabaseHandle()
|
||||
log.Printf("Handler SVEJ %d initialized", id)
|
||||
return t
|
||||
}
|
||||
|
||||
func (self *SingleValueExtractorJsonpathHandler) ExtractionHelper(subTopics []string, jPayload interface{}, selector string, jp *jsonpath.Compiled) (interface{}, error) {
|
||||
var res interface{}
|
||||
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 = 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) (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) 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
|
||||
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
|
||||
}
|
||||
|
||||
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)
|
||||
if err1 != nil {
|
||||
self.Lost("Device extraction failed", err1, message)
|
||||
return
|
||||
}
|
||||
log.Printf("device: %s", device)
|
||||
|
||||
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)
|
||||
value, err2 := self.ExtractionHelper(subTopics, jPayload, self.valueSelector, self.valueJsonpath)
|
||||
if err2 != nil {
|
||||
self.Lost("Value extraction failed", err2, message)
|
||||
return
|
||||
}
|
||||
|
||||
value, err2 := self.ExtractionHelper(subTopics, jPayload, self.valueSelector, self.valueJsonpath)
|
||||
if err2 != nil {
|
||||
self.Lost("Value extraction failed", err2, message)
|
||||
return
|
||||
}
|
||||
unit, err3 := self.ExtractionHelper(subTopics, jPayload, self.unitSelector, self.unitJsonpath)
|
||||
if err3 != nil {
|
||||
self.Lost("Unit extraction failed", err3, message)
|
||||
return
|
||||
}
|
||||
|
||||
unit, err3 := self.ExtractionHelper(subTopics, jPayload, self.unitSelector, self.unitJsonpath)
|
||||
if err3 != nil {
|
||||
self.Lost("Unit extraction failed", err3, message)
|
||||
return
|
||||
}
|
||||
measurement.Device = device
|
||||
|
||||
measurement.Device = device.(string)
|
||||
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
|
||||
|
||||
var variable database.VariableType
|
||||
variable.Label = ""
|
||||
variable.Variable = ""
|
||||
variable.Unit = unit.(string)
|
||||
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()
|
||||
log.Printf("Prepared measurement item: %s", measurement)
|
||||
self.dbh.StoreMeasurement(&measurement)
|
||||
self.S()
|
||||
}
|
||||
|
||||
|
||||
@@ -1,22 +1,23 @@
|
||||
package sver
|
||||
|
||||
import (
|
||||
"log"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
"udi/config"
|
||||
"udi/database"
|
||||
"udi/handlers/handler"
|
||||
"time"
|
||||
"strconv"
|
||||
"strings"
|
||||
"regexp"
|
||||
"log"
|
||||
"udi/config"
|
||||
"udi/handlers/handler"
|
||||
"udi/database"
|
||||
)
|
||||
|
||||
|
||||
type SingleValueExtractorRegexHandler struct {
|
||||
handler.CommonHandler
|
||||
ready bool
|
||||
config localConfig
|
||||
payloadRegex *regexp.Regexp
|
||||
dbh *database.DatabaseHandle
|
||||
handler.CommonHandler
|
||||
ready bool
|
||||
config localConfig
|
||||
payloadRegex *regexp.Regexp
|
||||
dbh *database.DatabaseHandle
|
||||
}
|
||||
|
||||
const TOPIC_SEL = "topic"
|
||||
@@ -25,191 +26,175 @@ const PAYLOAD_FULL_SEL = "payload-full"
|
||||
const CONSTANT_SEL = "constant"
|
||||
|
||||
type localConfig struct {
|
||||
application string
|
||||
deviceFrom string
|
||||
devicePart int
|
||||
device string
|
||||
valueFrom string
|
||||
valuePart int
|
||||
valueType string
|
||||
unitFrom string
|
||||
unitPart int
|
||||
unit string
|
||||
application string
|
||||
deviceFrom string
|
||||
devicePart int
|
||||
device string
|
||||
valueFrom string
|
||||
valuePart int
|
||||
unitFrom string
|
||||
unitPart int
|
||||
unit string
|
||||
}
|
||||
|
||||
|
||||
func New(id string, config config.HandlerConfigT) handler.Handler {
|
||||
t := &SingleValueExtractorRegexHandler{
|
||||
ready: false,
|
||||
}
|
||||
t := &SingleValueExtractorRegexHandler {
|
||||
ready: false,
|
||||
}
|
||||
|
||||
var localConfig localConfig
|
||||
if config.Attributes["application"] == "" {
|
||||
log.Println("Error: application not configured")
|
||||
return t
|
||||
}
|
||||
localConfig.application = config.Attributes["application"]
|
||||
var localConfig localConfig
|
||||
if config.Attributes["application"] == "" {
|
||||
log.Println("Error: application not configured")
|
||||
return t
|
||||
}
|
||||
localConfig.application = config.Attributes["application"]
|
||||
|
||||
payloadRegex := config.Attributes["payloadRegex"]
|
||||
if payloadRegex != "" {
|
||||
t.payloadRegex = regexp.MustCompile(payloadRegex)
|
||||
} else {
|
||||
t.payloadRegex = nil
|
||||
}
|
||||
|
||||
payloadRegex := config.Attributes["payloadRegex"]
|
||||
if payloadRegex != "" {
|
||||
t.payloadRegex = regexp.MustCompile(payloadRegex)
|
||||
} else {
|
||||
t.payloadRegex = nil
|
||||
}
|
||||
if config.Attributes["deviceFrom"] != TOPIC_SEL && config.Attributes["deviceFrom"] != PAYLOAD_SEL && config.Attributes["deviceFrom"] != CONSTANT_SEL {
|
||||
log.Printf("Error: invalid value %s for deviceFrom", config.Attributes["deviceFrom"])
|
||||
return t
|
||||
}
|
||||
localConfig.deviceFrom = config.Attributes["deviceFrom"]
|
||||
|
||||
if config.Attributes["deviceFrom"] != TOPIC_SEL && config.Attributes["deviceFrom"] != PAYLOAD_SEL && config.Attributes["deviceFrom"] != CONSTANT_SEL {
|
||||
log.Printf("Error: invalid value %s for deviceFrom", config.Attributes["deviceFrom"])
|
||||
return t
|
||||
}
|
||||
localConfig.deviceFrom = config.Attributes["deviceFrom"]
|
||||
devicePart, err1 := strconv.Atoi(config.Attributes["devicePart"])
|
||||
if err1 != nil {
|
||||
log.Printf("Error: unable to convert devicePart to number: %s", err1)
|
||||
return t
|
||||
}
|
||||
localConfig.devicePart = devicePart
|
||||
|
||||
devicePart, err1 := strconv.Atoi(config.Attributes["devicePart"])
|
||||
if err1 != nil {
|
||||
log.Printf("Error: unable to convert devicePart to number: %s", err1)
|
||||
return t
|
||||
}
|
||||
localConfig.devicePart = devicePart
|
||||
// empty device is valid
|
||||
localConfig.device = config.Attributes["device"]
|
||||
|
||||
// empty device is valid
|
||||
localConfig.device = config.Attributes["device"]
|
||||
if config.Attributes["valueFrom"] != PAYLOAD_SEL && config.Attributes["valueFrom"] != PAYLOAD_FULL_SEL {
|
||||
log.Printf("Error: invalid value %s for valueFrom", config.Attributes["valueFrom"])
|
||||
return t
|
||||
}
|
||||
localConfig.valueFrom = config.Attributes["valueFrom"]
|
||||
|
||||
if config.Attributes["valueFrom"] != PAYLOAD_SEL && config.Attributes["valueFrom"] != PAYLOAD_FULL_SEL {
|
||||
log.Printf("Error: invalid value %s for valueFrom", config.Attributes["valueFrom"])
|
||||
return t
|
||||
}
|
||||
localConfig.valueFrom = config.Attributes["valueFrom"]
|
||||
if config.Attributes["valueFrom"] == PAYLOAD_SEL {
|
||||
valuePart, err2 := strconv.Atoi(config.Attributes["valuePart"])
|
||||
if err2 != nil {
|
||||
log.Printf("Error: unable to convert valuePart to number: %s", err2)
|
||||
return t
|
||||
}
|
||||
localConfig.valuePart = valuePart
|
||||
}
|
||||
|
||||
if config.Attributes["valueFrom"] == PAYLOAD_SEL {
|
||||
valuePart, err2 := strconv.Atoi(config.Attributes["valuePart"])
|
||||
if err2 != nil {
|
||||
log.Printf("Error: unable to convert valuePart to number: %s", err2)
|
||||
return t
|
||||
}
|
||||
localConfig.valuePart = valuePart
|
||||
}
|
||||
if config.Attributes["unitFrom"] != PAYLOAD_SEL && config.Attributes["unitFrom"] != CONSTANT_SEL {
|
||||
log.Printf("Error: invalid value %s for unitFrom", config.Attributes["unitFrom"])
|
||||
return t
|
||||
}
|
||||
localConfig.unitFrom = config.Attributes["unitFrom"]
|
||||
|
||||
if config.Attributes["valueType"] != "float" && config.Attributes["valueType"] != "string" {
|
||||
log.Printf("Error: invalid value %s for valueType", config.Attributes["valueType"])
|
||||
return t
|
||||
}
|
||||
localConfig.valueType = config.Attributes["valueType"]
|
||||
if config.Attributes["unitFrom"] == PAYLOAD_SEL {
|
||||
unitPart, err3 := strconv.Atoi(config.Attributes["unitPart"])
|
||||
if err3 != nil {
|
||||
log.Printf("Error: unable to convert unitPart to number: %s", err3)
|
||||
return t
|
||||
}
|
||||
localConfig.unitPart = unitPart
|
||||
}
|
||||
|
||||
if config.Attributes["unitFrom"] != PAYLOAD_SEL && config.Attributes["unitFrom"] != CONSTANT_SEL {
|
||||
log.Printf("Error: invalid value %s for unitFrom", config.Attributes["unitFrom"])
|
||||
return t
|
||||
}
|
||||
localConfig.unitFrom = config.Attributes["unitFrom"]
|
||||
// empty unit is valid
|
||||
localConfig.unit = config.Attributes["unit"]
|
||||
|
||||
if config.Attributes["unitFrom"] == PAYLOAD_SEL {
|
||||
unitPart, err3 := strconv.Atoi(config.Attributes["unitPart"])
|
||||
if err3 != nil {
|
||||
log.Printf("Error: unable to convert unitPart to number: %s", err3)
|
||||
return t
|
||||
}
|
||||
localConfig.unitPart = unitPart
|
||||
}
|
||||
t.config = localConfig
|
||||
|
||||
// empty unit is valid
|
||||
localConfig.unit = config.Attributes["unit"]
|
||||
|
||||
t.config = localConfig
|
||||
|
||||
t.Id = id
|
||||
t.ready = true
|
||||
t.dbh = database.NewDatabaseHandle()
|
||||
log.Printf("Handler SVER %d initialized", id)
|
||||
return t
|
||||
t.Id = id
|
||||
t.ready = true
|
||||
t.dbh = database.NewDatabaseHandle()
|
||||
log.Printf("Handler SVER %d initialized", id)
|
||||
return t
|
||||
}
|
||||
|
||||
func (self *SingleValueExtractorRegexHandler) Handle(message handler.MessageT) {
|
||||
if !self.ready {
|
||||
self.Lost("Handler is not marked as ready", nil, message)
|
||||
return
|
||||
}
|
||||
//log.Printf("Handler SingleValueExtractor %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 SingleValueExtractor %d processing %s -> %s", self.id, message.Topic, message.Payload)
|
||||
|
||||
var measurement database.Measurement
|
||||
measurement.Time = time.Now()
|
||||
measurement.Application = self.config.application
|
||||
var measurement database.Measurement
|
||||
measurement.Time = time.Now()
|
||||
measurement.Application = self.config.application
|
||||
|
||||
subTopics := strings.Split(message.Topic, "/")
|
||||
//log.Printf("Subtopics: %s", strings.Join(subTopics, ", "))
|
||||
|
||||
subTopics := strings.Split(message.Topic, "/")
|
||||
//log.Printf("Subtopics: %s", strings.Join(subTopics, ", "))
|
||||
var payloadMatches []string
|
||||
if self.payloadRegex != nil {
|
||||
payloadMatches = self.payloadRegex.FindStringSubmatch(message.Payload)
|
||||
//log.Printf("Matches: %s", strings.Join(payloadMatches, ", "))
|
||||
}
|
||||
|
||||
var payloadMatches []string
|
||||
if self.payloadRegex != nil {
|
||||
payloadMatches = self.payloadRegex.FindStringSubmatch(message.Payload)
|
||||
//log.Printf("Matches: %s", strings.Join(payloadMatches, ", "))
|
||||
}
|
||||
switch self.config.deviceFrom {
|
||||
case TOPIC_SEL:
|
||||
if self.config.devicePart >= len(subTopics) {
|
||||
self.Lost("devicePart out of range", nil, message)
|
||||
return
|
||||
}
|
||||
measurement.Device = subTopics[self.config.devicePart]
|
||||
case PAYLOAD_SEL:
|
||||
if self.payloadRegex == nil {
|
||||
self.Lost("no payloadRegex defined, devicePart can't be used", nil, message)
|
||||
return
|
||||
}
|
||||
if self.config.devicePart >= len(payloadMatches) {
|
||||
self.Lost("devicePart out of range", nil, message)
|
||||
return
|
||||
}
|
||||
measurement.Device = payloadMatches[self.config.devicePart]
|
||||
case CONSTANT_SEL:
|
||||
measurement.Device = self.config.device
|
||||
}
|
||||
|
||||
switch self.config.deviceFrom {
|
||||
case TOPIC_SEL:
|
||||
if self.config.devicePart >= len(subTopics) {
|
||||
self.Lost("devicePart out of range", nil, message)
|
||||
return
|
||||
}
|
||||
measurement.Device = subTopics[self.config.devicePart]
|
||||
case PAYLOAD_SEL:
|
||||
if self.payloadRegex == nil {
|
||||
self.Lost("no payloadRegex defined, devicePart can't be used", nil, message)
|
||||
return
|
||||
}
|
||||
if self.config.devicePart >= len(payloadMatches) {
|
||||
self.Lost("devicePart out of range", nil, message)
|
||||
return
|
||||
}
|
||||
measurement.Device = payloadMatches[self.config.devicePart]
|
||||
case CONSTANT_SEL:
|
||||
measurement.Device = self.config.device
|
||||
}
|
||||
measurement.Values = make(map[string]database.VariableType)
|
||||
var variable database.VariableType
|
||||
variable.Label = ""
|
||||
variable.Variable = ""
|
||||
|
||||
measurement.Values = make(map[string]database.VariableType)
|
||||
var variable database.VariableType
|
||||
variable.Label = ""
|
||||
variable.Variable = ""
|
||||
switch self.config.valueFrom {
|
||||
case PAYLOAD_SEL:
|
||||
if self.payloadRegex == nil {
|
||||
self.Lost("no payloadRegex defined, valuePart can't be used", nil, message)
|
||||
return
|
||||
}
|
||||
if self.config.valuePart >= len(payloadMatches) {
|
||||
self.Lost("valuePart out of range", nil, message)
|
||||
return
|
||||
}
|
||||
variable.Value = payloadMatches[self.config.valuePart]
|
||||
case PAYLOAD_FULL_SEL:
|
||||
variable.Value = message.Payload
|
||||
}
|
||||
|
||||
var value string
|
||||
switch self.config.valueFrom {
|
||||
case PAYLOAD_SEL:
|
||||
if self.payloadRegex == nil {
|
||||
self.Lost("no payloadRegex defined, valuePart can't be used", nil, message)
|
||||
return
|
||||
}
|
||||
if self.config.valuePart >= len(payloadMatches) {
|
||||
self.Lost("valuePart out of range", nil, message)
|
||||
return
|
||||
}
|
||||
value = payloadMatches[self.config.valuePart]
|
||||
case PAYLOAD_FULL_SEL:
|
||||
value = message.Payload
|
||||
}
|
||||
if self.config.valueType == "float" {
|
||||
fValue, err := strconv.ParseFloat(value, 64)
|
||||
if err != nil {
|
||||
self.Lost("Unable to convert value to float", err, message)
|
||||
return
|
||||
}
|
||||
variable.Value = fValue
|
||||
} else {
|
||||
variable.Value = value
|
||||
}
|
||||
switch self.config.unitFrom {
|
||||
case PAYLOAD_SEL:
|
||||
if self.payloadRegex == nil {
|
||||
self.Lost("no payloadRegex defined, unitPart can't be used", nil, message)
|
||||
return
|
||||
}
|
||||
if self.config.unitPart >= len(payloadMatches) {
|
||||
self.Lost("unitPart out of range", nil, message)
|
||||
return
|
||||
}
|
||||
variable.Unit = payloadMatches[self.config.unitPart]
|
||||
case CONSTANT_SEL:
|
||||
variable.Unit = self.config.unit
|
||||
}
|
||||
|
||||
switch self.config.unitFrom {
|
||||
case PAYLOAD_SEL:
|
||||
if self.payloadRegex == nil {
|
||||
self.Lost("no payloadRegex defined, unitPart can't be used", nil, message)
|
||||
return
|
||||
}
|
||||
if self.config.unitPart >= len(payloadMatches) {
|
||||
self.Lost("unitPart out of range", nil, message)
|
||||
return
|
||||
}
|
||||
variable.Unit = payloadMatches[self.config.unitPart]
|
||||
case CONSTANT_SEL:
|
||||
variable.Unit = self.config.unit
|
||||
}
|
||||
measurement.Values["Value"] = variable
|
||||
|
||||
measurement.Values["Value"] = variable
|
||||
|
||||
//log.Printf("Prepared measurement item: %s", measurement)
|
||||
self.dbh.StoreMeasurement(&measurement)
|
||||
self.S()
|
||||
//log.Printf("Prepared measurement item: %s", measurement)
|
||||
self.dbh.StoreMeasurement(&measurement)
|
||||
self.S()
|
||||
}
|
||||
|
||||
|
||||
15
src/udi/handlers/z2m/models/gs361ah04/gs361ah04.go
Normal file
15
src/udi/handlers/z2m/models/gs361ah04/gs361ah04.go
Normal file
@@ -0,0 +1,15 @@
|
||||
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"`
|
||||
}
|
||||
|
||||
11
src/udi/handlers/z2m/models/wsdcgq01lm/wsdcgq01lm.go
Normal file
11
src/udi/handlers/z2m/models/wsdcgq01lm/wsdcgq01lm.go
Normal file
@@ -0,0 +1,11 @@
|
||||
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"`
|
||||
}
|
||||
|
||||
10
src/udi/handlers/z2m/models/wsdcgq11lm/wsdcgq11lm.go
Normal file
10
src/udi/handlers/z2m/models/wsdcgq11lm/wsdcgq11lm.go
Normal file
@@ -0,0 +1,10 @@
|
||||
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"`
|
||||
}
|
||||
|
||||
@@ -1,80 +1,113 @@
|
||||
package z2m
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"strings"
|
||||
"time"
|
||||
"udi/config"
|
||||
"udi/database"
|
||||
"udi/handlers/handler"
|
||||
"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"
|
||||
)
|
||||
|
||||
|
||||
type Z2MHandler struct {
|
||||
handler.CommonHandler
|
||||
dbh *database.DatabaseHandle
|
||||
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
|
||||
}
|
||||
|
||||
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)
|
||||
|
||||
measurement.Device = 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)
|
||||
|
||||
// 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
|
||||
}
|
||||
measurement.Application = device.Application.Label
|
||||
measurement.Device = device.Attributes["Label"].(string)
|
||||
|
||||
measurement.Attributes = make(map[string]interface{})
|
||||
measurement.Values = make(map[string]database.VariableType)
|
||||
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
|
||||
}
|
||||
|
||||
// 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.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
|
||||
}
|
||||
|
||||
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()
|
||||
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()
|
||||
}
|
||||
|
||||
|
||||
|
||||
18
src/udi/migrate_schema.go
Normal file
18
src/udi/migrate_schema.go
Normal file
@@ -0,0 +1,18 @@
|
||||
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")
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user