Apply Ruff for better formatting

src/reachy_mini_conversation_demo/audio_sway.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import asyncio
 import base64
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from typing import Callable, Optional, Tuple
 
 import numpy as np

src/reachy_mini_conversation_demo/main.py

@@ -100,7 +100,9 @@ elif HEAD_TRACKING and SIM:
 else:
     logger.warning("Head tracking disabled")
 
-movement_manager = MovementManager(current_robot=current_robot, head_tracker=head_tracker, camera=camera)
+movement_manager = MovementManager(
+    current_robot=current_robot, head_tracker=head_tracker, camera=camera
+)
 robot_is_speaking = asyncio.Event()
 speaking_queue = asyncio.Queue()
 
@@ -410,7 +412,7 @@ async def control_mic_loop():
             audio_sync.on_response_completed()
             await asyncio.sleep(0)
             continue
-    
+
         await asyncio.sleep(block_time)
 
 
@@ -423,7 +425,7 @@ async def main():
     recorder.record()
     player = GstPlayer()
     player.play()
-    
+
     movement_manager.set_neutral()
     logger.info("Starting main audio loop. You can start to speak")
 
@@ -432,12 +434,16 @@ async def main():
         asyncio.create_task(emit_loop(player, openai), name="emit"),
         asyncio.create_task(receive_loop(recorder, openai), name="recv"),
         asyncio.create_task(control_mic_loop(), name="mic-mute"),
-        asyncio.create_task(movement_manager.enable(stop_event=stop_event), name="move"),        
+        asyncio.create_task(
+            movement_manager.enable(stop_event=stop_event), name="move"
+        ),
     ]
 
     if vision_manager:
         tasks.append(
-            asyncio.create_task(vision_manager.enable(stop_event=stop_event), name="vision"),
+            asyncio.create_task(
+                vision_manager.enable(stop_event=stop_event), name="vision"
+            ),
         )
 
     try:
@@ -445,7 +451,7 @@ async def main():
     except asyncio.CancelledError:
         logger.info("Shutting down")
         stop_event.set()
-    
+
     if camera:
         camera.release()
 
@@ -453,9 +459,10 @@ async def main():
     movement_manager.set_neutral()
     recorder.stop()
     player.stop()
-    
+
     current_robot.client.disconnect()
     logger.info("Stopped, robot disconected")
 
+
 if __name__ == "__main__":
     asyncio.run(main())

src/reachy_mini_conversation_demo/movement.py

@@ -15,7 +15,12 @@ logger = logging.getLogger(__name__)
 
 
 class MovementManager:
-    def __init__(self, current_robot: ReachyMini, head_tracker: HeadTracker | None, camera: cv2.VideoCapture| None):
+    def __init__(
+        self,
+        current_robot: ReachyMini,
+        head_tracker: HeadTracker | None,
+        camera: cv2.VideoCapture | None,
+    ):
         self.current_robot = current_robot
         self.head_tracker = head_tracker
         self.camera = camera
@@ -25,21 +30,20 @@ class MovementManager:
         self.moving_start = time.monotonic()
         self.moving_for = 0.0
         self.speech_head_offsets = [0.0] * 6
-        self.movement_loop_sleep = 0.05 # seconds 
+        self.movement_loop_sleep = 0.05  # seconds
 
     def set_offsets(self, offsets: list[float]) -> None:
         """Used by AudioSync callback to update speech offsets"""
         self.speech_head_offsets = list(offsets)
 
     def set_neutral(self) -> None:
-        """Set neutral robot position """
+        """Set neutral robot position"""
         self.speech_head_offsets = [0.0] * 6
         self.current_head_pose = create_head_pose(0, 0, 0, 0, 0, 0, degrees=True)
         self.current_robot.set_target(head=self.current_head_pose, antennas=(0.0, 0.0))
 
     def reset_head_pose(self) -> None:
         self.current_head_pose = np.eye(4)
-    
 
     async def enable(self, stop_event: threading.Event) -> None:
         logger.info("Starting head movement loop")
@@ -56,7 +60,9 @@ class MovementManager:
                         logger.warning("Camera read failed")
                         last_log_ts = current_time
                 else:
-                    eye_center, _ = self.head_tracker.get_head_position(im)  # as [-1, 1]
+                    eye_center, _ = self.head_tracker.get_head_position(
+                        im
+                    )  # as [-1, 1]
 
                     if eye_center is not None:
                         # Rescale target position into IMAGE_SIZE coordinates
@@ -68,22 +74,20 @@ class MovementManager:
                         # Bounds checking
                         eye_center = np.clip(eye_center, [0, 0], [w - 1, h - 1])
 
-                        current_head_pose = (
-                            self.current_robot.look_at_image(
-                                *eye_center, duration=0.0, apply=False
-                            )
+                        current_head_pose = self.current_robot.look_at_image(
+                            *eye_center, duration=0.0, apply=False
                         )
 
                         self.current_head_pose = current_head_pose
             # Pose calculation
             try:
-                current_x, current_y, current_z = self.current_head_pose[
-                    :3, 3
-                ]
+                current_x, current_y, current_z = self.current_head_pose[:3, 3]
 
-                current_roll, current_pitch, current_yaw = scipy.spatial.transform.Rotation.from_matrix(
-                    self.current_head_pose[:3, :3]
-                ).as_euler("xyz", degrees=False)
+                current_roll, current_pitch, current_yaw = (
+                    scipy.spatial.transform.Rotation.from_matrix(
+                        self.current_head_pose[:3, :3]
+                    ).as_euler("xyz", degrees=False)
+                )
 
                 if debug_frame_count % 50 == 0:
                     logger.debug(
@@ -102,15 +106,11 @@ class MovementManager:
             except Exception as e:
                 logger.exception("Invalid pose; resetting")
                 self.reset_head_pose()
-                current_x, current_y, current_z = self.current_head_pose[
-                    :3, 3
-                ]
+                current_x, current_y, current_z = self.current_head_pose[:3, 3]
                 current_roll = current_pitch = current_yaw = 0.0
 
             # Movement check
-            is_moving = (
-                time.monotonic() - self.moving_start < self.moving_for
-            )
+            is_moving = time.monotonic() - self.moving_start < self.moving_for
 
             if debug_frame_count % 50 == 0:
                 logger.debug(f"Robot moving: {is_moving}")
@@ -133,9 +133,7 @@ class MovementManager:
                         logger.debug(
                             "Final head pose with offsets: %s", head_pose[:3, 3]
                         )
-                        logger.debug(
-                            "Speech offsets: %s", self.speech_head_offsets
-                        )
+                        logger.debug("Speech offsets: %s", self.speech_head_offsets)
 
                     self.current_robot.set_target(head=head_pose, antennas=(0.0, 0.0))
 
@@ -146,5 +144,5 @@ class MovementManager:
                     logger.debug("Failed to set robot target: %s", e)
 
             await asyncio.sleep(self.movement_loop_sleep)
-        
+
         logger.info("Exited head movement loop")

src/reachy_mini_conversation_demo/vision.py

@@ -239,9 +239,10 @@ class VisionManager:
                 if current_time - self._last_processed_time >= self.vision_interval:
                     success, frame = await asyncio.to_thread(self.camera.read)
                     if success and frame is not None:
-                        
-                        description = await asyncio.to_thread(lambda: self.processor.process_image(
-                            frame, "Briefly describe what you see in one sentence.")
+                        description = await asyncio.to_thread(
+                            lambda: self.processor.process_image(
+                                frame, "Briefly describe what you see in one sentence."
+                            )
                         )
 
                         # Only update if we got a valid response
@@ -259,7 +260,7 @@ class VisionManager:
 
             except Exception as e:
                 logger.exception("Vision processing loop error")
-                await asyncio.sleep(5.0) # Longer sleep on error
+                await asyncio.sleep(5.0)  # Longer sleep on error
 
         logger.info(f"Vision loop finished")
 
@@ -275,8 +276,10 @@ class VisionManager:
             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,
@@ -287,8 +290,7 @@ class VisionManager:
         except Exception as e:
             logger.exception("Failed to process current frame")
             return {"error": f"Frame processing failed: {str(e)}"}
-        
-        
+
     async def get_status(self) -> Dict[str, Any]:
         """Get comprehensive status information"""
         return {
@@ -303,18 +305,14 @@ class VisionManager:
         }
 
 
-
-def init_camera(camera_index = 0, simulation=True):
-    
+def init_camera(camera_index=0, simulation=True):
     api_preference = cv2.CAP_AVFOUNDATION if sys.platform == "darwin" else 0
 
     if simulation:
         # Default build-in camera in SIM
         # TODO: please, test on Linux and Windows
         # TODO simulation in find_camera
-        camera = cv2.VideoCapture(
-            0, api_preference
-        )
+        camera = cv2.VideoCapture(0, api_preference)
     else:
         # TODO handle macos in find_camera
         if sys.platform == "darwin":
@@ -328,7 +326,6 @@ def init_camera(camera_index = 0, simulation=True):
 def init_vision(camera: cv2.VideoCapture) -> VisionManager:
     model_id = "HuggingFaceTB/SmolVLM2-2.2B-Instruct"
 
-
     cache_dir = os.path.expandvars(os.getenv("HF_HOME", "$HOME/.cache/huggingface"))
 
     try:
@@ -362,7 +359,7 @@ def init_vision(camera: cv2.VideoCapture) -> VisionManager:
 
     device_info = vision_manager.processor.get_model_info()
     logger.info(
-        f"Vision processing enabled: {device_info["model_path"]} on {device_info["device"]} memory: {device_info.get("gpu_memory", "N/A")}",
+        f"Vision processing enabled: {device_info['model_path']} on {device_info['device']} memory: {device_info.get('gpu_memory', 'N/A')}",
     )
 
     return vision_manager