refactor ei

Author: stefanotorneo

src/arduino/app_bricks/audio_classification/__init__.py

@@ -6,9 +6,10 @@
 import wave
 from typing import Callable
 
-from arduino.app_internal.core.audio import AudioDetector, NO_MIC as NO_MIC
+from arduino.app_internal.core.audio import AudioDetector
 from arduino.app_peripherals.microphone import Microphone
 from arduino.app_utils import brick, Logger
+from arduino.app_internal.core import EdgeImpulseRunnerFacade
 
 logger = Logger("AudioClassification")
 
@@ -28,8 +29,6 @@ def __init__(self, mic: Microphone = None, confidence: float = 0.8):
 
         Args:
             mic (Microphone, optional): Microphone instance used as the audio source. If None, a default Microphone will be initialized.
-                If NO_MIC is passed, no microphone will be initialized, and only file-based classification
-                will be available.
             confidence (float, optional): Minimum confidence threshold (0.0–1.0) required
                 for a detection to be considered valid. Defaults to 0.8 (80%).
 
@@ -68,7 +67,8 @@ def stop(self):
         """
         super().stop()
 
-    def classify_from_file(self, audio_path: str, confidence: int = None) -> dict | None:
+    @staticmethod
+    def classify_from_file(audio_path: str, confidence: int) -> dict | None:
         """Classify audio content from a WAV file.
 
         Supported sample widths:
@@ -91,10 +91,10 @@ def classify_from_file(self, audio_path: str, confidence: int = None) -> dict |
 
         Raises:
             AudioClassificationException: If the file cannot be found, read, or processed.
-            ValueError: If the file uses an unsupported sample width.
+            ValueError: If the file uses an unsupported sample width or if confidence is not specified.
         """
         if confidence is None:
-            confidence = self.confidence
+            raise ValueError("Confidence level must be specified.")
 
         try:
             with wave.open(audio_path, "rb") as wf:
@@ -127,8 +127,8 @@ def classify_from_file(self, audio_path: str, confidence: int = None) -> dict |
                 else:
                     raise ValueError(f"Unsupported sample width: {samp_width} bytes. Cannot process this WAV file.")
 
-                classification = super().infer_from_features(features[: int(self.model_info.input_features_count)])
-                best_match = super().get_best_match(classification, confidence)
+                classification = EdgeImpulseRunnerFacade.infer_from_features(features)
+                best_match = AudioDetector.get_best_match(classification, confidence)
                 if not best_match:
                     return None
                 keyword, confidence = best_match

src/arduino/app_bricks/audio_classification/examples/2_glass_breaking_from_file.py

@@ -4,9 +4,7 @@
 
 # EXAMPLE_NAME = "Detect the glass breaking sound from audio file"
 # EXAMPLE_REQUIRES = "Requires an audio file with the glass breaking sound."
-from arduino.app_bricks.audio_classification import AudioClassification, NO_MIC
+from arduino.app_bricks.audio_classification import AudioClassification
 
-classifier = AudioClassification(mic=NO_MIC)
-
-classification = classifier.classify_from_file("glass_breaking.wav")
+classification = AudioClassification.classify_from_file("glass_breaking.wav", confidence=0.8)
 print("Result:", classification)

src/arduino/app_bricks/motion_detection/__init__.py

@@ -26,7 +26,7 @@ def __init__(self, confidence: float = 0.4):
         """
         self._confidence = confidence
         super().__init__()
-        model_info = self.get_model_info()
+        model_info = EdgeImpulseRunnerFacade.get_model_info()
         if not model_info:
             raise ValueError("Failed to retrieve model information. Ensure the EI model runner service is running.")
         if model_info.frequency <= 0 or model_info.input_features_count <= 0:
@@ -133,7 +133,7 @@ def _detection_loop(self):
             return
 
         try:
-            ret = super().infer_from_features(features[: int(self._model_info.input_features_count)].flatten().tolist())
+            ret = EdgeImpulseRunnerFacade.infer_from_features(features.tolist())
             spotted_movement = self._movement_spotted(ret)
             if spotted_movement is not None:
                 keyword, confidence, complete_detection = spotted_movement

src/arduino/app_bricks/object_detection/__init__.py

@@ -31,7 +31,7 @@ def __init__(self, confidence: float = 0.3):
         """
         self.confidence = confidence
         super().__init__()
-        self._model_info = self.get_model_info()
+        self._model_info = EdgeImpulseRunnerFacade.get_model_info()
         if not self._model_info:
             raise ValueError("Failed to retrieve model information. Ensure the Edge Impulse service is running.")
 

src/arduino/app_bricks/vibration_anomaly_detection/__init__.py

@@ -48,7 +48,7 @@ def __init__(self, anomaly_detection_threshold: float = 1.0):
         """
         self._anomaly_detection_threshold = anomaly_detection_threshold
         super().__init__()
-        model_info = self.get_model_info()
+        model_info = EdgeImpulseRunnerFacade.get_model_info()
         if not model_info:
             raise ValueError("Failed to retrieve model information. Ensure the EI model runner service is running.")
         if model_info.frequency <= 0 or model_info.input_features_count <= 0:
@@ -133,7 +133,7 @@ def loop(self):
             if features is None or len(features) == 0:
                 return
 
-            ret = super().infer_from_features(features[: int(self._model_info.input_features_count)].flatten().tolist())
+            ret = EdgeImpulseRunnerFacade.infer_from_features(features.tolist())
             logger.debug(f"Inference result: {ret}")
             spotted_anomaly = self._extract_anomaly_score(ret)
             if spotted_anomaly is not None:
@@ -184,6 +184,14 @@ def stop(self):
         """
         self._clear()
 
+    def get_model_info(self):
+        """Get the Edge Impulse model information used by this detector.
+
+        Returns:
+            EdgeImpulseModelInfo: The model information object.
+        """
+        return self._model_info
+
     def _clear(self):
         """Internal helper: flush the sensor data buffer and log the action.
 

src/arduino/app_internal/core/audio.py

@@ -13,8 +13,6 @@
 
 logger = Logger(__name__)
 
-NO_MIC = object()  # Sentinel value for no microphone
-
 
 class AudioDetector(EdgeImpulseRunnerFacade):
     """AudioDetector module for detecting sounds and classifying audio using a specified model."""
@@ -37,14 +35,14 @@ def __init__(self, mic: Microphone = None, confidence: float = 0.8, debounce_sec
         self._debounce_sec = debounce_sec
         self._last_detected = {}
 
-        model_info = self.get_model_info()
+        model_info = EdgeImpulseRunnerFacade.get_model_info()
         if not model_info:
             raise ValueError("Failed to retrieve model information. Ensure the Edge Impulse service is running.")
         if model_info.frequency <= 0 or model_info.input_features_count <= 0:
             raise ValueError("Model parameters are missing or incomplete in the retrieved model information.")
         self.model_info = model_info
 
-        self._mic = None if mic is NO_MIC else (mic or Microphone(sample_rate=model_info.frequency, channels=model_info.axis_count))
+        self._mic = mic if mic else Microphone(sample_rate=model_info.frequency, channels=model_info.axis_count)
         self._mic_lock = threading.Lock()
 
         self._window_size = int(model_info.input_features_count / model_info.axis_count)
@@ -87,17 +85,24 @@ def stop(self):
             self._mic.stop()
         self._buffer.flush()
 
-    def get_best_match(self, item: dict, confidence: int = None) -> tuple[str, float] | None:
+    @staticmethod
+    def get_best_match(item: dict, confidence: float) -> tuple[str, float] | None:
         """Extract the best matched keyword from the classification results.
 
         Args:
         item (dict): The classification result from the inference.
-        confidence (int): The confidence threshold for classification. If None, uses the instance's confidence level.
+        confidence (float): The confidence threshold for classification.
 
         Returns:
         tuple[str, float] | None: The best matched keyword and its confidence, or None if no match is found.
+
+        Raises:
+        ValueError: If confidence level is not provided.
         """
-        classification = _extract_classification(item, confidence or self.confidence)
+        if confidence is None:
+            raise ValueError("Confidence level must be provided.")
+
+        classification = _extract_classification(item, confidence)
         if not classification:
             return None
 
@@ -143,8 +148,8 @@ def _inference_loop(self):
 
         logger.debug(f"Processing sensor data with {len(features)} features.")
         try:
-            ret = super().infer_from_features(features[: int(self.model_info.input_features_count)].tolist())
-            spotted_keyword = self.get_best_match(ret)
+            ret = EdgeImpulseRunnerFacade.infer_from_features(features.tolist())
+            spotted_keyword = AudioDetector.get_best_match(ret, self.confidence)
             if spotted_keyword:
                 keyword, confidence = spotted_keyword
                 keyword = keyword.lower()

src/arduino/app_internal/core/ei.py

@@ -43,17 +43,8 @@ class EdgeImpulseRunnerFacade:
 
     def __init__(self):
         """Initialize the EdgeImpulseRunnerFacade with the API path."""
-        infra = load_brick_compose_file(self.__class__)
-        for k, v in infra["services"].items():
-            self.host = k
-            self.infra = v
-            break  # Only one service is expected
-
-        self.host = resolve_address(self.host)
-
-        self.port = 1337  # Default EI HTTP port
-        self.url = f"http://{self.host}:{self.port}"
-        logger.warning(f"[{self.__class__.__name__}] Host: {self.host} - Ports: {self.port} - URL: {self.url}")
+        self.url = _get_ei_url(self.__class__)
+        logger.warning(f"[{self.__class__.__name__}] URL: {self.url}")
 
     def infer_from_file(self, image_path: str) -> dict | None:
         if not image_path or image_path == "":
@@ -124,47 +115,56 @@ def process(self, item):
             logger.error(f"[{self.__class__}] Error processing file {item}: {e}")
         return None
 
-    def infer_from_features(self, features: list) -> dict | None:
-        """Infer from features using the Edge Impulse API.
+    @classmethod
+    def infer_from_features(cls, features: list) -> dict | None:
+        """
+        Infer from features using the Edge Impulse API.
 
         Args:
+            cls: The class method caller.
             features (list): A list of features to send to the Edge Impulse API.
 
         Returns:
             dict | None: The response from the Edge Impulse API as a dictionary, or None if an error occurs.
         """
         try:
-            response = requests.post(f"{self.url}/api/features", json={"features": features})
+            url = _get_ei_url(cls)
+            model_info = EdgeImpulseRunnerFacade.get_model_info()
+            features = features[: int(model_info.input_features_count)]
+
+            response = requests.post(f"{url}/api/features", json={"features": features})
             if response.status_code == 200:
                 return response.json()
             else:
-                logger.warning(f"[{self.__class__}] error: {response.status_code}. Message: {response.text}")
+                logger.warning(f"[{cls.__name__}] error: {response.status_code}. Message: {response.text}")
                 return None
         except Exception as e:
-            logger.error(f"[{self.__class__.__name__}] Error: {e}")
+            logger.error(f"[{cls.__name__}] Error: {e}")
             return None
 
-    def get_model_info(self) -> EdgeImpulseModelInfo | None:
+    @classmethod
+    def get_model_info(cls) -> EdgeImpulseModelInfo | None:
         """Get model information from the Edge Impulse API.
 
+        Args:
+            cls: The class method caller.
+
         Returns:
             model_info (EdgeImpulseModelInfo | None): An instance of EdgeImpulseModelInfo containing model details, None if an error occurs.
         """
-        if not self.host or not self.port:
-            logger.error(f"[{self.__class__}] Host or port not set. Cannot fetch model info.")
-            return None
+        url = _get_ei_url(cls)
 
         http_client = HttpClient(total_retries=6)  # Initialize the HTTP client with retry logic
         try:
-            response = http_client.request_with_retry(f"{self.url}/api/info")
+            response = http_client.request_with_retry(f"{url}/api/info")
             if response.status_code == 200:
-                logger.debug(f"[{self.__class__.__name__}] Fetching model info from {self.url}/api/info -> {response.status_code} {response.json}")
+                logger.debug(f"[{cls.__name__}] Fetching model info from {url}/api/info -> {response.status_code} {response.json}")
                 return EdgeImpulseModelInfo(response.json())
             else:
-                logger.warning(f"[{self.__class__}] Error fetching model info: {response.status_code}. Message: {response.text}")
+                logger.warning(f"[{cls}] Error fetching model info: {response.status_code}. Message: {response.text}")
                 return None
         except Exception as e:
-            logger.error(f"[{self.__class__}] Error fetching model info: {e}")
+            logger.error(f"[{cls}] Error fetching model info: {e}")
             return None
         finally:
             http_client.close()  # Close the HTTP client session
@@ -237,3 +237,14 @@ def _extract_anomaly_score(self, item: dict):
                 return class_results["anomaly"]
 
         return None
+
+
+def _get_ei_url(cls):
+    infra = load_brick_compose_file(cls)
+    for k, v in infra["services"].items():
+        host = k
+        break
+    host = resolve_address(host)
+    port = 1337
+    url = f"http://{host}:{port}"
+    return url

tests/arduino/app_bricks/motion_detection/test_motion_detection.py

@@ -24,10 +24,10 @@ def app_instance(monkeypatch):
 @pytest.fixture(autouse=True)
 def mock_dependencies(monkeypatch: pytest.MonkeyPatch):
     """Mock out docker-compose lookups and image helpers."""
-    fake_compose = {"services": {"models-runner": {"ports": ["${BIND_ADDRESS:-127.0.0.1}:${BIND_PORT:-8100}:8100"]}}}
-    monkeypatch.setattr("arduino.app_internal.core.load_brick_compose_file", lambda cls: fake_compose)
+    fake_compose = {"services": {"ei-inference": {"ports": ["${BIND_ADDRESS:-127.0.0.1}:${BIND_PORT:-1337}:1337"]}}}
+    monkeypatch.setattr("arduino.app_internal.core.ei.load_brick_compose_file", lambda cls: fake_compose)
     monkeypatch.setattr("arduino.app_internal.core.resolve_address", lambda host: "127.0.0.1")
-    monkeypatch.setattr("arduino.app_internal.core.parse_docker_compose_variable", lambda x: [(None, None), (None, "8200")])
+    monkeypatch.setattr("arduino.app_internal.core.parse_docker_compose_variable", lambda x: [(None, None), (None, "1337")])
 
     class FakeResp:
         status_code = 200

tests/arduino/app_bricks/objectdetection/test_objectdetection.py

@@ -7,6 +7,7 @@
 import io
 from PIL import Image
 from arduino.app_bricks.object_detection import ObjectDetection
+from arduino.app_utils import HttpClient
 
 
 class ModelInfo:
@@ -20,12 +21,65 @@ def mock_dependencies(monkeypatch: pytest.MonkeyPatch):
 
     This is needed to avoid network calls and other side effects.
     """
-    fake_compose = {"services": {"models-runner": {"ports": ["${BIND_ADDRESS:-127.0.0.1}:${BIND_PORT:-8100}:8100"]}}}
-    monkeypatch.setattr("arduino.app_internal.core.load_brick_compose_file", lambda cls: fake_compose)
+    fake_compose = {"services": {"ei-inference": {"ports": ["${BIND_ADDRESS:-127.0.0.1}:${BIND_PORT:-1337}:1337"]}}}
+    monkeypatch.setattr("arduino.app_internal.core.ei.load_brick_compose_file", lambda cls: fake_compose)
     monkeypatch.setattr("arduino.app_internal.core.resolve_address", lambda host: "127.0.0.1")
-    monkeypatch.setattr("arduino.app_internal.core.parse_docker_compose_variable", lambda x: [(None, None), (None, "8100")])
+    monkeypatch.setattr("arduino.app_internal.core.parse_docker_compose_variable", lambda x: [(None, None), (None, "1337")])
     monkeypatch.setattr("arduino.app_bricks.object_detection.ObjectDetection.get_model_info", lambda self: ModelInfo("object-detection"))
 
+    class FakeResp:
+        status_code = 200
+
+        def json(self):
+            return {
+                "project": {
+                    "deploy_version": 11,
+                    "id": 774707,
+                    "impulse_id": 1,
+                    "impulse_name": "Time series data, Spectral Analysis, Classification (Keras), Anomaly Detection (K-means)",
+                    "name": "Fan Monitoring - Advanced Anomaly Detection",
+                    "owner": "Arduino",
+                },
+                "modelParameters": {
+                    "has_visual_anomaly_detection": False,
+                    "axis_count": 3,
+                    "frequency": 100,
+                    "has_anomaly": 1,
+                    "has_object_tracking": False,
+                    "has_performance_calibration": False,
+                    "image_channel_count": 0,
+                    "image_input_frames": 0,
+                    "image_input_height": 0,
+                    "image_input_width": 0,
+                    "image_resize_mode": "none",
+                    "inferencing_engine": 4,
+                    "input_features_count": 600,
+                    "interval_ms": 10,
+                    "label_count": 2,
+                    "labels": ["nominal", "off"],
+                    "model_type": "classification",
+                    "sensor": 2,
+                    "slice_size": 50,
+                    "thresholds": [],
+                    "use_continuous_mode": False,
+                    "sensorType": "accelerometer",
+                },
+            }
+
+    def fake_get(
+        self,
+        url: str,
+        method: str = "GET",
+        data: dict | str = None,
+        json: dict = None,
+        headers: dict = None,
+        timeout: int = 5,
+    ):
+        return FakeResp()
+
+    # Mock the requests.get method to return a fake response
+    monkeypatch.setattr(HttpClient, "request_with_retry", fake_get)
+
 
 @pytest.fixture
 def detector():