inserts into database works

src/DummyEGM1.py

@@ -37,7 +37,7 @@ def sendValue(client):
     msg = {
         "ts": str(datetime.now()),
         "dt": 100,
-        "v": sample(range(0xffffffff), 3)
+        "v": sample(range(0xffffffff), 3000)
     }
     topic = f"{TOPIC_PRE}/rig01/dev05/md"
     client.publish(topic, json.dumps(msg))
@@ -48,14 +48,9 @@ def perform(client, params):
     logger.info("Start")
     sendCmd(client, "start")
 
-    for i in range(2):
+    for i in range(10):
         sendValue(client)
         sleep(1.0)
-        sendValue(client)
+    
-        sleep(1.0)
-        sendValue(client)
-        sendLog(client)
-        sleep(1.0)
-
     logger.info("Stop")
     sendCmd(client, "stop")

src/EGM1mdTsDbMixin.py

@@ -1,3 +1,4 @@
+import psycopg2
 from AbstractTsDbMixin import AbstractTsDbMixin
 from loguru import logger
 import iso8601
@@ -6,31 +7,28 @@ from io import StringIO
 
 class EGM1mdTsDbMixin(AbstractTsDbMixin):
     def write(self, experiment, dataObject):
-        logger.info(f"Db write: {experiment} {dataObject}")
+        logger.info("database write operation start")
         try:
             ts = iso8601.parse_date(dataObject.payload["ts"])
             dt = datetime.timedelta(microseconds=dataObject.payload["dt"])
             values = dataObject.payload["v"]
-            logger.info(f"ts: {ts}")
-            logger.info(f"dt: {dt}")
-            logger.info(f"v: {values}") 
         
             stringBuffer = ''
+            cnt = 0
             for rv in values:
-                statusOctet = (rv & 0xff000000) >> 24
                 unsignedValue = rv & 0x00ffffff
-                valid = (statusOctet & 0x04) >> 2
+                valid = (rv & 0x04000000) >> 26
-                sign = (statusOctet & 0x02) >> 1
+                sign = (rv & 0x02000000) >> 25
                 signedValue = unsignedValue if not sign else unsignedValue * -1
-                
-                logger.info(f"store: ts: {ts}, rv: {rv:08x}, status: {statusOctet:02x}, valid: {valid}, sign: {sign}, unsigned: {unsignedValue}, signed: {signedValue}")
 
                 stringBuffer += f"{ts}\t{experiment}\t{rv}\t{valid}\t{signedValue}\n"
                 ts += dt
+                cnt += 1
+            logger.info(f"{cnt} items")
 
             stringIO = StringIO(stringBuffer)
 
-            logger.info("database copy started")
+            logger.info("database copy start")
             with self.connection:
                 with self.connection.cursor() as cursor:
                     cursor.copy_from(stringIO, 'egm1md', columns=('time', 'experiment', 'rawvalue', 'valid', 'value'))
@@ -38,6 +36,10 @@ class EGM1mdTsDbMixin(AbstractTsDbMixin):
 
 
         except iso8601.ParseError as e:
-            logger.error(f"unable to parse timestamp, item dropped: {f}")
+            logger.error(f"unable to parse timestamp, items dropped: {e}")
         except KeyError as e:
-            logger.error(f"missing attribute in payload, item dropped: {e}")
+            logger.error(f"missing attribute in payload, items dropped: {e}")
+        except psycopg2.Error as e:
+            logger.error(f"error {e.pgcode} when talking to database, items dropped: {e}")
+        
+        logger.info("database write operation done")

src/EGM1mdTsDbSinkHandler.py

@@ -16,5 +16,5 @@ class EGM1mdTsDbSinkHandler(AbstractSinkHandler, EGM1mdTsDbMixin):
         self.disconnectDb()
 
     def sinkAction(self, dataObject):
-        logger.info(f"sink {self.name} received {dataObject} for experiment {self.experiment}")
+        # logger.info(f"sink {self.name} received {dataObject} for experiment {self.experiment}")
         self.write(self.experiment, dataObject)

src/GenericMqttSubscriber.py

@@ -21,7 +21,8 @@ class GenericMqttSubscriber(AbstractMqttHandler):
 
     def onMessage(self, topic, payload):
         try: 
-            logger.info("message received: {} -> {}".format(topic, str(payload)))
+            # logger.info("message received: {} -> {}".format(topic, str(payload)))
+            logger.info(f"message received on topic {topic}")
             dataObject = DataObject(name=self.dataObjectName, topic=topic, payload=json.loads(payload))
             self.queue.put(dataObject)
         except JSONDecodeError as e: