150 lines
4.5 KiB
Python
150 lines
4.5 KiB
Python
"""Camera input adapters for the local motion capture agent."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Protocol
|
|
|
|
|
|
class MotionAgentCameraError(RuntimeError):
|
|
"""Raised when a configured camera source cannot be opened."""
|
|
|
|
|
|
class MotionAgentDependencyError(RuntimeError):
|
|
"""Raised when an optional runtime dependency is missing."""
|
|
|
|
|
|
class CameraInput(Protocol):
|
|
camera_id: str
|
|
|
|
def open(self) -> None:
|
|
"""Open the camera source."""
|
|
|
|
def read(self) -> Any:
|
|
"""Return the next frame or raise a camera error."""
|
|
|
|
def close(self) -> None:
|
|
"""Release the camera source."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class UsbCameraSpec:
|
|
index: int
|
|
width: int = 1280
|
|
height: int = 720
|
|
fps: int = 30
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class UrlCameraSpec:
|
|
url: str
|
|
camera_id: str | None = None
|
|
|
|
|
|
class UsbCameraInput:
|
|
"""OpenCV-backed USB camera input.
|
|
|
|
OpenCV is intentionally imported lazily so tests and protocol-only runs do
|
|
not require native camera dependencies.
|
|
"""
|
|
|
|
def __init__(self, spec: UsbCameraSpec):
|
|
self.spec = spec
|
|
self.camera_id = f"usb:{spec.index}"
|
|
self._capture: Any = None
|
|
|
|
def open(self) -> None:
|
|
device_path = Path(f"/dev/video{self.spec.index}")
|
|
if Path("/dev").exists() and not device_path.exists():
|
|
raise MotionAgentCameraError(
|
|
f"USB camera index {self.spec.index} was not found at {device_path}. "
|
|
"Check camera passthrough or set MOTION_AGENT_CAMERA_INDEXES to the available indexes."
|
|
)
|
|
|
|
try:
|
|
import cv2 # type: ignore
|
|
except ImportError as exc:
|
|
raise MotionAgentDependencyError(
|
|
"OpenCV is required for camera capture. Add opencv-python with uv "
|
|
"or start the agent with --dry-run for protocol testing."
|
|
) from exc
|
|
|
|
capture = cv2.VideoCapture(self.spec.index, cv2.CAP_V4L2)
|
|
if not capture or not capture.isOpened():
|
|
raise MotionAgentCameraError(f"Unable to open USB camera index {self.spec.index}.")
|
|
|
|
capture.set(cv2.CAP_PROP_FRAME_WIDTH, self.spec.width)
|
|
capture.set(cv2.CAP_PROP_FRAME_HEIGHT, self.spec.height)
|
|
capture.set(cv2.CAP_PROP_FPS, self.spec.fps)
|
|
capture.set(cv2.CAP_PROP_BUFFERSIZE, 1)
|
|
self._capture = capture
|
|
|
|
def read(self) -> Any:
|
|
if self._capture is None:
|
|
raise MotionAgentCameraError(f"Camera {self.camera_id} is not open.")
|
|
ok, frame = self._capture.read()
|
|
if not ok:
|
|
raise MotionAgentCameraError(f"Camera {self.camera_id} did not return a frame.")
|
|
return frame
|
|
|
|
def close(self) -> None:
|
|
if self._capture is not None:
|
|
self._capture.release()
|
|
self._capture = None
|
|
|
|
|
|
class UrlCameraInput:
|
|
"""OpenCV-backed URL camera input for RTSP/HTTP/Web camera bridges."""
|
|
|
|
def __init__(self, spec: UrlCameraSpec):
|
|
self.spec = spec
|
|
self.camera_id = spec.camera_id or f"url:{spec.url}"
|
|
self._capture: Any = None
|
|
|
|
def open(self) -> None:
|
|
try:
|
|
import cv2 # type: ignore
|
|
except ImportError as exc:
|
|
raise MotionAgentDependencyError(
|
|
"OpenCV is required for URL camera capture. Add opencv-python with uv "
|
|
"or start the agent with --dry-run for protocol testing."
|
|
) from exc
|
|
|
|
capture = cv2.VideoCapture(self.spec.url)
|
|
if not capture or not capture.isOpened():
|
|
raise MotionAgentCameraError(
|
|
f"Unable to open camera URL {self.spec.url}. Check the stream URL, "
|
|
"firewall, and LAN reachability."
|
|
)
|
|
capture.set(cv2.CAP_PROP_BUFFERSIZE, 1)
|
|
self._capture = capture
|
|
|
|
def read(self) -> Any:
|
|
if self._capture is None:
|
|
raise MotionAgentCameraError(f"Camera {self.camera_id} is not open.")
|
|
ok, frame = self._capture.read()
|
|
if not ok:
|
|
raise MotionAgentCameraError(f"Camera {self.camera_id} did not return a frame.")
|
|
return frame
|
|
|
|
def close(self) -> None:
|
|
if self._capture is not None:
|
|
self._capture.release()
|
|
self._capture = None
|
|
|
|
|
|
class NullCameraInput:
|
|
"""No-op camera source used by dry-run and tests."""
|
|
|
|
camera_id = "dry-run:null-camera"
|
|
|
|
def open(self) -> None:
|
|
return None
|
|
|
|
def read(self) -> None:
|
|
return None
|
|
|
|
def close(self) -> None:
|
|
return None
|