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

LSE01
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful
Details

Browse Source
This commit is contained in:
Wolfgang Hottgenroth
2023-12-22 17:28:32 +01:00
parent 8e6bea3f19
commit a8db62ea52
8 changed files with 75 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
src/udi/config-iot2.json
View File

@ -1,21 +0,0 @@
{
"mqtt": {
"broker": "mqtt://172.23.1.102:1883",
"tlsEnable": "false"
},
"topicMappings": [
{
"topics": [ "ttn/#" ],
"handler": "TTN",
"id": "TTN0",
"config": {
"databaseConnStr": "",
"attributes": {
}
}
}
],
"archiver": {
"dir": "./tmp/udi"
}
}

23
src/udi/config-level.json
View File

@ -1,23 +0,0 @@
{
"mqtt": {
"broker": "ssl://eu1.cloud.thethings.network:8883",
"username": "de-hottis-level-monitoring@ttn",
"passwordEnvVar": "MQTT_PASSWORD",
"tlsEnable": "true"
},
"topicMappings": [
{
"topics": [ "v3/#" ],
"handler": "TTN",
"id": "TTN0",
"config": {
"databaseConnStr": "",
"attributes": {
}
}
}
],
"archiver": {
"dir": "./tmp/udi"
}
}

23
src/udi/config-pg-cluster.json
View File

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

23
src/udi/config-pg.json
View File

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

4
src/udi/config-level-berresheim.json → src/udi/config-test-ttn.json
View File

@ -1,8 +1,7 @@
{
"mqtt": {
"broker": "ssl://eu1.cloud.thethings.network:8883",
"username": "de-hottis-level-monitoring@ttn",
"password": "ENV",
"username": "de-hottis-lora-test1@ttn",
"tlsEnable": "true"
},
"topicMappings": [
@ -11,7 +10,6 @@
"handler": "TTN",
"id": "TTN0",
"config": {
"databaseConnStr": "",
"attributes": {
}
}

29
src/udi/config-ttn.json
View File

@ -1,29 +0,0 @@
{
"mqtt": {
"broker": "mqtt://172.23.1.102:1883",
"tlsEnable": "false"
},
"topicMappings": [
{
"topics": [ "v3/de-hottis-level-monitoring@ttn/devices/+/up" ],
"handler": "TTN"
}
],
"handlers": [
{
"name": "TTN",
"databaseConnStr": "",
"attributes": {
}
},
{
"name": "IoT",
"databaseConnStr": "",
"attributes": {
}
}
],
"archiver": {
"dir": "./tmp/udi"
}
}

63
src/udi/handlers/ttn/models/draginoLse01/draginoLse01.go Normal file
View File

@ -0,0 +1,63 @@
package draginoLse01
import (
"fmt"
"encoding/json"
"udi/database"
)
/*
{
"Bat":3.211,
"TempC_DS18B20":"0.0",
"conduct_SOIL":32,
"temp_SOIL":"7.56",
"water_SOIL":"25.92"
}
*/
type message struct {
Bat float32 `json:"Bat"`
TempC_DS18B20 string `json:"TempC_DS18B20"`
Conduct_SOIL int `json:"conduct_SOIL"`
Temp_SOIL string `json:"temp_SOIL"`
Water_SOIL string `json:"water_SOIL"`
}
func Parse(fPort int, decodedPayload []byte, variables *map[string]database.VariableType, device *database.Device) error {
if fPort != 2 {
return fmt.Errorf("Unexpected fPort %d", fPort)
}
var message message
err := json.Unmarshal(decodedPayload, &message)
if err != nil {
return fmt.Errorf("Unable to parse payload, fPort %d, error %s", fPort, err)
}
(*variables)["Battery"] = database.VariableType {
Label: "Battery",
Variable: "Voltage",
Unit: "V",
Value: message.Bat,
}
(*variables)["Conductance"] = database.VariableType {
Label: "Conductance",
Variable: "Conductance",
Unit: "uS/cm",
Value: message.Conduct_SOIL,
}
(*variables)["Temperature"] = database.VariableType {
Label: "Temperature",
Variable: "Temperature",
Unit: "°C",
Value: message.Temp_SOIL,
}
(*variables)["Water"] = database.VariableType {
Label: "Water",
Variable: "Water",
Unit: "%",
Value: message.Water_SOIL,
}
return nil
}

18
src/udi/handlers/ttn/ttn.go
View File

@ -2,6 +2,7 @@ package ttn
import (
"fmt"
"log"
"time"
"encoding/json"
"udi/config"
@ -9,6 +10,7 @@ import (
"udi/handlers/ttn/models/emuProfIILoRaCfg1"
"udi/handlers/ttn/models/draginoLdds75"
"udi/handlers/ttn/models/draginoLmds200"
"udi/handlers/ttn/models/draginoLse01"
"udi/database"
)
@ -83,7 +85,7 @@ func New(id string, config config.HandlerConfigT) handler.Handler {
}
func (self *TTNHandler) Handle(message handler.MessageT) {
// log.Printf("Handler TTN %d processing %s -> %s", self.id, message.Topic, message.Payload)
log.Printf("Handler TTN %d processing %s -> %s", self.Id, message.Topic, message.Payload)
var measurement database.Measurement
measurement.Time = time.Now()
@ -94,7 +96,7 @@ func (self *TTNHandler) Handle(message handler.MessageT) {
self.Lost("Error when unmarshaling message", err, message)
return
}
//log.Printf("Parsed message: %s", uplinkMessage)
log.Printf("Parsed message: %s", uplinkMessage)
var attributes attributes
attributes.DeviceId = uplinkMessage.EndDeviceIds.DeviceId
@ -104,11 +106,11 @@ func (self *TTNHandler) Handle(message handler.MessageT) {
attributes.FrmPayload = uplinkMessage.UplinkMessage.FrmPayload
attributes.ConsumedAirtime = uplinkMessage.UplinkMessage.ConsumedAirtime
for _, rxm := range uplinkMessage.UplinkMessage.RxMetadata {
//log.Printf("RXM: %s", rxm)
log.Printf("RXM: %s", rxm)
g := gatewayAttributes { GatewayId: rxm.GatewayIds.GatewayId, Rssi: rxm.Rssi, Snr: rxm.Snr }
attributes.Gateways = append(attributes.Gateways, g)
}
//log.Printf("Attributes: %s", attributes)
log.Printf("Attributes: %s", attributes)
measurement.Attributes = map[string]interface{} {
"DeviceId": attributes.DeviceId,
"ApplicationId": attributes.ApplicationId,
@ -119,7 +121,7 @@ func (self *TTNHandler) Handle(message handler.MessageT) {
"ConsumedAirtime": attributes.ConsumedAirtime,
}
//log.Printf("ApplicationId: %s, DeviceId: %s", attributes.ApplicationId, attributes.DeviceId)
log.Printf("ApplicationId: %s, DeviceId: %s", attributes.ApplicationId, attributes.DeviceId)
device, err2 := self.dbh.GetDeviceByLabelAndApplication(attributes.ApplicationId, attributes.DeviceId)
if err2 != nil {
self.Lost("Error when loading device", err2, message)
@ -129,7 +131,7 @@ func (self *TTNHandler) Handle(message handler.MessageT) {
measurement.Device = attributes.DeviceId
measurement.Attributes["DeviceType"] = device.DeviceType.ModelIdentifier
//log.Printf("DeviceLabel: %s, DeviceType: %s", device.Label, device.DeviceType.ModelIdentifier)
log.Printf("DeviceLabel: %s, DeviceType: %s", device.Label, device.DeviceType.ModelIdentifier)
var parser func(int, []byte, *map[string]database.VariableType, *database.Device) error
switch device.DeviceType.ModelIdentifier {
@ -139,6 +141,8 @@ func (self *TTNHandler) Handle(message handler.MessageT) {
parser = draginoLdds75.Parse
case "dragino-lmds200":
parser = draginoLmds200.Parse
case "dragino-lse01":
parser = draginoLse01.Parse
default:
self.Lost(fmt.Sprintf("No parser found for %s", device.DeviceType.ModelIdentifier), nil, message)
return
@ -150,7 +154,7 @@ func (self *TTNHandler) Handle(message handler.MessageT) {
self.Lost("Model parser failed", err3, message)
return
}
//log.Printf("Prepared measurement item: %s", measurement)
log.Printf("Prepared measurement item: %s", measurement)
self.dbh.StoreMeasurement(&measurement)
self.S()
}