#!/usr/bin/env python3 """ RoboPark Preview Agent — runs on each robot/satellite Pi. Polls the scheduler for an active operator preview request. When active, opens the selected camera + microphone and publishes them to the returned LiveKit room. When the preview expires or is stopped, the agent leaves the room and releases the hardware. Also runs a small local HTTP listener for RoboVisionAI_PI's motion webhook (POST /api/motion/webhook target — see RoboVisionAI_PI's app_pi_clean.py). On a motion event it calls the scheduler's presence-triggered POST /api/devices/{id}/request-session and publishes into the returned room using the same LiveKitPublisher as the operator preview — this is the "scene detection" trigger point the scheduler's request-session docstring describes; RoboVisionAI_PI supplies the detection, this agent supplies the bridge to an actual session. Configuration (env / ~/.robopark/preview_agent.json): SCHEDULER_URL base URL of the RoboPark scheduler ROBOT_ID robot identity in the scheduler (defaults to hostname) DEVICE_TOKEN long-lived device token (after enrollment) ENROLLMENT_TOKEN one-time enrollment token (device token will be saved) VIDEO_DEVICE camera path/index/name (e.g. /dev/video0, 0, "PiCamera") AUDIO_DEVICE microphone name/index or "default" VIDEO_WIDTH / HEIGHT capture resolution (default 640x480) VIDEO_FPS capture fps (default 15) POLL_INTERVAL seconds between scheduler polls (default 3) HEARTBEAT_INTERVAL seconds between device heartbeats (default 30) VISION_WEBHOOK_PORT local port for the RoboVision motion webhook (default 5057, 0 disables) VISION_TRIGGER_COOLDOWN min seconds between motion-triggered sessions (default 20) VISION_SESSION_SECONDS HARD ceiling on how long a motion-triggered session can hold the publisher, regardless of activity (default 90) VISION_SILENCE_TIMEOUT how long a motion-triggered session may go without an activity signal before it's ended early (default 15). Activity is reported via POST /api/sessions/{id}/keepalive, which this agent polls for via GET /api/sessions/{id}/status on every poll tick. The actual caller of keepalive (the voice agent, VAD-driven) lives in a different repo — see main.py's keepalive endpoint docstring for the integration contract. Without any caller ever hitting keepalive, a vision session simply ends after VISION_SILENCE_TIMEOUT. """ from __future__ import annotations import argparse import asyncio import glob import json import logging import math import os import signal import socket import struct import sys import threading import time from dataclasses import dataclass from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Optional import httpx logger = logging.getLogger("robopark.preview_agent") _DEVICE_INVENTORY_CACHE: Optional[dict] = None _DEVICE_INVENTORY_CACHE_AT = 0.0 DEVICE_INVENTORY_CACHE_SECONDS = 5.0 ROBOVISION_MEDIA_URL = os.getenv("ROBOVISION_MEDIA_URL", "http://127.0.0.1:5000/api/media/inventory") def _stable_audio_label(value: object) -> str: """Compare USB product names without volatile ALSA card coordinates.""" import re return " ".join(re.sub(r"\s*\(hw:\d+,\d+\)\s*$", "", str(value), flags=re.I).lower().split()) def _resolve_inventory_audio(items: list, selected: object, preferred: str) -> str: if preferred: wanted = _stable_audio_label(preferred) match = next((item for item in items if _stable_audio_label(item.get("name")) == wanted), None) if match and match.get("name"): return str(match["name"]) selected_text = str(selected) selected_label = _stable_audio_label(selected_text) match = next( ( item for item in items if str(item.get("id")) == selected_text or _stable_audio_label(item.get("name")) == selected_label ), None, ) return str(match.get("name")) if match and match.get("name") else selected_text def _normalize_livekit_url(url: Optional[str]) -> Optional[str]: """Avoid Windows localhost IPv6/IPv4 ambiguity for local LiveKit.""" if not url: return url for scheme in ("ws", "wss"): prefix = f"{scheme}://localhost" if url.startswith(prefix): return f"{scheme}://127.0.0.1" + url[len(prefix):] return url def _pcm16_scale_and_peak(data: bytes, gain: float) -> tuple[bytes, int]: """Apply gain and measure peak without audioop (removed in Python 3.13).""" from array import array samples = array("h") samples.frombytes(data[:len(data) - (len(data) % 2)]) if sys.byteorder != "little": samples.byteswap() peak = 0 for index, value in enumerate(samples): scaled = max(-32768, min(32767, int(value * gain))) if gain != 1.0 else value samples[index] = scaled peak = max(peak, abs(scaled)) if sys.byteorder != "little": samples.byteswap() return samples.tobytes(), peak def _pcm16_resample_mono(data: bytes, source_rate: int, target_rate: int) -> bytes: """Linearly resample a PCM16 mono chunk using only the standard library.""" from array import array if source_rate == target_rate or len(data) < 4: return data source = array("h") source.frombytes(data[:len(data) - (len(data) % 2)]) if sys.byteorder != "little": source.byteswap() target_count = max(1, round(len(source) * target_rate / source_rate)) target = array("h", [0]) * target_count scale = source_rate / target_rate last = len(source) - 1 for index in range(target_count): position = min(last, index * scale) left = int(position) right = min(last, left + 1) fraction = position - left target[index] = max(-32768, min(32767, round( source[left] + (source[right] - source[left]) * fraction ))) if sys.byteorder != "little": target.byteswap() return target.tobytes() def _play_audio_effect(selected_output: str | None, effect: str) -> None: """Play a short local cue without involving the voice pipeline.""" sample_rate = 48000 channels = 2 if effect == "motion": notes = ((880, 0.09), (1320, 0.13)) else: notes = ((660, 0.10), (440, 0.16)) selected = str(selected_output or "default") frames = bytearray() for frequency, duration in notes: count = int(sample_rate * duration) for n in range(count): envelope = min(1.0, n / 240.0, (count - n) / 1200.0) value = int(5000 * envelope * math.sin(2 * math.pi * frequency * n / sample_rate)) frames.extend(struct.pack(" 0 and needle in str(info.get("name", "")).lower(): device_index = i break stream = pa.open( format=pyaudio.paInt16, channels=channels, rate=sample_rate, output=True, output_device_index=device_index, ) stream.write(bytes(frames)) stream.stop_stream() stream.close() except Exception as e: logger.debug(f"audio effect unavailable: {e}") finally: pa.terminate() def _get_device_inventory() -> dict: """Return discoverable camera and audio devices for dashboard selection.""" global _DEVICE_INVENTORY_CACHE, _DEVICE_INVENTORY_CACHE_AT now = time.monotonic() if (_DEVICE_INVENTORY_CACHE is not None and now - _DEVICE_INVENTORY_CACHE_AT < DEVICE_INVENTORY_CACHE_SECONDS): return _DEVICE_INVENTORY_CACHE inventory = {"video": [], "audio_input": [], "audio_output": [], "platform": sys.platform} inventory["video"].append({"id": "auto", "name": "Auto detect"}) inventory["video"].append({"id": "none", "name": "Disable camera"}) # RoboVisionAI_PI owns the production camera. Prefer its native inventory # so this process never probes an already-open V4L2 device just to fill a # dashboard dropdown. robovision_inventory = None try: response = httpx.get(ROBOVISION_MEDIA_URL, timeout=0.8) if response.is_success: robovision_inventory = response.json() for key in ("video", "audio_input", "audio_output"): if isinstance(robovision_inventory.get(key), list): inventory[key] = robovision_inventory[key] inventory["source"] = robovision_inventory.get("source", "robovision_pi") except Exception: pass if not robovision_inventory: try: import cv2 candidates = sorted(glob.glob("/dev/video*")) if os.name != "nt" else [str(i) for i in range(10)] for candidate in candidates: value = int(candidate) if os.name == "nt" else candidate backend = cv2.CAP_DSHOW if os.name == "nt" else cv2.CAP_ANY cap = cv2.VideoCapture(value, backend) if cap.isOpened(): device_id = str(value) inventory["video"].append({"id": device_id, "name": f"Camera {candidate}", "backend": "dshow" if os.name == "nt" else "v4l2"}) cap.release() except Exception as e: logger.debug(f"camera inventory unavailable: {e}") if not robovision_inventory: try: import pyaudio pa = pyaudio.PyAudio() default_in = None default_out = None try: wasapi = pa.get_host_api_info_by_type(pyaudio.paWASAPI) default_in = wasapi.get("defaultInputDevice") default_out = wasapi.get("defaultOutputDevice") except Exception: pass inventory["audio_input"].append({"id": "default", "name": "System default input"}) inventory["audio_output"].append({"id": "default", "name": "System default output"}) for i in range(pa.get_device_count()): info = pa.get_device_info_by_index(i) name = str(info.get("name", f"Audio device {i}")) item = {"id": str(i), "name": name, "host_api": str(info.get("hostApi", ""))} if info.get("maxInputChannels", 0) > 0: item["default"] = i == default_in inventory["audio_input"].append(item.copy()) if info.get("maxOutputChannels", 0) > 0: item["default"] = i == default_out inventory["audio_output"].append(item.copy()) pa.terminate() except Exception as e: logger.debug(f"audio inventory unavailable: {e}") media_health = { "service_uid": os.geteuid() if hasattr(os, "geteuid") else None, "camera_access": None, "audio_access": None, "camera_worker": None, "camera_stalled": None, "camera_frame_age_seconds": None, } if sys.platform.startswith("linux"): camera_nodes = [ str(item.get("id")) for item in inventory.get("video", []) if str(item.get("id", "")).startswith("/dev/video") ] sound_nodes = glob.glob("/dev/snd/pcm*") media_health["camera_access"] = bool(camera_nodes) and all( os.access(path, os.R_OK | os.W_OK) for path in camera_nodes[:1] ) media_health["audio_access"] = bool(sound_nodes) and all( os.access(path, os.R_OK | os.W_OK) for path in sound_nodes ) try: camera_response = httpx.get("http://127.0.0.1:5000/api/camera/status", timeout=0.6) if camera_response.is_success: camera_status = camera_response.json() media_health["camera_worker"] = bool(camera_status.get("worker_started")) media_health["camera_stalled"] = bool(camera_status.get("read_stalled")) media_health["camera_frame_age_seconds"] = camera_status.get("last_frame_age_seconds") except Exception: pass inventory["media_health"] = media_health _DEVICE_INVENTORY_CACHE = inventory _DEVICE_INVENTORY_CACHE_AT = now logger.info( f"device inventory: {len(inventory['video']) - 2} cameras, " f"{len(inventory['audio_input']) - 1} inputs, " f"{len(inventory['audio_output']) - 1} outputs" ) return inventory CONFIG_DIR = Path.home() / ".robopark" CONFIG_FILE = CONFIG_DIR / "preview_agent.json" TOKEN_FILE = CONFIG_DIR / "device_token" DEFAULT_VIDEO_WIDTH = 640 DEFAULT_VIDEO_HEIGHT = 480 DEFAULT_FPS = 15 DEFAULT_POLL_INTERVAL = 3.0 DEFAULT_HEARTBEAT_INTERVAL = 5.0 DEFAULT_VISION_WEBHOOK_PORT = 5057 DEFAULT_VISION_TRIGGER_COOLDOWN = 20.0 DEFAULT_VISION_SESSION_SECONDS = 90.0 # How long a motion-triggered session may go without an activity signal # (POST /api/sessions/{id}/keepalive, called by the voice agent — a # different repo, not this one) before preview_agent tears it down. # vision_session_seconds remains a hard ceiling regardless of activity. # Cloud STT/LLM/TTS can legitimately leave a session quiet for several # seconds. The voice agent has its own VAD-aware 60s idle policy; this remote # safety timeout must not preempt a turn while it is thinking or speaking. DEFAULT_VISION_SILENCE_TIMEOUT = 60.0 def _load_config() -> dict: cfg: dict = {} if CONFIG_FILE.exists(): try: cfg = json.loads(CONFIG_FILE.read_text(encoding="utf8")) except Exception as e: logger.warning(f"failed to read {CONFIG_FILE}: {e}") return cfg def _save_config(cfg: dict) -> None: CONFIG_DIR.mkdir(parents=True, exist_ok=True) CONFIG_FILE.write_text(json.dumps(cfg, indent=2), encoding="utf8") def _load_token() -> Optional[str]: if TOKEN_FILE.exists(): return TOKEN_FILE.read_text(encoding="utf8").strip() or None return None def _save_token(token: str) -> None: CONFIG_DIR.mkdir(parents=True, exist_ok=True) TOKEN_FILE.write_text(token, encoding="utf8") os.chmod(TOKEN_FILE, 0o600) def _mesh_proxy_headers() -> dict: """Add hub authentication only when the runtime supplied it. Direct LAN scheduler calls work unchanged. Tailscale calls use the hub's /robopark proxy, which needs this separate mesh credential while the normal Authorization header remains the scheduler device token. """ token = os.getenv("ROBOPARK_MESH_TOKEN", "").strip() return {"X-RoboPark-Mesh-Token": token} if token else {} async def _bootstrap_mesh_device(scheduler_url: str, robot_id: str) -> tuple[str, str]: """Create or recover this robot's scheduler identity using mesh auth.""" async with httpx.AsyncClient(headers=_mesh_proxy_headers()) as client: response = await client.post( f"{scheduler_url.rstrip('/')}/api/devices/bootstrap", json={"name": robot_id, "lan_ip": _get_lan_ip(), "livekit_url": os.getenv("ROBOPARK_LIVEKIT_URL") or None}, timeout=30.0, ) response.raise_for_status() data = response.json() device_id = str(data.get("device_id") or "").strip() device_token = str(data.get("device_token") or "").strip() if not device_id or not device_token: raise RuntimeError("mesh bootstrap did not return device credentials") _save_token(device_token) cfg = _load_config() cfg["device_id"] = device_id cfg["device_token"] = device_token cfg["scheduler_url"] = scheduler_url _save_config(cfg) logger.info("mesh bootstrap resolved scheduler device %s", device_id) return device_id, device_token async def _enroll(scheduler_url: str, enrollment_token: str, robot_id: str) -> tuple[str, str]: """Enroll this Pi and return the exact scheduler identity and token.""" import socket payload = { "enrollment_token": enrollment_token, "name": robot_id, "lan_ip": _get_lan_ip(), "livekit_url": os.getenv("ROBOPARK_LIVEKIT_URL") or None, } async with httpx.AsyncClient(headers=_mesh_proxy_headers()) as client: r = await client.post( f"{scheduler_url.rstrip('/')}/api/devices/enroll", json=payload, timeout=30.0, ) r.raise_for_status() data = r.json() device_id = str(data.get("device_id") or "").strip() device_token = str(data.get("device_token") or "").strip() if not device_id or not device_token: raise RuntimeError("enrollment did not return device credentials") _save_token(device_token) cfg = _load_config() cfg["device_id"] = device_id cfg["scheduler_url"] = data.get("scheduler_url", scheduler_url) _save_config(cfg) logger.info("enrolled as device %s", device_id) return device_id, device_token def _get_lan_ip() -> Optional[str]: try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.settimeout(0.5) s.connect(("10.255.255.255", 1)) ip = s.getsockname()[0] s.close() return ip except Exception: return None async def _send_heartbeat( scheduler_url: str, device_id: str, token: str ) -> tuple[Optional[bool], bool]: try: inventory = _get_device_inventory() headers = {"Authorization": f"Bearer {token}", **_mesh_proxy_headers()} async with httpx.AsyncClient() as client: response = await client.post( f"{scheduler_url.rstrip('/')}/api/devices/{device_id}/heartbeat", json={"status": "online", "ip": _get_lan_ip(), "device_inventory": inventory, "livekit_url": os.getenv("ROBOPARK_LIVEKIT_URL") or None}, headers=headers, timeout=10.0, ) if response.status_code in (401, 404): logger.warning( "heartbeat credentials rejected for %s (%s)", device_id, response.status_code, ) return None, False response.raise_for_status() logger.info( "heartbeat accepted: inventory source=%s camera=%d mic=%d speaker=%d", inventory.get("source", "local"), len(inventory.get("video", [])), len(inventory.get("audio_input", [])), len(inventory.get("audio_output", [])), ) return bool(response.json().get("production_mode", False)), True except Exception as e: logger.warning(f"heartbeat failed: {e}") return None, True @dataclass class PreviewState: active: bool = False url: Optional[str] = None token: Optional[str] = None room: Optional[str] = None mode: str = "preview" session_id: Optional[str] = None class PreviewAgent: def __init__(self, cfg: dict): self.scheduler_url = cfg.get("scheduler_url", os.getenv("SCHEDULER_URL", "http://localhost:8080")) self.robot_id = cfg.get("robot_id", os.getenv("ROBOT_ID", _hostname())) self.device_token: Optional[str] = cfg.get("device_token") or os.getenv("DEVICE_TOKEN") or _load_token() self.device_id: Optional[str] = cfg.get("device_id") self.enrollment_token: Optional[str] = cfg.get("enrollment_token") self.video_device = cfg.get("video_device", os.getenv("VIDEO_DEVICE", "auto")) self.audio_device = cfg.get("audio_device", os.getenv("AUDIO_DEVICE", "default")) # Scheduler inventory IDs come from RoboVision/sounddevice. They are # not guaranteed to equal PyAudio's device indexes, so capture must # use the resolved hardware name reported in that same inventory. self.audio_capture_device = self.audio_device self.audio_output_device = cfg.get("audio_output_device", os.getenv("AUDIO_OUTPUT_DEVICE", "default")) # Keep RoboVision/sounddevice inventory IDs out of PyAudio. Both APIs # number the same ALSA cards differently, so resolve the selected ID # to its hardware name before opening any playback stream. self.audio_playback_device = self.audio_output_device self.robovision_url = cfg.get("robovision_url", os.getenv("ROBOVISION_URL", "http://127.0.0.1:5000")) self.use_robovision_camera = bool(cfg.get("use_robovision_camera", True)) self.width = int(cfg.get("video_width", os.getenv("VIDEO_WIDTH", DEFAULT_VIDEO_WIDTH))) self.height = int(cfg.get("video_height", os.getenv("VIDEO_HEIGHT", DEFAULT_VIDEO_HEIGHT))) self.fps = int(cfg.get("video_fps", os.getenv("VIDEO_FPS", DEFAULT_FPS))) self.poll_interval = float(cfg.get("poll_interval", os.getenv("POLL_INTERVAL", DEFAULT_POLL_INTERVAL))) self.heartbeat_interval = float(cfg.get("heartbeat_interval", os.getenv("HEARTBEAT_INTERVAL", DEFAULT_HEARTBEAT_INTERVAL))) self.vision_webhook_port = int(cfg.get("vision_webhook_port", os.getenv("VISION_WEBHOOK_PORT", DEFAULT_VISION_WEBHOOK_PORT))) self.vision_trigger_cooldown = float(cfg.get("vision_trigger_cooldown", os.getenv("VISION_TRIGGER_COOLDOWN", DEFAULT_VISION_TRIGGER_COOLDOWN))) self.vision_session_seconds = float(cfg.get("vision_session_seconds", os.getenv("VISION_SESSION_SECONDS", DEFAULT_VISION_SESSION_SECONDS))) self.vision_silence_timeout = float(cfg.get("vision_silence_timeout", os.getenv("VISION_SILENCE_TIMEOUT", DEFAULT_VISION_SILENCE_TIMEOUT))) self.local_camera_motion = str( # Production needs an always-on detector. The preview publisher # remains the single camera owner when a session starts, so this # does not require the separate OpenCV vision server. cfg.get("local_camera_motion", os.getenv("LOCAL_CAMERA_MOTION", "true")) ).lower() in ("1", "true", "yes", "on") # The character greeting is the production acknowledgement. A local # motion beep serializes on the same ALSA output and delays allocation. self.motion_cue_enabled = str( cfg.get("motion_cue_enabled", os.getenv("ROBOPARK_MOTION_CUE", "false")) ).lower() in ("1", "true", "yes", "on") self._shutdown = asyncio.Event() self._task: Optional[asyncio.Task] = None self._session: Optional[httpx.AsyncClient] = None self._current_state: PreviewState = PreviewState() self._publisher: Optional["LiveKitPublisher"] = None self._loop: Optional[asyncio.AbstractEventLoop] = None self._vision_server: Optional[ThreadingHTTPServer] = None self._last_vision_trigger: float = 0.0 # Motion-triggered ("vision") session bookkeeping. Silence-timeout # with a hard ceiling — see DEFAULT_VISION_SILENCE_TIMEOUT above and # _check_vision_session() below for the full design. self._vision_session_id: Optional[str] = None self._vision_trigger_in_flight = False self._speaker_test_in_flight = False # The USB adapter is an exclusive ALSA endpoint. Keep operator tests # and motion cues from racing each other while LiveKit is paused. self._speaker_operation_lock = asyncio.Lock() self._vision_hard_deadline: float = 0.0 self._vision_last_activity: float = 0.0 self.production_mode = False self._robovision_motion_state: Optional[bool] = None self._remote_session_ended = False self._last_device_config_poll = 0.0 self._motion_reference = None self._last_motion_sample = 0.0 self._motion_capture: Optional[VideoCapture] = None self._motion_capture_lock = asyncio.Lock() self._mesh_bootstrap_ready = False self._next_mesh_bootstrap = 0.0 self._reported_pipeline: set[tuple[str, str]] = set() async def _report_pipeline(self, stage: str, status: str = "ok", message: str = "", details: Optional[dict] = None, once: bool = True) -> None: """Report a real robot-side transition for the Park test timeline.""" if not self._session or not self.device_id or not self.device_token: return session_id = self._vision_session_id or self._current_state.session_id key = (session_id or "idle", stage) if once and key in self._reported_pipeline: return try: response = await self._session.post( f"{self.scheduler_url.rstrip('/')}/api/devices/{self.device_id}/pipeline-events", headers={"Authorization": f"Bearer {self.device_token}"}, json={"stage": stage, "status": status, "message": message, "session_id": session_id, "source": "preview_agent", "details": details or {}}, timeout=5.0, ) response.raise_for_status() if once: self._reported_pipeline.add(key) except Exception as exc: logger.debug("pipeline event %s was not accepted: %s", stage, exc) async def _ensure_mesh_identity(self) -> bool: if not _mesh_proxy_headers(): return bool(self.device_id and self.device_token) now = time.monotonic() if self._mesh_bootstrap_ready: return True if now < self._next_mesh_bootstrap: return False self._next_mesh_bootstrap = now + 5.0 try: self.device_id, self.device_token = await _bootstrap_mesh_device( self.scheduler_url, self.robot_id ) self._mesh_bootstrap_ready = True return True except Exception as exc: logger.warning("mesh device bootstrap failed: %s", exc) return False async def run(self) -> None: enrollment_token = os.getenv("ENROLLMENT_TOKEN") or self.enrollment_token # A UI-minted token identifies a specific pre-created device row. # Consume it before generic mesh recovery so heartbeats cannot bind to # a stale same-name identity. if not self.device_token and enrollment_token: try: self.device_id, self.device_token = await _enroll( self.scheduler_url, enrollment_token, self.robot_id ) self._mesh_bootstrap_ready = True except Exception as exc: if not _mesh_proxy_headers(): raise # Enrollment tokens are intentionally one-time. A service # reinstall can retain the original systemd argument after # its credential file was removed; recover through the hub's # authenticated mesh path rather than crash-loop forever. logger.warning("device enrollment failed; recovering through mesh bootstrap: %s", exc) await self._ensure_mesh_identity() elif _mesh_proxy_headers(): await self._ensure_mesh_identity() if not self.device_token: logger.error("no DEVICE_TOKEN and no ENROLLMENT_TOKEN; cannot poll scheduler") sys.exit(1) self._session = httpx.AsyncClient(headers=_mesh_proxy_headers()) # Must come after self._session exists — _resolve_device_id() guards # on it and silently no-ops otherwise. On a fresh enroll (no cached # device_id in ~/.robopark/preview_agent.json) that made this ALWAYS # no-op, so device_id never resolved and preview/agent polling fell # back to robot_id (the display name) instead of the real device_id # for the entire session — the same id-mismatch bug fixed elsewhere, # recurring here only on a first-ever enroll. if not self.device_id: await self._resolve_device_id() self._loop = asyncio.get_running_loop() self._start_vision_webhook_server() tasks = [ asyncio.create_task(self._poll_loop()), asyncio.create_task(self._heartbeat_loop()), ] # In production the robot's motion detector sends the webhook. Do not # open the same Windows camera locally unless explicitly requested; # doing both creates a DirectShow ownership conflict with LiveKit. if self.local_camera_motion: tasks.append(asyncio.create_task(self._motion_loop())) await self._shutdown.wait() for t in tasks: t.cancel() try: await t except asyncio.CancelledError: pass if self._vision_server: self._vision_server.shutdown() await self._stop_publisher() await self._stop_motion_capture() await self._session.aclose() async def _stop_motion_capture(self) -> None: async with self._motion_capture_lock: capture = self._motion_capture self._motion_capture = None if capture is not None: await asyncio.to_thread(capture.stop) async def _motion_loop(self) -> None: """Detect motion from RoboVision's shared stream without owning V4L2.""" while not self._shutdown.is_set(): if not self.production_mode: await self._stop_motion_capture() await asyncio.sleep(0.5) continue if self._vision_session_id or self._current_state.active: await self._stop_motion_capture() await asyncio.sleep(0.25) continue try: async with self._motion_capture_lock: if self._motion_capture is None: self._motion_capture = await asyncio.to_thread( create_video_capture, self.video_device, self.width, self.height, self.fps, self.robovision_url, ) self._motion_reference = None if self._motion_capture is None: continue logger.info("motion sampler connected to RoboVision shared stream") frame = await asyncio.to_thread(self._motion_capture.read) if frame is not None: self._detect_motion(frame) except Exception as e: logger.warning(f"motion camera unavailable: {e}") await self._stop_motion_capture() await asyncio.sleep(2.0) await asyncio.sleep(0.1) def _detect_motion(self, frame) -> None: """Compare sparse RGB samples and trigger only meaningful scene changes.""" now = time.monotonic() if now - self._last_motion_sample < 0.25: return self._last_motion_sample = now try: import numpy as np data = np.frombuffer(bytes(frame.data), dtype=np.uint8) sample = data.reshape(frame.height, frame.width, 3)[::12, ::12].mean(axis=2) previous = self._motion_reference self._motion_reference = sample if previous is None or previous.shape != sample.shape: return change = float(np.abs(sample - previous).mean()) threshold = float(os.getenv("VISION_MOTION_THRESHOLD", "12")) if change >= threshold: asyncio.create_task( self._on_vision_motion({"source": "preview_camera", "change": change}) ) except Exception as e: logger.debug(f"preview motion sampling failed: {e}") async def _resolve_device_id(self) -> None: """Look up our device_id from the scheduler using the token.""" if not self._session or not self.device_token: return try: r = await self._session.get( f"{self.scheduler_url.rstrip('/')}/api/devices", headers={"Authorization": f"Bearer {self.device_token}"}, timeout=10.0, ) r.raise_for_status() for d in r.json(): if d.get("name") == self.robot_id: self.device_id = d.get("id") cfg = _load_config() cfg["device_id"] = self.device_id _save_config(cfg) return except Exception as e: logger.warning(f"could not resolve device_id: {e}") async def _poll_loop(self) -> None: while not self._shutdown.is_set(): try: if self._vision_session_id: # A vision-triggered session currently owns the publisher. # Decide whether it's still "active" (silence timeout vs. # hard ceiling) instead of letting the unrelated # operator-preview state tear it down. await self._check_vision_session() if not self._vision_session_id: await self._poll_trigger_command() if not self._vision_session_id: await self._poll_device_config() state = await self._fetch_preview_state() await self._apply_state(state) except Exception as e: logger.warning(f"poll error: {e}") try: await asyncio.wait_for(self._shutdown.wait(), timeout=self.poll_interval) except asyncio.TimeoutError: pass async def _poll_trigger_command(self) -> None: """Consume dashboard test triggers through the authenticated device path.""" if not self._session or not self.device_id or not self.device_token: return r = await self._session.get( f"{self.scheduler_url.rstrip('/')}/api/devices/{self.device_id}/trigger-command", headers={"Authorization": f"Bearer {self.device_token}"}, timeout=10.0, ) r.raise_for_status() data = r.json() self.production_mode = bool(data.get("production_mode", self.production_mode)) if data.get("trigger"): await self._on_vision_motion({"source": data.get("source", "dashboard")}) async def _poll_device_config(self) -> None: """Apply dashboard-selected camera and microphone IDs before preview.""" if not self._session or not self.device_id or not self.device_token: return now = time.monotonic() if now - self._last_device_config_poll < 5.0: return self._last_device_config_poll = now try: r = await self._session.get( f"{self.scheduler_url.rstrip('/')}/api/devices/{self.device_id}/config", headers={"Authorization": f"Bearer {self.device_token}"}, timeout=10.0, ) r.raise_for_status() data = r.json() changed = False if data.get("video_device") is not None: changed = changed or self.video_device != data["video_device"] self.video_device = data["video_device"] if data.get("audio_device") is not None: changed = changed or self.audio_device != data["audio_device"] self.audio_device = data["audio_device"] inventory = data.get("device_inventory") or {} selected = str(self.audio_device) self.audio_capture_device = _resolve_inventory_audio( inventory.get("audio_input", []), selected, os.getenv("ROBOPARK_AUDIO_INPUT_MATCH", ""), ) if data.get("audio_output_device") is not None: changed = changed or self.audio_output_device != data["audio_output_device"] self.audio_output_device = data["audio_output_device"] inventory = data.get("device_inventory") or {} selected = str(self.audio_output_device) self.audio_playback_device = _resolve_inventory_audio( inventory.get("audio_output", []), selected, os.getenv("ROBOPARK_AUDIO_OUTPUT_MATCH", ""), ) if changed: await self._sync_robovision_media_config() except Exception as e: logger.debug(f"device config poll failed: {e}") async def _sync_robovision_media_config(self) -> None: """Apply scheduler media choices to the local RoboVisionAI_PI owner.""" try: response = await self._session.post( f"{self.robovision_url.rstrip('/')}/api/media/config", json={ "video_device": self.video_device, "audio_device": self.audio_device, "audio_output_device": self.audio_output_device, }, timeout=1.0, ) if response.is_success: global _DEVICE_INVENTORY_CACHE _DEVICE_INVENTORY_CACHE = None except Exception: # RoboVision is optional on laptops and simulation nodes. pass async def _check_vision_session(self) -> None: """Decide whether the active motion-triggered session should keep holding the publisher. Two limits apply, whichever comes first: 1. HARD ceiling — vision_session_seconds after the session started. Enforced locally, no scheduler round-trip needed. Guarantees a stuck/misbehaving caller can never hold the publisher forever. 2. SILENCE timeout — vision_silence_timeout since last_activity_at last moved. last_activity_at lives on the scheduler and is only ever bumped by POST /api/sessions/{id}/keepalive — a call this agent does NOT make itself. In production that call is made by the VOICE AGENT (a separate process/repo, VAD-driven), which is the only thing that actually knows whether the user/agent are still talking. We poll GET /api/sessions/{id}/status here every poll_interval to pull the latest value. If nothing ever calls keepalive, last_activity_at never advances past started_at and the session ends after vision_silence_timeout — a safe default, not a malfunction. Wiring the voice agent to call keepalive is required to make this genuinely silence-aware; that integration is out of scope for this repo. """ now = time.time() if now >= self._vision_hard_deadline: logger.info("vision session hit its hard ceiling (vision_session_seconds) — ending") await self._end_vision_session() return try: last_activity = await self._fetch_session_last_activity(self._vision_session_id) if self._remote_session_ended: logger.info("scheduler ended the vision session — stopping publisher") await self._end_vision_session() return if last_activity is not None and last_activity > self._vision_last_activity: self._vision_last_activity = last_activity except Exception as e: logger.debug(f"could not refresh session activity: {e}") if now - self._vision_last_activity >= self.vision_silence_timeout: logger.info( f"vision session silent for {now - self._vision_last_activity:.0f}s " f"(>= {self.vision_silence_timeout}s timeout) — ending" ) await self._end_vision_session() async def _fetch_session_last_activity(self, session_id: str) -> Optional[float]: """GET /api/sessions/{id}/status and return last_activity_at as a Unix timestamp, or None if unavailable. Naive UTC ISO strings (as written by main.py's datetime.utcnow().isoformat()) are interpreted as UTC explicitly — treating them as local time would silently skew the silence calculation on any host not running in UTC.""" if not self._session or not self.device_token: return None r = await self._session.get( f"{self.scheduler_url.rstrip('/')}/api/sessions/{session_id}/status", headers={"Authorization": f"Bearer {self.device_token}"}, timeout=10.0, ) r.raise_for_status() data = r.json() if data.get("ended_at"): self._remote_session_ended = True return None ts = data.get("last_activity_at") or data.get("started_at") if not ts: return None dt = datetime.fromisoformat(ts) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.timestamp() async def _end_vision_session(self) -> None: # Tell the scheduler too, not just the local publisher/LiveKit # connection — otherwise the sessions row never gets ended_at set, # and the server's active_sessions count climbs forever until it # hits max_sessions and silently blocks the room getting genuinely # picked up (unrelated to this specific dispatch, but a real bug: # every silence-timeout before this fix leaked a permanently # "active" session). remote_session_ended = self._remote_session_ended if not remote_session_ended and self._session and self.device_token: try: await self._session.post( f"{self.scheduler_url.rstrip('/')}/api/robots/{self.device_id or self.robot_id}/end-session", params={"reason": "silence"}, headers={"Authorization": f"Bearer {self.device_token}"}, timeout=10.0, ) except Exception as e: logger.warning(f"failed to notify scheduler of session end: {e}") await self._report_pipeline("session_ended", "ok", "Robot conversation loop stopped") self._vision_session_id = None self._vision_trigger_in_flight = False self._vision_hard_deadline = 0.0 self._vision_last_activity = 0.0 self._remote_session_ended = False await self._stop_publisher() await asyncio.to_thread(_play_audio_effect, self.audio_playback_device, "disconnect") self._current_state = PreviewState() async def _heartbeat_loop(self) -> None: while not self._shutdown.is_set(): if _mesh_proxy_headers() and not self._mesh_bootstrap_ready: await self._ensure_mesh_identity() if self.device_id and self.device_token: production_mode, credentials_ok = await _send_heartbeat( self.scheduler_url, self.device_id, self.device_token ) if not credentials_ok: self._mesh_bootstrap_ready = False self._next_mesh_bootstrap = 0.0 if production_mode is not None: self.production_mode = production_mode # Exactly one motion detector is active. With the default # local sampler enabled, preview reads RoboVision's shared # MJPEG stream and RoboVision only owns capture/encoding. robovision_motion = bool(production_mode and not self.local_camera_motion) if self._robovision_motion_state != robovision_motion: try: response = await self._session.post( f"{self.robovision_url.rstrip('/')}/api/motion/toggle", json={"enabled": robovision_motion}, timeout=1.5, ) response.raise_for_status() self._robovision_motion_state = robovision_motion except Exception as exc: logger.warning(f"failed to synchronize RoboVision motion mode: {exc}") await self._report_pipeline("robot_online", message="Robot heartbeat accepted") inventory = _get_device_inventory() real_video = [d for d in inventory.get("video", []) if str(d.get("id", "")).lower() not in ("auto", "none")] real_inputs = [d for d in inventory.get("audio_input", []) if str(d.get("id", "")).lower() not in ("default", "none")] real_outputs = [d for d in inventory.get("audio_output", []) if str(d.get("id", "")).lower() not in ("default", "none")] await self._report_pipeline("camera_ready", "ok" if real_video else "blocked", f"{len(real_video)} camera device(s) detected") await self._report_pipeline("microphone_ready", "ok" if real_inputs else "blocked", f"{len(real_inputs)} microphone device(s) detected") await self._report_pipeline("speaker_ready", "ok" if real_outputs else "blocked", f"{len(real_outputs)} speaker device(s) detected") if not production_mode and self._vision_session_id: await self._end_vision_session() await self._poll_speaker_test() try: await asyncio.wait_for(self._shutdown.wait(), timeout=self.heartbeat_interval) except asyncio.TimeoutError: pass async def _poll_speaker_test(self) -> None: """Execute queued speaker tests even when no optional supervisor runs.""" if self._speaker_test_in_flight or not self._session or not self.device_id or not self.device_token: return headers = {"Authorization": f"Bearer {self.device_token}", **_mesh_proxy_headers()} try: response = await self._session.get( f"{self.scheduler_url.rstrip('/')}/api/devices/{self.device_id}/shell/next-speaker-test", headers=headers, timeout=5.0, ) response.raise_for_status() request = response.json().get("request") if not request: return self._speaker_test_in_flight = True from robot_supervisor import _speaker_roundtrip_test async with self._speaker_operation_lock: # LiveKit owns the USB microphone and speaker continuously. # Pause it while holding the same lock as the motion cue so # nothing can reopen ALSA before the explicit test starts. restore_state = self._current_state if self._publisher else None if restore_state: await self._stop_publisher() await asyncio.sleep(1.0) try: result = await asyncio.to_thread(_speaker_roundtrip_test, request.get("params") or {}) result["media_owner_paused"] = bool(restore_state) finally: if restore_state and restore_state.active: await self._start_publisher(restore_state) result_response = await self._session.post( f"{self.scheduler_url.rstrip('/')}/api/devices/{self.device_id}/supervisor-output", json={"kind": "speaker_test", "service": None, "payload": result, "request_id": request.get("id")}, headers=headers, timeout=8.0, ) result_response.raise_for_status() except Exception as e: logger.warning(f"speaker test execution failed: {e}") finally: self._speaker_test_in_flight = False async def _fetch_preview_state(self) -> PreviewState: if not self._session or not self.device_token: return PreviewState() r = await self._session.get( f"{self.scheduler_url.rstrip('/')}/api/robots/{self.device_id or self.robot_id}/preview/agent", headers={"Authorization": f"Bearer {self.device_token}"}, timeout=10.0, ) r.raise_for_status() data = r.json() if not data.get("active"): return PreviewState() return PreviewState( active=True, url=_normalize_livekit_url(data.get("url")), token=data.get("token"), room=data.get("room"), mode=data.get("mode", "preview"), session_id=data.get("session_id"), ) async def _apply_state(self, state: PreviewState) -> None: same = ( state.active == self._current_state.active and state.room == self._current_state.room and state.url == self._current_state.url and state.mode == self._current_state.mode and state.session_id == self._current_state.session_id ) if same: return self._current_state = state if not state.active: await self._stop_publisher() return await self._start_publisher(state) async def _start_publisher(self, state: PreviewState) -> bool: await self._stop_publisher() if not state.url or not state.token or not state.room: return False pub = None try: pub = LiveKitPublisher(state.url, state.token, state.room, self) await pub.start() self._publisher = pub logger.info(f"joined preview room {state.room}") return True except Exception as e: logger.error(f"failed to start publisher: {e}") if pub is not None: try: await pub.stop() except Exception as cleanup_error: logger.debug(f"publisher cleanup after start failure: {cleanup_error}") return False async def _stop_publisher(self) -> None: if self._publisher: try: await self._publisher.stop() except Exception as e: logger.warning(f"publisher stop error: {e}") self._publisher = None # ---- vision (RoboVisionAI_PI motion webhook) -> presence-triggered session ---- def _start_vision_webhook_server(self) -> None: if not self.vision_webhook_port: return agent = self class Handler(BaseHTTPRequestHandler): def log_message(self, fmt, *a): # noqa: A002 — quiet by default logger.debug("vision webhook: " + fmt, *a) def do_POST(self): # noqa: N802 — http.server's required method name try: length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(length) if length else b"{}" payload = json.loads(body or b"{}") except Exception as e: self.send_response(400) self.end_headers() self.wfile.write(str(e).encode()) return self.send_response(200) self.end_headers() self.wfile.write(b'{"ok":true}') if agent._loop: asyncio.run_coroutine_threadsafe(agent._on_vision_motion(payload), agent._loop) def do_GET(self): # noqa: N802 self.send_response(200) self.end_headers() self.wfile.write(b'{"status":"ok","listening_for":"robovision motion webhook"}') try: self._vision_server = ThreadingHTTPServer(("0.0.0.0", self.vision_webhook_port), Handler) thread = threading.Thread(target=self._vision_server.serve_forever, daemon=True) thread.start() logger.info( f"vision webhook listening on :{self.vision_webhook_port} — " f"point RoboVisionAI_PI's POST /api/motion/webhook at " f"http://:{self.vision_webhook_port}/" ) except Exception as e: logger.error(f"could not start vision webhook server: {e}") self._vision_server = None async def _on_vision_motion(self, payload: dict) -> None: # vision_trigger_cooldown is a MINIMUM SPACING between trigger # attempts, not "an active session is fine to interrupt" -- without # this separate check, continuous ambient motion (someone standing # in frame) re-fires every cooldown window regardless of whether a # conversation is already in progress, tearing down the room and # restarting it before the greeting/response ever finishes playing. # Only allow a new trigger once the previous vision session has # actually ended (silence timeout, hard ceiling, or remote end). if self._vision_session_id or self._vision_trigger_in_flight: logger.debug("vision trigger suppressed (a vision session is already active)") return now = time.time() if now - self._last_vision_trigger < self.vision_trigger_cooldown: logger.debug("vision trigger suppressed (cooldown)") return self._last_vision_trigger = now if not self.production_mode: logger.info("motion event ignored because production mode is OFF") return if not self.device_id or not self.device_token or not self._session: logger.warning("vision motion event received but not enrolled yet — ignoring") return self._vision_trigger_in_flight = True # Observability must not add a scheduler round trip before the actual # session request. The reporter handles and logs its own errors. asyncio.create_task(self._report_pipeline( "motion_detected", "ok", f"Motion received from {payload.get('source', 'camera')}", once=False, )) logger.info("motion detected by RoboVisionAI_PI — requesting a session") # RoboVision owns the physical camera and exposes a shared stream, so # there is no capture handle to tear down and no reason to sleep here. if self.motion_cue_enabled: logger.warning("ROBOPARK_MOTION_CUE is enabled; the diagnostic cue adds greeting latency") async with self._speaker_operation_lock: await asyncio.to_thread(_play_audio_effect, self.audio_playback_device, "motion") try: r = await self._session.post( f"{self.scheduler_url.rstrip('/')}/api/devices/{self.device_id}/request-session", headers={"Authorization": f"Bearer {self.device_token}"}, timeout=10.0, ) r.raise_for_status() data = r.json() except Exception as e: logger.error(f"request-session failed: {e}") await self._report_pipeline("scheduler_session", "failed", f"Session request failed: {type(e).__name__}", once=False) self._vision_trigger_in_flight = False return state = PreviewState( active=True, url=_normalize_livekit_url(data.get("server_url")), token=data.get("token"), room=data.get("room_name"), ) session_id = data.get("session_id") voice_config = data.get("voice_config") now = time.time() self._vision_session_id = session_id self._vision_trigger_in_flight = False self._vision_hard_deadline = now + self.vision_session_seconds self._vision_last_activity = now self._remote_session_ended = False self._current_state = state await self._report_pipeline("scheduler_session", "ok", "Scheduler allocated a conversation session") # If the scheduler returned a voice config, let the agent know by # posting it to our local webhook endpoint. The preview agent itself # does not consume it, but this makes the config observable locally # and lets downstream components (audio server, vision, etc.) adapt. if voice_config and session_id: logger.info(f"scheduler voice config for session {session_id}: {voice_config}") joined = await self._start_publisher(state) if not joined: await self._report_pipeline( "livekit_join", "failed", "Robot could not connect to its assigned LiveKit route", details={"server_url": state.url}, once=False, ) if session_id and self._session: try: response = await self._session.post( f"{self.scheduler_url.rstrip('/')}/api/robots/{self.device_id}/end-session", params={"reason": "livekit_join_failed"}, headers={"Authorization": f"Bearer {self.device_token}"}, timeout=10.0, ) response.raise_for_status() except Exception as e: logger.debug(f"could not end failed LiveKit session: {e}") self._vision_session_id = None self._vision_hard_deadline = 0.0 self._vision_last_activity = 0.0 self._current_state = PreviewState() return if session_id and self._session: try: response = await self._session.post( f"{self.scheduler_url.rstrip('/')}/api/sessions/{session_id}/joined", headers={"Authorization": f"Bearer {self.device_token}"}, timeout=10.0, ) response.raise_for_status() except Exception as e: logger.debug(f"could not mark session joined: {e}") def shutdown(self) -> None: self._shutdown.set() class LiveKitPublisher: """Encapsulates LiveKit room connection, camera capture and mic capture.""" def __init__(self, url: str, token: str, room_name: str, agent: PreviewAgent): from livekit import rtc self.url = url self.token = token self.room_name = room_name self.agent = agent self.rtc = rtc self.room: Optional[rtc.Room] = None self.video_source: Optional[rtc.VideoSource] = None self.video_track: Optional[rtc.LocalVideoTrack] = None self.audio_source: Optional[rtc.AudioSource] = None self.audio_track: Optional[rtc.LocalAudioTrack] = None self._stop_event = asyncio.Event() self._tasks: list[asyncio.Task] = [] self._capture: Optional["VideoCapture"] = None self._mic_capture: Optional["AudioCapture"] = None self._mic_streaming = asyncio.Event() self._mic_error: Optional[str] = None # The launch hardware has no acoustic echo cancellation. Publishing # the amplified USB microphone while the robot speaker plays TTS makes # the voice worker hear itself and trigger barge-in, truncating or # chopping its own response. Keep the track alive with silence while # playback is active, plus a short room-echo decay tail. self._half_duplex = str(os.getenv("ROBOPARK_HALF_DUPLEX", "true")).lower() in ( "1", "true", "yes", "on", ) self._speaker_playback_active = threading.Event() self._speaker_gate_until = 0.0 self._speaker_started_at = 0.0 self._barge_in_until = 0.0 self._echo_mic_floor = 0.0 self._barge_in_candidate_frames = 0 self._speaker_echo_tail = max( 0.1, min(float(os.getenv("ROBOPARK_ECHO_TAIL_MS", "350")) / 1000.0, 2.0) ) self._adaptive_barge_in = str( os.getenv("ROBOPARK_ADAPTIVE_BARGE_IN", "true") ).lower() in ("1", "true", "yes", "on") self._barge_in_min_peak = max( 256, min(int(os.getenv("ROBOPARK_BARGE_IN_MIN_PEAK", "2200")), 20000) ) self._barge_in_ratio = max( 1.25, min(float(os.getenv("ROBOPARK_BARGE_IN_ECHO_RATIO", "2.4")), 8.0) ) self._barge_in_hold = max( 0.4, min(float(os.getenv("ROBOPARK_BARGE_IN_HOLD_MS", "1400")) / 1000.0, 3.0) ) # Motion sampling state belongs to the publisher instance. Keeping it # initialized here prevents shutdown/reopen paths from raising while # the camera is being handed between preview and voice sessions. self._last_motion_sample = 0.0 async def start(self) -> None: self.room = self.rtc.Room() await self.room.connect(self.url, self.token) await self.agent._report_pipeline("livekit_join", "ok", "Robot joined the LiveKit room") # Register after signaling. Registering during Room.connect can invoke # callbacks while the native LiveKit participant state is incomplete; # on Windows that has caused an unrecoverable native client abort. self._playback_streams: dict = {} def _on_track_subscribed(track, publication, participant): logger.info(f"track_subscribed: kind={track.kind} from={participant.identity} sid={publication.sid}") if track.kind != self.rtc.TrackKind.KIND_AUDIO: return if participant.identity == self.room.local_participant.identity: return self._tasks.append(asyncio.create_task(self.agent._report_pipeline("tts_subscribed", "ok", "Subscribed to remote voice audio"))) self._tasks.append(asyncio.create_task(self._play_remote_audio(track, publication.sid))) self.room.on("track_subscribed", _on_track_subscribed) logger.info("audio out: track_subscribed listener registered") # Open and warm the camera before creating the native LiveKit source. # Windows webcams may ignore the requested 640x480 mode and return a # different size (for example 480x360); capturing that frame into a # mismatched VideoSource can abort the native SDK. video_enabled = str(self.agent.video_device).lower() not in ("none", "", "false", "null") first_frame = None if video_enabled: try: self._capture = await asyncio.to_thread( create_video_capture, self.agent.video_device, self.agent.width, self.agent.height, self.agent.fps, self.agent.robovision_url if self.agent.use_robovision_camera else None, ) except Exception as e: # Camera availability must not block the microphone/session. # Windows camera drivers can fail during a reopen while the # audio path remains healthy and should still accept speech. logger.warning(f"camera unavailable for this session; continuing audio-only: {e}") self._capture = None if self._capture: for _ in range(12): try: first_frame = await asyncio.to_thread(self._capture.read) except Exception as e: logger.warning(f"camera read failed during warm-up; continuing audio-only: {e}") first_frame = None break if first_frame is not None: logger.info( "camera warm-up produced a valid frame (%sx%s)", first_frame.width, first_frame.height, ) break await asyncio.sleep(0.08) if first_frame is None: self._capture.stop() self._capture = None # Do not publish dummy tracks for explicitly disabled devices. Apart # from misleading the worker, creating native LiveKit sources for a # disabled Windows device has caused unstable track publication. audio_enabled = str(self.agent.audio_device).lower() not in ("none", "", "false", "null") if first_frame is not None: self.video_source = self.rtc.VideoSource(first_frame.width, first_frame.height) self.video_track = self.rtc.LocalVideoTrack.create_video_track("camera", self.video_source) vopts = self.rtc.TrackPublishOptions() vopts.source = self.rtc.TrackSource.SOURCE_CAMERA await self.room.local_participant.publish_track(self.video_track, vopts) self.video_source.capture_frame(first_frame) await self.agent._report_pipeline("camera_published", "ok", "Camera track published") if audio_enabled: self.audio_source = self.rtc.AudioSource(48000, 1) self.audio_track = self.rtc.LocalAudioTrack.create_audio_track("microphone", self.audio_source) aopts = self.rtc.TrackPublishOptions() aopts.source = self.rtc.TrackSource.SOURCE_MICROPHONE await self.room.local_participant.publish_track(self.audio_track, aopts) # Register speaker playback (subscribe to the voice agent's TTS audio # track) BEFORE opening the camera. Camera open is a slow/occasionally # hanging blocking call (see create_video_capture below), and since # asyncio is single-threaded, running it inline here would stall the # entire event loop — including receiving the agent's greeting audio # — until it finished, so a short "Hello friend!" greeting could be # over and gone before we ever got a chance to subscribe to it. if audio_enabled: self._tasks.append(asyncio.create_task(self._audio_retry_loop())) try: await asyncio.wait_for(self._mic_streaming.wait(), timeout=5.0) except asyncio.TimeoutError: detail = self._mic_error or f"no PCM received from {self.agent.audio_capture_device}" await self.agent._report_pipeline( "microphone_published", "blocked", f"Microphone track published but PCM capture did not start: {detail}", {"device": self.agent.audio_capture_device, "error": detail}, ) logger.warning( "microphone PCM is not ready (%s); keeping the session alive while capture retries", detail, ) if self._capture and self.video_source: self._tasks.append(asyncio.create_task(self._video_loop())) async def _play_remote_audio(self, track, sid: str) -> None: OUT_RATE = 48000 OUT_CHANNELS = 2 selected_output = str(self.agent.audio_playback_device or "default") pa = None output_device_index = None speaker_guard = None # RoboVision inventories Linux devices through sounddevice/PortAudio, # but the numeric indices are not stable across PyAudio builds. More # importantly, BMW's production USB adapter is already proven through # ALSA's plughw conversion path. Use that exact endpoint for live TTS # instead of reopening the unrelated PyAudio index. if sys.platform.startswith("linux") and "hw:" in selected_output: import re import subprocess from media_lock import media_lock match = re.search(r"\b(hw:\d+,\d+)\b", selected_output) if not match: logger.warning(f"audio out: no ALSA hardware address in {selected_output!r}") return alsa_device = f"plug{match.group(1)}" try: speaker_guard = media_lock("speaker", timeout=8.0).acquire() process = subprocess.Popen( [ "aplay", "-q", "-D", alsa_device, "-t", "raw", "-f", "S16_LE", "-r", str(OUT_RATE), "-c", str(OUT_CHANNELS), ], stdin=subprocess.PIPE, stderr=subprocess.PIPE, ) except Exception as exc: if speaker_guard is not None: speaker_guard.release() logger.warning(f"audio out: could not acquire {alsa_device}: {exc}") return if process.stdin is None: logger.warning(f"audio out: aplay did not expose stdin for {alsa_device}") process.kill() process.wait(timeout=1) speaker_guard.release() return class _AplayOutput: def write(self, chunk: bytes) -> None: if process.poll() is not None: detail = "" if process.stderr is not None: detail = process.stderr.read().decode("utf-8", errors="replace").strip() raise OSError(detail or f"aplay exited {process.returncode}") process.stdin.write(chunk) process.stdin.flush() def stop_stream(self) -> None: if process.stdin and not process.stdin.closed: process.stdin.close() try: process.wait(timeout=2) except subprocess.TimeoutExpired: process.terminate() try: process.wait(timeout=1) except subprocess.TimeoutExpired: process.kill() process.wait(timeout=1) def close(self) -> None: return out = _AplayOutput() output_device_index = alsa_device else: try: import pyaudio except Exception as e: logger.warning(f"pyaudio unavailable for playback: {e}") return pa = pyaudio.PyAudio() # Same MME-vs-WASAPI gotcha as mic capture (see # PyAudioCapture._resolve_device): prefer the endpoint backing the # Windows volume mixer instead of the silent MME default. if selected_output.strip().isdigit(): output_device_index = int(selected_output.strip()) try: if selected_output.lower() == "default": wasapi = pa.get_host_api_info_by_type(pyaudio.paWASAPI) idx = wasapi.get("defaultOutputDevice") if idx is not None and idx >= 0: output_device_index = idx elif output_device_index is None: needle = selected_output.lower() for i in range(pa.get_device_count()): info = pa.get_device_info_by_index(i) if info.get("maxOutputChannels", 0) > 0 and needle in str(info.get("name", "")).lower(): output_device_index = i break except Exception as e: logger.debug(f"audio out: WASAPI default output lookup failed, using PyAudio default: {e}") out = pa.open( format=pyaudio.paInt16, channels=OUT_CHANNELS, rate=OUT_RATE, output=True, output_device_index=output_device_index, frames_per_buffer=960, ) logger.info(f"audio out: opened playback stream for {sid} (device_index={output_device_index})") frame_count = 0 mismatch_logged = False try: import numpy as np except Exception as e: np = None logger.warning(f"numpy unavailable for playback resampling: {e}") # Frames arrive over the network in irregular bursts (TTS streaming, # scheduling jitter); writing each one straight to a blocking PyAudio # stream inline ties the audio device's write timing to that jitter, # which is what produced "heavily distorted/staticky" playback even # with matching rate/channels. Decouple the two with a small jitter # buffer: a writer thread drains a queue into PyAudio on its own # steady pace, independent of how unevenly frames actually arrive. import queue import threading write_queue: "queue.Queue[Optional[bytes]]" = queue.Queue() written_frames = [0] PREBUFFER_CHUNKS = 3 stream_failed = threading.Event() playback_reported = threading.Event() event_loop = asyncio.get_running_loop() echo_gate_peak = max(16, min(int(os.getenv("ROBOPARK_ECHO_GATE_PEAK", "96")), 4096)) def _outbound_peak(chunk: bytes) -> int: if len(chunk) < 2: return 0 try: samples = memoryview(chunk).cast("h") # Stereo duplication means sampling every eighth value is # sufficient and keeps the writer thread lightweight. return max((abs(int(value)) for value in samples[::8]), default=0) except (TypeError, ValueError): return 0 def _open_echo_gate() -> None: if not self._half_duplex: return if not self._speaker_playback_active.is_set(): self._speaker_started_at = time.monotonic() self._echo_mic_floor = 0.0 self._barge_in_candidate_frames = 0 self._speaker_playback_active.set() self._speaker_gate_until = time.monotonic() + self._speaker_echo_tail def _extend_echo_gate() -> None: if self._half_duplex: self._speaker_gate_until = time.monotonic() + self._speaker_echo_tail def _safe_write(chunk: bytes) -> bool: if stream_failed.is_set(): return False try: audible = _outbound_peak(chunk) >= echo_gate_peak if audible: _open_echo_gate() elif time.monotonic() >= self._speaker_gate_until: self._speaker_playback_active.clear() out.write(chunk) if audible: _extend_echo_gate() if not playback_reported.is_set(): playback_reported.set() asyncio.run_coroutine_threadsafe( self.agent._report_pipeline( "playback_started", "ok", "First TTS audio chunk written to robot speaker" ), event_loop, ) return True except Exception as e: stream_failed.set() logger.warning(f"audio out stream closed; disabling playback for this track: {e}") return False def _writer(): buffered = [] started = False while True: try: chunk = write_queue.get() except queue.Empty: continue if chunk is None: # Drain any short tail when the remote track ends. for pending in buffered: if not _safe_write(pending): break break if not started: buffered.append(chunk) if len(buffered) < PREBUFFER_CHUNKS: continue for pending in buffered: if not _safe_write(pending): break buffered.clear() if stream_failed.is_set(): break started = True continue try: if _safe_write(chunk): written_frames[0] += 1 if written_frames[0] == 1: logger.info(f"audio out: writer thread wrote its first chunk for {sid}") elif stream_failed.is_set(): break except Exception as e: # A single write failing (e.g. a transient device hiccup) must # not silently kill the whole thread — that leaves every # later frame queued with nothing left to consume them, # which looks exactly like "no audio at all" even though # frames kept arriving fine. Log loudly and keep going. logger.warning(f"audio out write error (continuing): {e}") writer_thread = threading.Thread(target=_writer, daemon=True) writer_thread.start() try: stream = self.rtc.AudioStream(track=track) async for frame in stream: af = frame.frame if hasattr(frame, "frame") else frame frame_count += 1 if frame_count == 1: logger.info( f"audio out: first frame received for {sid} " f"(rate={af.sample_rate}, channels={getattr(af, 'num_channels', 1)})" ) in_rate = af.sample_rate in_channels = int(getattr(af, "num_channels", 1) or 1) raw = bytes(af.data) if np is not None: in_data = np.frombuffer(raw, dtype=np.int16) if in_channels > 1: in_data = in_data.reshape(-1, in_channels).mean(axis=1).astype(np.int16) else: in_data = raw if in_rate == OUT_RATE or np is None: if in_rate != OUT_RATE and not mismatch_logged: mismatch_logged = True logger.warning( f"audio out: sample rate {in_rate} != {OUT_RATE} and numpy " f"unavailable — playing unresampled (expect distortion)" ) mono_data = in_data else: # Mirror the real Pi client's resampling (pi-client/livekit_bridge.py): # writing raw bytes straight to a fixed-rate output stream when the # source rate differs (e.g. Kokoro's native rate vs our 48kHz stream) # is what produced the "heavily distorted/staticky" playback — every # frame needs resampling to OUT_RATE first, not just the ones that # happen to already match. if not mismatch_logged: mismatch_logged = True logger.info(f"audio out: resampling {in_rate}Hz -> {OUT_RATE}Hz for {sid}") ratio = OUT_RATE / in_rate n_out = int(len(in_data) * ratio) indices = np.linspace(0, len(in_data) - 1, n_out) out_data = np.interp(indices, np.arange(len(in_data)), in_data.astype(np.float64)).astype(np.int16) mono_data = out_data if np is not None: # The Windows endpoint is stereo; duplicate mono LiveKit audio explicitly. data = np.repeat(mono_data[:, None], OUT_CHANNELS, axis=1).astype(np.int16).tobytes() else: data = b"".join(raw[i:i + 2] * OUT_CHANNELS for i in range(0, len(raw), 2)) write_queue.put(data) except Exception as e: logger.warning(f"audio out stream error: {e}") finally: cancelling = bool(asyncio.current_task() and asyncio.current_task().cancelling()) if cancelling: # On publisher teardown, close ALSA first. Waiting for the # writer while aplay still owns the device leaves hw:X,Y busy # long enough for the queued dashboard test to fail. stream_failed.set() out.stop_stream() write_queue.put(None) writer_thread.join(timeout=2.0) logger.info( f"audio out: {sid} received {frame_count} frames, " f"writer wrote {written_frames[0]} chunks, queue backlog at close={write_queue.qsize()}" ) if not cancelling: out.stop_stream() out.close() if pa is not None: pa.terminate() if speaker_guard is not None: speaker_guard.release() if self._half_duplex: self._speaker_gate_until = time.monotonic() + self._speaker_echo_tail self._speaker_playback_active.clear() async def stop(self) -> None: self._stop_event.set() for t in self._tasks: t.cancel() try: await t except asyncio.CancelledError: pass self._tasks.clear() if self._mic_capture: await asyncio.to_thread(self._mic_capture.stop) self._mic_capture = None if self._capture: self._capture.stop() self._capture = None if self.room: await self.room.disconnect() self.room = None async def _video_loop(self) -> None: assert self._capture is not None and self.video_source is not None frame_interval = 1.0 / self.agent.fps while not self._stop_event.is_set(): start = time.monotonic() # self._capture.read() is a synchronous, occasionally slow cv2 # call (same USB/driver flakiness as the initial camera open). # Calling it inline here blocks the whole single-threaded event # loop every ~66ms, starving the mic audio loop running on the # same loop — audio throughput was observed collapsing from # ~25 frames/sec to ~1 frame/sec whenever this stalled. frame = await asyncio.to_thread(self._capture.read) if frame is not None: self.video_source.capture_frame(frame) self._detect_motion(frame) elapsed = time.monotonic() - start sleep_for = frame_interval - elapsed if sleep_for > 0: try: await asyncio.wait_for(self._stop_event.wait(), timeout=sleep_for) except asyncio.TimeoutError: pass def _detect_motion(self, frame: "rtc.VideoFrame") -> None: """Detect motion from the already-published camera frame. The preview publisher is the sole camera owner. Keeping motion detection here avoids opening the Windows camera a second time from a separate detector process, which caused black frames and crashes. """ now = time.monotonic() if now - self._last_motion_sample < 0.25: return self._last_motion_sample = now try: import numpy as np data = np.frombuffer(bytes(frame.data), dtype=np.uint8) sample = data.reshape(frame.height, frame.width, 3)[::12, ::12].mean(axis=2) previous = self._motion_reference self._motion_reference = sample if previous is None or previous.shape != sample.shape: return change = float(np.abs(sample - previous).mean()) if change >= float(os.getenv("VISION_MOTION_THRESHOLD", "12")): asyncio.create_task( self._on_vision_motion({"source": "preview_camera", "change": change}) ) except Exception as e: logger.debug(f"preview motion sampling failed: {e}") async def _audio_retry_loop(self) -> None: """Keep microphone capture alive across transient ALSA ownership errors.""" while not self._stop_event.is_set(): try: await self._audio_loop() except asyncio.CancelledError: raise except Exception as e: self._mic_error = str(e) logger.warning( f"audio_loop: capture failed on {self.agent.audio_capture_device}: {e}; retrying" ) finally: mic = self._mic_capture self._mic_capture = None if mic is not None: await asyncio.to_thread(mic.stop) if not self._stop_event.is_set(): try: await asyncio.wait_for(self._stop_event.wait(), timeout=1.0) except asyncio.TimeoutError: pass async def _audio_loop(self) -> None: assert self.audio_source is not None mic = create_audio_capture(self.agent.audio_capture_device) if mic is None: logger.warning("audio_loop: create_audio_capture returned None, mic will not publish") return self._mic_capture = mic self._mic_error = None logger.info(f"audio_loop: mic capture started ({type(mic).__name__})") # Read in bigger batches (100ms) instead of one 20ms frame per # asyncio.to_thread() dispatch. Each dispatch/poll cycle has a fixed # overhead (~30ms observed on this machine) that dominates when the # payload itself is only 20ms — real capture fell to ~11 real # frames/sec against a 50/sec target, and the resulting audio was so # gappy (mostly *missing* frames, not silence) that the agent's VAD # never saw enough continuous signal to trigger, even with loud, # close-mic speech landing clean peaks in the frames that did arrive. # Batching amortizes that overhead over 5 frames per dispatch. BATCH_MS = 100 frame_ms = 20 samples_per_frame = int(48000 * frame_ms / 1000) samples_per_batch = int(48000 * BATCH_MS / 1000) bytes_per_frame = samples_per_frame * 2 # int16 mono mic_gain = max(1.0, min(float(os.getenv("ROBOPARK_MIC_GAIN", "4.0")), 12.0)) _diag_peak = 0 _diag_count = 0 _diag_last_log = time.monotonic() _diag_read_ms = 0.0 _diag_publish_ms = 0.0 _echo_gate_was_active = False batch_interval = BATCH_MS / 1000 while not self._stop_event.is_set(): _iter_start = time.monotonic() _t0 = time.monotonic() batch = await asyncio.to_thread(mic.read, samples_per_batch) _t1 = time.monotonic() if batch is not None: data = bytes(batch.data) _pub_start = time.monotonic() for off in range(0, len(data) - bytes_per_frame + 1, bytes_per_frame): chunk = data[off:off + bytes_per_frame] now = time.monotonic() echo_gate_active = self._half_duplex and ( self._speaker_playback_active.is_set() or now < self._speaker_gate_until ) # Learn the microphone's speaker-echo floor while output is # active. A nearby visitor speaking produces a fast peak well # above that floor; reopen the mic briefly so LiveKit VAD can # cancel normal TTS. The initial 300 ms remains protected. _, raw_peak = _pcm16_scale_and_peak(chunk, 1.0) if echo_gate_active and self._adaptive_barge_in: if self._echo_mic_floor <= 0: self._echo_mic_floor = float(raw_peak) else: self._echo_mic_floor = self._echo_mic_floor * 0.92 + raw_peak * 0.08 threshold = max( self._barge_in_min_peak, int(self._echo_mic_floor * self._barge_in_ratio), ) warmed = now - self._speaker_started_at >= 0.3 if warmed and raw_peak >= threshold: self._barge_in_candidate_frames += 1 else: self._barge_in_candidate_frames = 0 if self._barge_in_candidate_frames >= 3: if now >= self._barge_in_until: logger.info( "audio_loop: adaptive barge-in opened mic " "(peak=%d threshold=%d echo_floor=%d)", raw_peak, threshold, int(self._echo_mic_floor), ) self._barge_in_until = now + self._barge_in_hold barge_in_active = self._adaptive_barge_in and now < self._barge_in_until if echo_gate_active and not barge_in_active: # Preserve 20 ms frame cadence; only suppress content. # Stopping publication would create gaps and destabilize # VAD/endpointing when listening resumes. chunk = b"\x00" * len(chunk) p = 0 else: chunk, p = _pcm16_scale_and_peak(chunk, mic_gain) if echo_gate_active != _echo_gate_was_active: logger.info( "audio_loop: speaker echo gate %s", "active" if echo_gate_active else "released", ) _echo_gate_was_active = echo_gate_active frame = self.rtc.AudioFrame( data=chunk, sample_rate=48000, num_channels=1, samples_per_channel=samples_per_frame, ) await self.audio_source.capture_frame(frame) if not self._mic_streaming.is_set(): self._mic_streaming.set() await self.agent._report_pipeline( "microphone_published", "ok", "Microphone track published with live PCM frames", ) _diag_peak = max(_diag_peak, p) _diag_count += 1 _diag_read_ms += (_t1 - _t0) * 1000 _diag_publish_ms += (time.monotonic() - _pub_start) * 1000 if time.monotonic() - _diag_last_log > 1.0: logger.info(f"audio_loop: {_diag_count} frames sent in last 1s, peak={_diag_peak}/32767, read={_diag_read_ms:.0f}ms, publish={_diag_publish_ms:.0f}ms") _diag_peak = 0 _diag_count = 0 _diag_read_ms = 0.0 _diag_publish_ms = 0.0 _diag_last_log = time.monotonic() # Pace to real wall-clock time per batch. mic.read()'s own # buffer-polling wait is not a reliable substitute — a driver # that briefly double-buffers can let it return a full batch # almost instantly, and capture_frame() (which expects real-time # delivery, blocking internally if fed faster) then accumulates # backpressure that snowballs into multi-second stalls per call. elapsed = time.monotonic() - _iter_start sleep_for = batch_interval - elapsed if sleep_for > 0: try: await asyncio.wait_for(self._stop_event.wait(), timeout=sleep_for) except asyncio.TimeoutError: pass # ----------------------------------------------------------------------------- # Video capture abstraction: V4L2/USB via OpenCV, or picamera2 on Pi. # ----------------------------------------------------------------------------- class VideoCapture: def read(self) -> Optional["rtc.VideoFrame"]: raise NotImplementedError def stop(self) -> None: raise NotImplementedError class OpencvVideoCapture(VideoCapture): def __init__(self, device: int | str, width: int, height: int, fps: int): import cv2 self.cv2 = cv2 if isinstance(device, str) and device.startswith("/dev/video"): device = int(device.replace("/dev/video", "")) elif isinstance(device, str) and device.isdigit(): device = int(device) # MSMF frequently returns intermittent grab failures for USB cameras # on Windows. DirectShow is more stable for this long-lived stream; # retain the default backend as a fallback for unusual devices. if os.name == "nt" and isinstance(device, int): self.cap = cv2.VideoCapture(device, cv2.CAP_DSHOW) if not self.cap.isOpened(): self.cap.release() self.cap = cv2.VideoCapture(device) else: self.cap = cv2.VideoCapture(device) self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, width) self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, height) self.cap.set(cv2.CAP_PROP_FPS, fps) self._last_black_log = 0.0 self._ensure_frame() def _ensure_frame(self): ok, _ = self.cap.read() if not ok: raise RuntimeError("cannot read from camera") def read(self): from livekit import rtc ok, bgr = self.cap.read() if not ok or bgr is None: return None # OpenCV can report a successful read while a Windows camera driver # returns an effectively black frame (privacy shutter, wrong device, # or a stale DirectShow handle). Do not publish misleading video. mean = float(bgr.mean()) if mean <= 1.0 and float(bgr.max()) <= 8.0: now = time.monotonic() if now - self._last_black_log >= 10.0: logger.error( "camera returned black frames from device %s; check camera selection, " "privacy shutter, and Windows camera permissions", self.cap, ) self._last_black_log = now return None # The current LiveKit RTC SDK exposes RGB/RGBA frame types, not BGR. # OpenCV captures BGR, so convert before constructing the frame. rgb = self.cv2.cvtColor(bgr, self.cv2.COLOR_BGR2RGB) return rtc.VideoFrame( width=rgb.shape[1], height=rgb.shape[0], type=rtc.VideoBufferType.RGB24, data=rgb.tobytes(), ) def stop(self): self.cap.release() class Picamera2Capture(VideoCapture): def __init__(self, width: int, height: int, fps: int): from picamera2 import Picamera2 self.picam = Picamera2() config = self.picam.create_video_configuration( main={"format": "RGB888", "size": (width, height)}, controls={"FrameRate": fps}, ) self.picam.configure(config) self.picam.start() self.width = width self.height = height def read(self): from livekit import rtc arr = self.picam.capture_array() if arr is None: return None return rtc.VideoFrame( width=self.width, height=self.height, type=rtc.VideoBufferType.RGB24, data=arr.tobytes(), ) def stop(self): try: self.picam.stop() except Exception: pass def create_video_capture(device: str, width: int, height: int, fps: int, robovision_url: Optional[str] = None) -> Optional[VideoCapture]: if device.lower() in ("none", "", "false", "null"): return None try: from livekit import rtc _ = rtc.VideoSource except Exception as e: logger.error(f"livekit python sdk not installed: {e}") return None if robovision_url: stream_url = f"{robovision_url.rstrip('/')}/video_feed" try: cap = OpencvVideoCapture(stream_url, width, height, fps) logger.info(f"using RoboVisionAI_PI camera stream {stream_url}") return cap except Exception as e: logger.error(f"RoboVision camera stream unavailable: {e}") return None # Auto-detect: prefer first V4L2 device, fall back to picamera2 if available. if device.lower() in ("auto", "default", "first"): for i in range(4): try: cap = OpencvVideoCapture(i, width, height, fps) logger.info(f"auto-selected USB camera /dev/video{i}") return cap except Exception: continue try: cap = Picamera2Capture(width, height, fps) logger.info("auto-selected Pi Camera") return cap except Exception: pass logger.error("no camera found (tried V4L2 and picamera2)") return None if device.lower() in ("picamera", "picamera2", "pi", "rpi"): return Picamera2Capture(width, height, fps) # Treat as V4L2 index or path return OpencvVideoCapture(device, width, height, fps) # ----------------------------------------------------------------------------- # Audio capture abstraction: PyAudio or sounddevice -> LiveKit AudioFrame. # ----------------------------------------------------------------------------- class AudioCapture: def read(self, samples_per_frame: int) -> Optional["rtc.AudioFrame"]: raise NotImplementedError def stop(self) -> None: raise NotImplementedError class AlsaAudioCapture(AudioCapture): """Capture Linux PCM through the same ALSA path used by onsite tests.""" def __init__(self, device: str): import os import re import select import subprocess from media_lock import media_lock match = re.search(r"\b(hw:\d+,\d+)\b", device) if not match: raise ValueError(f"no ALSA hardware address in {device!r}") self.buffer = bytearray() self.media_guard = media_lock("microphone", timeout=1.5).acquire() self.process = None self.source_rate = 48000 errors = [] # The fleet USB microphone normally accepts 48 kHz through ALSA's # plug layer. Some firmware revisions expose only native 44.1 kHz; # accept that rate and resample below rather than publishing silence. for source_rate in (48000, 44100): alsa_device = f"plug{match.group(1)}" process = None try: process = subprocess.Popen( [ "arecord", "-q", "-D", alsa_device, "-t", "raw", "-f", "S16_LE", "-r", str(source_rate), "-c", "1", "--period-size", str(max(256, source_rate // 50)), ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=0, ) if process.stdout is None: raise RuntimeError("arecord did not provide a PCM stream") ready, _, _ = select.select([process.stdout], [], [], 2.0) if not ready: if process.poll() is None: raise TimeoutError("arecord produced no PCM within 2 seconds") detail = process.stderr.read().decode("utf-8", errors="replace").strip() if process.stderr else "" raise OSError(detail or f"arecord exited {process.returncode}") first = os.read(process.stdout.fileno(), max(2048, source_rate // 25 * 2)) if not first: detail = process.stderr.read().decode("utf-8", errors="replace").strip() if process.stderr else "" raise OSError(detail or "arecord returned an empty PCM frame") self.process = process self.source_rate = source_rate self.alsa_device = alsa_device if source_rate != 48000: first = _pcm16_resample_mono(first, source_rate, 48000) self.buffer.extend(first) break except Exception as exc: errors.append(f"{source_rate}Hz: {exc}") if process is not None: if process.poll() is None: process.terminate() try: process.wait(timeout=1) except subprocess.TimeoutExpired: process.kill() process.wait(timeout=1) if self.process is None: self.media_guard.release() raise OSError(f"ALSA capture failed on plug{match.group(1)} ({'; '.join(errors)})") def read(self, samples_per_frame: int): import os from livekit import rtc bytes_needed = samples_per_frame * 2 source_bytes_needed = max(2, int(samples_per_frame * self.source_rate / 48000) * 2) while len(self.buffer) < bytes_needed: chunk = os.read(self.process.stdout.fileno(), source_bytes_needed) if not chunk: detail = "" if self.process.stderr is not None: detail = self.process.stderr.read().decode("utf-8", errors="replace").strip() raise OSError(f"ALSA capture stopped on {self.alsa_device}: {detail or 'no PCM data'}") if self.source_rate != 48000: chunk = _pcm16_resample_mono(chunk, self.source_rate, 48000) self.buffer.extend(chunk) data = bytes(self.buffer[:bytes_needed]) del self.buffer[:bytes_needed] return rtc.AudioFrame( data=data, sample_rate=48000, num_channels=1, samples_per_channel=samples_per_frame, ) def stop(self): try: if self.process.poll() is None: self.process.terminate() try: self.process.wait(timeout=2) except Exception: self.process.kill() self.process.wait(timeout=1) finally: self.media_guard.release() class PyAudioCapture(AudioCapture): def __init__(self, device: str | int | None): import pyaudio self.pa = pyaudio.PyAudio() self.device_index = self._resolve_device(device) self.buffer = bytearray() self.stream = self.pa.open( format=pyaudio.paInt16, channels=1, rate=48000, input=True, input_device_index=self.device_index, frames_per_buffer=960, stream_callback=self._callback, ) self.stream.start_stream() def _resolve_device(self, device: str | int | None) -> Optional[int]: if isinstance(device, int): return device if isinstance(device, str) and device.strip().isdigit(): return int(device.strip()) if device is None or device == "default": # PyAudio's global default resolves through the MME host API, # which on Windows can silently capture pure silence (no error, # no exception — the stream just never has any real signal in # it) even though the same physical mic works fine everywhere # else, including Windows' own input meter. WASAPI is the API # actually backing that meter, so prefer its default input. import pyaudio try: wasapi = self.pa.get_host_api_info_by_type(pyaudio.paWASAPI) idx = wasapi.get("defaultInputDevice") if idx is not None and idx >= 0: return idx except Exception: pass return None # Try exact name match for i in range(self.pa.get_device_count()): info = self.pa.get_device_info_by_index(i) if info.get("maxInputChannels", 0) > 0 and device.lower() in str(info.get("name", "")).lower(): return i return None def _callback(self, in_data, frame_count, time_info, status): import pyaudio self.buffer.extend(in_data) return (None, pyaudio.paContinue) def read(self, samples_per_frame: int): from livekit import rtc bytes_needed = samples_per_frame * 2 # int16 mono while len(self.buffer) < bytes_needed: time.sleep(0.005) chunk = bytes(self.buffer[:bytes_needed]) self.buffer = self.buffer[bytes_needed:] return rtc.AudioFrame( data=chunk, sample_rate=48000, num_channels=1, samples_per_channel=samples_per_frame, ) def stop(self): if self.stream: self.stream.stop_stream() self.stream.close() self.pa.terminate() def has_audio() -> bool: try: import pyaudio # noqa: F401 return True except Exception: return False def create_audio_capture(device: str) -> Optional[AudioCapture]: if device.lower() in ("none", "", "false", "null"): return None if sys.platform.startswith("linux") and "hw:" in device: try: return AlsaAudioCapture(device) except Exception as e: logger.warning(f"ALSA capture unavailable for {device}: {e}") raise try: import pyaudio # noqa: F401 return PyAudioCapture(device if device != "default" else None) except Exception as e: logger.warning(f"pyaudio not available: {e}") return None def _hostname() -> str: import socket return socket.gethostname().split(".")[0] def main() -> None: parser = argparse.ArgumentParser(description="RoboPark preview agent") parser.add_argument("--scheduler-url", default=os.getenv("SCHEDULER_URL", "http://localhost:8080")) parser.add_argument("--robot-id", default=os.getenv("ROBOT_ID", _hostname())) parser.add_argument("--device-token", default=os.getenv("DEVICE_TOKEN")) parser.add_argument("--enrollment-token", default=os.getenv("ENROLLMENT_TOKEN")) parser.add_argument("--video-device", default=os.getenv("VIDEO_DEVICE", "auto")) parser.add_argument("--audio-device", default=os.getenv("AUDIO_DEVICE", "default")) parser.add_argument("--robovision-url", default=None, help="use RoboVisionAI_PI's /video_feed as the LiveKit camera source") parser.add_argument("--width", type=int, default=int(os.getenv("VIDEO_WIDTH", DEFAULT_VIDEO_WIDTH))) parser.add_argument("--height", type=int, default=int(os.getenv("VIDEO_HEIGHT", DEFAULT_VIDEO_HEIGHT))) parser.add_argument("--fps", type=int, default=int(os.getenv("VIDEO_FPS", DEFAULT_FPS))) parser.add_argument("--poll-interval", type=float, default=float(os.getenv("POLL_INTERVAL", DEFAULT_POLL_INTERVAL))) parser.add_argument("--heartbeat-interval", type=float, default=float(os.getenv("HEARTBEAT_INTERVAL", DEFAULT_HEARTBEAT_INTERVAL))) parser.add_argument("--vision-webhook-port", type=int, default=int(os.getenv("VISION_WEBHOOK_PORT", DEFAULT_VISION_WEBHOOK_PORT)), help="local port for RoboVisionAI_PI's motion webhook (0 disables)") parser.add_argument("--vision-trigger-cooldown", type=float, default=float(os.getenv("VISION_TRIGGER_COOLDOWN", DEFAULT_VISION_TRIGGER_COOLDOWN))) parser.add_argument("--vision-session-seconds", type=float, default=float(os.getenv("VISION_SESSION_SECONDS", DEFAULT_VISION_SESSION_SECONDS))) parser.add_argument("--save-config", action="store_true", help="write CLI args to ~/.robopark/preview_agent.json") parser.add_argument("-v", "--verbose", action="store_true") args = parser.parse_args() logging.basicConfig( level=logging.DEBUG if args.verbose else logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s", ) cfg = _load_config() cfg.update({ "scheduler_url": args.scheduler_url, "robot_id": args.robot_id, "video_device": args.video_device, "audio_device": args.audio_device, "robovision_url": args.robovision_url or os.getenv("ROBOVISION_URL", "http://127.0.0.1:5000"), "use_robovision_camera": bool(args.robovision_url) or os.getenv( "ROBOVISION_CAMERA", "true" ).lower() in ("1", "true", "yes", "on"), "video_width": args.width, "video_height": args.height, "video_fps": args.fps, "poll_interval": args.poll_interval, "heartbeat_interval": args.heartbeat_interval, "vision_webhook_port": args.vision_webhook_port, "vision_trigger_cooldown": args.vision_trigger_cooldown, "vision_session_seconds": args.vision_session_seconds, }) if args.device_token: cfg["device_token"] = args.device_token _save_token(args.device_token) if args.enrollment_token: cfg["enrollment_token"] = args.enrollment_token if args.save_config: _save_config(cfg) logger.info(f"saved config to {CONFIG_FILE}") agent = PreviewAgent(cfg) loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) for sig in (signal.SIGINT, signal.SIGTERM): try: loop.add_signal_handler(sig, agent.shutdown) except NotImplementedError: # add_signal_handler is POSIX-only (raises on Windows) — fall back to # signal.signal, dispatched back onto the loop thread-safely. signal.signal(sig, lambda *_: loop.call_soon_threadsafe(agent.shutdown)) try: loop.run_until_complete(agent.run()) finally: loop.close() if __name__ == "__main__": main()