Refactoring

Author: rjtokenring

appslab_modules/db_storage/README.md …odules/sources/user_text_input/README.mdappslab_modules/db_storage/README.md renamed to appslab_modules/sources/user_text_input/README.md appslab_modules/db_storage/README.md renamed to appslab_modules/sources/user_text_input/README.md

@@ -0,0 +1,12 @@
+class UserTextInput:
+    def __init__(self, prompt: str):
+        self.prompt = prompt
+
+    def get(self):
+        return input(self.prompt)
+    
+    def produce(self):
+        try:
+            return input(self.prompt)
+        except KeyboardInterrupt:
+            return StopIteration

appslab_modules/sources/user_text_input/__init__.py

@@ -0,0 +1,2 @@
+name: user_text_input
+module_description: "User text input source."

appslab_modules/sources/user_text_input/module_config.yaml

@@ -0,0 +1,3 @@
+# Database storage module
+
+TODO: document usage

appslab_modules/steps/db_storage/README.md

@@ -27,10 +27,9 @@ def _connect(self):
                 self.client = client
                 self.write_api = client.write_api(write_precision=WritePrecision.MS)
                 self.query_api = client.query_api()
-                #Test connection
+                # Test connection
                 query = f'from(bucket: "{self.bucket}") |> range(start: -1m)'
                 self.query_api.query(query)
-                print("Connection and query successful!")
 
         except Exception as e:
             print(f"Error: {e}")
@@ -42,23 +41,30 @@ class DatabasePersistence(_InfluxPersistence):
     def __init__(self, host:str, port:int):
         super().__init__(host, port)
 
-    def write_sample(self, measurement_name :str, measure :str, value :any, ts :int = 0):
+    def write_sample(self, measure :str, value :any, ts :int = 0, measurement_name :str = "arduino"):
         try:
             if ts <= 0:
                 ts = int(time.time_ns() / 1000000)
             point = Point(measurement_name).field(measure, value).time(ts, WritePrecision.MS)
 
             self.write_api.write(bucket=self.bucket, record=point)
-            print("Data written successfully!")
 
         except Exception as e:
             print(f"Error: {e}")
 
+    def process(self, item):
+        if isinstance(item, dict):
+            for k, v in item.items():
+                self.write_sample(k, v)
+
+        # After processing, return the original item
+        return item
+
 class DatabaseRetrieval(_InfluxPersistence):
     def __init__(self, host:str, port:int):
         super().__init__(host, port)
 
-    def read_last_sample(self, measurement_name :str, measure :str):
+    def read_last_sample(self, measure :str, measurement_name :str = "arduino") -> tuple:
         try:
             query = f"""
             from(bucket: "{self.bucket}")
@@ -73,10 +79,24 @@ def read_last_sample(self, measurement_name :str, measure :str):
             if result:
               for table in result:
                   for record in table.records:
-                      return record.get_time(), record.get_value()
+                      return measure, record.get_time(), record.get_value()
             else:
-                print(f"No data found for {measurement_name} and {measure}")
+                return None
 
         except Exception as e:
             print(f"Error: {e}")
+
+    def process(self, item):
+        output = {}
+        if isinstance(item, str):
+            data = self.read_last_sample(item)
+            if data:
+                output[data[0]] = (data[1], data[2])
+        elif isinstance(item, dict):
+            for k in item:
+                data = self.read_last_sample(k)
+                if data:
+                    output[data[0]] = (data[1], data[2])
+
+        return output