added json parser step

Author: rjtokenring

appslab_modules/sinks/mqtt/__init__.py

@@ -1,20 +1,65 @@
 import paho.mqtt.client as mqtt
+import json
+import logging
+import time
+
+logger = logging.getLogger(__name__)
 
 class MQTTWriter:
-    def __init__(self, broker_address:str, broker_port:int, topic:str, username:str, password:str):
-        self.client = mqtt.Client()
+    def __init__(self, broker_address:str, broker_port:int, topic:str, username:str, password:str, client_id:str = "Arduino_MQTTWriter"):
+        self.topic = topic
+        self.client = mqtt.Client(client_id)
         self.client.username_pw_set(username, password)
+
+        def on_connect(client, userdata, flags, rc):
+            if rc != 0:
+                logger.error("Failed to connect, return code %d\n", rc)
+
+        FIRST_RECONNECT_DELAY = 1
+        RECONNECT_RATE = 2
+        MAX_RECONNECT_COUNT = 12
+        MAX_RECONNECT_DELAY = 60
+
+        def on_disconnect(client, userdata, rc):
+            logging.warning("Disconnected with result code: %s", rc)
+            reconnect_count, reconnect_delay = 0, FIRST_RECONNECT_DELAY
+            while reconnect_count < MAX_RECONNECT_COUNT:
+                logging.info("Reconnecting in %d seconds...", reconnect_delay)
+                time.sleep(reconnect_delay)
+
+                try:
+                    client.reconnect()
+                    logging.info("Reconnected successfully!")
+                    return
+                except Exception as err:
+                    logging.error("%s. Reconnect failed. Retrying...", err)
+
+                reconnect_delay *= RECONNECT_RATE
+                reconnect_delay = min(reconnect_delay, MAX_RECONNECT_DELAY)
+                reconnect_count += 1
+            logging.info("Reconnect failed after %s attempts. Exiting...", reconnect_count)
+
+        self.client.on_connect = on_connect
+        self.client.on_disconnect = on_disconnect
+
         self.client.connect(broker_address, broker_port, 60)
-        self.topic = topic
+        self.client.loop_start()
 
     def close(self):
+        self.client.loop_stop()
         self.client.disconnect()
 
     def write(self, message:str):
         if message and message != "":
             self.client.publish(self.topic, message)
 
     def consume(self, item):
+        if item == None:
+            return None
+        
         if isinstance(item, str):
             self.write(item)
+        elif isinstance(item, dict) and len(item) > 0:
+            self.write(json.dumps(item))
+
         return item

…db_storage/test/test_database_storage.py …teps/db_storage/test_database_storage.pyappslab_modules/steps/db_storage/test/test_database_storage.py renamed to appslab_modules/steps/db_storage/test_database_storage.py appslab_modules/steps/db_storage/test/test_database_storage.py renamed to appslab_modules/steps/db_storage/test_database_storage.py

@@ -0,0 +1,23 @@
+import json
+import logging
+
+logger = logging.getLogger(__name__)
+
+class JSONParser:
+    def __init__(self, silent:bool = False):
+        self.silent = silent
+
+    def parse(self, item:str) -> dict:
+        try:
+            return json.loads(item)
+        except Exception as e:
+            if self.silent == False:
+                logger.error(f"Error parsing content: {e}")
+            return None
+        
+    def process(self, item):
+        if isinstance(item, str):
+            return self.parse(item)
+
+        return item # No processing needed
+

appslab_modules/steps/json_parser/__init__.py

@@ -0,0 +1,2 @@
+name: json_parser
+module_description: "JSON parser module. Parses JSON data and outputs it as a dictionary. Drop strings that cannot be converted to JSON."

appslab_modules/steps/json_parser/module_config.yaml

@@ -0,0 +1,19 @@
+import unittest
+from __init__ import JSONParser
+
+class TestJsonParser(unittest.TestCase):
+
+    def test_successful_parser(self):
+        """Test successfull parse."""
+        ps = JSONParser()
+        out = ps.parse('{"test_key": "test_val"}')
+        self.assertEqual(out["test_key"], "test_val")
+
+    def test_drop_data_parser(self):
+        """Test parse of not valid json data."""
+        ps = JSONParser(silent=True)
+        out = ps.parse('not json text')
+        self.assertEqual(out,None)
+
+if __name__ == '__main__':
+    unittest.main()