bug #397: new --fix-audio argument for fixing wrongly initialized respeaker

Author: FabienDanieau

src/reachy_mini/daemon/app/main.py

@@ -55,6 +55,8 @@ class Args:
 
     localhost_only: bool | None = None
 
+    fix_audio: bool = False
+
 
 def create_app(args: Args, health_check_event: asyncio.Event | None = None) -> FastAPI:
     """Create and configure the FastAPI application."""
@@ -152,6 +154,17 @@ def run_app(args: Args) -> None:
     """Run the FastAPI app with Uvicorn."""
     logging.basicConfig(level=logging.INFO)
 
+    if args.fix_audio:
+        """Temporary fix for audio issues with beta version."""
+        logging.info("Applying audio fix for beta version.")
+
+        from reachy_mini.media.audio_control_utils import init_respeaker_usb
+
+        respeaker = init_respeaker_usb()
+        respeaker.write("REBOOT", [1])
+        respeaker.close()
+        logging.debug("Respeaker rebooted.")
+
     health_check_event = asyncio.Event()
     app = create_app(args, health_check_event)
 
@@ -314,6 +327,12 @@ def main() -> None:
         help="Set the logging level (default: INFO).",
     )
 
+    parser.add_argument(
+        "--fix-audio",
+        action="store_true",
+        help="Fix audio issues with beta version (default: False).",
+    )
+
     args = parser.parse_args()
     run_app(Args(**vars(args)))
 

src/reachy_mini/media/audio_base.py

@@ -5,39 +5,30 @@
 """
 
 import logging
-import struct
 from abc import ABC, abstractmethod
-from typing import List, Optional
+from typing import Optional
 
 import numpy as np
 import numpy.typing as npt
-import usb
-from libusb_package import get_libusb1_backend
+
+from reachy_mini.media.audio_control_utils import ReSpeaker, init_respeaker_usb
 
 
 class AudioBase(ABC):
     """Abstract class for opening and managing audio devices."""
 
     SAMPLE_RATE = 16000  # respeaker samplerate
-    TIMEOUT = 100000
-    PARAMETERS = {
-        "VERSION": (48, 0, 4, "ro", "uint8"),
-        "AEC_AZIMUTH_VALUES": (33, 75, 16 + 1, "ro", "radians"),
-        "DOA_VALUE": (20, 18, 4 + 1, "ro", "uint16"),
-        "DOA_VALUE_RADIANS": (20, 19, 8 + 1, "ro", "radians"),
-    }
 
     def __init__(self, log_level: str = "INFO") -> None:
         """Initialize the audio device."""
         self.logger = logging.getLogger(__name__)
         self.logger.setLevel(log_level)
-        self._respeaker = self._init_respeaker_usb()
-        # name, resid, cmdid, length, type
+        self._respeaker: ReSpeaker = init_respeaker_usb()
 
     def __del__(self) -> None:
         """Destructor to ensure resources are released."""
         if self._respeaker:
-            usb.util.dispose_resources(self._respeaker)
+            self._respeaker.close()
 
     @abstractmethod
     def start_recording(self) -> None:
@@ -79,63 +70,6 @@ def play_sound(self, sound_file: str) -> None:
         """
         pass
 
-    def _init_respeaker_usb(self) -> Optional[usb.core.Device]:
-        try:
-            dev = usb.core.find(
-                idVendor=0x2886, idProduct=0x001A, backend=get_libusb1_backend()
-            )
-            return dev
-        except usb.core.NoBackendError:
-            self.logger.error(
-                "No USB backend was found ! Make sure libusb_package is correctly installed with `pip install libusb_package`."
-            )
-            return None
-
-    def _read_usb(self, name: str) -> Optional[List[int] | List[float]]:
-        try:
-            data = self.PARAMETERS[name]
-        except KeyError:
-            self.logger.error(f"Unknown parameter: {name}")
-            return None
-
-        if not self._respeaker:
-            self.logger.warning("ReSpeaker device not found.")
-            return None
-
-        resid = data[0]
-        cmdid = 0x80 | data[1]
-        length = data[2]
-
-        response = self._respeaker.ctrl_transfer(
-            usb.util.CTRL_IN
-            | usb.util.CTRL_TYPE_VENDOR
-            | usb.util.CTRL_RECIPIENT_DEVICE,
-            0,
-            cmdid,
-            resid,
-            length,
-            self.TIMEOUT,
-        )
-
-        self.logger.debug(f"Response for {name}: {response}")
-
-        result: Optional[List[float] | List[int]] = None
-        if data[4] == "uint8":
-            result = response.tolist()
-        elif data[4] == "radians":
-            byte_data = response.tobytes()
-            num_values = (data[2] - 1) / 4
-            match_str = "<"
-            for i in range(int(num_values)):
-                match_str += "f"
-            result = [
-                float(x) for x in struct.unpack(match_str, byte_data[1 : data[2]])
-            ]
-        elif data[4] == "uint16":
-            result = response.tolist()
-
-        return result
-
     def get_DoA(self) -> tuple[float, bool] | None:
         """Get the Direction of Arrival (DoA) value from the ReSpeaker device.
 
@@ -153,7 +87,8 @@ def get_DoA(self) -> tuple[float, bool] | None:
         if not self._respeaker:
             self.logger.warning("ReSpeaker device not found.")
             return None
-        result = self._read_usb("DOA_VALUE_RADIANS")
+
+        result = self._respeaker.read("DOA_VALUE_RADIANS")
         if result is None:
             return None
         return float(result[0]), bool(result[1])

src/reachy_mini/media/audio_sounddevice.py

@@ -30,8 +30,12 @@ def __init__(
         self.stream = None
         self._output_stream = None
         self._buffer: List[npt.NDArray[np.float32]] = []
-        self._output_device_id = self.get_output_device_id("respeaker")
-        self._input_device_id = self.get_input_device_id("respeaker")
+        self._output_device_id = self.get_device_id(
+            ["Reachy Mini Audio", "respeaker"], device_io_type="output"
+        )
+        self._input_device_id = self.get_device_id(
+            ["Reachy Mini Audio", "respeaker"], device_io_type="input"
+        )
 
     def start_recording(self) -> None:
         """Open the audio input stream, using ReSpeaker card if available."""
@@ -181,43 +185,27 @@ def _clean_up_thread() -> None:
                 daemon=True,
             ).start()
 
-    def get_output_device_id(self, name_contains: str) -> int:
+    def get_device_id(
+        self, names_contains: List[str], device_io_type: str = "output"
+    ) -> int:
         """Return the output device id whose name contains the given string (case-insensitive).
 
         If not found, return the default output device id.
         """
         devices = sd.query_devices()
 
         for idx, dev in enumerate(devices):
-            if (
-                name_contains.lower() in dev["name"].lower()
-                and dev["max_output_channels"] > 0
-            ):
-                return idx
+            for name_contains in names_contains:
+                if (
+                    name_contains.lower() in dev["name"].lower()
+                    and dev[f"max_{device_io_type}_channels"] > 0
+                ):
+                    return idx
         # Return default output device if not found
         self.logger.warning(
-            f"No output device found containing '{name_contains}', using default."
-        )
-        return self._safe_query_device("output")
-
-    def get_input_device_id(self, name_contains: str) -> int:
-        """Return the input device id whose name contains the given string (case-insensitive).
-
-        If not found, return the default input device id.
-        """
-        devices = sd.query_devices()
-
-        for idx, dev in enumerate(devices):
-            if (
-                name_contains.lower() in dev["name"].lower()
-                and dev["max_input_channels"] > 0
-            ):
-                return idx
-        # Return default input device if not found
-        self.logger.warning(
-            f"No input device found containing '{name_contains}', using default."
+            f"No {device_io_type} device found containing '{name_contains}', using default."
         )
-        return self._safe_query_device("input")
+        return self._safe_query_device(device_io_type)
 
     def _safe_query_device(self, kind: str) -> int:
         try:

src/reachy_mini/media/audio_utils.py

@@ -14,12 +14,20 @@ def get_respeaker_card_number() -> int:
 
         lines = output.split("\n")
         for line in lines:
-            if "respeaker" in line.lower() and "card" in line:
+            if "reachy mini audio" in line.lower() and "card" in line:
                 card_number = line.split(":")[0].split("card ")[1].strip()
-                logging.debug(f"Found ReSpeaker sound card: {card_number}")
+                logging.debug(f"Found Reachy Mini Audio sound card: {card_number}")
+                return int(card_number)
+            elif "respeaker" in line.lower() and "card" in line:
+                card_number = line.split(":")[0].split("card ")[1].strip()
+                logging.warning(
+                    f"Found ReSpeaker sound card: {card_number}. Please update firmware!"
+                )
                 return int(card_number)
 
-        logging.warning("ReSpeaker sound card not found. Returning default card")
+        logging.warning(
+            "Reachy Mini Audio sound card not found. Returning default card"
+        )
         return 0  # default sound card
 
     except subprocess.CalledProcessError as e: