add status in ttn handlers

migration/migrate-level.py

@@ -4,8 +4,10 @@ from loguru import logger
 try:
     srcConn = psycopg2.connect(database="level_monitoring_berresheim")
     srcConn.autocommit = False
+    destConn = psycopg2.connect(database="udi-berresheim")
+    destConn.autocommit = False
 
-    with srcConn.cursor() as srcCur:
+    with srcConn.cursor() as srcCur, destConn.cursor() as destCur:
         srcCur.execute("select time, application_name, raw_level, level, status, battery from measurement_t")
         for srcObj in srcCur:
             timestamp = srcObj[0]
@@ -21,9 +23,16 @@ try:
             destApplication = "de-hottis-level-monitoring"
             destDevice = "eui-a84041a2c18341d6"
             destAttributes = '{"ApplicationId":"de-hottis-level-monitoring", "DeviceType":"dragino-ldds75", "Hint": "Migrated"}'
-            destValues = '{"Battery":{"unit":"V","label":"Battery","value":' + str(battery) + ',"variable":"Voltage"}, "Distance":{"unit":mm","label":"Distance","variable":"Level","value":' + str(rawLevel) + '}, "CorrectedDistance":{"unit":"mm", "label":"CorrectedDistance", "variable":"Level","value":' + str(level) + '}}'
+            destValues = '{"Battery":{"unit":"V","label":"Battery","value":' + str(battery) + ',"variable":"Voltage"}, "Distance":{"unit":"mm","label":"Distance","variable":"Level","value":' + str(rawLevel) + '}, "CorrectedDistance":{"unit":"mm", "label":"CorrectedDistance", "variable":"Level","value":' + str(level) + '}}'
             logger.info(f"{destTime=}, {destApplication=}, {destDevice=}, {destAttributes=}, {destValues=}")
+
+            destCur.execute("insert into measurements (time, application, device, attributes, values) values(%s, %s, %s, %s, %s)",
+                            (destTime, destApplication, destDevice, destAttributes, destValues))
+            destConn.commit()
 finally:
     if srcConn:
         srcConn.close()
+    if destConn:
+        destConn.close()
+
 

queries/hottis.sql

@@ -45,3 +45,10 @@ create or replace view temperature_v as
     from measurements
     where application in ('Temperature Multisensor', 'Temperature Shelly Plus HT');
 
+create or replace view humidity_v as
+  select time,
+         cast(values->'Value'->>'value' as float) as humidity,
+         device
+    from measurements
+    where application in ('Humidity Multisensor');
+

queries/old-pv-yield-query.sql

@@ -0,0 +1,73 @@
+-- query
+
+with 
+    first_day_in_year as (
+        select 
+                date_trunc('day', min(time)) as day
+            from pv_power_measurement_t
+            where 
+                time between date_trunc('year', time) and now()
+    ),
+    first_value_in_year as (
+        select 
+                time_bucket('1 day', time) as interval,
+                first(exportenergyactive, time) as energy
+            from pv_power_measurement_t
+            where 
+                time between (select day from first_day_in_year) and (select day from first_day_in_year) + interval '1 day' and
+                status = 'Ok'
+            group by interval
+    ),
+    first_day_in_month as (
+        select 
+                date_trunc('day', min(time)) as day
+            from pv_power_measurement_t
+            where 
+                time between date_trunc('month', now()) and now()
+    ),
+    first_value_in_month as (
+        select 
+                time_bucket('1 day', time) as interval,
+                first(exportenergyactive, time) as energy
+            from pv_power_measurement_t
+            where 
+                time between (select day from first_day_in_month) and (select day from first_day_in_month) + interval '1 day' and
+                status = 'Ok'
+            group by interval
+    ),
+    first_value_in_day as (
+        select
+                time_bucket('1 day', time) as interval,
+                first(exportenergyactive, time) as energy
+            from pv_power_measurement_t
+            where time >= date_trunc('day', now())
+            group by interval
+    ),
+    last_value as (
+        select  
+                time_bucket('1 day', time) as interval,
+                last(exportenergyactive, time) as energy
+            from pv_power_measurement_t
+            where 
+                time between date_trunc('day', now()) and date_trunc('day', now()) + interval '1 day' and
+                status = 'Ok'
+            group by interval
+    )
+    select
+        extract(year from (select day from first_day_in_year))::text as period_value,
+        'Year' as period_name,
+        round(((select energy from last_value) - (select energy from first_value_in_year))::numeric, 2) as yield
+    union
+    select
+        to_char((select day from first_day_in_month), 'Month') as period_value,
+        'Month' as period_name,
+        round(((select energy from last_value) - (select energy from first_value_in_month))::numeric, 2) as yield
+    union
+    select
+        now()::date::text as period_value,
+        'Day' as period_name,
+        round(((select energy from last_value) - (select energy from first_value_in_day))::numeric, 2) as yield;
+
+-- output format
+-- wn@atuin:~/Workspace/go-workspace/src/universal-data-ingest  [main ≡ +0 ~1 -0 !]$ mosquitto_sub -h 172.23.1.102 -v -t IoT/PV/Yields
+-- IoT/PV/Yields {"Month":"1.43","Year":"285.39","Day":"0.00"}

src/udi/handlers/ttn/models/draginoLdds75/draginoLdds75.go

@@ -2,7 +2,7 @@ package draginoLdds75
 
 import (
   "fmt"
-  "log"
+  // "log"
   "strings"
   "strconv"
   "encoding/json"
@@ -26,7 +26,7 @@ type message struct {
   TempC_DS18B20 string `json:"TempC_DS18B20"`
 }
 
-func Parse(fPort int, decodedPayload []byte, _ string, variables *map[string]database.VariableType, device *database.Device) error {
+func Parse(fPort int, decodedPayload []byte, _ string, variables *map[string]database.VariableType, attributes *map[string]interface{}, device *database.Device) error {
   if fPort != 2 {
     return fmt.Errorf("Unexpected fPort %d", fPort)
   }
@@ -55,11 +55,20 @@ func Parse(fPort int, decodedPayload []byte, _ string, variables *map[string]dat
     Unit: "mm",
     Value: distance,
   }
+
+  if distance == 20 {
+    (*attributes)["Status"] = "invalid value"
+  } else if distance == 0 {
+    (*attributes)["Status"] = "no sensor detected"
+  } else {
+    (*attributes)["Status"] = "Ok"
+  }
+
   groundLevelI, exists := device.Attributes["GroundLevel"]
   groundLevelS, ok := groundLevelI.(string)
   groundLevel, err3 := strconv.Atoi(groundLevelS)
   if exists && err3 == nil && ok {
-    log.Println("add corrected distance")
+    //log.Println("add corrected distance")
     correctedDistance := groundLevel - distance
     (*variables)["CorrectedDistance"] = database.VariableType {
       Label: "CorrectedDistance",
@@ -67,11 +76,11 @@ func Parse(fPort int, decodedPayload []byte, _ string, variables *map[string]dat
       Unit: "mm",
       Value: correctedDistance,
     }
-  } else {
+  } /* else {
     log.Printf("no ground level: %s %s %s", exists, err3, ok)
     log.Printf("Device: %s", device)
     log.Printf("Attributes: %s", device.Attributes)
-  }
+  } */
 
   return nil
 }

src/udi/handlers/ttn/models/draginoLmds200/draginoLmds200.go

@@ -26,7 +26,7 @@ type message struct {
   Dis2 int `json:"dis2"`
 }
 
-func Parse(fPort int, decodedPayload []byte, _ string, variables *map[string]database.VariableType, device *database.Device) error {
+func Parse(fPort int, decodedPayload []byte, _ string, variables *map[string]database.VariableType, attributes *map[string]interface{}, device *database.Device) error {
   if fPort != 2 {
     return fmt.Errorf("Unexpected fPort %d", fPort)
   }
@@ -55,6 +55,15 @@ func Parse(fPort int, decodedPayload []byte, _ string, variables *map[string]dat
     Unit: "mm",
     Value: distance2,
   }
+
+  if distance1 == 2 {
+    (*attributes)["Status"] = "invalid value"
+  } else if distance1 == 1 {
+    (*attributes)["Status"] = "no sensor detected"
+  } else {
+    (*attributes)["Status"] = "Ok"
+  }
+
   groundLevelI, exists := device.Attributes["GroundLevel"]
   groundLevelS, ok := groundLevelI.(string)
   groundLevel, err3 := strconv.Atoi(groundLevelS)

src/udi/handlers/ttn/models/draginoLse01/draginoLse01.go

@@ -24,7 +24,7 @@ type message struct {
   Water_SOIL string `json:"water_SOIL"`
 }
 
-func Parse(fPort int, decodedPayload []byte, _ string, variables *map[string]database.VariableType, device *database.Device) error {
+func Parse(fPort int, decodedPayload []byte, _ string, variables *map[string]database.VariableType, _ *map[string]interface{}, _ *database.Device) error {
   if fPort != 2 {
     return fmt.Errorf("Unexpected fPort %d", fPort)
   }

src/udi/handlers/ttn/models/emuProfIILoRaCfg1/emuProfIILoRaCfg1.go

@@ -177,7 +177,7 @@ type emuMessage1 struct {
 
 
 
-func Parse(fPort int, decodedPayload []byte, _ string, variables *map[string]database.VariableType, _ *database.Device) error {
+func Parse(fPort int, decodedPayload []byte, _ string, variables *map[string]database.VariableType, _ *map[string]interface{}, _ *database.Device) error {
   //log.Printf("Parse input: %d, %s", fPort, decodedPayload)
   switch fPort {
   case 1:

src/udi/handlers/ttn/models/rawPayloadPrinter/rawPayloadPrinter.go

@@ -3,15 +3,23 @@ package rawPayloadPrinter
 import (
   "log"
   "fmt"
+  "encoding/base64"
+  "encoding/hex"
   "udi/database"
 )
 
 
-func Parse(fPort int, _ []byte, frmPayload string, variables *map[string]database.VariableType, device *database.Device) error {
+func Parse(fPort int, _ []byte, frmPayload string, variables *map[string]database.VariableType, _ *map[string]interface{}, _ *database.Device) error {
   if fPort != 2 {
     return fmt.Errorf("Unexpected fPort %d", fPort)
   }
 
-  log.Printf("frmPayload: %s", frmPayload)
+  bytes, err := base64.StdEncoding.DecodeString(frmPayload)
+  if err != nil {
+    return fmt.Errorf("Unable to base64-decode payload: %v", err)
+  }
+  hexString := hex.EncodeToString(bytes)
+
+  log.Printf("Payload: %s", hexString)
   return nil
 }

src/udi/handlers/ttn/ttn.go

@@ -2,7 +2,7 @@ package ttn
 
 import (
   "fmt"
-  "log"
+  //"log"
   "time"
   "encoding/json"
   "udi/config"
@@ -86,7 +86,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()
@@ -97,7 +97,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
@@ -107,11 +107,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,
@@ -122,7 +122,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)
@@ -132,9 +132,9 @@ 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, string, *map[string]database.VariableType, *database.Device) error
+  var parser func(int, []byte, string, *map[string]database.VariableType, *map[string]interface{}, *database.Device) error
   switch device.DeviceType.ModelIdentifier {
   case "emu-prof-ii-lora-cfg1":
     parser = emuProfIILoRaCfg1.Parse
@@ -156,12 +156,13 @@ func (self *TTNHandler) Handle(message handler.MessageT) {
                  uplinkMessage.UplinkMessage.DecodedPayload.Payload, 
                  uplinkMessage.UplinkMessage.FrmPayload,
                  &(measurement.Values), 
+                 &(measurement.Attributes),
                  device)
   if err3 != nil {
     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()
 }