66 lines
2.4 KiB
Python
66 lines
2.4 KiB
Python
"""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"], 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)))
|