Add local face tracking (YOLO) and vision processing (SmolVLM2)

Face Tracking:
- Initialize YOLO HeadTracker for local face detection
- Face tracking now works out of the box with --no-face-tracking to disable
- CameraWorker properly integrates with head tracker

Local Vision (optional):
- Add VisionProcessor using SmolVLM2 for on-device image understanding
- VisionManager for periodic scene description
- Enable with --local-vision flag or ENABLE_LOCAL_VISION=true
- Falls back to OpenClaw for vision if local not available

Configuration:
- ENABLE_FACE_TRACKING (default: true)
- ENABLE_LOCAL_VISION (default: false)
- LOCAL_VISION_MODEL (default: SmolVLM2-256M-Video-Instruct)
- VISION_DEVICE (auto, cuda, mps, cpu)
- HF_HOME for model cache directory

Based on pollen-robotics/reachy_mini_conversation_app implementation.

src/reachy_mini_openclaw/config.py

@@ -42,7 +42,13 @@ class Config:
     
     # Face Tracking Configuration
     # Options: "yolo", "mediapipe", or None for auto-detect
-    HEAD_TRACKER_TYPE: Optional[str] = field(default_factory=lambda: os.getenv("HEAD_TRACKER_TYPE"))
+    HEAD_TRACKER_TYPE: Optional[str] = field(default_factory=lambda: os.getenv("HEAD_TRACKER_TYPE", "yolo"))
+    
+    # Local Vision Processing
+    ENABLE_LOCAL_VISION: bool = field(default_factory=lambda: os.getenv("ENABLE_LOCAL_VISION", "false").lower() == "true")
+    LOCAL_VISION_MODEL: str = field(default_factory=lambda: os.getenv("LOCAL_VISION_MODEL", "HuggingFaceTB/SmolVLM2-256M-Video-Instruct"))
+    VISION_DEVICE: str = field(default_factory=lambda: os.getenv("VISION_DEVICE", "auto"))  # "auto", "cuda", "mps", "cpu"
+    HF_HOME: str = field(default_factory=lambda: os.getenv("HF_HOME", os.path.expanduser("~/.cache/huggingface")))
     
     # Custom Profile (for personality customization)
     CUSTOM_PROFILE: Optional[str] = field(default_factory=lambda: os.getenv("REACHY_MINI_CUSTOM_PROFILE"))
@@ -64,3 +70,15 @@ def set_custom_profile(profile: Optional[str]) -> None:
     global config
     config.CUSTOM_PROFILE = profile
     os.environ["REACHY_MINI_CUSTOM_PROFILE"] = profile or ""
+
+
+def set_face_tracking_enabled(enabled: bool) -> None:
+    """Enable or disable face tracking at runtime."""
+    global config
+    config.ENABLE_FACE_TRACKING = enabled
+
+
+def set_local_vision_enabled(enabled: bool) -> None:
+    """Enable or disable local vision processing at runtime."""
+    global config
+    config.ENABLE_LOCAL_VISION = enabled

src/reachy_mini_openclaw/main.py

@@ -111,6 +111,16 @@ Examples:
         action="store_true",
         help="Disable OpenClaw integration"
     )
+    parser.add_argument(
+        "--no-face-tracking",
+        action="store_true",
+        help="Disable face tracking"
+    )
+    parser.add_argument(
+        "--local-vision",
+        action="store_true",
+        help="Enable local vision processing with SmolVLM2"
+    )
     parser.add_argument(
         "--profile",
         type=str,
@@ -208,16 +218,29 @@ class ClawBodyCore:
         
         # Camera worker for video streaming and frame capture
         self.camera_worker = None
+        self.head_tracker = None
+        self.vision_manager = None
+        
         if enable_camera:
             logger.info("Initializing camera worker...")
             from reachy_mini_openclaw.camera_worker import CameraWorker
-            # Initialize without face tracking (face tracking is optional)
+            
+            # Initialize head tracker for local face tracking
+            if config.ENABLE_FACE_TRACKING:
+                self.head_tracker = self._initialize_head_tracker(config.HEAD_TRACKER_TYPE)
+            
+            # Initialize camera worker with head tracker
             self.camera_worker = CameraWorker(
                 reachy_mini=self.robot,
-                head_tracker=None,  # No face tracking by default
+                head_tracker=self.head_tracker,
             )
-            # Disable head tracking since we don't have a tracker
-            self.camera_worker.set_head_tracking_enabled(False)
+            
+            # Enable/disable head tracking based on whether we have a tracker
+            self.camera_worker.set_head_tracking_enabled(self.head_tracker is not None)
+            
+            # Initialize local vision processor if enabled
+            if config.ENABLE_LOCAL_VISION:
+                self.vision_manager = self._initialize_vision_manager()
         
         # Create tool dependencies
         self.deps = ToolDependencies(
@@ -226,6 +249,7 @@ class ClawBodyCore:
             robot=self.robot,
             camera_worker=self.camera_worker,
             openclaw_bridge=self.openclaw_bridge,
+            vision_manager=self.vision_manager,
         )
         
         # Initialize OpenAI Realtime handler with OpenClaw bridge
@@ -238,6 +262,88 @@ class ClawBodyCore:
         self._stop_event = asyncio.Event()
         self._tasks: list[asyncio.Task] = []
         
+    def _initialize_vision_manager(self) -> Optional[Any]:
+        """Initialize local vision processor (SmolVLM2).
+        
+        Returns:
+            VisionManager instance or None if initialization fails
+        """
+        if self.camera_worker is None:
+            logger.warning("Cannot initialize vision manager without camera worker")
+            return None
+        
+        try:
+            from reachy_mini_openclaw.vision.processors import (
+                VisionConfig, 
+                initialize_vision_manager,
+            )
+            from reachy_mini_openclaw.config import config
+            
+            vision_config = VisionConfig(
+                model_path=config.LOCAL_VISION_MODEL,
+                device_preference=config.VISION_DEVICE,
+                hf_home=config.HF_HOME,
+            )
+            
+            logger.info("Initializing local vision processor (SmolVLM2)...")
+            vision_manager = initialize_vision_manager(self.camera_worker, vision_config)
+            
+            if vision_manager is not None:
+                logger.info("Local vision processor initialized")
+            else:
+                logger.warning("Local vision processor failed to initialize")
+            
+            return vision_manager
+            
+        except ImportError as e:
+            logger.warning(f"Local vision not available: {e}")
+            logger.warning("Install with: pip install torch transformers")
+            return None
+        except Exception as e:
+            logger.error(f"Failed to initialize vision manager: {e}")
+            return None
+    
+    def _initialize_head_tracker(self, tracker_type: Optional[str] = None) -> Optional[Any]:
+        """Initialize head tracker for local face tracking.
+        
+        Args:
+            tracker_type: Type of tracker ("yolo", "mediapipe", or None for auto)
+            
+        Returns:
+            Initialized head tracker or None if initialization fails
+        """
+        # Default to YOLO if not specified
+        if tracker_type is None:
+            tracker_type = "yolo"
+        
+        if tracker_type == "yolo":
+            try:
+                from reachy_mini_openclaw.vision.yolo_head_tracker import HeadTracker
+                logger.info("Initializing YOLO face tracker...")
+                tracker = HeadTracker(device="cpu")  # CPU is fast enough for face detection
+                logger.info("YOLO face tracker initialized")
+                return tracker
+            except ImportError as e:
+                logger.warning(f"YOLO tracker not available: {e}")
+                logger.warning("Install with: pip install ultralytics supervision")
+            except Exception as e:
+                logger.error(f"Failed to initialize YOLO tracker: {e}")
+        
+        elif tracker_type == "mediapipe":
+            try:
+                from reachy_mini_openclaw.vision.mediapipe_tracker import HeadTracker
+                logger.info("Initializing MediaPipe face tracker...")
+                tracker = HeadTracker()
+                logger.info("MediaPipe face tracker initialized")
+                return tracker
+            except ImportError as e:
+                logger.warning(f"MediaPipe tracker not available: {e}")
+            except Exception as e:
+                logger.error(f"Failed to initialize MediaPipe tracker: {e}")
+        
+        logger.warning("No face tracker available - face tracking disabled")
+        return None
+        
     def _should_stop(self) -> bool:
         """Check if we should stop."""
         if self._stop_event.is_set():
@@ -305,6 +411,11 @@ class ClawBodyCore:
             logger.info("Starting camera worker...")
             self.camera_worker.start()
         
+        # Start local vision processor if available
+        if self.vision_manager is not None:
+            logger.info("Starting local vision processor...")
+            self.vision_manager.start()
+        
         # Start audio
         logger.info("Starting audio...")
         self.robot.media.start_recording()
@@ -342,6 +453,10 @@ class ClawBodyCore:
         self.head_wobbler.stop()
         self.movement_manager.stop()
         
+        # Stop vision manager
+        if self.vision_manager is not None:
+            self.vision_manager.stop()
+        
         # Stop camera worker
         if self.camera_worker is not None:
             self.camera_worker.stop()
@@ -404,6 +519,16 @@ def main() -> None:
         from reachy_mini_openclaw.config import set_custom_profile
         set_custom_profile(args.profile)
     
+    # Configure face tracking and local vision from args
+    from reachy_mini_openclaw.config import (
+        set_face_tracking_enabled, 
+        set_local_vision_enabled,
+    )
+    if args.no_face_tracking:
+        set_face_tracking_enabled(False)
+    if args.local_vision:
+        set_local_vision_enabled(True)
+    
     if args.gradio:
         # Launch Gradio UI
         logger.info("Starting Gradio UI...")

src/reachy_mini_openclaw/tools/core_tools.py

@@ -36,6 +36,7 @@ class ToolDependencies:
     robot: Any  # ReachyMini instance
     camera_worker: Optional[Any] = None
     openclaw_bridge: Optional["OpenClawBridge"] = None
+    vision_manager: Optional[Any] = None  # Local vision processor (SmolVLM2)
 
 
 # Tool specifications in OpenAI format
@@ -211,12 +212,12 @@ async def _handle_look(args: dict, deps: ToolDependencies) -> dict:
 
 
 async def _handle_camera(args: dict, deps: ToolDependencies) -> dict:
-    """Handle the camera tool - capture image and get description from OpenClaw.
+    """Handle the camera tool - capture image and get description.
     
-    Since OpenAI Realtime doesn't have vision, we send the image to OpenClaw
-    for analysis and return the description.
+    Uses local vision (SmolVLM2) if available, otherwise falls back to OpenClaw.
     """
-    logger.info("Camera tool called, camera_worker=%s", deps.camera_worker is not None)
+    logger.info("Camera tool called, camera_worker=%s, vision_manager=%s", 
+                deps.camera_worker is not None, deps.vision_manager is not None)
     
     if deps.camera_worker is None:
         logger.warning("Camera worker is None")
@@ -239,16 +240,31 @@ async def _handle_camera(args: dict, deps: ToolDependencies) -> dict:
         if frame is None:
             return {"error": "No frame available from camera"}
         
-        # Encode frame as JPEG base64
-        import cv2
-        logger.info("Encoding frame, shape=%s", frame.shape)
-        _, buffer = cv2.imencode('.jpg', frame, [cv2.IMWRITE_JPEG_QUALITY, 85])
-        b64_image = base64.b64encode(buffer).decode('utf-8')
-        logger.info("Frame encoded, size=%d bytes", len(b64_image))
+        logger.info("Got frame, shape=%s", frame.shape)
         
-        # Send to OpenClaw for vision analysis (OpenAI Realtime doesn't have vision)
+        # Option 1: Use local vision processor (SmolVLM2) if available
+        if deps.vision_manager is not None:
+            logger.info("Using local vision processor (SmolVLM2)...")
+            description = deps.vision_manager.process_now(
+                "Describe what you see in this image. Be specific about people, objects, and the environment. Keep it concise (2-3 sentences)."
+            )
+            if description and not description.startswith(("Vision", "Failed", "Error", "GPU", "No camera")):
+                logger.info("Local vision response: %s", description[:100])
+                return {
+                    "status": "success",
+                    "description": description,
+                    "source": "local_vision"
+                }
+            else:
+                logger.warning("Local vision failed: %s", description)
+        
+        # Option 2: Fall back to OpenClaw for vision analysis
         if deps.openclaw_bridge is not None and deps.openclaw_bridge.is_connected:
-            logger.info("Sending image to OpenClaw for vision analysis...")
+            logger.info("Using OpenClaw for vision analysis...")
+            import cv2
+            _, buffer = cv2.imencode('.jpg', frame, [cv2.IMWRITE_JPEG_QUALITY, 85])
+            b64_image = base64.b64encode(buffer).decode('utf-8')
+            
             response = await deps.openclaw_bridge.chat(
                 "Describe what you see in this image. Be specific about people, objects, and the environment. Keep it concise (2-3 sentences).",
                 image_b64=b64_image,
@@ -259,14 +275,15 @@ async def _handle_camera(args: dict, deps: ToolDependencies) -> dict:
                 return {
                     "status": "success",
                     "description": response.content,
+                    "source": "openclaw"
                 }
             else:
                 logger.warning("OpenClaw vision failed: %s", response.error)
         
-        # Fallback if OpenClaw not available
+        # Fallback if neither is available
         return {
-            "status": "success",
-            "description": "I captured an image but couldn't analyze it. My vision system needs OpenClaw to process what I see."
+            "status": "partial",
+            "description": "I captured an image but couldn't analyze it. No vision processing available."
         }
     except Exception as e:
         logger.error("Camera tool error: %s", e, exc_info=True)
@@ -281,11 +298,12 @@ async def _handle_face_tracking(args: dict, deps: ToolDependencies) -> dict:
         return {"error": "Camera not available for face tracking"}
     
     try:
-        if hasattr(deps.camera_worker, 'set_face_tracking'):
-            deps.camera_worker.set_face_tracking(enabled)
-            return {"status": "success", "face_tracking": enabled}
-        else:
-            return {"error": "Face tracking not supported"}
+        # Check if head tracker is available
+        if deps.camera_worker.head_tracker is None:
+            return {"error": "Face tracking not available - no head tracker initialized"}
+        
+        deps.camera_worker.set_head_tracking_enabled(enabled)
+        return {"status": "success", "face_tracking": enabled}
     except Exception as e:
         return {"error": str(e)}
 

src/reachy_mini_openclaw/vision/__init__.py

@@ -1,5 +1,18 @@
-"""Vision modules for face tracking and detection."""
+"""Vision modules for face tracking, detection, and image understanding."""
 
 from reachy_mini_openclaw.vision.head_tracker import get_head_tracker
 
-__all__ = ["get_head_tracker"]
+__all__ = [
+    "get_head_tracker",
+]
+
+# Lazy imports for optional heavy dependencies
+def get_vision_processor():
+    """Get the VisionProcessor class (requires torch, transformers)."""
+    from reachy_mini_openclaw.vision.processors import VisionProcessor
+    return VisionProcessor
+
+def get_vision_manager():
+    """Get the VisionManager class (requires torch, transformers)."""
+    from reachy_mini_openclaw.vision.processors import VisionManager
+    return VisionManager

src/reachy_mini_openclaw/vision/processors.py

@@ -0,0 +1,419 @@
+"""Local vision processing with SmolVLM2.
+
+Provides on-device image understanding using the SmolVLM2 model
+for scene description and visual analysis.
+
+Based on pollen-robotics/reachy_mini_conversation_app vision processors.
+"""
+
+import os
+import time
+import base64
+import logging
+import threading
+from typing import Any, Dict, Optional
+from dataclasses import dataclass, field
+
+import cv2
+import numpy as np
+from numpy.typing import NDArray
+
+try:
+    import torch
+    from transformers import AutoProcessor, AutoModelForImageTextToText
+    from huggingface_hub import snapshot_download
+    VISION_AVAILABLE = True
+except ImportError:
+    VISION_AVAILABLE = False
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class VisionConfig:
+    """Configuration for vision processing."""
+
+    model_path: str = "HuggingFaceTB/SmolVLM2-256M-Video-Instruct"
+    vision_interval: float = 5.0
+    max_new_tokens: int = 64
+    jpeg_quality: int = 85
+    max_retries: int = 3
+    retry_delay: float = 1.0
+    device_preference: str = "auto"  # "auto", "cuda", "mps", "cpu"
+    hf_home: str = field(default_factory=lambda: os.path.expanduser("~/.cache/huggingface"))
+
+
+class VisionProcessor:
+    """Handles SmolVLM2 model loading and inference for local vision."""
+
+    def __init__(self, vision_config: Optional[VisionConfig] = None):
+        """Initialize the vision processor.
+        
+        Args:
+            vision_config: Vision configuration settings
+        """
+        if not VISION_AVAILABLE:
+            raise ImportError(
+                "Vision processing requires: pip install torch transformers huggingface-hub"
+            )
+        
+        self.vision_config = vision_config or VisionConfig()
+        self.model_path = self.vision_config.model_path
+        self.device = self._determine_device()
+        self.processor = None
+        self.model = None
+        self._initialized = False
+
+    def _determine_device(self) -> str:
+        """Determine the best device for inference."""
+        pref = self.vision_config.device_preference
+        
+        if pref == "cpu":
+            return "cpu"
+        if pref == "cuda":
+            return "cuda" if torch.cuda.is_available() else "cpu"
+        if pref == "mps":
+            return "mps" if torch.backends.mps.is_available() else "cpu"
+        
+        # auto: prefer mps on Apple, then cuda, else cpu
+        if torch.backends.mps.is_available():
+            return "mps"
+        return "cuda" if torch.cuda.is_available() else "cpu"
+
+    def initialize(self) -> bool:
+        """Load model and processor onto the selected device.
+        
+        Returns:
+            True if initialization successful, False otherwise
+        """
+        try:
+            cache_dir = self.vision_config.hf_home
+            os.makedirs(cache_dir, exist_ok=True)
+            os.environ["HF_HOME"] = cache_dir
+            
+            logger.info(f"Loading SmolVLM2 model on {self.device} (HF_HOME={cache_dir})")
+            
+            # Download model to cache first
+            logger.info(f"Downloading vision model {self.model_path}...")
+            snapshot_download(
+                repo_id=self.model_path,
+                repo_type="model",
+                cache_dir=cache_dir,
+            )
+            
+            self.processor = AutoProcessor.from_pretrained(self.model_path)
+
+            # Select dtype depending on device
+            if self.device == "cuda":
+                dtype = torch.bfloat16
+            elif self.device == "mps":
+                dtype = torch.float32  # best for MPS
+            else:
+                dtype = torch.float32
+
+            model_kwargs: Dict[str, Any] = {"torch_dtype": dtype}
+
+            # flash_attention_2 is CUDA-only; skip on MPS/CPU
+            if self.device == "cuda":
+                model_kwargs["_attn_implementation"] = "flash_attention_2"
+
+            # Load model weights
+            self.model = AutoModelForImageTextToText.from_pretrained(
+                self.model_path, **model_kwargs
+            ).to(self.device)
+
+            if self.model is not None:
+                self.model.eval()
+                self._initialized = True
+                logger.info(f"Vision model loaded successfully on {self.device}")
+                return True
+
+        except Exception as e:
+            logger.error(f"Failed to initialize vision model: {e}")
+            return False
+        
+        return False
+
+    def process_image(
+        self,
+        cv2_image: NDArray[np.uint8],
+        prompt: str = "Briefly describe what you see in one sentence.",
+    ) -> str:
+        """Process CV2 image and return description with retry logic.
+        
+        Args:
+            cv2_image: OpenCV image (BGR format)
+            prompt: Question/prompt to ask about the image
+            
+        Returns:
+            Text description of the image
+        """
+        if not self._initialized or self.processor is None or self.model is None:
+            return "Vision model not initialized"
+
+        for attempt in range(self.vision_config.max_retries):
+            try:
+                # Convert to JPEG bytes
+                success, jpeg_buffer = cv2.imencode(
+                    ".jpg",
+                    cv2_image,
+                    [cv2.IMWRITE_JPEG_QUALITY, self.vision_config.jpeg_quality],
+                )
+                if not success:
+                    return "Failed to encode image"
+
+                # Convert to base64
+                image_base64 = base64.b64encode(jpeg_buffer.tobytes()).decode("utf-8")
+
+                messages = [
+                    {
+                        "role": "user",
+                        "content": [
+                            {
+                                "type": "image",
+                                "url": f"data:image/jpeg;base64,{image_base64}",
+                            },
+                            {"type": "text", "text": prompt},
+                        ],
+                    },
+                ]
+
+                inputs = self.processor.apply_chat_template(
+                    messages,
+                    add_generation_prompt=True,
+                    tokenize=True,
+                    return_dict=True,
+                    return_tensors="pt",
+                )
+
+                # 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()
+                }
+
+                with torch.no_grad():
+                    generated_ids = self.model.generate(
+                        **inputs,
+                        do_sample=False,
+                        max_new_tokens=self.vision_config.max_new_tokens,
+                        pad_token_id=self.processor.tokenizer.eos_token_id,
+                    )
+
+                generated_texts = self.processor.batch_decode(
+                    generated_ids,
+                    skip_special_tokens=True,
+                )
+
+                # Extract just the response part
+                full_text = generated_texts[0]
+                response = self._extract_response(full_text)
+
+                # Clean up GPU memory if using CUDA
+                if self.device == "cuda":
+                    torch.cuda.empty_cache()
+                elif self.device == "mps":
+                    torch.mps.empty_cache()
+
+                return response.replace(chr(10), " ").strip()
+
+            except Exception as e:
+                if "OutOfMemory" in str(type(e).__name__):
+                    logger.error(f"GPU OOM on attempt {attempt + 1}: {e}")
+                    if self.device == "cuda":
+                        torch.cuda.empty_cache()
+                    if attempt < self.vision_config.max_retries - 1:
+                        time.sleep(self.vision_config.retry_delay * (attempt + 1))
+                    else:
+                        return "GPU out of memory - vision processing failed"
+                else:
+                    logger.error(f"Vision processing failed (attempt {attempt + 1}): {e}")
+                    if attempt < self.vision_config.max_retries - 1:
+                        time.sleep(self.vision_config.retry_delay)
+                    else:
+                        return f"Vision processing error after {self.vision_config.max_retries} attempts"
+
+        return "Vision processing failed"
+
+    def _extract_response(self, full_text: str) -> str:
+        """Extract the assistant's response from the full generated text."""
+        # Handle different response formats
+        markers = ["assistant\n", "Assistant:", "Response:", "\n\n"]
+
+        for marker in markers:
+            if marker in full_text:
+                response = full_text.split(marker)[-1].strip()
+                if response:  # Ensure we got a meaningful response
+                    return response
+
+        # Fallback: return the full text cleaned up
+        return full_text.strip()
+
+    def get_model_info(self) -> Dict[str, Any]:
+        """Get information about the loaded model."""
+        info = {
+            "initialized": self._initialized,
+            "device": self.device,
+            "model_path": self.model_path,
+            "cuda_available": torch.cuda.is_available() if VISION_AVAILABLE else False,
+        }
+        
+        if VISION_AVAILABLE and torch.cuda.is_available():
+            info["gpu_memory_gb"] = torch.cuda.get_device_properties(0).total_memory // (1024**3)
+        else:
+            info["gpu_memory_gb"] = "N/A"
+            
+        return info
+
+
+class VisionManager:
+    """Manages periodic vision processing and scene understanding.
+    
+    This runs in the background, periodically capturing frames and
+    generating scene descriptions that can be queried.
+    """
+
+    def __init__(
+        self, 
+        camera_worker: Any, 
+        vision_config: Optional[VisionConfig] = None,
+    ):
+        """Initialize vision manager.
+        
+        Args:
+            camera_worker: CameraWorker instance for frame capture
+            vision_config: Vision configuration settings
+        """
+        self.camera_worker = camera_worker
+        self.vision_config = vision_config or VisionConfig()
+        self.vision_interval = self.vision_config.vision_interval
+        self.processor = VisionProcessor(self.vision_config)
+
+        self._last_processed_time = 0.0
+        self._last_description = ""
+        self._description_lock = threading.Lock()
+        self._stop_event = threading.Event()
+        self._thread: Optional[threading.Thread] = None
+
+        # Initialize processor
+        if not self.processor.initialize():
+            logger.error("Failed to initialize vision processor")
+            raise RuntimeError("Vision processor initialization failed")
+
+    def start(self) -> None:
+        """Start the vision processing loop in a background thread."""
+        self._stop_event.clear()
+        self._thread = threading.Thread(target=self._working_loop, daemon=True)
+        self._thread.start()
+        logger.info("Local vision processing started")
+
+    def stop(self) -> None:
+        """Stop the vision processing loop."""
+        self._stop_event.set()
+        if self._thread is not None:
+            self._thread.join(timeout=5.0)
+        logger.info("Local vision processing stopped")
+
+    def get_latest_description(self) -> str:
+        """Get the most recent scene description.
+        
+        Returns:
+            Latest scene description or empty string if none available
+        """
+        with self._description_lock:
+            return self._last_description
+
+    def process_now(self, prompt: str = "Briefly describe what you see in one sentence.") -> str:
+        """Process the current frame immediately with a custom prompt.
+        
+        Args:
+            prompt: Question/prompt to ask about the image
+            
+        Returns:
+            Description of what the camera sees
+        """
+        frame = self.camera_worker.get_latest_frame()
+        if frame is None:
+            return "No camera frame available"
+        
+        return self.processor.process_image(frame, prompt)
+
+    def _working_loop(self) -> None:
+        """Vision processing loop (runs in separate thread)."""
+        while not self._stop_event.is_set():
+            try:
+                current_time = time.time()
+
+                if current_time - self._last_processed_time >= self.vision_interval:
+                    frame = self.camera_worker.get_latest_frame()
+                    if frame is not None:
+                        description = self.processor.process_image(
+                            frame,
+                            "Briefly describe what you see in one sentence.",
+                        )
+
+                        # Only update if we got a valid response
+                        if description and not description.startswith(
+                            ("Vision", "Failed", "Error", "GPU")
+                        ):
+                            with self._description_lock:
+                                self._last_description = description
+                            self._last_processed_time = current_time
+                            logger.debug(f"Vision update: {description}")
+                        else:
+                            logger.warning(f"Invalid vision response: {description}")
+
+                time.sleep(1.0)  # Check every second
+
+            except Exception:
+                logger.exception("Vision processing loop error")
+                time.sleep(5.0)  # Longer sleep on error
+
+        logger.info("Vision loop finished")
+
+    def get_status(self) -> Dict[str, Any]:
+        """Get comprehensive status information."""
+        return {
+            "last_processed": self._last_processed_time,
+            "last_description": self.get_latest_description(),
+            "processor_info": self.processor.get_model_info(),
+            "config": {
+                "interval": self.vision_interval,
+            },
+        }
+
+
+def initialize_vision_manager(
+    camera_worker: Any, 
+    config: Optional[VisionConfig] = None,
+) -> Optional[VisionManager]:
+    """Initialize vision manager with model download and configuration.
+
+    Args:
+        camera_worker: CameraWorker instance for frame capture
+        config: Optional vision configuration
+        
+    Returns:
+        VisionManager instance or None if initialization fails
+    """
+    if not VISION_AVAILABLE:
+        logger.warning("Vision dependencies not available. Install: pip install torch transformers")
+        return None
+    
+    try:
+        vision_config = config or VisionConfig()
+
+        # Initialize vision manager
+        vision_manager = VisionManager(camera_worker, vision_config)
+
+        # Log device info
+        device_info = vision_manager.processor.get_model_info()
+        logger.info(
+            f"Local vision enabled: {device_info.get('model_path')} on {device_info.get('device')}"
+        )
+
+        return vision_manager
+
+    except Exception as e:
+        logger.error(f"Failed to initialize vision manager: {e}")
+        return None