Apply Ruff

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

@@ -25,18 +25,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 +102,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 +115,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 +179,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 +199,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/speech_tapper.py

@@ -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

@@ -114,17 +114,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 +130,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 +150,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 +165,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 +181,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 +200,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

@@ -15,7 +15,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

@@ -30,9 +30,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 +43,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 +64,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 +77,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 +112,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 +128,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/moves.py

@@ -27,9 +27,7 @@ from reachy_mini.utils.interpolation import (
 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 +66,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 +79,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 +114,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 +211,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 +222,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 +233,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 +289,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 +297,7 @@ class MovementManager:
                 if self.move_queue:
                     self.state.current_move = self.move_queue.popleft()
                     self.state.move_start_time = current_time
-                    logger.debug(
-                        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 +308,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 +318,7 @@ class MovementManager:
                         )
                         self.move_queue.append(breathing_move)
                         self.state.update_activity()
-                        logger.debug(
-                            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 +335,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 +356,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
 
@@ -496,9 +453,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 +494,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 +509,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

@@ -93,35 +93,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),
                         ),
                     )
 
@@ -155,9 +142,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:
@@ -178,9 +163,7 @@ class OpenaiRealtimeHandler(AsyncStreamHandler):
                             {
                                 "role": "assistant",
                                 "content": json.dumps(tool_result),
-                                "metadata": dict(
-                                    title="🛠️ Used tool " + tool_name, status="done"
-                                ),
+                                "metadata": dict(title="🛠️ Used tool " + tool_name, status="done"),
                             },
                         )
                     )
@@ -232,11 +215,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:
@@ -259,9 +238,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/tools.py

@@ -183,9 +183,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 +234,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
@@ -436,9 +430,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 +515,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 +594,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 +669,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

@@ -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

@@ -61,9 +61,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 +79,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 +134,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 +239,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 +259,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 +322,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

@@ -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: