"""Command line entrypoint for the local motion capture agent.""" from __future__ import annotations import argparse import asyncio import sys from dataclasses import replace from .cameras import MotionAgentCameraError, MotionAgentDependencyError from .config import MotionAgentConfig from .server import MotionAgentServer def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Run the Planet local motion capture agent.") parser.add_argument("--host", default=None, help="WebSocket bind host.") parser.add_argument("--port", type=int, default=None, help="WebSocket bind port.") parser.add_argument("--camera-indexes", default=None, help="Comma-separated USB camera indexes.") parser.add_argument("--camera-urls", default=None, help="Comma-separated RTSP/HTTP camera URLs.") parser.add_argument( "--mode", choices=["auto", "single", "dual", "dual_redundant", "single_fallback", "calibrated_3d"], default=None, ) parser.add_argument("--dry-run", action="store_true", help="Start without camera/CV dependencies.") return parser def _parse_indexes(raw: str | None, fallback: tuple[int, ...]) -> tuple[int, ...]: if raw is None: return fallback values = [item.strip() for item in raw.split(",") if item.strip()] return tuple(int(item) for item in values) or fallback def _parse_urls(raw: str | None, fallback: tuple[str, ...]) -> tuple[str, ...]: if raw is None: return fallback return tuple(item.strip() for item in raw.split(",") if item.strip()) def config_from_args(args: argparse.Namespace) -> MotionAgentConfig: config = MotionAgentConfig.from_env() return replace( config, host=args.host or config.host, port=args.port or config.port, camera_indexes=_parse_indexes(args.camera_indexes, config.camera_indexes), camera_urls=_parse_urls(args.camera_urls, config.camera_urls), mode=args.mode or config.mode, dry_run=args.dry_run or config.dry_run, ) async def async_main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) config = config_from_args(args) try: server = MotionAgentServer(config) await server.run() except (MotionAgentCameraError, MotionAgentDependencyError, RuntimeError) as exc: print(f"Motion agent failed: {exc}", file=sys.stderr) return 2 return 0 def main(argv: list[str] | None = None) -> None: raise SystemExit(asyncio.run(async_main(argv)))