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(AudioClassification, 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

@@ -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(self.__class__, features)
             spotted_movement = self._movement_spotted(ret)
             if spotted_movement is not None:
                 keyword, confidence, complete_detection = spotted_movement

src/arduino/app_bricks/vibration_anomaly_detection/__init__.py

@@ -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(self.__class__, features)
             logger.debug(f"Inference result: {ret}")
             spotted_anomaly = self._extract_anomaly_score(ret)
             if spotted_anomaly is not None:

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."""
@@ -44,7 +42,7 @@ def __init__(self, mic: Microphone = None, confidence: float = 0.8, debounce_sec
             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(self.__class__, features)
+            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,53 @@ 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.
+    @staticmethod
+    def infer_from_features(cls, features: list) -> dict | None:
+        """
+        Infer from features using the Edge Impulse API.
 
         Args:
+            cls: The class to use for infra loading.
             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(cls)
+            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:
+    @staticmethod
+    def get_model_info(cls) -> EdgeImpulseModelInfo | None:
         """Get model information from the Edge Impulse API.
 
         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 +234,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