Some updates

Author: rjtokenring

appslab_modules/sinks/mqtt/__init__.py

@@ -2,26 +2,33 @@
 import json
 import logging
 import time
+import uuid
 
 logger = logging.getLogger(__name__)
 
-class MQTTWriter:
-    def __init__(self, broker_address:str, broker_port:int, topic:str, username:str, password:str, client_id:str = "Arduino_MQTTWriter"):
+def generate_client_id():
+    return "Arduino_MQTTSink-" + str(uuid.uuid4())
+
+class MQTTSink:
+    def __init__(self, broker_address:str, broker_port:int, topic:str, username:str, password:str, client_id:str = ""):
         self.topic = topic
-        self.client = mqtt.Client(client_id)
-        self.client.username_pw_set(username, password)
+        if not client_id or client_id == "":
+            client_id = generate_client_id()
+        self.client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION2, client_id=client_id)
+        if username and password:
+            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)
+        def on_connect(client, userdata, flags, reason_code, properties):
+            if reason_code != 0:
+                logger.error("Failed to connect, return code %d\n", reason_code)
 
         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)
+        def on_disconnect(client, userdata, flags, reason_code, properties):
+            logging.info("Disconnected with result code: %s", reason_code)
             reconnect_count, reconnect_delay = 0, FIRST_RECONNECT_DELAY
             while reconnect_count < MAX_RECONNECT_COUNT:
                 logging.info("Reconnecting in %d seconds...", reconnect_delay)
@@ -51,7 +58,8 @@ def close(self):
 
     def write(self, message:str):
         if message and message != "":
-            self.client.publish(self.topic, message)
+            return self.client.publish(self.topic, message)
+        return None
 
     def consume(self, item):
         if item == None:
@@ -60,6 +68,4 @@ def consume(self, item):
         if isinstance(item, str):
             self.write(item)
         elif isinstance(item, dict) and len(item) > 0:
-            self.write(json.dumps(item))
-
-        return item
+            self.write(json.dumps(item))

appslab_modules/sinks/mqtt/test_mqtt_sink.py

@@ -0,0 +1,23 @@
+import unittest
+from __init__ import MQTTSink
+
+class TestMQTTSink(unittest.TestCase):
+
+    def setUp(self):
+        """Set up test fixtures."""
+        self.mqtt = MQTTSink(broker_address = "192.168.1.11", broker_port = 1883, topic = "/test/tp1", username = "admin", password = "admin")
+
+    def tearDown(self):
+        """Clean up after tests."""
+        self.mqtt.close()
+
+    def test_write_to_topic(self):
+        """Test successfull write to topic."""
+        msginfo = self.mqtt.write("Hello, World!")
+        print(msginfo)
+        msginfo.wait_for_publish(2)
+        self.assertTrue(msginfo.is_published())
+
+
+if __name__ == '__main__':
+    unittest.main()

appslab_modules/steps/log/__init__.py

@@ -0,0 +1,16 @@
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+class LogStep:
+    def __init__(self):
+        pass
+
+    def process(self, item):
+        if item == None:
+            return None
+        
+        logger.info(item)
+
+        return item

appslab_modules/steps/log/module_config.yaml

@@ -0,0 +1,2 @@
+name: log_step
+module_description: "Logger step"

design/declarative.md

@@ -29,7 +29,24 @@ pipe.add_sink(DBStore())
 pipe.process()
 ```
 
-### Adapter functions
+### Adapters - explicit mapping steps
+```python
+
+def user_mapping_function(input :dic) - dic
+    # User defined logic
+    return input
+
+pipe = Pipeline()
+pipe.add_source(UserInputText())
+pipe.add_source(JSONParser()) # To properly filter input - or
+pipe.add_source(ExecuteFunction(user_mapping_function)) # To let use inline a used defined function
+pipe.add_step(WeatherForecast())
+pipe.add_step(print)
+pipe.add_sink(DBStore())
+pipe.process()
+```
+
+### Adapter functions - inlining
 ```python
 pipe = Pipeline()
 pipe.add_source(UserInputText(), map = lambda x: x.strip())

pyproject.toml

@@ -23,10 +23,10 @@ dependencies = [
 
 [project.optional-dependencies]
 db_storage = [
-    "influxdb_client>=1.48.0"
+    "influxdb_client>=1.48.0",
 ]
 mqtt = [
-    "paho_mqtt>=2.1.0"
+    "paho_mqtt>=2.1.0",
 ]
 all = [
     "appslab_modules[db_storage]",