Module refactoring

Author: rjtokenring

appslab_modules/sinks/mqtt/__init__.py

@@ -0,0 +1,20 @@
+import paho.mqtt.client as mqtt
+
+class MQTTWriter:
+    def __init__(self, broker_address:str, broker_port:int, topic:str, username:str, password:str):
+        self.client = mqtt.Client()
+        self.client.username_pw_set(username, password)
+        self.client.connect(broker_address, broker_port, 60)
+        self.topic = topic
+
+    def close(self):
+        self.client.disconnect()
+
+    def write(self, message:str):
+        if message and message != "":
+            self.client.publish(self.topic, message)
+
+    def consume(self, item):
+        if isinstance(item, str):
+            self.write(item)
+        return item

appslab_modules/sinks/mqtt/module_config.yaml

@@ -0,0 +1,2 @@
+name: mqtt_sink
+module_description: "MQTT output writer"

appslab_modules/steps/db_storage/README.md

@@ -1,3 +1,12 @@
 # Database storage module
 
 TODO: document usage
+
+
+# Local testing
+
+Startup containarized infrastructure using
+
+```sh
+docker compose -f module_compose.yaml up -d
+```

appslab_modules/steps/db_storage/database_storage.py

@@ -17,7 +17,7 @@ def __init__(self, host:str = "db_storage", port:int = 8086):
         self._connect()
 
     def load_default_infra(self):
-        with open("compose.yaml", "r") as file:
+        with open("module_compose.yaml", "r") as file:
             config = yaml.safe_load(file)
             return config
 
@@ -27,9 +27,6 @@ def _connect(self):
                 self.client = client
                 self.write_api = client.write_api(write_precision=WritePrecision.MS)
                 self.query_api = client.query_api()
-                # Test connection
-                query = f'from(bucket: "{self.bucket}") |> range(start: -1m)'
-                self.query_api.query(query)
 
         except Exception as e:
             print(f"Error: {e}")
@@ -38,7 +35,7 @@ def close(self):
         self.client.close()        
 
 class DatabasePersistence(_InfluxPersistence):
-    def __init__(self, host:str, port:int):
+    def __init__(self, host:str = "db_storage", port:int = 8086):
         super().__init__(host, port)
 
     def write_sample(self, measure :str, value :any, ts :int = 0, measurement_name :str = "arduino"):
@@ -61,14 +58,14 @@ def process(self, item):
         return item
 
 class DatabaseRetrieval(_InfluxPersistence):
-    def __init__(self, host:str, port:int):
+    def __init__(self, host:str = "db_storage", port:int = 8086):
         super().__init__(host, port)
 
     def read_last_sample(self, measure :str, measurement_name :str = "arduino") -> tuple:
         try:
             query = f"""
             from(bucket: "{self.bucket}")
-            |> range(start: -1d) # look back 1 day
+            |> range(start: -1d)
             |> filter(fn: (r) => r["_measurement"] == "{measurement_name}")
             |> filter(fn: (r) => r["_field"] == "{measure}")
             |> last()
@@ -79,7 +76,7 @@ def read_last_sample(self, measure :str, measurement_name :str = "arduino") -> t
             if result:
               for table in result:
                   for record in table.records:
-                      return measure, record.get_time(), record.get_value()
+                      return measure, record.get_time().isoformat(), record.get_value()
             else:
                 return None
 

appslab_modules/steps/db_storage/test/test_database_storage.py

@@ -0,0 +1,37 @@
+import unittest
+from database_storage import DatabasePersistence, DatabaseRetrieval  # Assuming this is in database_storage.py
+
+# NOTE: before test, start dependencies with "docker compose -f module_compose.yaml up -d"
+class TestDatabaseStorage(unittest.TestCase):
+
+    def setUp(self):
+        """Set up test fixtures."""
+        self.db = DatabasePersistence(host="localhost")
+        self.dbread = DatabaseRetrieval(host="localhost")
+
+    def tearDown(self):
+        """Clean up after tests."""
+        self.db.close()
+        self.dbread.close()
+
+    def test_write_and_read_string(self):
+        """Test writing and reading a string sample."""
+        self.db.write_sample("test_measurement_str", "test_string")
+        val = self.dbread.read_last_sample("test_measurement_str")
+        self.assertEqual(val[2], "test_string")
+        self.assertEqual(val[0], "test_measurement_str")
+
+    def test_write_and_read_integer(self):
+        """Test writing and reading an integer sample."""
+        self.db.write_sample("test_measurement", 1111)
+        self.db.write_sample("test_measurement", 1234)
+        val = self.dbread.read_last_sample("test_measurement")
+        self.assertEqual(val[2], 1234)
+
+    def test_read_nonexistent_measurement(self):
+        """Test reading a non-existent measurement."""
+        val = self.dbread.read_last_sample("nonexistent_measurement")
+        self.assertIsNone(val) #or whatever the expected behavior is in your case.
+
+if __name__ == '__main__':
+    unittest.main()

appslab_modules/steps/db_storage/test_database_storage.py

@@ -25,8 +25,12 @@ dependencies = [
 db_storage = [
     "influxdb_client>=1.48.0"
 ]
+mqtt = [
+    "paho_mqtt>=2.1.0"
+]
 all = [
     "appslab_modules[db_storage]",
+    "appslab_modules[mqtt]",
 ]
 
 [project.urls]