Initial commit

.env.example

@@ -0,0 +1,21 @@
+# Remote Control App Configuration
+
+# Profile to use (default, example_remote, or custom)
+REMOTE_CONTROL_PROFILE=default
+
+# Robot configuration
+ROBOT_NAME=reachy_mini
+
+# Logging
+LOG_LEVEL=INFO
+
+# Video settings (can override profile settings)
+VIDEO_JPEG_QUALITY=80
+
+# Monitoring
+LATENCY_WARNING_MS=200
+
+# REST API settings (optional)
+ENABLE_REST_API=true
+REST_API_HOST=0.0.0.0
+REST_API_PORT=7860

README.md

@@ -0,0 +1,327 @@
+# Reachy Mini Remote Control App
+
+A standalone app that connects Reachy Mini robot to a remote server for WebSocket-based control. The robot acts as a WebSocket **client** connecting to a remote server (e.g., Hugging Face Space), enabling remote control, video streaming, and bidirectional audio.
+
+## Features
+
+- **WebSocket Client Architecture**: Robot connects to remote server (works behind NAT/firewalls)
+- **Profile-Based Configuration**: Easily switch between different remote servers
+- **Bidirectional Streaming**: Video (robot → server) and Audio (robot ↔ server)
+- **Movement Control**: Receive movement commands from remote server
+- **Connection Monitoring**: Track latency, frames, and connection health
+- **Optional REST API**: Query status and manage profiles via HTTP endpoints
+
+## Architecture
+
+```
+Remote Server                    Robot (this app)                Reachy Mini Daemon
+(Hugging Face Space)
+
+WebSocket Server     ←──────     WebSocket Clients    ←──────    ReachyMini SDK
+- /robot                         - robot_control.py              (Zenoh protocol)
+- /video_stream                  - video_stream.py
+- /audio_stream                  - audio_stream.py
+```
+
+The app connects TO a remote server as a client, not the other way around. This simplifies network setup since the robot only needs outbound connections.
+
+## Installation
+
+### From Source
+
+```bash
+cd reachy_mini_remote_control_app
+pip install -e .
+```
+
+### Requirements
+
+- Python >= 3.10
+- Reachy Mini daemon running (`reachy-mini-daemon`)
+- Remote server running (e.g., [reachy_mini_remote_control/app.py](../reachy_mini_remote_control/app.py))
+
+## Usage
+
+### 1. Start the Remote Server
+
+First, start the remote server that the robot will connect to:
+
+```bash
+cd ../reachy_mini_remote_control
+python app.py
+# Server starts on ws://localhost:8000
+```
+
+### 2. Start the Reachy Mini Daemon
+
+```bash
+reachy-mini-daemon --mockup-sim
+```
+
+### 3. Start the Remote Control App
+
+```bash
+reachy-mini-remote-control-app --profile default
+```
+
+The app will:
+- Connect to the remote server at `ws://localhost:8000`
+- Stream video and audio
+- Receive and execute movement commands
+
+### Command Line Options
+
+```bash
+reachy-mini-remote-control-app [OPTIONS]
+
+Options:
+  --profile PROFILE            Profile to use (default: from .env)
+  --websocket-uri URI          WebSocket URI to connect to (overrides profile)
+  --robot-name NAME            Robot name (default: reachy_mini)
+  --log-level LEVEL            Logging level (DEBUG, INFO, WARNING, ERROR)
+  --no-rest-api                Disable REST API for status endpoints
+```
+
+## Configuration
+
+### Profiles
+
+Profiles are stored in `src/reachy_mini_remote_control_app/profiles/`. Each profile is a directory with a `config.txt` file.
+
+**Default Profile** (`profiles/default/config.txt`):
+```
+WEBSOCKET_URI=ws://localhost:8000
+VIDEO_JPEG_QUALITY=80
+VIDEO_WITH_TIMESTAMP=false
+VIDEO_FPS=25
+AUDIO_WITH_TIMESTAMP=false
+AUDIO_BATCH_SIZE=4096
+ENABLE_METRICS_LOGGING=true
+METRICS_LOG_INTERVAL_SEC=5
+```
+
+**Example Remote Profile** (`profiles/example_remote/config.txt`):
+```
+WEBSOCKET_URI=wss://YOUR-SPACE.hf.space
+VIDEO_JPEG_QUALITY=75
+VIDEO_WITH_TIMESTAMP=true
+VIDEO_FPS=25
+AUDIO_WITH_TIMESTAMP=true
+AUDIO_BATCH_SIZE=4096
+ENABLE_METRICS_LOGGING=true
+METRICS_LOG_INTERVAL_SEC=10
+```
+
+### Environment Variables
+
+Create a `.env` file (copy from `.env.example`):
+
+```bash
+REMOTE_CONTROL_PROFILE=default
+ROBOT_NAME=reachy_mini
+LOG_LEVEL=INFO
+VIDEO_JPEG_QUALITY=80
+LATENCY_WARNING_MS=200
+ENABLE_REST_API=true
+REST_API_HOST=0.0.0.0
+REST_API_PORT=7860
+```
+
+## REST API (Optional)
+
+If `ENABLE_REST_API=true`, the app provides HTTP endpoints:
+
+### Get Status
+
+```bash
+curl http://localhost:7860/status
+```
+
+Response:
+```json
+{
+  "robot_connected": true,
+  "control": {
+    "connected": true,
+    "commands_received": 42,
+    "last_command_type": "movement",
+    "last_activity": 1705123456.78
+  },
+  "video": {
+    "connected": true,
+    "frames_sent": 1234,
+    "frames_dropped": 5,
+    "fps": 24.8,
+    "latency_ms": 45.2,
+    "latency_avg_ms": 42.1,
+    "latency_min_ms": 38.5,
+    "latency_max_ms": 52.3,
+    "last_activity": 1705123456.78
+  },
+  "audio": {
+    "connected": true,
+    "chunks_sent": 567,
+    "chunks_received": 234,
+    "latency_avg_ms": 38.5,
+    "last_activity": 1705123456.78
+  }
+}
+```
+
+### List Profiles
+
+```bash
+curl http://localhost:7860/profiles
+```
+
+Response:
+```json
+{
+  "profiles": ["default", "example_remote"],
+  "current": "default"
+}
+```
+
+### Switch Profile
+
+```bash
+curl -X POST http://localhost:7860/profile/example_remote
+```
+
+**Note**: Requires app restart to take effect.
+
+### Health Check
+
+```bash
+curl http://localhost:7860/health
+```
+
+## Integration with Reachy Mini Dashboard
+
+This app integrates with the Reachy Mini dashboard as a `ReachyMiniApp`:
+
+1. Install the app: `pip install -e reachy_mini_remote_control_app`
+2. The app will appear in the dashboard
+3. Start from the dashboard UI
+4. The dashboard provides the `ReachyMini` instance and `stop_event`
+
+## Deployment to Production
+
+### Hugging Face Space Setup
+
+1. Deploy the remote server ([reachy_mini_remote_control/app.py](../reachy_mini_remote_control/app.py)) to Hugging Face Space
+2. Note the Space URL: `wss://YOUR-SPACE.hf.space`
+3. Create a new profile:
+
+```bash
+mkdir -p src/reachy_mini_remote_control_app/profiles/production
+```
+
+Edit `profiles/production/config.txt`:
+```
+WEBSOCKET_URI=wss://YOUR-SPACE.hf.space
+VIDEO_JPEG_QUALITY=75
+VIDEO_WITH_TIMESTAMP=true
+VIDEO_FPS=25
+AUDIO_WITH_TIMESTAMP=true
+AUDIO_BATCH_SIZE=4096
+ENABLE_METRICS_LOGGING=true
+METRICS_LOG_INTERVAL_SEC=10
+```
+
+4. Set environment variable:
+```bash
+export REMOTE_CONTROL_PROFILE=production
+```
+
+5. Run the app:
+```bash
+reachy-mini-remote-control-app --profile production
+```
+
+### Network Requirements
+
+- **Robot side**: Only **outbound** connections needed (no port forwarding required)
+- **Server side**: WebSocket server must be publicly accessible
+- **Protocol**: Use WSS (WebSocket Secure) for production
+
+### Security Considerations
+
+- Use WSS (not WS) for encrypted connections
+- Consider implementing authentication tokens
+- Monitor connection logs for suspicious activity
+- Rate limiting on the server side
+
+## Troubleshooting
+
+### Robot fails to connect to daemon
+
+```
+Failed to connect to robot: Connection timeout
+```
+
+**Solution**: Make sure the daemon is running:
+```bash
+reachy-mini-daemon --mockup-sim
+```
+
+### WebSocket connection fails
+
+```
+[Robot Control] Connection failed: Cannot connect to host
+```
+
+**Solutions**:
+1. Verify the remote server is running and accessible
+2. Check the `WEBSOCKET_URI` in your profile
+3. Test connectivity: `curl http://YOUR-SERVER:8000/health` (if server has health endpoint)
+4. Check firewall rules
+
+### No video/audio streaming
+
+**Solutions**:
+1. Ensure media is started: check daemon logs
+2. Verify WebSocket connections are established: check `/status` endpoint
+3. Check frame/audio publisher threads are running (look for log messages)
+
+### High latency
+
+Check the `/status` endpoint for latency metrics:
+```bash
+curl http://localhost:7860/status | jq '.video.latency_avg_ms'
+```
+
+**Solutions**:
+- Reduce `VIDEO_JPEG_QUALITY` in profile (e.g., from 80 to 60)
+- Lower `VIDEO_FPS` (e.g., from 25 to 15)
+- Use a server geographically closer to the robot
+- Check network bandwidth
+
+## Development
+
+### Running Tests
+
+```bash
+pytest tests/
+```
+
+### Code Structure
+
+- `config.py` - Configuration management and profile loading
+- `monitoring/` - Metrics tracking and connection monitoring
+- `websocket_clients/` - WebSocket client implementations
+  - `robot_control.py` - Receive movement commands
+  - `video_stream.py` - Send video frames
+  - `audio_stream.py` - Bidirectional audio
+- `app.py` - Optional REST API for status
+- `main.py` - Entry point and ReachyMiniApp implementation
+
+## License
+
+Same as Reachy Mini SDK.
+
+## Support
+
+For issues and questions:
+- GitHub Issues: [pollen-robotics/reachy_mini](https://github.com/pollen-robotics/reachy_mini/issues)
+- Email: contact@pollen-robotics.com

pyproject.toml

@@ -0,0 +1,39 @@
+[build-system]
+requires = ["setuptools"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "reachy_mini_remote_control_app"
+version = "0.1.0"
+authors = [{ name = "Pollen Robotics", email = "contact@pollen-robotics.com" }]
+description = "Remote control WebSocket bridge for Reachy Mini"
+readme = "README.md"
+requires-python = ">=3.10"
+dependencies = [
+    "fastapi>=0.109.0",
+    "uvicorn>=0.27.0",
+    "websockets>=12.0",
+    "opencv-python>=4.8.0",
+    "numpy>=1.24.0",
+    "python-dotenv>=1.0.0",
+    "reachy-mini @ git+https://github.com/pollen-robotics/reachy_mini.git@develop",
+]
+
+[project.scripts]
+reachy-mini-remote-control-app = "reachy_mini_remote_control_app.main:main"
+
+[project.entry-points."reachy_mini_apps"]
+reachy_mini_remote_control_app = "reachy_mini_remote_control_app.main:ReachyMiniRemoteControlApp"
+
+[tool.setuptools]
+package-dir = { "" = "src" }
+include-package-data = true
+
+[tool.setuptools.packages.find]
+where = ["src"]
+
+[tool.setuptools.package-data]
+reachy_mini_remote_control_app = [
+    "profiles/**/*.txt",
+    ".env.example",
+]

src/reachy_mini_remote_control_app/__init__.py

@@ -0,0 +1,3 @@
+"""Reachy Mini Remote Control App."""
+
+__version__ = "0.1.0"

src/reachy_mini_remote_control_app/app.py

@@ -0,0 +1,108 @@
+"""Optional FastAPI application for REST status endpoints."""
+
+import logging
+from pathlib import Path
+from typing import Dict, List
+
+from fastapi import FastAPI, HTTPException
+from fastapi.responses import JSONResponse
+
+from reachy_mini_remote_control_app.config import Config, load_profile, set_profile
+from reachy_mini_remote_control_app.monitoring.connection_monitor import ConnectionMonitor
+
+
+logger = logging.getLogger("reachy_mini_remote_control_app.app")
+
+
+def create_fastapi_app(monitor: ConnectionMonitor, config: Config) -> FastAPI:
+    """Create FastAPI application with status endpoints.
+
+    Args:
+        monitor: ConnectionMonitor instance for metrics.
+        config: Config instance.
+
+    Returns:
+        FastAPI application.
+
+    """
+    app = FastAPI(
+        title="Reachy Mini Remote Control App",
+        description="REST API for remote control status and configuration",
+        version="0.1.0",
+    )
+
+    @app.get("/health")
+    async def health_check() -> Dict[str, str]:
+        """Health check endpoint."""
+        return {"status": "ok"}
+
+    @app.get("/status")
+    async def get_status() -> JSONResponse:
+        """Get current system status with metrics.
+
+        Returns:
+            JSON response with system status.
+
+        """
+        try:
+            status = monitor.get_status()
+            return JSONResponse(content=status.to_dict())
+        except Exception as e:
+            logger.error(f"Error getting status: {e}")
+            raise HTTPException(status_code=500, detail=str(e))
+
+    @app.get("/profiles")
+    async def list_profiles() -> Dict[str, List[str]]:
+        """List available profiles.
+
+        Returns:
+            Dictionary with list of available profiles.
+
+        """
+        try:
+            profiles_dir = Path(__file__).parent / "profiles"
+            profiles = [p.name for p in profiles_dir.iterdir() if p.is_dir()]
+            return {"profiles": profiles, "current": config.REMOTE_CONTROL_PROFILE}
+        except Exception as e:
+            logger.error(f"Error listing profiles: {e}")
+            raise HTTPException(status_code=500, detail=str(e))
+
+    @app.post("/profile/{profile_name}")
+    async def switch_profile(profile_name: str) -> Dict[str, str]:
+        """Switch to a different profile.
+
+        Note: Requires app restart to take effect.
+
+        Args:
+            profile_name: Name of the profile to switch to.
+
+        Returns:
+            Success message.
+
+        """
+        try:
+            # Verify profile exists
+            profiles_dir = Path(__file__).parent / "profiles"
+            profile_path = profiles_dir / profile_name
+
+            if not profile_path.exists():
+                raise HTTPException(status_code=404, detail=f"Profile '{profile_name}' not found")
+
+            # Load profile to verify it's valid
+            load_profile(profile_name)
+
+            # Set as active profile
+            set_profile(profile_name)
+
+            return {
+                "status": "success",
+                "message": f"Profile switched to '{profile_name}'. Restart app to apply changes.",
+            }
+
+        except HTTPException:
+            raise
+        except Exception as e:
+            logger.error(f"Error switching profile: {e}")
+            raise HTTPException(status_code=500, detail=str(e))
+
+    return app

src/reachy_mini_remote_control_app/config.py

@@ -0,0 +1,114 @@
+"""Configuration management for the remote control app."""
+
+import os
+from pathlib import Path
+from typing import Dict, Optional
+
+from dotenv import find_dotenv, load_dotenv
+
+
+# Load environment variables
+load_dotenv(find_dotenv())
+
+
+class Config:
+    """Configuration class for remote control app."""
+
+    def __init__(self, profile: Optional[str] = None):
+        """Initialize configuration.
+
+        Args:
+            profile: Profile name to load. If None, uses REMOTE_CONTROL_PROFILE env var.
+
+        """
+        # Basic settings from environment
+        self.ROBOT_NAME = os.getenv("ROBOT_NAME", "reachy_mini")
+        self.LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
+
+        # Profile management
+        profile_name = profile or os.getenv("REMOTE_CONTROL_PROFILE", "default")
+        self.REMOTE_CONTROL_PROFILE = profile_name
+
+        # Load profile-specific configuration
+        self.profile_config = load_profile(profile_name)
+
+        # WebSocket URI (from profile or environment)
+        self.WEBSOCKET_URI = self.profile_config.get(
+            "WEBSOCKET_URI", os.getenv("WEBSOCKET_URI", "ws://localhost:8000")
+        )
+
+        # Video settings
+        self.VIDEO_JPEG_QUALITY = int(
+            self.profile_config.get("VIDEO_JPEG_QUALITY", os.getenv("VIDEO_JPEG_QUALITY", "80"))
+        )
+        self.VIDEO_WITH_TIMESTAMP = (
+            self.profile_config.get("VIDEO_WITH_TIMESTAMP", "true").lower() == "true"
+        )
+        self.VIDEO_FPS = int(self.profile_config.get("VIDEO_FPS", "25"))
+
+        # Audio settings
+        self.AUDIO_SAMPLE_RATE = 16000  # Fixed
+        self.AUDIO_BATCH_SIZE = int(self.profile_config.get("AUDIO_BATCH_SIZE", "4096"))
+        self.AUDIO_WITH_TIMESTAMP = (
+            self.profile_config.get("AUDIO_WITH_TIMESTAMP", "true").lower() == "true"
+        )
+
+        # Monitoring
+        self.ENABLE_METRICS_LOGGING = (
+            self.profile_config.get("ENABLE_METRICS_LOGGING", "true").lower() == "true"
+        )
+        self.METRICS_LOG_INTERVAL_SEC = int(
+            self.profile_config.get("METRICS_LOG_INTERVAL_SEC", "5")
+        )
+        self.LATENCY_WARNING_MS = float(os.getenv("LATENCY_WARNING_MS", "200"))
+
+        # REST API settings
+        self.ENABLE_REST_API = os.getenv("ENABLE_REST_API", "true").lower() == "true"
+        self.REST_API_HOST = os.getenv("REST_API_HOST", "0.0.0.0")
+        self.REST_API_PORT = int(os.getenv("REST_API_PORT", "7860"))
+
+
+def load_profile(profile_name: str) -> Dict[str, str]:
+    """Load profile configuration from profiles/{profile_name}/config.txt.
+
+    Args:
+        profile_name: Name of the profile to load.
+
+    Returns:
+        Dictionary of configuration key-value pairs.
+
+    """
+    profile_path = Path(__file__).parent / "profiles" / profile_name / "config.txt"
+
+    if not profile_path.exists():
+        print(f"Warning: Profile '{profile_name}' not found at {profile_path}, using defaults")
+        return {}
+
+    config = {}
+    with open(profile_path) as f:
+        for line in f:
+            line = line.strip()
+            # Skip empty lines and comments
+            if not line or line.startswith("#"):
+                continue
+            # Parse KEY=VALUE
+            if "=" in line:
+                key, value = line.split("=", 1)
+                config[key.strip()] = value.strip()
+
+    return config
+
+
+def set_profile(profile_name: str) -> None:
+    """Switch active profile at runtime.
+
+    Args:
+        profile_name: Name of the profile to switch to.
+
+    """
+    # Update environment variable for consistency
+    os.environ["REMOTE_CONTROL_PROFILE"] = profile_name
+
+
+# Global config instance
+config = Config()

src/reachy_mini_remote_control_app/main.py

@@ -0,0 +1,297 @@
+"""Main entry point for the Reachy Mini Remote Control App."""
+
+import argparse
+import logging
+import sys
+import threading
+import time
+from typing import Optional
+
+from reachy_mini import ReachyMini, ReachyMiniApp
+
+from reachy_mini_remote_control_app.app import create_fastapi_app
+from reachy_mini_remote_control_app.config import Config
+from reachy_mini_remote_control_app.monitoring.connection_monitor import ConnectionMonitor
+from reachy_mini_remote_control_app.websocket_clients.audio_stream import AsyncWebSocketAudioStreamer
+from reachy_mini_remote_control_app.websocket_clients.robot_control import AsyncWebSocketController
+from reachy_mini_remote_control_app.websocket_clients.video_stream import AsyncWebSocketFrameSender
+
+
+logger = logging.getLogger("reachy_mini_remote_control_app")
+
+
+def setup_logging(log_level: str) -> None:
+    """Setup logging configuration.
+
+    Args:
+        log_level: Logging level (DEBUG, INFO, WARNING, ERROR).
+
+    """
+    logging.basicConfig(
+        level=getattr(logging, log_level.upper()),
+        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    )
+
+
+def parse_args() -> argparse.Namespace:
+    """Parse command line arguments.
+
+    Returns:
+        Parsed arguments.
+
+    """
+    parser = argparse.ArgumentParser(description="Reachy Mini Remote Control App")
+    parser.add_argument(
+        "--profile", default=None, help="Profile to use (overrides REMOTE_CONTROL_PROFILE env var)"
+    )
+    parser.add_argument(
+        "--websocket-uri", default=None, help="WebSocket URI to connect to (overrides profile)"
+    )
+    parser.add_argument(
+        "--robot-name", default=None, help="Robot name (overrides ROBOT_NAME env var)"
+    )
+    parser.add_argument("--log-level", default=None, help="Logging level (DEBUG, INFO, WARNING, ERROR)")
+    parser.add_argument("--no-rest-api", action="store_true", help="Disable REST API for status")
+
+    return parser.parse_args()
+
+
+def _publish_frames(
+    robot: ReachyMini,
+    ws_video: AsyncWebSocketFrameSender,
+    stop_flag: threading.Event,
+    fps: int = 25,
+) -> None:
+    """Publish video frames to WebSocket.
+
+    Args:
+        robot: ReachyMini instance.
+        ws_video: AsyncWebSocketFrameSender instance.
+        stop_flag: Threading event to signal stop.
+        fps: Target frames per second.
+
+    """
+    frame_interval = 1.0 / fps
+    logger.info(f"[Frame Publisher] Starting video streaming at {fps} FPS")
+
+    while not stop_flag.is_set():
+        try:
+            # Get frame from robot
+            frame = robot.media.get_frame()
+
+            if frame is not None:
+                # Send frame
+                ws_video.send_frame(frame)
+
+            # Rate limiting
+            time.sleep(frame_interval)
+
+        except Exception as e:
+            logger.error(f"[Frame Publisher] Error: {e}")
+            time.sleep(1)
+
+
+def _publish_audio(
+    robot: ReachyMini,
+    ws_audio: AsyncWebSocketAudioStreamer,
+    stop_flag: threading.Event,
+) -> None:
+    """Publish and receive audio to/from WebSocket.
+
+    Args:
+        robot: ReachyMini instance.
+        ws_audio: AsyncWebSocketAudioStreamer instance.
+        stop_flag: Threading event to signal stop.
+
+    """
+    logger.info("[Audio Publisher] Starting bidirectional audio streaming")
+
+    while not stop_flag.is_set():
+        try:
+            # Send robot audio to remote server
+            audio_sample = robot.media.get_audio_sample()
+            if audio_sample is not None:
+                ws_audio.send_audio_chunk(audio_sample)
+
+            # Receive audio from remote server and play on robot
+            received_audio = ws_audio.get_audio_chunk(timeout=0.01)
+            if received_audio is not None:
+                robot.media.push_audio_sample(received_audio)
+
+        except Exception as e:
+            logger.error(f"[Audio Publisher] Error: {e}")
+            time.sleep(0.1)
+
+
+def run(
+    args: argparse.Namespace,
+    robot: Optional[ReachyMini] = None,
+    app_stop_event: Optional[threading.Event] = None,
+) -> None:
+    """Run the Reachy Mini remote control app.
+
+    Args:
+        args: Command line arguments.
+        robot: Optional pre-initialized ReachyMini instance.
+        app_stop_event: Optional stop event for graceful shutdown.
+
+    """
+    # Load configuration
+    config = Config(profile=args.profile)
+
+    # Override config with command line arguments
+    if args.websocket_uri:
+        config.WEBSOCKET_URI = args.websocket_uri
+    if args.robot_name:
+        config.ROBOT_NAME = args.robot_name
+    if args.log_level:
+        config.LOG_LEVEL = args.log_level
+    if args.no_rest_api:
+        config.ENABLE_REST_API = False
+
+    # Setup logging
+    setup_logging(config.LOG_LEVEL)
+
+    logger.info("Starting Reachy Mini Remote Control App")
+    logger.info(f"Profile: {config.REMOTE_CONTROL_PROFILE}")
+    logger.info(f"WebSocket URI: {config.WEBSOCKET_URI}")
+
+    # Initialize robot if not provided
+    if robot is None:
+        try:
+            logger.info(f"Connecting to robot: {config.ROBOT_NAME}")
+            robot = ReachyMini(robot_name=config.ROBOT_NAME)
+            logger.info("Robot connected successfully")
+        except Exception as e:
+            logger.error(f"Failed to connect to robot: {e}")
+            logger.error("Make sure the daemon is running (reachy-mini-daemon --mockup-sim)")
+            sys.exit(1)
+
+    # Start media recording and playback
+    try:
+        robot.media.start_recording()
+        robot.media.start_playing()
+    except Exception as e:
+        logger.warning(f"Failed to start media: {e}")
+
+    # Initialize connection monitor
+    monitor = ConnectionMonitor()
+    monitor.set_robot_connected(True)
+
+    # Initialize WebSocket clients
+    logger.info("Initializing WebSocket clients...")
+
+    ws_controller = AsyncWebSocketController(
+        ws_uri=config.WEBSOCKET_URI + "/robot", robot=robot, monitor=monitor
+    )
+
+    ws_video = AsyncWebSocketFrameSender(
+        ws_uri=config.WEBSOCKET_URI + "/video_stream",
+        jpeg_quality=config.VIDEO_JPEG_QUALITY,
+        with_timestamp=config.VIDEO_WITH_TIMESTAMP,
+        monitor=monitor,
+    )
+
+    ws_audio = AsyncWebSocketAudioStreamer(
+        ws_uri=config.WEBSOCKET_URI + "/audio_stream", monitor=monitor
+    )
+
+    logger.info("WebSocket clients initialized")
+
+    # Create stop event if not provided
+    if app_stop_event is None:
+        app_stop_event = threading.Event()
+
+    # Start frame and audio publishing threads
+    frame_thread = threading.Thread(
+        target=_publish_frames,
+        args=(robot, ws_video, app_stop_event, config.VIDEO_FPS),
+        daemon=True,
+    )
+    audio_thread = threading.Thread(
+        target=_publish_audio, args=(robot, ws_audio, app_stop_event), daemon=True
+    )
+
+    frame_thread.start()
+    audio_thread.start()
+
+    logger.info("Frame and audio publishers started")
+
+    try:
+        # Start REST API if enabled
+        if config.ENABLE_REST_API:
+            logger.info(f"Starting REST API on {config.REST_API_HOST}:{config.REST_API_PORT}")
+            app = create_fastapi_app(monitor, config)
+
+            import uvicorn
+
+            uvicorn.run(app, host=config.REST_API_HOST, port=config.REST_API_PORT)
+        else:
+            logger.info("REST API disabled, waiting for stop signal...")
+            app_stop_event.wait()
+
+    except KeyboardInterrupt:
+        logger.info("Keyboard interrupt received, shutting down...")
+
+    finally:
+        # Cleanup
+        logger.info("Cleaning up...")
+        app_stop_event.set()
+
+        # Stop WebSocket clients
+        ws_controller.stop()
+        ws_video.close()
+        ws_audio.close()
+
+        # Give threads time to finish
+        time.sleep(1)
+
+        # Close robot media
+        try:
+            robot.media.close()
+        except Exception as e:
+            logger.debug(f"Error closing media: {e}")
+
+        # Disconnect robot
+        try:
+            robot.client.disconnect()
+        except Exception as e:
+            logger.debug(f"Error disconnecting robot: {e}")
+
+        logger.info("Shutdown complete")
+
+
+def main() -> None:
+    """CLI entry point."""
+    args = parse_args()
+    run(args)
+
+
+class ReachyMiniRemoteControlApp(ReachyMiniApp):  # type: ignore[misc]
+    """Reachy Mini Apps entry point for the remote control app."""
+
+    custom_app_url = "http://0.0.0.0:7860/"
+    dont_start_webserver = False
+
+    def run(self, reachy_mini: ReachyMini, stop_event: threading.Event) -> None:
+        """Run the remote control app from the dashboard.
+
+        Args:
+            reachy_mini: ReachyMini instance provided by dashboard.
+            stop_event: Stop event for graceful shutdown.
+
+        """
+        # Create minimal args for run()
+        args = argparse.Namespace(
+            profile=None,
+            websocket_uri=None,
+            robot_name=None,
+            log_level=None,
+            no_rest_api=False,
+        )
+
+        run(args, robot=reachy_mini, app_stop_event=stop_event)
+
+
+if __name__ == "__main__":
+    main()

src/reachy_mini_remote_control_app/monitoring/__init__.py

@@ -0,0 +1 @@
+"""Monitoring and metrics tracking."""

src/reachy_mini_remote_control_app/monitoring/connection_monitor.py

@@ -0,0 +1,200 @@
+"""Connection monitoring with thread-safe metrics tracking."""
+
+import logging
+import threading
+import time
+from collections import deque
+from typing import Deque, Optional
+
+from reachy_mini_remote_control_app.monitoring.metrics import (
+    AudioMetrics,
+    ControlMetrics,
+    SystemStatus,
+    VideoMetrics,
+)
+
+
+logger = logging.getLogger("reachy_mini_remote_control_app.monitor")
+
+
+class ConnectionMonitor:
+    """Thread-safe connection metrics tracking."""
+
+    def __init__(self, latency_window_size: int = 100):
+        """Initialize the connection monitor.
+
+        Args:
+            latency_window_size: Number of latency samples to keep for statistics.
+
+        """
+        self._lock = threading.Lock()
+
+        # Metrics
+        self.control_metrics = ControlMetrics()
+        self.video_metrics = VideoMetrics()
+        self.audio_metrics = AudioMetrics()
+        self.robot_connected = False
+
+        # Rolling windows for latency tracking
+        self._video_latencies: Deque[float] = deque(maxlen=latency_window_size)
+        self._audio_latencies: Deque[float] = deque(maxlen=latency_window_size)
+
+        # FPS tracking for video
+        self._frame_timestamps: Deque[float] = deque(maxlen=100)
+
+    def update_control_metrics(
+        self,
+        connected: Optional[bool] = None,
+        command_type: Optional[str] = None,
+    ) -> None:
+        """Update control metrics.
+
+        Args:
+            connected: Connection status.
+            command_type: Type of command received.
+
+        """
+        with self._lock:
+            if connected is not None:
+                self.control_metrics.connected = connected
+
+            if command_type:
+                self.control_metrics.commands_received += 1
+                self.control_metrics.last_command_type = command_type
+
+            self.control_metrics.last_activity = time.time()
+
+    def update_video_metrics(
+        self,
+        connected: Optional[bool] = None,
+        frame_sent: bool = False,
+        frame_dropped: bool = False,
+        latency_ms: Optional[float] = None,
+    ) -> None:
+        """Update video metrics.
+
+        Args:
+            connected: Connection status.
+            frame_sent: Whether a frame was sent.
+            frame_dropped: Whether a frame was dropped.
+            latency_ms: Latency in milliseconds.
+
+        """
+        with self._lock:
+            if connected is not None:
+                self.video_metrics.connected = connected
+
+            if frame_sent:
+                self.video_metrics.frames_sent += 1
+                now = time.time()
+                self._frame_timestamps.append(now)
+
+                # Calculate FPS from last 100 frames
+                if len(self._frame_timestamps) > 1:
+                    time_span = self._frame_timestamps[-1] - self._frame_timestamps[0]
+                    if time_span > 0:
+                        self.video_metrics.fps = (len(self._frame_timestamps) - 1) / time_span
+
+            if frame_dropped:
+                self.video_metrics.frames_dropped += 1
+
+            if latency_ms is not None:
+                self._video_latencies.append(latency_ms)
+                self.video_metrics.latency_ms = latency_ms
+
+                # Update statistics
+                if self._video_latencies:
+                    self.video_metrics.latency_avg_ms = sum(self._video_latencies) / len(
+                        self._video_latencies
+                    )
+                    self.video_metrics.latency_min_ms = min(self._video_latencies)
+                    self.video_metrics.latency_max_ms = max(self._video_latencies)
+
+            self.video_metrics.last_activity = time.time()
+
+    def update_audio_metrics(
+        self,
+        connected: Optional[bool] = None,
+        chunk_sent: bool = False,
+        chunk_received: bool = False,
+        latency_ms: Optional[float] = None,
+    ) -> None:
+        """Update audio metrics.
+
+        Args:
+            connected: Connection status.
+            chunk_sent: Whether a chunk was sent.
+            chunk_received: Whether a chunk was received.
+            latency_ms: Latency in milliseconds.
+
+        """
+        with self._lock:
+            if connected is not None:
+                self.audio_metrics.connected = connected
+
+            if chunk_sent:
+                self.audio_metrics.chunks_sent += 1
+
+            if chunk_received:
+                self.audio_metrics.chunks_received += 1
+
+            if latency_ms is not None:
+                self._audio_latencies.append(latency_ms)
+
+                # Update statistics
+                if self._audio_latencies:
+                    self.audio_metrics.latency_avg_ms = sum(self._audio_latencies) / len(
+                        self._audio_latencies
+                    )
+                    self.audio_metrics.latency_min_ms = min(self._audio_latencies)
+                    self.audio_metrics.latency_max_ms = max(self._audio_latencies)
+
+            self.audio_metrics.last_activity = time.time()
+
+    def set_robot_connected(self, connected: bool) -> None:
+        """Set robot connection status.
+
+        Args:
+            connected: Whether the robot is connected.
+
+        """
+        with self._lock:
+            self.robot_connected = connected
+
+    def get_status(self) -> SystemStatus:
+        """Get current system status.
+
+        Returns:
+            SystemStatus with current metrics.
+
+        """
+        with self._lock:
+            return SystemStatus(
+                robot_connected=self.robot_connected,
+                control=ControlMetrics(
+                    connected=self.control_metrics.connected,
+                    commands_received=self.control_metrics.commands_received,
+                    last_command_type=self.control_metrics.last_command_type,
+                    last_activity=self.control_metrics.last_activity,
+                ),
+                video=VideoMetrics(
+                    connected=self.video_metrics.connected,
+                    frames_sent=self.video_metrics.frames_sent,
+                    frames_dropped=self.video_metrics.frames_dropped,
+                    fps=self.video_metrics.fps,
+                    latency_ms=self.video_metrics.latency_ms,
+                    latency_avg_ms=self.video_metrics.latency_avg_ms,
+                    latency_min_ms=self.video_metrics.latency_min_ms,
+                    latency_max_ms=self.video_metrics.latency_max_ms,
+                    last_activity=self.video_metrics.last_activity,
+                ),
+                audio=AudioMetrics(
+                    connected=self.audio_metrics.connected,
+                    chunks_sent=self.audio_metrics.chunks_sent,
+                    chunks_received=self.audio_metrics.chunks_received,
+                    latency_avg_ms=self.audio_metrics.latency_avg_ms,
+                    latency_min_ms=self.audio_metrics.latency_min_ms,
+                    latency_max_ms=self.audio_metrics.latency_max_ms,
+                    last_activity=self.audio_metrics.last_activity,
+                ),
+            )

src/reachy_mini_remote_control_app/monitoring/metrics.py

@@ -0,0 +1,100 @@
+"""Metrics dataclasses for connection monitoring."""
+
+from dataclasses import dataclass, field
+from typing import Dict, Any, Optional
+
+
+@dataclass
+class VideoMetrics:
+    """Metrics for video streaming."""
+
+    connected: bool = False
+    frames_sent: int = 0
+    frames_dropped: int = 0
+    fps: float = 0.0
+    latency_ms: Optional[float] = None
+    latency_avg_ms: float = 0.0
+    latency_min_ms: float = float("inf")
+    latency_max_ms: float = 0.0
+    last_activity: float = 0.0
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary for JSON serialization."""
+        return {
+            "connected": self.connected,
+            "frames_sent": self.frames_sent,
+            "frames_dropped": self.frames_dropped,
+            "fps": round(self.fps, 2),
+            "latency_ms": round(self.latency_ms, 2) if self.latency_ms is not None else None,
+            "latency_avg_ms": round(self.latency_avg_ms, 2),
+            "latency_min_ms": round(self.latency_min_ms, 2)
+            if self.latency_min_ms != float("inf")
+            else None,
+            "latency_max_ms": round(self.latency_max_ms, 2),
+            "last_activity": self.last_activity,
+        }
+
+
+@dataclass
+class AudioMetrics:
+    """Metrics for audio streaming."""
+
+    connected: bool = False
+    chunks_sent: int = 0
+    chunks_received: int = 0
+    latency_avg_ms: float = 0.0
+    latency_min_ms: float = float("inf")
+    latency_max_ms: float = 0.0
+    last_activity: float = 0.0
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary for JSON serialization."""
+        return {
+            "connected": self.connected,
+            "chunks_sent": self.chunks_sent,
+            "chunks_received": self.chunks_received,
+            "latency_avg_ms": round(self.latency_avg_ms, 2),
+            "latency_min_ms": round(self.latency_min_ms, 2)
+            if self.latency_min_ms != float("inf")
+            else None,
+            "latency_max_ms": round(self.latency_max_ms, 2),
+            "last_activity": self.last_activity,
+        }
+
+
+@dataclass
+class ControlMetrics:
+    """Metrics for robot control commands."""
+
+    connected: bool = False
+    commands_received: int = 0
+    last_command_type: Optional[str] = None
+    last_activity: float = 0.0
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary for JSON serialization."""
+        return {
+            "connected": self.connected,
+            "commands_received": self.commands_received,
+            "last_command_type": self.last_command_type,
+            "last_activity": self.last_activity,
+        }
+
+
+@dataclass
+class SystemStatus:
+    """Overall system status."""
+
+    robot_connected: bool = False
+    control: ControlMetrics = field(default_factory=ControlMetrics)
+    video: VideoMetrics = field(default_factory=VideoMetrics)
+    audio: AudioMetrics = field(default_factory=AudioMetrics)
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary for JSON serialization."""
+        return {
+            "robot_connected": self.robot_connected,
+            "control": self.control.to_dict(),
+            "video": self.video.to_dict(),
+            "audio": self.audio.to_dict(),
+        }

src/reachy_mini_remote_control_app/profiles/default/config.txt

@@ -0,0 +1,17 @@
+# Remote server URI to connect to
+# Example: ws://YOUR-SPACE.hf.space (Hugging Face Space)
+# For local testing: ws://localhost:8000
+WEBSOCKET_URI=ws://localhost:8000
+
+# Video settings
+VIDEO_JPEG_QUALITY=80
+VIDEO_WITH_TIMESTAMP=false
+VIDEO_FPS=25
+
+# Audio settings
+AUDIO_WITH_TIMESTAMP=false
+AUDIO_BATCH_SIZE=4096
+
+# Monitoring
+ENABLE_METRICS_LOGGING=true
+METRICS_LOG_INTERVAL_SEC=5

src/reachy_mini_remote_control_app/profiles/example_remote/config.txt

@@ -0,0 +1,18 @@
+# Example profile for remote Hugging Face Space deployment
+# Replace YOUR-SPACE with your actual Hugging Face Space name
+
+# Remote server URI (Hugging Face Space)
+WEBSOCKET_URI=wss://YOUR-SPACE.hf.space
+
+# Video settings
+VIDEO_JPEG_QUALITY=75
+VIDEO_WITH_TIMESTAMP=true
+VIDEO_FPS=25
+
+# Audio settings
+AUDIO_WITH_TIMESTAMP=true
+AUDIO_BATCH_SIZE=4096
+
+# Monitoring
+ENABLE_METRICS_LOGGING=true
+METRICS_LOG_INTERVAL_SEC=10

src/reachy_mini_remote_control_app/websocket_clients/__init__.py

@@ -0,0 +1 @@
+"""WebSocket client implementations."""

src/reachy_mini_remote_control_app/websocket_clients/audio_stream.py

@@ -0,0 +1,259 @@
+"""WebSocket client for bidirectional audio streaming."""
+
+import asyncio
+import logging
+import threading
+import time
+from queue import Empty, Queue
+from typing import Optional, Union
+
+import numpy as np
+import numpy.typing as npt
+from websockets.asyncio.client import ClientConnection, connect
+
+from reachy_mini_remote_control_app.monitoring.connection_monitor import ConnectionMonitor
+
+
+logger = logging.getLogger("reachy_mini_remote_control_app.audio_stream")
+
+
+class AsyncWebSocketAudioStreamer:
+    """WebSocket client for bidirectional audio streaming with remote server."""
+
+    # Target ~2048 samples per packet (approx 128ms)
+    # 2048 samples * 2 bytes (int16) = 4096 bytes
+    BATCH_SIZE_BYTES = 4096
+    # Don't hold audio longer than 200ms even if buffer isn't full
+    BATCH_TIMEOUT = 0.2
+
+    def __init__(
+        self,
+        ws_uri: str,
+        keep_alive_interval: float = 2.0,
+        monitor: Optional[ConnectionMonitor] = None,
+    ) -> None:
+        """Initialize the WebSocket audio streamer.
+
+        Args:
+            ws_uri: WebSocket URI to connect to (e.g., ws://server:8000/audio_stream).
+            keep_alive_interval: Interval in seconds to send keep-alive pings when no audio.
+            monitor: Optional ConnectionMonitor for metrics tracking.
+
+        """
+        self.ws_uri = ws_uri
+        self.keep_alive_interval = keep_alive_interval
+        self.monitor = monitor
+        self.send_queue: "Queue[bytes]" = Queue()
+        self.recv_queue: "Queue[bytes]" = Queue()
+        self.loop = asyncio.new_event_loop()
+        self.thread = threading.Thread(target=self._run_loop, daemon=True)
+        self.connected = threading.Event()
+        self.stop_flag = False
+        self.thread.start()
+
+    def _run_loop(self) -> None:
+        """Run the WebSocket streamer loop in a background thread."""
+        asyncio.set_event_loop(self.loop)
+        self.loop.run_until_complete(self._run())
+
+    async def _run(self) -> None:
+        """Run the main reconnect loop."""
+        while not self.stop_flag:
+            try:
+                async with connect(self.ws_uri) as ws:
+                    logger.info("[Audio Stream] Connected to remote server")
+                    self.connected.set()
+
+                    # Update connection status
+                    if self.monitor:
+                        self.monitor.update_audio_metrics(connected=True)
+
+                    # Run send and receive tasks concurrently
+                    send_task = asyncio.create_task(self._send_loop(ws))
+                    recv_task = asyncio.create_task(self._recv_loop(ws))
+
+                    done, pending = await asyncio.wait(
+                        {send_task, recv_task},
+                        return_when=asyncio.FIRST_EXCEPTION,
+                    )
+
+                    # Cancel the other task if one fails or finishes
+                    for task in pending:
+                        task.cancel()
+                        try:
+                            await task
+                        except Exception:
+                            pass
+
+            except Exception as e:
+                logger.info(f"[Audio Stream] Connection failed: {e}")
+
+                # Update connection status
+                if self.monitor:
+                    self.monitor.update_audio_metrics(connected=False)
+
+                await asyncio.sleep(1.0)
+
+            self.connected.clear()
+
+    async def _send_loop(self, ws: ClientConnection) -> None:
+        """Send outgoing audio chunks and keep-alive pings.
+
+        Args:
+            ws: WebSocket connection.
+
+        """
+        last_activity = time.time()
+        batch_buffer = bytearray()
+        batch_start_time = time.time()
+
+        while not self.stop_flag:
+            try:
+                # Try to pull data from the queue
+                chunk = self.send_queue.get(timeout=0.01)
+
+                # If this is the first chunk in the buffer, reset timer
+                if len(batch_buffer) == 0:
+                    batch_start_time = time.time()
+
+                batch_buffer.extend(chunk)
+
+            except Empty:
+                pass
+            except Exception as e:
+                logger.info(f"[Audio Stream] Queue error: {e}")
+                break
+
+            # Check if we should send the batch
+            now = time.time()
+
+            # Condition A: Buffer is full enough (size based)
+            is_full = len(batch_buffer) >= self.BATCH_SIZE_BYTES
+
+            # Condition B: Buffer has data but it's getting too old (time based)
+            is_timed_out = (len(batch_buffer) > 0) and ((now - batch_start_time) > self.BATCH_TIMEOUT)
+
+            if is_full or is_timed_out:
+                try:
+                    # Send the aggregated buffer
+                    await ws.send(batch_buffer)  # type: ignore
+
+                    # Update metrics
+                    if self.monitor:
+                        self.monitor.update_audio_metrics(chunk_sent=True)
+
+                    # Reset
+                    batch_buffer = bytearray()
+                    last_activity = now
+
+                except Exception as e:
+                    logger.info(f"[Audio Stream] Send error: {e}")
+                    break
+
+            # Keep-alive ping (only if completely idle)
+            if len(batch_buffer) == 0 and (now - last_activity) > self.keep_alive_interval:
+                try:
+                    await ws.send("ping")
+                    last_activity = now
+                    logger.debug("[Audio Stream] Sent keep-alive ping")
+                except Exception as e:
+                    logger.info(f"[Audio Stream] Ping failed: {e}")
+                    break
+
+            # Tiny sleep to yield control if we are just spinning
+            if len(batch_buffer) == 0:
+                await asyncio.sleep(0.001)
+
+    async def _recv_loop(self, ws: ClientConnection) -> None:
+        """Receive incoming audio chunks.
+
+        Args:
+            ws: WebSocket connection.
+
+        """
+        while not self.stop_flag:
+            try:
+                msg = await ws.recv()
+            except Exception as e:
+                logger.info(f"[Audio Stream] Receive error: {e}")
+                break
+
+            if isinstance(msg, bytes):
+                try:
+                    self.recv_queue.put_nowait(msg)
+
+                    # Update metrics
+                    if self.monitor:
+                        self.monitor.update_audio_metrics(chunk_received=True)
+
+                except Exception as e:
+                    logger.debug(f"[Audio Stream] Failed to enqueue received audio: {e}")
+            else:
+                logger.debug(f"[Audio Stream] Received non-binary message: {msg}")
+
+    def send_audio_chunk(
+        self,
+        audio: Union[bytes, npt.NDArray[np.int16], npt.NDArray[np.float32]],
+    ) -> None:
+        """Queue an audio chunk to be sent.
+
+        Args:
+            audio: Either raw bytes or a numpy array of int16 or float32.
+                   Float32 arrays are assumed to be in [-1, 1] and will be converted to int16 PCM.
+
+        """
+        if self.stop_flag:
+            return
+
+        if isinstance(audio, bytes):
+            data = audio
+        else:
+            # Convert only if needed
+            arr = np.asarray(audio)
+            # Handle stereo if accidentally passed (take channel 0)
+            if arr.ndim > 1:
+                arr = arr[:, 0]
+
+            if arr.dtype == np.float32 or arr.dtype == np.float64:
+                # Scale if any value is above 1 or below -1
+                max_abs = np.max(np.abs(arr))
+                if max_abs > 1.0:
+                    arr = arr / max_abs
+                # Convert float audio [-1,1] to int16 PCM
+                arr = np.clip(arr, -1.0, 1.0)
+                arr = (arr * 32767.0).astype(np.int16)  # type: ignore
+            elif arr.dtype != np.int16:
+                arr = arr.astype(np.int16)
+
+            data = arr.tobytes()
+
+        self.send_queue.put(data)
+
+    def get_audio_chunk(self, timeout: Optional[float] = 0.01) -> Optional[npt.NDArray[np.float32]]:
+        """Retrieve a received audio chunk, if any.
+
+        Args:
+            timeout: Timeout in seconds for waiting for a chunk.
+
+        Returns:
+            Audio chunk as float32 array in [-1, 1] range, or None if no chunk available.
+
+        """
+        try:
+            if timeout == 0:
+                audio_bytes = self.recv_queue.get_nowait()
+            else:
+                audio_bytes = self.recv_queue.get(timeout=timeout)
+
+            # bytes -> int16 -> float32 in [-1, 1]
+            int16_arr = np.frombuffer(audio_bytes, dtype=np.int16)
+            float_arr = int16_arr.astype(np.float32) / 32767.0
+            return float_arr
+        except Empty:
+            return None
+
+    def close(self) -> None:
+        """Close the WebSocket audio streamer."""
+        self.stop_flag = True
+        if self.loop.is_running():
+            self.loop.call_soon_threadsafe(self.loop.stop)

src/reachy_mini_remote_control_app/websocket_clients/robot_control.py

@@ -0,0 +1,136 @@
+"""WebSocket client for receiving robot control commands."""
+
+import asyncio
+import json
+import logging
+import threading
+from typing import Any, Dict, Optional
+
+import numpy as np
+from reachy_mini import ReachyMini
+from websockets.asyncio.client import ClientConnection, connect
+
+from reachy_mini_remote_control_app.monitoring.connection_monitor import ConnectionMonitor
+
+
+logger = logging.getLogger("reachy_mini_remote_control_app.robot_control")
+
+
+class AsyncWebSocketController:
+    """WebSocket client for receiving robot control commands from remote server."""
+
+    def __init__(
+        self,
+        ws_uri: str,
+        robot: ReachyMini,
+        monitor: Optional[ConnectionMonitor] = None,
+    ) -> None:
+        """Initialize the WebSocket controller.
+
+        Args:
+            ws_uri: WebSocket URI to connect to (e.g., ws://server:8000/robot).
+            robot: ReachyMini instance for robot control.
+            monitor: Optional ConnectionMonitor for metrics tracking.
+
+        """
+        self.ws_uri = ws_uri
+        self.robot = robot
+        self.monitor = monitor
+        self.loop = asyncio.new_event_loop()
+        self.thread = threading.Thread(target=self._run_loop, daemon=True)
+        self.stop_flag = False
+        self.thread.start()
+
+    def _run_loop(self) -> None:
+        """Run the WebSocket controller loop in a background thread."""
+        asyncio.set_event_loop(self.loop)
+        self.loop.run_until_complete(self._run())
+
+    async def on_command(self, cmd: Dict[str, Any]) -> None:
+        """Handle a command from the WebSocket.
+
+        Args:
+            cmd: Command dictionary from remote server.
+
+        """
+        typ = cmd.get("type")
+
+        if typ == "movement":
+            logger.debug("[Robot Control] Movement command received")
+            mov = cmd.get("movement", {})
+            logger.debug("[Robot Control] Movement command: %s", mov)
+
+            # Parse head matrix
+            head = mov.get("head")
+            if head is not None:
+                head_arr = np.array(head, dtype=float).reshape(4, 4)
+            else:
+                head_arr = None
+
+            # Parse antennas
+            antennas = mov.get("antennas")
+            if antennas is not None:
+                antennas_arr = np.array(antennas, dtype=float)
+            else:
+                antennas_arr = None
+
+            try:
+                # Execute movement command
+                await self.robot.goto(
+                    head=head_arr,
+                    antennas=antennas_arr,
+                    duration=mov.get("duration", 1.0),
+                    body_yaw=mov.get("body_yaw", 0.0),
+                )
+
+                # Update metrics
+                if self.monitor:
+                    self.monitor.update_control_metrics(command_type="movement")
+
+            except Exception as e:
+                logger.error("[Robot Control] Error in goto: %s", e)
+
+        elif typ == "ping":
+            logger.debug("[Robot Control] Ping received")
+            return
+        else:
+            logger.debug("[Robot Control] Unknown command type: %s", typ)
+
+    async def _run(self) -> None:
+        """Run the WebSocket controller loop with automatic reconnection."""
+        while not self.stop_flag:
+            try:
+                ws: ClientConnection
+                async with connect(self.ws_uri, ping_interval=5, ping_timeout=10) as ws:
+                    logger.info("[Robot Control] Connected to remote server")
+
+                    # Update connection status
+                    if self.monitor:
+                        self.monitor.update_control_metrics(connected=True)
+
+                    # Receive and process commands
+                    async for msg in ws:
+                        try:
+                            data = json.loads(msg)
+                        except Exception as e:
+                            logger.debug("[Robot Control] Bad JSON: %s raw: %s", e, msg)
+                            continue
+
+                        # Process command
+                        await self.on_command(data)
+
+            except Exception as e:
+                logger.info("[Robot Control] Connection failed: %s", e)
+
+                # Update connection status
+                if self.monitor:
+                    self.monitor.update_control_metrics(connected=False)
+
+                # Backoff before reconnect
+                await asyncio.sleep(1)
+
+    def stop(self) -> None:
+        """Stop the WebSocket controller."""
+        self.stop_flag = True
+        if self.loop.is_running():
+            self.loop.call_soon_threadsafe(lambda: None)

src/reachy_mini_remote_control_app/websocket_clients/video_stream.py

@@ -0,0 +1,188 @@
+"""WebSocket client for sending video frames to remote server."""
+
+import asyncio
+import base64
+import json
+import logging
+import threading
+import time
+from queue import Empty, Full, Queue
+from typing import Optional
+
+import cv2
+import numpy as np
+import numpy.typing as npt
+from websockets.asyncio.client import ClientConnection, connect
+from websockets.exceptions import ConnectionClosed
+
+from reachy_mini_remote_control_app.monitoring.connection_monitor import ConnectionMonitor
+
+
+logger = logging.getLogger("reachy_mini_remote_control_app.video_stream")
+
+
+class AsyncWebSocketFrameSender:
+    """WebSocket client for sending video frames to remote server."""
+
+    def __init__(
+        self,
+        ws_uri: str,
+        jpeg_quality: int = 80,
+        with_timestamp: bool = False,
+        monitor: Optional[ConnectionMonitor] = None,
+    ) -> None:
+        """Initialize the WebSocket frame sender.
+
+        Args:
+            ws_uri: WebSocket URI to connect to (e.g., ws://server:8000/video_stream).
+            jpeg_quality: JPEG encoding quality (0-100).
+            with_timestamp: Whether to send timestamp with frames.
+            monitor: Optional ConnectionMonitor for metrics tracking.
+
+        """
+        self.ws_uri = ws_uri
+        self.jpeg_quality = jpeg_quality
+        self.with_timestamp = with_timestamp
+        self.monitor = monitor
+        self.queue: "Queue[bytes]" = Queue(maxsize=2)
+        self.loop = asyncio.new_event_loop()
+        self.thread = threading.Thread(target=self._run_loop, daemon=True)
+        self.connected = threading.Event()
+        self.stop_flag = False
+        self._last_frame: Optional[npt.NDArray[np.uint8]] = None
+        self.thread.start()
+
+    def _run_loop(self) -> None:
+        """Run the WebSocket loop in a background thread."""
+        asyncio.set_event_loop(self.loop)
+        self.loop.run_until_complete(self._run())
+
+    def _clear_queue(self) -> None:
+        """Empty the queue so we don't send old video on reconnect."""
+        while not self.queue.empty():
+            try:
+                self.queue.get_nowait()
+            except Empty:
+                break
+
+    async def _run(self) -> None:
+        """Run the WebSocket frame sender loop with automatic reconnection."""
+        while not self.stop_flag:
+            try:
+                ws: ClientConnection
+                async with connect(
+                    self.ws_uri,
+                    ping_interval=5,
+                    ping_timeout=10,
+                    close_timeout=1,
+                ) as ws:
+                    logger.info("[Video Stream] Connected to remote server")
+                    self.connected.set()
+                    self._clear_queue()
+
+                    # Update connection status
+                    if self.monitor:
+                        self.monitor.update_video_metrics(connected=True)
+
+                    while not self.stop_flag:
+                        try:
+                            frame_data = self.queue.get_nowait()
+
+                            # Send frame
+                            await ws.send(frame_data)
+
+                            # Update metrics
+                            if self.monitor:
+                                self.monitor.update_video_metrics(frame_sent=True)
+
+                        except Empty:
+                            # Queue is empty, yield to event loop
+                            await asyncio.sleep(0.05)
+                            continue
+
+                        except Exception as e:
+                            logger.error(f"[Video Stream] Send error: {e}")
+                            break
+
+            except ConnectionClosed as e:
+                logger.error(f"[Video Stream] Connection lost ({type(e).__name__}). Retrying...")
+                self.connected.clear()
+                self._last_frame = None
+
+                # Update connection status
+                if self.monitor:
+                    self.monitor.update_video_metrics(connected=False)
+
+                await asyncio.sleep(0.5)
+
+            except Exception as e:
+                logger.error(f"[Video Stream] Unexpected error: {e}")
+
+                # Update connection status
+                if self.monitor:
+                    self.monitor.update_video_metrics(connected=False)
+
+                await asyncio.sleep(1)
+
+            self.connected.clear()
+            self._last_frame = None
+
+    def send_frame(self, frame: npt.NDArray[np.uint8]) -> None:
+        """Send a frame to the WebSocket (non-blocking).
+
+        Args:
+            frame: Video frame as numpy array (BGR format).
+
+        """
+        if self.stop_flag:
+            return
+
+        # Frame deduplication
+        if self._last_frame is not None:
+            if np.array_equal(frame, self._last_frame):
+                return
+        self._last_frame = frame.copy()
+
+        # Encode to JPEG
+        ok, jpeg_bytes = cv2.imencode(
+            ".jpg",
+            frame,
+            [int(cv2.IMWRITE_JPEG_QUALITY), self.jpeg_quality],
+        )
+        if not ok:
+            return
+
+        # Prepare data for sending
+        if self.with_timestamp:
+            # Send as JSON with timestamp
+            data = json.dumps(
+                {
+                    "timestamp": time.time(),
+                    "frame": base64.b64encode(jpeg_bytes.tobytes()).decode(),
+                }
+            ).encode()
+        else:
+            # Send as raw bytes
+            data = jpeg_bytes.tobytes()
+
+        # Add to queue (drop oldest if full)
+        try:
+            self.queue.put_nowait(data)
+        except Full:
+            # Queue is full, network is slower than camera
+            # Drop the oldest frame to make room for the newest
+            try:
+                self.queue.get_nowait()  # Pop old frame
+                self.queue.put_nowait(data)  # Push new frame
+
+                # Track dropped frame
+                if self.monitor:
+                    self.monitor.update_video_metrics(frame_dropped=True)
+
+            except Empty:
+                logger.error("[Video Stream] Queue is full and empty, this should not happen.")
+
+    def close(self) -> None:
+        """Close the WebSocket frame sender."""
+        self.stop_flag = True
+        self.loop.call_soon_threadsafe(self.loop.stop)