Edge Impulse container POC (#5)

Author: rjtokenring

containers/python-apps-base/appslab_modules-0.0.1-py3-none-any.whl

@@ -4,7 +4,7 @@ services:
     ports:
       - "${BIND_ADDRESS:-127.0.0.1}:${BIND_PORT:-8086}:8086"
     volumes:
-      - ./influxdb_data:/app/dbstorage/data
+      - "${APP_PATH:-.}/influx-data:/var/lib/influxdb2"
     environment:
       DOCKER_INFLUXDB_INIT_MODE: setup
       DOCKER_INFLUXDB_INIT_USERNAME: "${USERNAME:-admin}"

src/appslab/modules/dbstorage/module_compose.yaml

@@ -1,2 +1,4 @@
 name: db_storage
 module_description: "Simplified database storage layer for Arduino sensor data"
+require_container: true
+

src/appslab/modules/dbstorage/module_config.yaml

@@ -0,0 +1,66 @@
+import logging
+import requests
+from appslab.core import load_module_compose_file, parse_docker_compose_variable
+import logging
+import io
+
+logger = logging.getLogger(__name__)
+
+class EdgeImpulseObjectDetection:
+    def __init__(self):
+        infra = load_module_compose_file(self.__class__)
+        for k, v in infra['services'].items():
+            self.host = k
+            self.infra = v
+            break # Only one service is expected
+
+        self.port =  parse_docker_compose_variable(self.infra['ports'][0])[1][1] # Get the port, discard the host
+        self.url = f"http://{self.host}:{self.port}/api/image"
+        logger.info(f"[{self.__class__.__name__}] Host: {self.host} - Ports: {self.infra['ports']} - URL: {self.url}")
+
+    def detect_from_file(self, image_path :str) -> dict:
+        if not image_path or image_path == "":
+            return None
+        with open(image_path, 'rb') as f:
+            try:
+                return self.detect(image_bytes = f.read(), image_type = image_path.split('.')[-1])
+            except Exception as e:
+                logger.error(f"Error: {e}")
+                return None
+
+    def detect(self, image_bytes, image_type:str = "jpg") -> dict:
+        if not image_bytes or not image_type:
+            return None
+        if image_type not in ["jpg", "jpeg", "png"]:
+            logger.warning(f"[{self.__class__.__name__}] Invalid image type: {image_type}. Discarding image.")
+            return None
+        elif image_type == "jpg":
+            image_type = "jpeg"
+        
+        try:
+            logger.debug(f"[{self.__class__.__name__}] Detecting image of type: {image_type} -> {len(image_bytes)} bytes")
+
+            files = {'image': (f"image.{image_type}", io.BytesIO(image_bytes), f"image/{image_type}")}
+            response = requests.post(self.url, files=files)
+
+        except Exception as e:
+            logger.error(f"[{self.__class__.__name__}] Error: {e}")
+            return None
+
+        # Check the response
+        if response.status_code == 200:
+            return response.json()
+        else:
+            logger.warning(f"[{self.__class__}] error: {response.status_code}. Message: {response.text}")
+            return None
+        
+    def process(self, item):
+        if isinstance(item, str):
+            return self.parse(item)
+        elif isinstance(item, dict) and 'image' in item and item['image'] != "":
+            image = item['image']
+            image_type = image.split('.')[-1]
+            return self.detect(image, image_type.lower())
+
+        return item # No processing needed
+

src/appslab/modules/edge_impulse_classification/__init__.py

@@ -0,0 +1,7 @@
+services:
+  ei-inference-classification:
+    image: public.ecr.aws/g7a8t7v6/inference-container:48cd3f0d76c701a6070a27a8d9487d1733c155aa
+    restart: unless-stopped
+    ports:
+      - ${BIND_ADDRESS:-127.0.0.1}:${BIND_PORT:-1337}:1337
+    command: ["--api-key", "${EI_PROJECT_API_KEY:-x}", "--force-variant", "float32", "--force-engine", "tflite", "--run-http-server", "1337"]

src/appslab/modules/edge_impulse_classification/model_compose.yaml

@@ -0,0 +1,8 @@
+name: edge_impulse_classification
+module_description: "Image classification using Edge Impulse"
+require_container: true
+configuration_variables:
+  - name: EI_PROJECT_API_KEY
+    description: "Edge Impulse project API key"
+    default_value:
+    required: true

src/appslab/modules/edge_impulse_classification/module_config.yaml

@@ -2,22 +2,33 @@
 import logging
 
 logger = logging.getLogger(__name__)
-
 class Log:
-    def __init__(self):
+    def __init__(self, level = logging.INFO):
+        logging.basicConfig(level=level) 
         pass
 
+    def info(self, msg):
+        logger.info(msg)
+
+    def debug(self, msg):
+        logger.debug(msg)
+
+    def error(self, msg):
+        logger.error(msg)
+
+    def warning(self, msg):
+        logger.warning(msg)
+
     def print(self, item):
+        self.consume(item)
+
+    def consume(self, item):
         if item == None:
             return None
 
         logger.info(item)
 
         return item
 
-    def consume(self, item):
-        self.print(item)
-
     def process(self, item):
-        self.print(item)
-        return item
+        return self.consume(item)

src/appslab/modules/log/__init__.py

@@ -30,6 +30,10 @@ def on_connect(client, userdata, flags, reason_code, properties):
     MAX_RECONNECT_DELAY = 60
 
     def on_disconnect(client, userdata, flags, reason_code, properties):
+        if reason_code == 0:
+            logger.info("MQTT client disconnected gracefully")
+            return
+        
         logger.info("Disconnected with result code: %s", reason_code)
         reconnect_count, reconnect_delay = 0, FIRST_RECONNECT_DELAY
         while reconnect_count < MAX_RECONNECT_COUNT:
@@ -93,7 +97,7 @@ def __init__(self, broker_address:str, broker_port:int, topic:str, username:str,
 
         self.client.on_message = self._on_message
 
-        err = self.client.connect(broker_address, broker_port, 60)
+        self.client.connect(broker_address, broker_port, 60)
 
         self.client.loop_start()
 

src/appslab/modules/mqtt/__init__.py

@@ -1,2 +1,3 @@
 name: object_detection
 module_description: "Simple module for general purpose object detection."
+require_container: true