rename snmp handler to prepared handler, 4

deployment/instances/udi/default/config.json

@@ -6,13 +6,22 @@
   "topicMappings": [
     {
       "topics": [ "snmp" ],
-      "handler": "SNMP",
+      "handler": "PREP",
       "id": "SNMP",
       "config": {
         "attributes": {
         }
       }
     },
+    {
+      "topics": [ "tsm" ],
+      "handler": "PREP",
+      "id": "TSM",
+      "config": {
+        "attributes": {
+        }
+      }
+    },
     {
       "topics": [ "dt1/ai/periodic/1" ],
       "handler": "DT1T",

migration/.gitignore

@@ -1 +0,0 @@
-.venv

migration/migrate-level.py

@@ -1,38 +0,0 @@
-import psycopg2
-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, 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]
-            deviceName = srcObj[1]
-            rawLevel = srcObj[2]
-            level = srcObj[3]
-            status = srcObj[4]
-            battery = srcObj[5]
-
-            logger.info(f"{timestamp=}, {deviceName=}, {rawLevel=}, {level=}, {status=}, {battery=}")
-
-            destTime = timestamp
-            destApplication = "de-hottis-level-monitoring"
-            destDevice = "eui-a84041a2c18341d6"
-            destAttributes = '{"ApplicationId":"de-hottis-level-monitoring", "DeviceType":"dragino-ldds75", "Status":"' + status + '","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) + '}}'
-            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()
-
-

migration/migrate-pv.py

@@ -1,79 +0,0 @@
-import psycopg2
-from loguru import logger
-import os
-
-srcPgHost = os.environ["SRC_PGHOST"]
-srcPgUser = os.environ["SRC_PGUSER"]
-srcPgPassword = os.environ["SRC_PGPASSWORD"]
-srcPgDatabase = os.environ["SRC_PGDATABASE"]
-destPgHost = os.environ["DEST_PGHOST"]
-destPgUser = os.environ["DEST_PGUSER"]
-destPgPassword = os.environ["DEST_PGPASSWORD"]
-destPgDatabase = os.environ["DEST_PGDATABASE"]
-
-try:
-    srcConn = psycopg2.connect(
-                host=srcPgHost,
-                dbname=srcPgDatabase,
-                user=srcPgUser,
-                password=srcPgPassword,
-                sslmode='require'
-            )
-    srcConn.autocommit = False
-
-    destConn = psycopg2.connect(
-                host=destPgHost,
-                dbname=destPgDatabase,
-                user=destPgUser,
-                password=destPgPassword,
-                sslmode='require'
-            )
-    destConn.autocommit = False
-
-    with srcConn.cursor() as srcCur, destConn.cursor() as destCur:
-        srcCur.execute("select time, deviceid, status, state, importenergyactive, importenergyreactive, exportenergyactive, exportenergyreactive, powerapparent, poweractive, powerreactive, powerdemandpositive, powerdemandreverse, factor, angle, voltage, current, powerdemand from pv_power_measurement_t order by time")
-        for srcObj in srcCur:
-            timestamp = srcObj[0]
-            deviceName = srcObj[1]
-            status = srcObj[2]
-            state = srcObj[3]
-            importenergyactive = srcObj[4]
-            importenergyreactive = srcObj[5]
-            exportenergyactive = srcObj[6]
-            exportenergyreactive = srcObj[7]
-            powerapparent = srcObj[8]
-            poweractive = srcObj[9]
-            powerreactive = srcObj[10]
-            powerdemandpositive = srcObj[11]
-            powerdemandreverse = srcObj[12]
-            factor = srcObj[13]
-            angle = srcObj[14]
-            voltage = srcObj[15]
-            current = srcObj[16]
-            powerdemand = srcObj[17]
-
-
-            logger.info(f"{timestamp=}, {deviceName=}")
-
-            destTime = timestamp
-            destApplication = "PV"
-            destDevice = "Powermeter"
-            destAttributes = f"{{\"ApplicationId\":\"PV\", \"Status\":\"{status}\",\"Hint\": \"Migrated\"}}"
-            destValues = f"{{\"Cnt\": {{\"unit\": \"\", \"label\": \"\", \"value\": \"-1\", \"variable\": \"Cnt\"}}, \"Angle\": {{\"unit\": \"degree\", \"label\": \"\", \"value\": \"{angle}\", \"variable\": \"Angle\"}}, \"State\": {{\"unit\": \"\", \"label\": \"\", \"value\": \"{state}\", \"variable\": \"State\"}}, \"Factor\": {{\"unit\": \"\", \"label\": \"\", \"value\": \"{factor}\", \"variable\": \"Factor\"}}, \"Current\": {{\"unit\": \"A\", \"label\": \"\", \"value\": \"{current}\", \"variable\": \"Current\"}}, \"Voltage\": {{\"unit\": \"V\", \"label\": \"\", \"value\": \"{voltage}\", \"variable\": \"Voltage\"}}, \"PowerActive\": {{\"unit\": \"W\", \"label\": \"\", \"value\": \"{poweractive}\", \"variable\": \"PowerActive\"}}, \"PowerApparent\": {{\"unit\": \"VA\", \"label\": \"\", \"value\": \"{powerapparent}\", \"variable\": \"PowerApparent\"}}, \"PowerReactive\": {{\"unit\": \"VA\", \"label\": \"\", \"value\": \"{powerreactive}\", \"variable\": \"PowerReactive\"}}, \"ExportEnergyActive\": {{\"unit\": \"Wh\", \"label\": \"\", \"value\": \"{exportenergyactive}\", \"variable\": \"ExportEnergyActive\"}}, \"ImportEnergyActive\": {{\"unit\": \"Wh\", \"label\": \"\", \"value\": \"{importenergyactive}\", \"variable\": \"ImportEnergyActive\"}}, \"PowerDemandReverse\": {{\"unit\": \"W\", \"label\": \"\", \"value\": \"{powerdemandreverse}\", \"variable\": \"PowerDemandReverse\"}}, \"PowerDemandPositive\": {{\"unit\": \"W\", \"label\": \"\", \"value\": \"{powerdemandpositive}\", \"variable\": \"PowerDemandPositive\"}}, \"ExportEnergyReactive\": {{\"unit\": \"VAh\", \"label\": \"\", \"value\": \"{exportenergyreactive}\", \"variable\": \"ExportEnergyReactive\"}}, \"ImportEnergyReactive\": {{\"unit\": \"VAh\", \"label\": \"\", \"value\": \"{importenergyreactive}\", \"variable\": \"ImportEnergyReactive\"}}}}"
-            logger.info(f"{destTime=}, {destApplication=}, {destDevice=}, {destAttributes=}, {destValues=}")
-
-
-            try:
-                destCur.execute("insert into measurements (time, application, device, attributes, values) values(%s, %s, %s, %s, %s)",
-                                (destTime, destApplication, destDevice, destAttributes, destValues))
-                destConn.commit()
-            except Exception as e:
-                destConn.rollback()
-                logger.error(f"Error {e} when inserted time {destTime}")
-finally:
-    if srcConn:
-        srcConn.close()
-    if destConn:
-        destConn.close()
-
-

migration/migrate-temperature.py

@@ -1,78 +0,0 @@
-import psycopg2
-from loguru import logger
-import os
-
-srcPgHost = os.environ["SRC_PGHOST"]
-srcPgUser = os.environ["SRC_PGUSER"]
-srcPgPassword = os.environ["SRC_PGPASSWORD"]
-srcPgDatabase = os.environ["SRC_PGDATABASE"]
-destPgHost = os.environ["DEST_PGHOST"]
-destPgUser = os.environ["DEST_PGUSER"]
-destPgPassword = os.environ["DEST_PGPASSWORD"]
-destPgDatabase = os.environ["DEST_PGDATABASE"]
-
-try:
-    srcConn = psycopg2.connect(
-                host=srcPgHost,
-                dbname=srcPgDatabase,
-                user=srcPgUser,
-                password=srcPgPassword,
-                sslmode='require'
-            )
-    srcConn.autocommit = False
-
-    destConn = psycopg2.connect(
-                host=destPgHost,
-                dbname=destPgDatabase,
-                user=destPgUser,
-                password=destPgPassword,
-                sslmode='require'
-            )
-    destConn.autocommit = False
-
-    with srcConn.cursor() as srcCur, destConn.cursor() as destCur:
-        srcCur.execute("select time, location, status, temperature, category from room_climate_measurement_t where category = 'heating' and time > '2023-12-19 05:20:00' order by time")
-        for srcObj in srcCur:
-            timestamp = srcObj[0]
-            location = srcObj[1]
-            status = srcObj[2]
-            temperature = srcObj[3]
-            category = srcObj[4]
-
-            logger.info(f"{timestamp=}, {location=}, {status=}, {temperature=}, {category=}")
-
-            destTime = timestamp
-
-            match category:
-                case 'heating':
-                    destApplication = 'Temperature Heating'
-                case 'Outdoor':
-                    destApplication = 'Temperature Wago'
-                case 'Device':
-                    destApplication = 'Temperature Wago'
-                case 'Indoor':
-                    destApplication = 'Temperature Multisensor' if location != 'Anna-Koeln-2' else 'Temperature Shelly Plus HT'
-                case 'Special':
-                    destApplication = 'Temperature Multisensor'
-
-            destDevice = location
-            destAttributes = '{"ApplicationId":"temperature-imported", "Status":"' + status + '","Location":"' + location + '","Category":"' + category + '","Hint": "Migrated"}'
-            destValues = '{"Value": {"unit": "°C", "label": "", "value": "' + str(temperature) + '", "variable": ""}}'
-
-            logger.info(f"{destTime=}, {destApplication=}, {destDevice=}, {destAttributes=}, {destValues=}")
-
-            try:
-                destCur.execute("insert into measurements (time, application, device, attributes, values) values(%s, %s, %s, %s, %s)",
-                                (destTime, destApplication, destDevice, destAttributes, destValues))
-                destConn.commit()
-            except Exception as e:
-                destConn.rollback()
-                logger.error(f"Error {e} when inserted time {destTime}")
-
-finally:
-    if srcConn:
-        srcConn.close()
-
-    if destConn:
-        destConn.close()
-

migration/requirements.txt

@@ -1,2 +0,0 @@
-loguru==0.7.3
-psycopg2==2.9.9

src/udi/database/abstract_database.go

@@ -9,6 +9,7 @@ type VariableType struct {
   Variable string `json:"variable"`
   Unit string `json:"unit"`
   Value interface{} `json:"value,omitempty"`
+  Status string `json:"status,omitempty"`
 }
 
 type Measurement struct {

src/udi/dispatcher/dispatcher.go

@@ -17,7 +17,7 @@ import "udi/handlers/sver"
 import "udi/handlers/svej"
 import "udi/handlers/dt1t"
 import "udi/handlers/locative"
-import "udi/handlers/snmp"
+import "udi/handlers/prepared"
 import "udi/handlers/z2m"
 
 
@@ -49,8 +49,8 @@ func InitDispatcher() {
       factory = dt1t.New
     case "Locative":
       factory = locative.New
-    case "SNMP":
+    case "PREP":
-      factory = snmp.New
+      factory = prepared.New
     case "Z2M":
       factory = z2m.New
     default:

src/udi/handlers/prepared/prepared.go

@@ -1,4 +1,4 @@
-package snmp
+package prepared
 
 import (
   "time"
@@ -10,7 +10,7 @@ import (
 )
 
 
-type SnmpHandler struct {
+type PreparedHandler struct {
   handler.CommonHandler
   dbh *database.DatabaseHandle
 }
@@ -19,6 +19,8 @@ type endpoint_t struct {
   Label string `json:"label"`
   Variable string `json:"variable"`
   Value string `json:"value"`
+  Unit string `json:"unit"`
+  Status string `json:"status"`
 }
 
 type observation_t struct {
@@ -29,16 +31,16 @@ type observation_t struct {
 
 
 func New(id string, config config.HandlerConfigT) handler.Handler {
-  t := &SnmpHandler {
+  t := &PreparedHandler {
   }
   t.Id = id
   t.dbh = database.NewDatabaseHandle()
-  log.Printf("Handler SNMP %d initialized", id)
+  log.Printf("Handler Prepared %d initialized", id)
   return t
 }
 
-func (self *SnmpHandler) Handle(message handler.MessageT) {
+func (self *PreparedHandler) Handle(message handler.MessageT) {
-  // log.Printf("Handler SNMP %d processing %s -> %s", self.Id, message.Topic, message.Payload)
+  log.Printf("Handler Prepared %d processing %s -> %s", self.Id, message.Topic, message.Payload)
 
   var observation observation_t
   err := json.Unmarshal([]byte(message.Payload), &observation)
@@ -50,7 +52,7 @@ func (self *SnmpHandler) Handle(message handler.MessageT) {
   var measurement database.Measurement
   measurement.Time = time.Now()
 
-  measurement.Application = "SNMP"
+  measurement.Application = self.Id
   measurement.Device = observation.Device
 
   measurement.Attributes = map[string]interface{} {
@@ -62,12 +64,13 @@ func (self *SnmpHandler) Handle(message handler.MessageT) {
     measurement.Values[k] = database.VariableType {
       Label: v.Label,
       Variable: v.Variable,
-      Unit: "",
+      Unit: v.Unit,
       Value: v.Value,
+      Status: v.Status,
     }
   }
 
-  // log.Printf("Prepared measurement item: %s", measurement)
+  log.Printf("Prepared measurement item: %s", measurement)
 
   self.dbh.StoreMeasurement(&measurement)
   self.S()