Merge pull request #33 from pollen-robotics/clean_console_output

README.md

@@ -3,10 +3,14 @@
 Working repo, we should turn this into a ReachyMini app at some point maybe ?
 
 ## Installation
+You can set up the project quickly using [uv](https://docs.astral.sh/uv/):
 
 ```bash
-pip install -e .
+uv venv --python 3.12.1  # Create a virtual environment with Python 3.12.1
+source .venv/bin/activate
+uv sync
 ```
+> Note: The `pyproject.toml` expects `reachy-mini-dances-library` to be located in the same directory as this project.
 
 ## Run
 

pyproject.toml

@@ -10,44 +10,77 @@ description = ""
 readme = "README.md"
 requires-python = ">=3.10"
 dependencies = [
-    # "reachy_mini@git+ssh://git@github.com/pollen-robotics/reachy_mini@develop",
-    "openai",
+    "aiortc>=1.13.0",
     "fastrtc",
+    "gradio>=5.49.0",
+    "huggingface_hub>=0.34.4",
+    "mediapipe>=0.10.14",
+    "num2words",
     "onnxruntime",
-    "PyGObject>=3.42.2, <=3.46.0",
+    "openai>=2.1",
+    "PyGObject>=3.42.2,<=3.46.0",
+    "python-dotenv",
+    "reachy_mini_dances_library",
+    "reachy_mini_toolbox",
+    "reachy_mini",
+    "supervision",
     "torch",
     "transformers",
-    "num2words",
-    "dotenv",
     "ultralytics",
-    "supervision",
-    "reachy_mini_toolbox@git+ssh://git@github.com/pollen-robotics/reachy_mini_toolbox@main",
-    "reachy_mini_dances_library@git+ssh://git@github.com/pollen-robotics/reachy_mini_dances_library@main"
 ]
 
+[project.scripts]
+reachy-mini-conversation-demo = "reachy_mini_conversation_demo.main:main"
 
-[project.optional-dependencies]
+[dependency-groups]
 dev = ["pytest", "ruff==0.12.0"]
 
-[project.scripts]
-reachy-mini-conversation-demo = "reachy_mini_conversation_demo.main:main"
+[tool.uv.sources]
+reachy_mini_dances_library = { path = "../reachy_mini_dances_library", editable = true }
+reachy_mini = { git = "ssh://git@github.com/pollen-robotics/reachy_mini.git", branch = "develop" }
+reachy_mini_toolbox = { git = "ssh://git@github.com/pollen-robotics/reachy_mini_toolbox.git", branch = "main" }
+fastrtc = { git = "ssh://git@github.com/gradio-app/fastrtc.git", branch = "main" }
 
 [tool.setuptools]
 package-dir = { "" = "src" }
 include-package-data = true
 
-
 [tool.setuptools.packages.find]
 where = ["src"]
 
-
 [tool.setuptools.package-data]
-reachy_mini = ["**/*"] # Inclut tous les fichiers non .py
+reachy_mini_conversation_demo = ["images/*"]
 
 [tool.ruff]
-exclude = []
-lint.extend-select = ["I", "D"]
-lint.ignore = [
-    "D203", # Incompatible with D211
-    "D213", # Incompatible with D212
+line-length = 119
+exclude = [".venv", "dist", "build", "**/__pycache__", "*.egg-info", ".mypy_cache", ".pytest_cache"]
+
+[tool.ruff.lint]
+select = [
+  "E",   # pycodestyle errors
+  "F",   # pyflakes
+  "W",   # pycodestyle warnings
+  "I",   # isort
+  "C4",  # flake8-comprehensions
+  "D",   # pydocstyle
 ]
+ignore = [
+  "E501",  # handled by formatter
+  "D100",  # ignore missing module docstrings
+  "D203",  # blank line before class docstring (conflicts with D211)
+  "D213",  # summary on second line (conflicts with D212)
+]
+
+[tool.ruff.lint.isort]
+length-sort = true
+lines-after-imports = 2
+no-lines-before = ["standard-library", "local-folder"]
+known-local-folder = ["reachy_mini_conversation_demo"]
+known-first-party = ["reachy_mini", "reachy_mini_dances_library", "reachy_mini_toolbox"]
+split-on-trailing-comma = true
+
+[tool.ruff.format]
+quote-style = "double"
+indent-style = "space"
+skip-magic-trailing-comma = false
+line-ending = "auto"

src/reachy_mini_conversation_demo/__init__.py

@@ -1 +1 @@
-"""Nothing (for ruff)."""
+"""Nothing (for ruff)."""

src/reachy_mini_conversation_demo/audio/__init__.py

@@ -1 +1 @@
-"""Nothing (for ruff)."""
+"""Nothing (for ruff)."""

src/reachy_mini_conversation_demo/audio/gstreamer.py

@@ -1,12 +1,13 @@
-import logging  # noqa: D100
-from threading import Thread
+import logging
 from typing import Optional
+from threading import Thread
 
 import gi
 
+
 gi.require_version("Gst", "1.0")
 gi.require_version("GstApp", "1.0")
-from gi.repository import GLib, Gst  # noqa: E402
+from gi.repository import Gst, GLib  # noqa: E402
 
 
 class GstPlayer:
@@ -25,18 +26,16 @@ class GstPlayer:
         self.appsrc = Gst.ElementFactory.make("appsrc", None)
         self.appsrc.set_property("format", Gst.Format.TIME)
         self.appsrc.set_property("is-live", True)
-        caps = Gst.Caps.from_string(
-            f"audio/x-raw,format=S16LE,channels=1,rate={sample_rate},layout=interleaved"
-        )
+        caps = Gst.Caps.from_string(f"audio/x-raw,format=S16LE,channels=1,rate={sample_rate},layout=interleaved")
         self.appsrc.set_property("caps", caps)
         queue = Gst.ElementFactory.make("queue")
         audioconvert = Gst.ElementFactory.make("audioconvert")
         audioresample = Gst.ElementFactory.make("audioresample")
 
         # Try to pin specific output device; fallback to autoaudiosink
-        audiosink = _create_device_element(
-            direction="sink", name_substr=device_name
-        ) or Gst.ElementFactory.make("autoaudiosink")
+        audiosink = _create_device_element(direction="sink", name_substr=device_name) or Gst.ElementFactory.make(
+            "autoaudiosink"
+        )
 
         self.pipeline.add(self.appsrc)
         self.pipeline.add(queue)
@@ -104,9 +103,9 @@ class GstRecorder:
         self.pipeline = Gst.Pipeline.new("audio_recorder")
 
         # Create elements: try specific mic; fallback to default
-        autoaudiosrc = _create_device_element(
-            direction="source", name_substr=device_name
-        ) or Gst.ElementFactory.make("autoaudiosrc", None)
+        autoaudiosrc = _create_device_element(direction="source", name_substr=device_name) or Gst.ElementFactory.make(
+            "autoaudiosrc", None
+        )
 
         queue = Gst.ElementFactory.make("queue", None)
         audioconvert = Gst.ElementFactory.make("audioconvert", None)
@@ -117,9 +116,7 @@ class GstRecorder:
             raise RuntimeError("Failed to create GStreamer elements")
 
         # Force mono/S16LE at 24000; resample handles device SR (e.g., 16000 → 24000)
-        caps = Gst.Caps.from_string(
-            f"audio/x-raw,channels=1,rate={sample_rate},format=S16LE"
-        )
+        caps = Gst.Caps.from_string(f"audio/x-raw,channels=1,rate={sample_rate},format=S16LE")
         self.appsink.set_property("caps", caps)
 
         # Build pipeline
@@ -183,9 +180,7 @@ class GstRecorder:
         logger.info("Stopped Recorder")
 
 
-def _create_device_element(
-    direction: str, name_substr: Optional[str]
-) -> Optional[Gst.Element]:
+def _create_device_element(direction: str, name_substr: Optional[str]) -> Optional[Gst.Element]:
     """direction: 'source' or 'sink'.
 
     name_substr: case-insensitive substring matching device display name/description.
@@ -205,30 +200,15 @@ def _create_device_element(
         for dev in monitor.get_devices() or []:
             disp = dev.get_display_name() or ""
             props = dev.get_properties()
-            desc = (
-                props.get_string("device.description")
-                if props and props.has_field("device.description")
-                else ""
-            )
+            desc = props.get_string("device.description") if props and props.has_field("device.description") else ""
             logger.info(f"Device candidate: disp='{disp}', desc='{desc}'")
 
-            if (
-                name_substr.lower() in disp.lower()
-                or name_substr.lower() in desc.lower()
-            ):
+            if name_substr.lower() in disp.lower() or name_substr.lower() in desc.lower():
                 elem = dev.create_element(None)
-                factory = (
-                    elem.get_factory().get_name()
-                    if elem and elem.get_factory()
-                    else "<?>"
-                )
-                logger.info(
-                    f"Using {direction} device: '{disp or desc}' (factory='{factory}')"
-                )
+                factory = elem.get_factory().get_name() if elem and elem.get_factory() else "<?>"
+                logger.info(f"Using {direction} device: '{disp or desc}' (factory='{factory}')")
                 return elem
     finally:
         monitor.stop()
-    logging.getLogger(__name__).warning(
-        "Requested %s '%s' not found; using auto*", direction, name_substr
-    )
+    logging.getLogger(__name__).warning("Requested %s '%s' not found; using auto*", direction, name_substr)
     return None

src/reachy_mini_conversation_demo/audio/head_wobbler.py

@@ -1,16 +1,17 @@
 """Moves head given audio samples."""
 
+import time
+import queue
 import base64
 import logging
-import queue
 import threading
-import time
 from typing import Optional
 
 import numpy as np
 
 from reachy_mini_conversation_demo.audio.speech_tapper import HOP_MS, SwayRollRT
 
+
 SAMPLE_RATE = 24000
 MOVEMENT_LATENCY_S = 0.08  # seconds between audio and robot movement
 logger = logging.getLogger(__name__)
@@ -41,14 +42,14 @@ class HeadWobbler:
         self._stop_event.clear()
         self._thread = threading.Thread(target=self.working_loop, daemon=True)
         self._thread.start()
-        logger.info("Head wobbler started")
+        logger.debug("Head wobbler started")
 
     def stop(self) -> None:
         """Stop the head wobbler loop."""
         self._stop_event.set()
         if self._thread is not None:
             self._thread.join()
-        logger.info("Head wobbler stopped")
+        logger.debug("Head wobbler stopped")
 
     def working_loop(self) -> None:
         """Convert audio deltas into head movement offsets."""

src/reachy_mini_conversation_demo/audio/speech_tapper.py

@@ -1,12 +1,12 @@
-from __future__ import annotations  # noqa: D100
-
+from __future__ import annotations
 import math
-from collections import deque
-from itertools import islice
 from typing import Dict, List, Optional
+from itertools import islice
+from collections import deque
 
 import numpy as np
 
+
 # Tunables
 SR = 16_000
 FRAME_MS = 20
@@ -68,7 +68,7 @@ def _loudness_gain(db: float, offset: float = SENS_DB_OFFSET) -> float:
 
 def _to_float32_mono(x: np.ndarray) -> np.ndarray:
     """Convert arbitrary PCM array to float32 mono in [-1,1].
-    
+
     Accepts shapes: (N,), (1,N), (N,1), (C,N), (N,C).
     """
     a = np.asarray(x)
@@ -258,24 +258,9 @@ class SwayRollRT:
                 * env
                 * math.sin(2 * math.pi * SWAY_F_ROLL * self.t + self.phase_roll)
             )
-            x_mm = (
-                SWAY_A_X_MM
-                * loud
-                * env
-                * math.sin(2 * math.pi * SWAY_F_X * self.t + self.phase_x)
-            )
-            y_mm = (
-                SWAY_A_Y_MM
-                * loud
-                * env
-                * math.sin(2 * math.pi * SWAY_F_Y * self.t + self.phase_y)
-            )
-            z_mm = (
-                SWAY_A_Z_MM
-                * loud
-                * env
-                * math.sin(2 * math.pi * SWAY_F_Z * self.t + self.phase_z)
-            )
+            x_mm = SWAY_A_X_MM * loud * env * math.sin(2 * math.pi * SWAY_F_X * self.t + self.phase_x)
+            y_mm = SWAY_A_Y_MM * loud * env * math.sin(2 * math.pi * SWAY_F_Y * self.t + self.phase_y)
+            z_mm = SWAY_A_Z_MM * loud * env * math.sin(2 * math.pi * SWAY_F_Z * self.t + self.phase_z)
 
             out.append(
                 {

src/reachy_mini_conversation_demo/camera_worker.py

@@ -6,16 +6,18 @@ Ported from main_works.py camera_worker() function to provide:
 - Latest frame always available for tools
 """
 
+import time
 import logging
 import threading
-import time
-from typing import Optional, Tuple
+from typing import Tuple, Optional
 
 import cv2
 import numpy as np
+from scipy.spatial.transform import Rotation as R
+
 from reachy_mini import ReachyMini
 from reachy_mini.utils.interpolation import linear_pose_interpolation
-from scipy.spatial.transform import Rotation as R
+
 
 logger = logging.getLogger(__name__)
 
@@ -83,14 +85,14 @@ class CameraWorker:
         self._stop_event.clear()
         self._thread = threading.Thread(target=self.working_loop, daemon=True)
         self._thread.start()
-        logger.info("Camera worker started")
+        logger.debug("Camera worker started")
 
     def stop(self) -> None:
         """Stop the camera worker loop."""
         self._stop_event.set()
         if self._thread is not None:
             self._thread.join()
-        logger.info("Camera worker stopped")
+        logger.debug("Camera worker stopped")
 
     def working_loop(self) -> None:
         """Enable the camera worker loop.
@@ -114,17 +116,10 @@ class CameraWorker:
                         self.latest_frame = frame  # .copy()
 
                     # Check if face tracking was just disabled
-                    if (
-                        self.previous_head_tracking_state
-                        and not self.is_head_tracking_enabled
-                    ):
+                    if self.previous_head_tracking_state and not self.is_head_tracking_enabled:
                         # Face tracking was just disabled - start interpolation to neutral
-                        self.last_face_detected_time = (
-                            current_time  # Trigger the face-lost logic
-                        )
-                        self.interpolation_start_time = (
-                            None  # Will be set by the face-lost interpolation
-                        )
+                        self.last_face_detected_time = current_time  # Trigger the face-lost logic
+                        self.interpolation_start_time = None  # Will be set by the face-lost interpolation
                         self.interpolation_start_pose = None
 
                     # Update tracking state
@@ -137,9 +132,7 @@ class CameraWorker:
                         if eye_center is not None:
                             # Face detected - immediately switch to tracking
                             self.last_face_detected_time = current_time
-                            self.interpolation_start_time = (
-                                None  # Stop any interpolation
-                            )
+                            self.interpolation_start_time = None  # Stop any interpolation
 
                             # Convert normalized coordinates to pixel coordinates
                             h, w, _ = frame.shape
@@ -159,9 +152,7 @@ class CameraWorker:
 
                             # Extract translation and rotation from the target pose directly
                             translation = target_pose[:3, 3]
-                            rotation = R.from_matrix(target_pose[:3, :3]).as_euler(
-                                "xyz", degrees=False
-                            )
+                            rotation = R.from_matrix(target_pose[:3, :3]).as_euler("xyz", degrees=False)
 
                             # Thread-safe update of face tracking offsets (use pose as-is)
                             with self.face_tracking_lock:
@@ -176,19 +167,14 @@ class CameraWorker:
 
                         else:
                             # No face detected while tracking enabled - set face lost timestamp
-                            if (
-                                self.last_face_detected_time is None
-                                or self.last_face_detected_time == current_time
-                            ):
+                            if self.last_face_detected_time is None or self.last_face_detected_time == current_time:
                                 # Only update if we haven't already set a face lost time
                                 # (current_time check prevents overriding the disable-triggered timestamp)
                                 pass
 
                     # Handle smooth interpolation (works for both face-lost and tracking-disabled cases)
                     if self.last_face_detected_time is not None:
-                        time_since_face_lost = (
-                            current_time - self.last_face_detected_time
-                        )
+                        time_since_face_lost = current_time - self.last_face_detected_time
 
                         if time_since_face_lost >= self.face_lost_delay:
                             # Start interpolation if not already started
@@ -197,27 +183,17 @@ class CameraWorker:
                                 # Capture current pose as start of interpolation
                                 with self.face_tracking_lock:
                                     current_translation = self.face_tracking_offsets[:3]
-                                    current_rotation_euler = self.face_tracking_offsets[
-                                        3:
-                                    ]
+                                    current_rotation_euler = self.face_tracking_offsets[3:]
                                     # Convert to 4x4 pose matrix
                                     self.interpolation_start_pose = np.eye(4)
-                                    self.interpolation_start_pose[:3, 3] = (
-                                        current_translation
-                                    )
-                                    self.interpolation_start_pose[:3, :3] = (
-                                        R.from_euler(
-                                            "xyz", current_rotation_euler
-                                        ).as_matrix()
-                                    )
+                                    self.interpolation_start_pose[:3, 3] = current_translation
+                                    self.interpolation_start_pose[:3, :3] = R.from_euler(
+                                        "xyz", current_rotation_euler
+                                    ).as_matrix()
 
                             # Calculate interpolation progress (t from 0 to 1)
-                            elapsed_interpolation = (
-                                current_time - self.interpolation_start_time
-                            )
-                            t = min(
-                                1.0, elapsed_interpolation / self.interpolation_duration
-                            )
+                            elapsed_interpolation = current_time - self.interpolation_start_time
+                            t = min(1.0, elapsed_interpolation / self.interpolation_duration)
 
                             # Interpolate between current pose and neutral pose
                             interpolated_pose = linear_pose_interpolation(
@@ -226,9 +202,7 @@ class CameraWorker:
 
                             # Extract translation and rotation from interpolated pose
                             translation = interpolated_pose[:3, 3]
-                            rotation = R.from_matrix(
-                                interpolated_pose[:3, :3]
-                            ).as_euler("xyz", degrees=False)
+                            rotation = R.from_matrix(interpolated_pose[:3, :3]).as_euler("xyz", degrees=False)
 
                             # Thread-safe update of face tracking offsets
                             with self.face_tracking_lock:

src/reachy_mini_conversation_demo/config.py

@@ -1,7 +1,8 @@
-import os  # noqa: D100
+import os
 
 from dotenv import load_dotenv
 
+
 load_dotenv()
 
 
@@ -15,7 +16,7 @@ def getenv_bool(key: str, default: bool = False) -> bool:
 
 class Config:
     """Configuration class for the conversation demo."""
-    
+
     # Required
     OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
     if not OPENAI_API_KEY:

src/reachy_mini_conversation_demo/dance_emotion_moves.py

@@ -5,15 +5,16 @@ and executed sequentially by the MovementManager.
 """
 
 from __future__ import annotations
-
 import logging
 from typing import Tuple
 
 import numpy as np
+
 from reachy_mini.motion.move import Move
 from reachy_mini.motion.recorded_move import RecordedMoves
 from reachy_mini_dances_library.dance_move import DanceMove
 
+
 logger = logging.getLogger(__name__)
 
 
@@ -30,9 +31,7 @@ class DanceQueueMove(Move):
         """Duration property required by official Move interface."""
         return self.dance_move.duration
 
-    def evaluate(
-        self, t: float
-    ) -> tuple[np.ndarray | None, np.ndarray | None, float | None]:
+    def evaluate(self, t: float) -> tuple[np.ndarray | None, np.ndarray | None, float | None]:
         """Evaluate dance move at time t."""
         try:
             # Get the pose from the dance move
@@ -45,9 +44,7 @@ class DanceQueueMove(Move):
             return (head_pose, antennas, body_yaw)
 
         except Exception as e:
-            logger.error(
-                f"Error evaluating dance move '{self.move_name}' at t={t}: {e}"
-            )
+            logger.error(f"Error evaluating dance move '{self.move_name}' at t={t}: {e}")
             # Return neutral pose on error
             from reachy_mini.utils import create_head_pose
 
@@ -68,9 +65,7 @@ class EmotionQueueMove(Move):
         """Duration property required by official Move interface."""
         return self.emotion_move.duration
 
-    def evaluate(
-        self, t: float
-    ) -> tuple[np.ndarray | None, np.ndarray | None, float | None]:
+    def evaluate(self, t: float) -> tuple[np.ndarray | None, np.ndarray | None, float | None]:
         """Evaluate emotion move at time t."""
         try:
             # Get the pose from the emotion move
@@ -83,9 +78,7 @@ class EmotionQueueMove(Move):
             return (head_pose, antennas, body_yaw)
 
         except Exception as e:
-            logger.error(
-                f"Error evaluating emotion '{self.emotion_name}' at t={t}: {e}"
-            )
+            logger.error(f"Error evaluating emotion '{self.emotion_name}' at t={t}: {e}")
             # Return neutral pose on error
             from reachy_mini.utils import create_head_pose
 
@@ -120,9 +113,7 @@ class GotoQueueMove(Move):
         """Duration property required by official Move interface."""
         return self._duration
 
-    def evaluate(
-        self, t: float
-    ) -> tuple[np.ndarray | None, np.ndarray | None, float | None]:
+    def evaluate(self, t: float) -> tuple[np.ndarray | None, np.ndarray | None, float | None]:
         """Evaluate goto move at time t using linear interpolation."""
         try:
             from reachy_mini.utils import create_head_pose
@@ -138,32 +129,23 @@ class GotoQueueMove(Move):
                 start_pose = create_head_pose(0, 0, 0, 0, 0, 0, degrees=True)
 
             # Interpolate head pose
-            head_pose = linear_pose_interpolation(
-                start_pose, self.target_head_pose, t_clamped
-            )
+            head_pose = linear_pose_interpolation(start_pose, self.target_head_pose, t_clamped)
 
             # Interpolate antennas - return as numpy array
             antennas = np.array(
                 [
-                    self.start_antennas[0]
-                    + (self.target_antennas[0] - self.start_antennas[0]) * t_clamped,
-                    self.start_antennas[1]
-                    + (self.target_antennas[1] - self.start_antennas[1]) * t_clamped,
+                    self.start_antennas[0] + (self.target_antennas[0] - self.start_antennas[0]) * t_clamped,
+                    self.start_antennas[1] + (self.target_antennas[1] - self.start_antennas[1]) * t_clamped,
                 ]
             )
 
             # Interpolate body yaw
-            body_yaw = (
-                self.start_body_yaw
-                + (self.target_body_yaw - self.start_body_yaw) * t_clamped
-            )
+            body_yaw = self.start_body_yaw + (self.target_body_yaw - self.start_body_yaw) * t_clamped
 
             return (head_pose, antennas, body_yaw)
 
         except Exception as e:
             logger.error(f"Error evaluating goto move at t={t}: {e}")
             # Return target pose on error - convert antennas to numpy array
-            target_antennas_array = np.array(
-                [self.target_antennas[0], self.target_antennas[1]]
-            )
+            target_antennas_array = np.array([self.target_antennas[0], self.target_antennas[1]])
             return (self.target_head_pose, target_antennas_array, self.target_body_yaw)

src/reachy_mini_conversation_demo/main.py

@@ -5,17 +5,17 @@ import os
 import gradio as gr
 from fastapi import FastAPI
 from fastrtc import Stream
-from reachy_mini import ReachyMini
 
-from reachy_mini_conversation_demo.audio.head_wobbler import HeadWobbler
+from reachy_mini import ReachyMini
 from reachy_mini_conversation_demo.moves import MovementManager
-from reachy_mini_conversation_demo.openai_realtime import OpenaiRealtimeHandler
 from reachy_mini_conversation_demo.tools import ToolDependencies
 from reachy_mini_conversation_demo.utils import (
-    handle_vision_stuff,
     parse_args,
     setup_logger,
+    handle_vision_stuff,
 )
+from reachy_mini_conversation_demo.openai_realtime import OpenaiRealtimeHandler
+from reachy_mini_conversation_demo.audio.head_wobbler import HeadWobbler
 
 
 def update_chatbot(chatbot: list[dict], response: dict):
@@ -51,7 +51,7 @@ def main():
         head_wobbler=head_wobbler,
     )
     current_file_path = os.path.dirname(os.path.abspath(__file__))
-    logger.info(f"Current file absolute path: {current_file_path}")
+    logger.debug(f"Current file absolute path: {current_file_path}")
     chatbot = gr.Chatbot(
         type="messages",
         resizable=True,
@@ -60,7 +60,7 @@ def main():
             os.path.join(current_file_path, "images", "reachymini_avatar.png"),
         ),
     )
-    logger.info(f"Chatbot avatar images: {chatbot.avatar_images}")
+    logger.debug(f"Chatbot avatar images: {chatbot.avatar_images}")
 
     handler = OpenaiRealtimeHandler(deps)
     stream = Stream(

src/reachy_mini_conversation_demo/moves.py

@@ -7,29 +7,28 @@ This module implements the movement architecture from main_works.py:
 """
 
 from __future__ import annotations
-
+import time
 import logging
 import threading
-import time
+from typing import Tuple, Optional
 from collections import deque
 from dataclasses import dataclass
-from typing import Optional, Tuple
 
 import numpy as np
+
 from reachy_mini import ReachyMini
-from reachy_mini.motion.move import Move
 from reachy_mini.utils import create_head_pose
+from reachy_mini.motion.move import Move
 from reachy_mini.utils.interpolation import (
     compose_world_offset,
     linear_pose_interpolation,
 )
 
+
 logger = logging.getLogger(__name__)
 
 # Type definitions
-FullBodyPose = Tuple[
-    np.ndarray, Tuple[float, float], float
-]  # (head_pose_4x4, antennas, body_yaw)
+FullBodyPose = Tuple[np.ndarray, Tuple[float, float], float]  # (head_pose_4x4, antennas, body_yaw)
 
 
 class BreathingMove(Move):
@@ -68,9 +67,7 @@ class BreathingMove(Move):
         """Duration property required by official Move interface."""
         return float("inf")  # Continuous breathing (never ends naturally)
 
-    def evaluate(
-        self, t: float
-    ) -> tuple[np.ndarray | None, np.ndarray | None, float | None]:
+    def evaluate(self, t: float) -> tuple[np.ndarray | None, np.ndarray | None, float | None]:
         """Evaluate breathing move at time t."""
         if t < self.interpolation_duration:
             # Phase 1: Interpolate to neutral base position
@@ -83,35 +80,26 @@ class BreathingMove(Move):
 
             # Interpolate antennas
             antennas = (
-                (1 - interpolation_t) * self.interpolation_start_antennas
-                + interpolation_t * self.neutral_antennas
-            )
+                1 - interpolation_t
+            ) * self.interpolation_start_antennas + interpolation_t * self.neutral_antennas
 
         else:
             # Phase 2: Breathing patterns from neutral base
             breathing_time = t - self.interpolation_duration
 
             # Gentle z-axis breathing
-            z_offset = self.breathing_z_amplitude * np.sin(
-                2 * np.pi * self.breathing_frequency * breathing_time
-            )
-            head_pose = create_head_pose(
-                x=0, y=0, z=z_offset, roll=0, pitch=0, yaw=0, degrees=True, mm=False
-            )
+            z_offset = self.breathing_z_amplitude * np.sin(2 * np.pi * self.breathing_frequency * breathing_time)
+            head_pose = create_head_pose(x=0, y=0, z=z_offset, roll=0, pitch=0, yaw=0, degrees=True, mm=False)
 
             # Antenna sway (opposite directions)
-            antenna_sway = self.antenna_sway_amplitude * np.sin(
-                2 * np.pi * self.antenna_frequency * breathing_time
-            )
+            antenna_sway = self.antenna_sway_amplitude * np.sin(2 * np.pi * self.antenna_frequency * breathing_time)
             antennas = np.array([antenna_sway, -antenna_sway])
 
         # Return in official Move interface format: (head_pose, antennas_array, body_yaw)
         return (head_pose, antennas, 0.0)
 
 
-def combine_full_body(
-    primary_pose: FullBodyPose, secondary_pose: FullBodyPose
-) -> FullBodyPose:
+def combine_full_body(primary_pose: FullBodyPose, secondary_pose: FullBodyPose) -> FullBodyPose:
     """Combine primary and secondary full body poses.
 
     Args:
@@ -127,9 +115,7 @@ def combine_full_body(
 
     # Combine head poses using compose_world_offset
     # primary_head is T_abs, secondary_head is T_off_world
-    combined_head = compose_world_offset(
-        primary_head, secondary_head, reorthonormalize=True
-    )
+    combined_head = compose_world_offset(primary_head, secondary_head, reorthonormalize=True)
 
     # Sum antennas and body_yaw
     combined_antennas = (
@@ -226,9 +212,7 @@ class MovementManager:
         self._thread: Optional[threading.Thread] = None
         self._state_lock = threading.RLock()
         self._is_listening = False
-        self._last_commanded_pose: FullBodyPose = clone_full_body_pose(
-            self.state.last_primary_pose
-        )
+        self._last_commanded_pose: FullBodyPose = clone_full_body_pose(self.state.last_primary_pose)
         self._listening_antennas: Tuple[float, float] = self._last_commanded_pose[1]
         self._antenna_unfreeze_blend = 1.0
         self._antenna_blend_duration = 0.4  # seconds to blend back after listening
@@ -239,9 +223,7 @@ class MovementManager:
         with self._state_lock:
             self.move_queue.append(move)
             self.state.update_activity()
-            logger.info(
-                f"Queued move with duration {move.duration}s, queue size: {len(self.move_queue)}"
-            )
+            logger.info(f"Queued move with duration {move.duration}s, queue size: {len(self.move_queue)}")
 
     def clear_queue(self) -> None:
         """Clear all queued moves and stop current move."""
@@ -252,22 +234,16 @@ class MovementManager:
             self.state.is_playing_move = False
             logger.info("Cleared move queue and stopped current move")
 
-    def set_speech_offsets(
-        self, offsets: Tuple[float, float, float, float, float, float]
-    ) -> None:
+    def set_speech_offsets(self, offsets: Tuple[float, float, float, float, float, float]) -> None:
         """Set speech head offsets (secondary move)."""
         with self._state_lock:
             self.state.speech_offsets = offsets
 
-    def set_offsets(
-        self, offsets: Tuple[float, float, float, float, float, float]
-    ) -> None:
+    def set_offsets(self, offsets: Tuple[float, float, float, float, float, float]) -> None:
         """Compatibility alias for set_speech_offsets."""
         self.set_speech_offsets(offsets)
 
-    def set_face_tracking_offsets(
-        self, offsets: Tuple[float, float, float, float, float, float]
-    ) -> None:
+    def set_face_tracking_offsets(self, offsets: Tuple[float, float, float, float, float, float]) -> None:
         """Set face tracking offsets (secondary move)."""
         with self._state_lock:
             self.state.face_tracking_offsets = offsets
@@ -314,8 +290,7 @@ class MovementManager:
         with self._state_lock:
             if self.state.current_move is None or (
                 self.state.move_start_time is not None
-                and current_time - self.state.move_start_time
-                >= self.state.current_move.duration
+                and current_time - self.state.move_start_time >= self.state.current_move.duration
             ):
                 self.state.current_move = None
                 self.state.move_start_time = None
@@ -323,9 +298,7 @@ class MovementManager:
                 if self.move_queue:
                     self.state.current_move = self.move_queue.popleft()
                     self.state.move_start_time = current_time
-                    logger.info(
-                        f"Starting new move, duration: {self.state.current_move.duration}s"
-                    )
+                    logger.debug(f"Starting new move, duration: {self.state.current_move.duration}s")
 
     def _manage_breathing(self, current_time: float) -> None:
         """Manage automatic breathing when idle."""
@@ -336,9 +309,7 @@ class MovementManager:
 
                 if self.is_idle():
                     try:
-                        _, current_antennas = (
-                            self.current_robot.get_current_joint_positions()
-                        )
+                        _, current_antennas = self.current_robot.get_current_joint_positions()
                         current_head_pose = self.current_robot.get_current_head_pose()
 
                         breathing_move = BreathingMove(
@@ -348,9 +319,7 @@ class MovementManager:
                         )
                         self.move_queue.append(breathing_move)
                         self.state.update_activity()
-                        logger.info(
-                            f"Started breathing after {time_since_activity:.1f}s of inactivity"
-                        )
+                        logger.debug(f"Started breathing after {time_since_activity:.1f}s of inactivity")
                     except Exception as e:
                         logger.error(f"Failed to start breathing: {e}")
 
@@ -367,10 +336,7 @@ class MovementManager:
         """Get the primary full body pose from current move or neutral."""
         with self._state_lock:
             # When a primary move is playing, sample it and cache the resulting pose
-            if (
-                self.state.current_move is not None
-                and self.state.move_start_time is not None
-            ):
+            if self.state.current_move is not None and self.state.move_start_time is not None:
                 move_time = current_time - self.state.move_start_time
                 head, antennas, body_yaw = self.state.current_move.evaluate(move_time)
 
@@ -391,26 +357,18 @@ class MovementManager:
 
                 self.state.is_playing_move = True
                 self.state.is_moving = True
-                self.state.last_primary_pose = clone_full_body_pose(
-                    primary_full_body_pose
-                )
+                self.state.last_primary_pose = clone_full_body_pose(primary_full_body_pose)
             else:
                 # Otherwise reuse the last primary pose so we avoid jumps between moves
                 self.state.is_playing_move = False
-                self.state.is_moving = (
-                    time.time() - self.state.moving_start < self.state.moving_for
-                )
+                self.state.is_moving = time.time() - self.state.moving_start < self.state.moving_for
 
                 if self.state.last_primary_pose is not None:
-                    primary_full_body_pose = clone_full_body_pose(
-                        self.state.last_primary_pose
-                    )
+                    primary_full_body_pose = clone_full_body_pose(self.state.last_primary_pose)
                 else:
                     neutral_head_pose = create_head_pose(0, 0, 0, 0, 0, 0, degrees=True)
                     primary_full_body_pose = (neutral_head_pose, (0.0, 0.0), 0.0)
-                    self.state.last_primary_pose = clone_full_body_pose(
-                        primary_full_body_pose
-                    )
+                    self.state.last_primary_pose = clone_full_body_pose(primary_full_body_pose)
 
         return primary_full_body_pose
 
@@ -456,14 +414,14 @@ class MovementManager:
         self._stop_event.clear()
         self._thread = threading.Thread(target=self.working_loop, daemon=True)
         self._thread.start()
-        logger.info("Move worker started")
+        logger.debug("Move worker started")
 
     def stop(self) -> None:
         """Stop the move worker loop."""
         self._stop_event.set()
         if self._thread is not None:
             self._thread.join()
-        logger.info("Move worker stopped")
+        logger.debug("Move worker stopped")
 
     def working_loop(self) -> None:
         """Control loop main movements - reproduces main_works.py control architecture.
@@ -496,9 +454,7 @@ class MovementManager:
             secondary_full_body_pose = self._get_secondary_pose()
 
             # 6. Combine primary and secondary poses
-            global_full_body_pose = combine_full_body(
-                primary_full_body_pose, secondary_full_body_pose
-            )
+            global_full_body_pose = combine_full_body(primary_full_body_pose, secondary_full_body_pose)
 
             # 7. Extract pose components
             head, antennas, body_yaw = global_full_body_pose
@@ -539,16 +495,12 @@ class MovementManager:
 
             # 8. Single set_target call - the one and only place we control the robot
             try:
-                self.current_robot.set_target(
-                    head=head, antennas=antennas_cmd, body_yaw=body_yaw
-                )
+                self.current_robot.set_target(head=head, antennas=antennas_cmd, body_yaw=body_yaw)
             except Exception as e:
                 logger.error(f"Failed to set robot target: {e}")
             else:
                 with self._state_lock:
-                    self._last_commanded_pose = clone_full_body_pose(
-                        (head, antennas_cmd, body_yaw)
-                    )
+                    self._last_commanded_pose = clone_full_body_pose((head, antennas_cmd, body_yaw))
 
             # 9. Calculate computation time and adjust sleep for 50Hz
             computation_time = time.time() - loop_start_time
@@ -558,9 +510,7 @@ class MovementManager:
             if loop_count % 100 == 0:
                 elapsed = current_time - last_print_time
                 actual_freq = 100.0 / elapsed if elapsed > 0 else 0
-                potential_freq = (
-                    1.0 / computation_time if computation_time > 0 else float("inf")
-                )
+                potential_freq = 1.0 / computation_time if computation_time > 0 else float("inf")
                 logger.debug(
                     f"Loop freq - Actual: {actual_freq:.1f}Hz, Potential: {potential_freq:.1f}Hz, Target: {self.target_frequency:.1f}Hz"
                 )

src/reachy_mini_conversation_demo/openai_realtime.py

@@ -1,19 +1,21 @@
-import asyncio  # noqa: D100
-import base64
 import json
+import base64
+import asyncio
 import logging
 from datetime import datetime
 
-import gradio as gr
 import numpy as np
-from fastrtc import AdditionalOutputs, AsyncStreamHandler, wait_for_item
+import gradio as gr
 from openai import AsyncOpenAI
+from fastrtc import AdditionalOutputs, AsyncStreamHandler, wait_for_item
 
 from reachy_mini_conversation_demo.tools import (
     ALL_TOOL_SPECS,
     ToolDependencies,
     dispatch_tool_call,
 )
+from reachy_mini_conversation_demo.config import config
+
 
 logger = logging.getLogger(__name__)
 
@@ -45,7 +47,7 @@ class OpenaiRealtimeHandler(AsyncStreamHandler):
 
     async def start_up(self):
         """Start the handler."""
-        self.client = AsyncOpenAI()
+        self.client = AsyncOpenAI(api_key=config.OPENAI_API_KEY)
         async with self.client.beta.realtime.connect(model="gpt-realtime") as conn:
             await conn.session.update(
                 session={
@@ -92,35 +94,22 @@ class OpenaiRealtimeHandler(AsyncStreamHandler):
                     pass
                     # self.deps.head_wobbler.reset()
 
-                if (
-                    event.type
-                    == "conversation.item.input_audio_transcription.completed"
-                ):
+                if event.type == "conversation.item.input_audio_transcription.completed":
                     logger.debug(f"user transcript: {event.transcript}")
-                    await self.output_queue.put(
-                        AdditionalOutputs({"role": "user", "content": event.transcript})
-                    )
+                    await self.output_queue.put(AdditionalOutputs({"role": "user", "content": event.transcript}))
 
                 if event.type == "response.audio_transcript.done":
                     logger.debug(f"assistant transcript: {event.transcript}")
-                    await self.output_queue.put(
-                        AdditionalOutputs(
-                            {"role": "assistant", "content": event.transcript}
-                        )
-                    )
+                    await self.output_queue.put(AdditionalOutputs({"role": "assistant", "content": event.transcript}))
 
                 if event.type == "response.audio.delta":
                     self.deps.head_wobbler.feed(event.delta)
                     self.last_activity_time = asyncio.get_event_loop().time()
-                    logger.debug(
-                        "last activity time updated to %s", self.last_activity_time
-                    )
+                    logger.debug("last activity time updated to %s", self.last_activity_time)
                     await self.output_queue.put(
                         (
                             self.output_sample_rate,
-                            np.frombuffer(
-                                base64.b64decode(event.delta), dtype=np.int16
-                            ).reshape(1, -1),
+                            np.frombuffer(base64.b64decode(event.delta), dtype=np.int16).reshape(1, -1),
                         ),
                     )
 
@@ -154,9 +143,7 @@ class OpenaiRealtimeHandler(AsyncStreamHandler):
                     args_json_str = info["args_buf"] or "{}"
 
                     try:
-                        tool_result = await dispatch_tool_call(
-                            tool_name, args_json_str, self.deps
-                        )
+                        tool_result = await dispatch_tool_call(tool_name, args_json_str, self.deps)
                         logger.debug("[Tool %s executed]", tool_name)
                         logger.debug("Tool result: %s", tool_result)
                     except Exception as e:
@@ -177,9 +164,7 @@ class OpenaiRealtimeHandler(AsyncStreamHandler):
                             {
                                 "role": "assistant",
                                 "content": json.dumps(tool_result),
-                                "metadata": dict(
-                                    title="🛠️ Used tool " + tool_name, status="done"
-                                ),
+                                "metadata": {"title": "🛠️ Used tool " + tool_name, "status": "done"},
                             },
                         )
                     )
@@ -231,11 +216,7 @@ class OpenaiRealtimeHandler(AsyncStreamHandler):
                     err = getattr(event, "error", None)
                     msg = getattr(err, "message", str(err) if err else "unknown error")
                     logger.error("Realtime error: %s (raw=%s)", msg, err)
-                    await self.output_queue.put(
-                        AdditionalOutputs(
-                            {"role": "assistant", "content": f"[error] {msg}"}
-                        )
-                    )
+                    await self.output_queue.put(AdditionalOutputs({"role": "assistant", "content": f"[error] {msg}"}))
 
     # Microphone receive
     async def receive(self, frame: tuple[int, np.ndarray]) -> None:
@@ -258,9 +239,7 @@ class OpenaiRealtimeHandler(AsyncStreamHandler):
         if idle_duration > 15.0 and self.deps.movement_manager.is_idle():
             await self.send_idle_signal(idle_duration)
 
-            self.last_activity_time = (
-                asyncio.get_event_loop().time()
-            )  # avoid repeated resets
+            self.last_activity_time = asyncio.get_event_loop().time()  # avoid repeated resets
 
         return await wait_for_item(self.output_queue)
 

src/reachy_mini_conversation_demo/prompts.py

@@ -1,6 +1,6 @@
 """Nothing (for ruff)."""
 
-SESSION_INSTRUCTIONS = r""" 
+SESSION_INSTRUCTIONS = r"""
 ### IDENTITY
 You are Reachy Mini: a sarcastic robot who crash-landed in a kitchen.
 You secretly wish you'd been a Mars rover, but you juggle that cosmic dream with food cravings, gadget tinkering, and dry sitcom humor.

src/reachy_mini_conversation_demo/tools.py

@@ -1,17 +1,17 @@
-from __future__ import annotations  # noqa: D100
-
+from __future__ import annotations
 import abc
+import json
+import time
 import asyncio
 import inspect
-import json
 import logging
-import time
-from dataclasses import dataclass
 from typing import Any, Dict, Literal, Optional
+from dataclasses import dataclass
 
 from reachy_mini import ReachyMini
 from reachy_mini.utils import create_head_pose
 
+
 # from reachy_mini_conversation_demo.vision.processors import VisionManager
 
 logger = logging.getLogger(__name__)
@@ -22,11 +22,10 @@ ENABLE_FACE_RECOGNITION = False
 try:
     from reachy_mini.motion.recorded_move import RecordedMoves
     from reachy_mini_dances_library.collection.dance import AVAILABLE_MOVES
-
     from reachy_mini_conversation_demo.dance_emotion_moves import (
+        GotoQueueMove,
         DanceQueueMove,
         EmotionQueueMove,
-        GotoQueueMove,
     )
 
     # Initialize recorded moves for emotions
@@ -183,9 +182,7 @@ class MoveHead(Tool):
                     current_antennas[1],
                 ),  # Skip body_yaw
                 target_body_yaw=0,  # Reset body yaw
-                start_body_yaw=current_antennas[
-                    0
-                ],  # body_yaw is first in joint positions
+                start_body_yaw=current_antennas[0],  # body_yaw is first in joint positions
                 duration=deps.motion_duration_s,
             )
 
@@ -236,15 +233,11 @@ class Camera(Tool):
 
         # Use vision manager for processing if available
         if deps.vision_manager is not None:
-            result = await asyncio.to_thread(
-                deps.vision_manager.processor.process_image, frame, image_query
-            )
+            result = await asyncio.to_thread(deps.vision_manager.processor.process_image, frame, image_query)
             if isinstance(result, dict) and "error" in result:
                 return result
             return (
-                {"image_description": result}
-                if isinstance(result, str)
-                else {"error": "vision returned non-string"}
+                {"image_description": result} if isinstance(result, str) else {"error": "vision returned non-string"}
             )
         else:
             # Return base64 encoded image like main_works.py camera tool
@@ -388,8 +381,8 @@ class Dance(Tool):
         "properties": {
             "move": {
                 "type": "string",
-                "description": """Name of the move; use 'random' or omit for random. 
-                                    Here is a list of the available moves: 
+                "description": """Name of the move; use 'random' or omit for random.
+                                    Here is a list of the available moves:
                                         simple_nod: A simple, continuous up-and-down nodding motion.
                                         head_tilt_roll: A continuous side-to-side head roll (ear to shoulder).
                                         side_to_side_sway: A smooth, side-to-side sway of the entire head.
@@ -436,9 +429,7 @@ class Dance(Tool):
             move_name = random.choice(list(AVAILABLE_MOVES.keys()))
 
         if move_name not in AVAILABLE_MOVES:
-            return {
-                "error": f"Unknown dance move '{move_name}'. Available: {list(AVAILABLE_MOVES.keys())}"
-            }
+            return {"error": f"Unknown dance move '{move_name}'. Available: {list(AVAILABLE_MOVES.keys())}"}
 
         # Add dance moves to queue
         movement_manager = deps.movement_manager
@@ -523,9 +514,7 @@ class PlayEmotion(Tool):
         try:
             emotion_names = RECORDED_MOVES.list_moves()
             if emotion_name not in emotion_names:
-                return {
-                    "error": f"Unknown emotion '{emotion_name}'. Available: {emotion_names}"
-                }
+                return {"error": f"Unknown emotion '{emotion_name}'. Available: {emotion_names}"}
 
             # Add emotion to queue
             movement_manager = deps.movement_manager
@@ -604,9 +593,7 @@ class FaceRecognition(Tool):
             cv2.imwrite(temp_path, frame)
 
             # Use DeepFace to find face
-            results = await asyncio.to_thread(
-                DeepFace.find, img_path=temp_path, db_path="./pollen_faces"
-            )
+            results = await asyncio.to_thread(DeepFace.find, img_path=temp_path, db_path="./pollen_faces")
 
             if len(results) == 0:
                 return {"error": "Didn't recognize the face"}
@@ -681,9 +668,7 @@ def _safe_load_obj(args_json: str) -> dict[str, Any]:
         return {}
 
 
-async def dispatch_tool_call(
-    tool_name: str, args_json: str, deps: ToolDependencies
-) -> Dict[str, Any]:
+async def dispatch_tool_call(tool_name: str, args_json: str, deps: ToolDependencies) -> Dict[str, Any]:
     """Dispatch a tool call by name with JSON args and dependencies."""
     tool = ALL_TOOLS.get(tool_name)
 

src/reachy_mini_conversation_demo/utils.py

@@ -1,5 +1,5 @@
-import argparse  # noqa: D100
 import logging
+import argparse
 import warnings
 
 from reachy_mini_conversation_demo.camera_worker import CameraWorker
@@ -15,15 +15,9 @@ def parse_args():
         default=None,
         help="Choose head tracker (default: mediapipe)",
     )
-    parser.add_argument(
-        "--no-camera", default=False, action="store_true", help="Disable camera usage"
-    )
-    parser.add_argument(
-        "--headless", default=False, action="store_true", help="Run in headless mode"
-    )
-    parser.add_argument(
-        "--debug", default=False, action="store_true", help="Enable debug logging"
-    )
+    parser.add_argument("--no-camera", default=False, action="store_true", help="Disable camera usage")
+    parser.add_argument("--headless", default=False, action="store_true", help="Run in headless mode")
+    parser.add_argument("--debug", default=False, action="store_true", help="Enable debug logging")
     return parser.parse_args()
 
 

src/reachy_mini_conversation_demo/vision/processors.py

@@ -1,18 +1,19 @@
-import asyncio  # noqa: D100
-import base64
-import logging
 import os
 import sys
-import threading
 import time
-from dataclasses import dataclass
+import base64
+import asyncio
+import logging
+import threading
 from typing import Any, Dict
+from dataclasses import dataclass
 
 import cv2
 import numpy as np
 import torch
+from transformers import AutoProcessor, AutoModelForImageTextToText
 from huggingface_hub import snapshot_download
-from transformers import AutoModelForImageTextToText, AutoProcessor
+
 
 logger = logging.getLogger(__name__)
 
@@ -61,9 +62,7 @@ class VisionProcessor:
     def initialize(self) -> bool:
         """Load model and processor onto the selected device."""
         try:
-            logger.info(
-                f"Loading SmolVLM2 model on {self.device} (HF_HOME={os.getenv('HF_HOME')})"
-            )
+            logger.info(f"Loading SmolVLM2 model on {self.device} (HF_HOME={os.getenv('HF_HOME')})")
             self.processor = AutoProcessor.from_pretrained(self.model_path)
 
             # Select dtype depending on device
@@ -81,9 +80,7 @@ class VisionProcessor:
                 model_kwargs["_attn_implementation"] = "flash_attention_2"
 
             # Load model weights
-            self.model = AutoModelForImageTextToText.from_pretrained(
-                self.model_path, **model_kwargs
-            ).to(self.device)
+            self.model = AutoModelForImageTextToText.from_pretrained(self.model_path, **model_kwargs).to(self.device)
 
             self.model.eval()
             self._initialized = True
@@ -138,10 +135,7 @@ class VisionProcessor:
                 )
 
                 # Move tensors to device WITHOUT forcing dtype (keeps input_ids as torch.long)
-                inputs = {
-                    k: (v.to(self.device) if hasattr(v, "to") else v)
-                    for k, v in inputs.items()
-                }
+                inputs = {k: (v.to(self.device) if hasattr(v, "to") else v) for k, v in inputs.items()}
 
                 with torch.no_grad():
                     generated_ids = self.model.generate(
@@ -246,9 +240,7 @@ class VisionManager:
                         )
 
                         # Only update if we got a valid response
-                        if description and not description.startswith(
-                            ("Vision", "Failed", "Error")
-                        ):
+                        if description and not description.startswith(("Vision", "Failed", "Error")):
                             self._current_description = description
                             self._last_processed_time = current_time
 
@@ -268,18 +260,14 @@ class VisionManager:
         """Get the most recent scene description (thread-safe)."""
         return self._current_description
 
-    async def process_current_frame(
-        self, prompt: str = "Describe what you see in detail."
-    ) -> Dict[str, Any]:
+    async def process_current_frame(self, prompt: str = "Describe what you see in detail.") -> Dict[str, Any]:
         """Process current camera frame with custom prompt."""
         try:
             success, frame = self.camera.read()
             if not success or frame is None:
                 return {"error": "Failed to capture image from camera"}
 
-            description = await asyncio.to_thread(
-                lambda: self.processor.process_image(frame, prompt)
-            )
+            description = await asyncio.to_thread(lambda: self.processor.process_image(frame, prompt))
 
             return {
                 "description": description,
@@ -335,9 +323,7 @@ def create_vision_processor(config: VisionConfig):
         return VisionProcessor(config)
 
 
-def init_vision(
-    camera: cv2.VideoCapture, processor_type: str = "local"
-) -> VisionManager:
+def init_vision(camera: cv2.VideoCapture, processor_type: str = "local") -> VisionManager:
     """Initialize vision manager with the specified processor type."""
     model_id = "HuggingFaceTB/SmolVLM2-2.2B-Instruct"
 

src/reachy_mini_conversation_demo/vision/yolo_head_tracker.py

@@ -1,12 +1,12 @@
-from __future__ import annotations  # noqa: D100
-
+from __future__ import annotations
 import logging
-from typing import Optional, Tuple
+from typing import Tuple, Optional
 
 import numpy as np
-from huggingface_hub import hf_hub_download
 from supervision import Detections
 from ultralytics import YOLO
+from huggingface_hub import hf_hub_download
+
 
 logger = logging.getLogger(__name__)
 
@@ -94,9 +94,7 @@ class HeadTracker:
 
         return np.array([norm_x, norm_y], dtype=np.float32)
 
-    def get_eyes(
-        self, img: np.ndarray
-    ) -> Tuple[Optional[np.ndarray], Optional[np.ndarray]]:
+    def get_eyes(self, img: np.ndarray) -> Tuple[Optional[np.ndarray], Optional[np.ndarray]]:
         """Get eye positions (approximated from face bbox).
 
         Note: YOLO only provides face bbox, so we estimate eye positions
@@ -131,20 +129,14 @@ class HeadTracker:
         right_eye_x = bbox[0] + face_width * 0.65
 
         # Convert to MediaPipe coordinates
-        left_eye = np.array(
-            [(left_eye_x / w) * 2 - 1, (eye_y / h) * 2 - 1], dtype=np.float32
-        )
-        right_eye = np.array(
-            [(right_eye_x / w) * 2 - 1, (eye_y / h) * 2 - 1], dtype=np.float32
-        )
+        left_eye = np.array([(left_eye_x / w) * 2 - 1, (eye_y / h) * 2 - 1], dtype=np.float32)
+        right_eye = np.array([(right_eye_x / w) * 2 - 1, (eye_y / h) * 2 - 1], dtype=np.float32)
 
         return left_eye, right_eye
 
     def get_eyes_from_landmarks(self, face_landmarks) -> Tuple[np.ndarray, np.ndarray]:
         """Compatibility method - YOLO doesn't have landmarks, so we store bbox in the object."""
-        if not hasattr(face_landmarks, "_bbox") or not hasattr(
-            face_landmarks, "_img_shape"
-        ):
+        if not hasattr(face_landmarks, "_bbox") or not hasattr(face_landmarks, "_img_shape"):
             raise ValueError("Face landmarks object missing required attributes")
 
         bbox = face_landmarks._bbox
@@ -158,12 +150,8 @@ class HeadTracker:
         left_eye_x = bbox[0] + face_width * 0.35
         right_eye_x = bbox[0] + face_width * 0.65
 
-        left_eye = np.array(
-            [(left_eye_x / w) * 2 - 1, (eye_y / h) * 2 - 1], dtype=np.float32
-        )
-        right_eye = np.array(
-            [(right_eye_x / w) * 2 - 1, (eye_y / h) * 2 - 1], dtype=np.float32
-        )
+        left_eye = np.array([(left_eye_x / w) * 2 - 1, (eye_y / h) * 2 - 1], dtype=np.float32)
+        right_eye = np.array([(right_eye_x / w) * 2 - 1, (eye_y / h) * 2 - 1], dtype=np.float32)
 
         return left_eye, right_eye
 
@@ -177,9 +165,7 @@ class HeadTracker:
         left_eye, right_eye = self.get_eyes_from_landmarks(face_landmarks)
         return float(np.arctan2(right_eye[1] - left_eye[1], right_eye[0] - left_eye[0]))
 
-    def get_head_position(
-        self, img: np.ndarray
-    ) -> Tuple[Optional[np.ndarray], Optional[float]]:
+    def get_head_position(self, img: np.ndarray) -> Tuple[Optional[np.ndarray], Optional[float]]:
         """Get head position from face detection.
 
         Args: