From 8c6103a093d707e674942acb492f1590ff0b33df Mon Sep 17 00:00:00 2001 From: Max Xiang Date: Mon, 27 Jul 2026 12:02:28 +0200 Subject: [PATCH 1/5] feat(capture): export config and cut over SharedCamera transport Decorate camera drivers with @export_config, add transport builtin type mapping, and cut over camera publisher stdin envelopes to camera: ComponentConfig. Update SharedCamera.from_config() and the tutorial notebook SharedCamera usage for the new construction API. Co-authored-by: Cursor --- docs/explanation/cameras.md | 27 +- examples/capture/shared_camera.ipynb | 8 +- examples/tutorials/collect_train_deploy.ipynb | 42 +- src/physicalai/capture/camera.py | 8 +- .../capture/cameras/basler/_camera.py | 2 + .../capture/cameras/realsense/_camera.py | 2 + src/physicalai/capture/cameras/uvc/_camera.py | 2 + src/physicalai/capture/factory.py | 61 +- src/physicalai/capture/transport/__init__.py | 7 +- .../capture/transport/_publisher.py | 15 +- .../capture/transport/_publisher_worker.py | 113 +++- .../capture/transport/_shared_camera.py | 221 +++++-- src/physicalai/capture/transport/_spec.py | 245 +++++++- src/physicalai/capture/transport/builtin.py | 91 +++ tests/unit/capture/conftest.py | 15 +- tests/unit/capture/fake.py | 2 + .../capture/test_camera_component_config.py | 180 ++++++ tests/unit/capture/test_factory.py | 70 ++- tests/unit/capture/test_transport.py | 551 +++++++++++++++--- 19 files changed, 1399 insertions(+), 263 deletions(-) create mode 100644 src/physicalai/capture/transport/builtin.py create mode 100644 tests/unit/capture/test_camera_component_config.py diff --git a/docs/explanation/cameras.md b/docs/explanation/cameras.md index 247eff77..c06fb5af 100644 --- a/docs/explanation/cameras.md +++ b/docs/explanation/cameras.md @@ -28,16 +28,37 @@ Camera instances are not thread-safe. Use one thread per camera instance or add ## SharedCamera -For multi-process or multi-thread access, use `SharedCamera`. It wraps any camera and handles IPC transport automatically. +`SharedCamera` lets one publisher subprocess own a camera's exclusive hardware +connection while any number of subscribers read frames over iceoryx2. It +satisfies the same `Camera` protocol as a direct driver. + +Construction uses `camera=` or `from_config()`, matching `SharedRobot`. Prefer +`from_config()` (or YAML) when sharing. A disconnected `@export_config` camera +can be passed directly to `camera=`; this exports its recipe rather than +handing an open device into the child. The publisher always opens fresh. Never +keep a direct camera connected to the same device while sharing: another +connected holder causes open failure. ```python -from physicalai.capture import SharedCamera, UVCCamera +from physicalai.capture import SharedCamera +from physicalai.config import to_config -shared = SharedCamera(UVCCamera(device="/dev/video0")) +# Prefer config-only / disconnected export — no live direct camera held open +shared = SharedCamera.from_config( + { + "class_path": "physicalai.capture.UVCCamera", + "init_args": {"device": "/dev/video0", "backend": "v4l2"}, + }, +) +# Equivalent after to_config(disconnected_driver): +# shared = SharedCamera.from_config(to_config(driver)) shared.connect() # Safe to read from multiple threads/processes frame = shared.read_latest() ``` +`create_camera(..., shared=True)` remains a convenience for shareable built-ins +(`uvc`, `realsense`, `basler`) that packs a type into `SharedCamera(camera=...)`. + `SharedCamera` is the recommended approach for production deployments where multiple consumers need camera frames. It avoids the need for manual synchronization and handles frame distribution efficiently. diff --git a/examples/capture/shared_camera.ipynb b/examples/capture/shared_camera.ipynb index 91ea7418..eb16dc55 100644 --- a/examples/capture/shared_camera.ipynb +++ b/examples/capture/shared_camera.ipynb @@ -14,7 +14,7 @@ "the camera, buffer management, or copy overhead.\n", "\n", "Because the transport layer is [iceoryx2](https://github.com/eclipse-iceoryx/iceoryx2),\n", - "any language with iceoryx2 bindings (Rust, C, C++, Python, …) can subscribe to\n", + "any language with iceoryx2 bindings (Rust, C, C++, Python, \u2026) can subscribe to\n", "the same camera stream.\n", "\n", "**Why use SharedCamera:**\n", @@ -81,16 +81,16 @@ "metadata": {}, "outputs": [], "source": [ - "from physicalai.capture import SharedCamera\n", + "from physicalai.capture import SharedCamera, create_camera\n", "\n", "# --- Edit these to match your setup ---\n", "CAMERA_TYPE = CameraType.UVC\n", "DEVICE_KWARGS = {\"device\": 0} # or {\"serial_number\": \"123456\"} for realsense/basler\n", "# ---------------------------------------\n", "\n", - "cam = SharedCamera(CAMERA_TYPE, zero_copy=True, **DEVICE_KWARGS)\n", + "cam: SharedCamera = create_camera(CAMERA_TYPE, shared=True, zero_copy=True, **DEVICE_KWARGS) # type: ignore[assignment]\n", "cam.connect()\n", - "print(f\"Connected to {cam.device_id}\")" + "print(f\"Connected to {cam.device_id}\")\n" ] }, { diff --git a/examples/tutorials/collect_train_deploy.ipynb b/examples/tutorials/collect_train_deploy.ipynb index dc61ebe0..0e5426cb 100644 --- a/examples/tutorials/collect_train_deploy.ipynb +++ b/examples/tutorials/collect_train_deploy.ipynb @@ -5,12 +5,12 @@ "id": "7c7e915d", "metadata": {}, "source": [ - "# Collect → Train → Deploy with SO-101\n", + "# Collect \u2192 Train \u2192 Deploy with SO-101\n", "\n", - "End-to-end workflow: collect teleoperation demos on an **SO-101** arm, fine-tune a **π0.5** policy, and deploy it back on the robot using `physicalai` runtime.\n", + "End-to-end workflow: collect teleoperation demos on an **SO-101** arm, fine-tune a **\u03c00.5** policy, and deploy it back on the robot using `physicalai` runtime.\n", "\n", "1. **Collect** demonstration data (teleoperation with leader/follower arms)\n", - "2. **Train** a π0.5 visuomotor diffusion policy\n", + "2. **Train** a \u03c00.5 visuomotor diffusion policy\n", "3. **Deploy** the trained policy on hardware via `physicalai` runtime\n", "\n", "---" @@ -28,7 +28,7 @@ "| | Requirement |\n", "|--|-------------|\n", "| **Robot** | SO-101 follower arm (+ leader arm for data collection) |\n", - "| **Cameras** | 1–2 USB cameras (UVC/RealSense/Basler compatible) |\n", + "| **Cameras** | 1\u20132 USB cameras (UVC/RealSense/Basler compatible) |\n", "| **Training GPU** | 40 GB+ VRAM (e.g. 2*B70 or A100) |\n", "| **Deploy machine** | Intel CPU, GPU, or NPU with OpenVINO support |\n", "| **OS** | Linux (Ubuntu 22.04+ recommended) |\n", @@ -36,8 +36,8 @@ "\n", "### Install packages\n", "\n", - "- **`physicalai`** — runtime for deployment (Step 3)\n", - "- **`physicalai-train`** — fine-tuning library (Step 2), distributed via [Physical AI Studio](https://github.com/open-edge-platform/physical-ai-studio)" + "- **`physicalai`** \u2014 runtime for deployment (Step 3)\n", + "- **`physicalai-train`** \u2014 fine-tuning library (Step 2), distributed via [Physical AI Studio](https://github.com/open-edge-platform/physical-ai-studio)" ] }, { @@ -87,7 +87,7 @@ "\n", "> **Note:** LeRobot uses `so100` as the robot type for both SO-100 and SO-101 hardware.\n", "\n", - "Aim for **50–100 demonstrations** of the target task with varied object positions." + "Aim for **50\u2013100 demonstrations** of the target task with varied object positions." ] }, { @@ -97,11 +97,11 @@ "source": [ "## Step 2: Train a Policy\n", "\n", - "Train a π0.5 (pi-zero-point-five) visuomotor diffusion policy on the collected demonstrations.\n", + "Train a \u03c00.5 (pi-zero-point-five) visuomotor diffusion policy on the collected demonstrations.\n", "\n", "### Option A: Physical AI Studio (recommended)\n", "\n", - "Launch a training job from the web UI — select your dataset, choose the π0.5 architecture, and configure training parameters visually.\n", + "Launch a training job from the web UI \u2014 select your dataset, choose the \u03c00.5 architecture, and configure training parameters visually.\n", "\n", "Recommended training parameters:\n", "| Parameter | Value |\n", @@ -113,7 +113,7 @@ "| Compile model | true |\n", "\n", "\n", - "> **Note:** Training π0.5 requires at least **40 GB of VRAM** (e.g. NVIDIA A100).\n", + "> **Note:** Training \u03c00.5 requires at least **40 GB of VRAM** (e.g. NVIDIA A100).\n", "\n", "### Option B: `physicalai-train` library" ] @@ -131,13 +131,13 @@ "from physicalai.policies import Pi05\n", "from physicalai.train import Trainer\n", "\n", - "# ── Config ──\n", + "# \u2500\u2500 Config \u2500\u2500\n", "# Physical AI Studio: ~/.cache/physicalai/datasets//\n", "# LeRobot: data/ (relative to where you ran the collection script)\n", "DATASET_PATH = \"~/.cache/physicalai/datasets//\"\n", "CHECKPOINT_DIR = \"./checkpoints\"\n", "\n", - "# Fine-tune π0.5 from the pretrained base\n", + "# Fine-tune \u03c00.5 from the pretrained base\n", "model = Pi05(\n", " pretrained_name_or_path=\"lerobot/pi05_base\",\n", " dtype=\"bfloat16\",\n", @@ -182,7 +182,7 @@ "source": [ "### Export to OpenVINO\n", "\n", - "If using **Physical AI Studio**, the model is automatically exported to OpenVINO format when you download it — no extra step needed.\n", + "If using **Physical AI Studio**, the model is automatically exported to OpenVINO format when you download it \u2014 no extra step needed.\n", "\n", "If training via the library, export manually:" ] @@ -226,7 +226,7 @@ "metadata": {}, "outputs": [], "source": [ - "# Discover available cameras — use the device IDs from this output in the config below\n", + "# Discover available cameras \u2014 use the device IDs from this output in the config below\n", "from physicalai.capture import discover_all\n", "\n", "for driver, devices in discover_all().items():\n", @@ -234,7 +234,7 @@ " continue\n", " print(f\"\\n[{driver}]\")\n", " for dev in devices:\n", - " print(f\" {dev.device_id} — {dev.name}\")" + " print(f\" {dev.device_id} \u2014 {dev.name}\")" ] }, { @@ -256,7 +256,7 @@ "# LeRobot: ~/.cache/calibration/so101_follower.json\n", "SO101_CALIBRATION = \"~/.local/share/physicalai/robots//calibrations/.json\"\n", "\n", - "# Cameras — use the same names and setup as during dataset collection\n", + "# Cameras \u2014 use the same names and setup as during dataset collection\n", "CAMERAS = [\n", " (\"overhead\", \"uvc\", \"/dev/video0\"),\n", " (\"arm\", \"uvc\", \"/dev/video1\"),\n", @@ -308,14 +308,14 @@ "source": [ "from typing import Any\n", "\n", - "from physicalai.capture import SharedCamera\n", + "from physicalai.capture import SharedCamera, create_camera\n", "from physicalai.robot import SO101\n", "\n", "# Cameras\n", "cameras = {}\n", "for name, driver, device_id in CAMERAS:\n", " kwargs: dict[str, Any] = {\"serial_number\": device_id} if driver == \"realsense\" else {\"device\": device_id}\n", - " cam = SharedCamera(driver, **kwargs, width=CAMERA_WIDTH, height=CAMERA_HEIGHT, fps=CAMERA_FPS)\n", + " cam: SharedCamera = create_camera(driver, shared=True, width=CAMERA_WIDTH, height=CAMERA_HEIGHT, fps=CAMERA_FPS, **kwargs) # type: ignore[assignment]\n", " cam.connect()\n", " cameras[name] = cam\n", " print(f\"Camera '{name}' connected: {cam.actual_width}x{cam.actual_height} @ {cam.actual_fps}fps\")\n", @@ -333,7 +333,7 @@ "source": [ "### Build and run the policy runtime\n", "\n", - "`RobotRuntime` runs the control loop: observe → infer → send actions at the target FPS. The action source (`PolicySource`) adapts the model + execution + action-queue pipeline.\n", + "`RobotRuntime` runs the control loop: observe \u2192 infer \u2192 send actions at the target FPS. The action source (`PolicySource`) adapts the model + execution + action-queue pipeline.\n", "\n", "> **Safety:** The robot will begin moving immediately. Keep your hand near the power switch or e-stop." ] @@ -361,10 +361,10 @@ ")\n", "\n", "with runtime:\n", - " print(f\"Running policy at {FPS} fps for {DURATION_S}s — task: {TASK!r}\")\n", + " print(f\"Running policy at {FPS} fps for {DURATION_S}s \u2014 task: {TASK!r}\")\n", " steps = runtime.run(duration_s=DURATION_S)\n", "\n", - "print(f\"\\nDone — {steps} steps, {policy_source.action_queue.total_pops} actions popped\")" + "print(f\"\\nDone \u2014 {steps} steps, {policy_source.action_queue.total_pops} actions popped\")" ] }, { diff --git a/src/physicalai/capture/camera.py b/src/physicalai/capture/camera.py index 064df50c..8c1d982f 100644 --- a/src/physicalai/capture/camera.py +++ b/src/physicalai/capture/camera.py @@ -63,14 +63,16 @@ class Camera(ABC): def __init__( self, *, - color_mode: ColorMode = ColorMode.RGB, + color_mode: ColorMode | str = ColorMode.RGB, ) -> None: """Store requested capture parameters. Args: - color_mode: Pixel format for colour image reads. + color_mode: Pixel format for colour image reads. Accepts + :class:`ColorMode` or its string value so + :func:`~physicalai.config.instantiate` round-trips work. """ - self._color_mode = color_mode + self._color_mode = ColorMode(color_mode) self.__executor: ThreadPoolExecutor | None = None # ------------------------------------------------------------------ diff --git a/src/physicalai/capture/cameras/basler/_camera.py b/src/physicalai/capture/cameras/basler/_camera.py index 2b50d719..2b932f71 100644 --- a/src/physicalai/capture/cameras/basler/_camera.py +++ b/src/physicalai/capture/cameras/basler/_camera.py @@ -13,6 +13,7 @@ from physicalai.capture.camera import Camera, ColorMode from physicalai.capture.errors import CaptureError, CaptureTimeoutError, NotConnectedError from physicalai.capture.frame import Frame +from physicalai.config import export_config if TYPE_CHECKING: import numpy as np @@ -20,6 +21,7 @@ from physicalai.capture.discovery import DeviceInfo +@export_config(class_path="physicalai.capture.BaslerCamera") class BaslerCamera(Camera): """Basler camera using pypylon SDK.""" diff --git a/src/physicalai/capture/cameras/realsense/_camera.py b/src/physicalai/capture/cameras/realsense/_camera.py index 6a2340f2..b325b392 100644 --- a/src/physicalai/capture/cameras/realsense/_camera.py +++ b/src/physicalai/capture/cameras/realsense/_camera.py @@ -14,11 +14,13 @@ from physicalai.capture.errors import CaptureError, CaptureTimeoutError, NotConnectedError from physicalai.capture.frame import Frame from physicalai.capture.mixins.depth import DepthMixin +from physicalai.config import export_config if TYPE_CHECKING: from physicalai.capture.discovery import DeviceInfo +@export_config(class_path="physicalai.capture.RealSenseCamera") class RealSenseCamera(DepthMixin, Camera): """RealSense color and depth camera.""" diff --git a/src/physicalai/capture/cameras/uvc/_camera.py b/src/physicalai/capture/cameras/uvc/_camera.py index 66d2d959..b8d6b719 100644 --- a/src/physicalai/capture/cameras/uvc/_camera.py +++ b/src/physicalai/capture/cameras/uvc/_camera.py @@ -18,6 +18,7 @@ from physicalai.capture.camera import Camera, ColorMode from physicalai.capture.cameras.uvc._camera_setting import CameraSetting # noqa: PLC2701 +from physicalai.config import export_config if TYPE_CHECKING: from physicalai.capture.cameras.uvc.v4l2 import V4L2Camera @@ -25,6 +26,7 @@ from physicalai.capture.frame import Frame +@export_config(class_path="physicalai.capture.UVCCamera") class UVCCamera(Camera): """Camera facade for UVC devices (USB Video Class). diff --git a/src/physicalai/capture/factory.py b/src/physicalai/capture/factory.py index 96abbece..3a901c32 100644 --- a/src/physicalai/capture/factory.py +++ b/src/physicalai/capture/factory.py @@ -5,7 +5,7 @@ from __future__ import annotations -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any from loguru import logger @@ -14,8 +14,17 @@ if TYPE_CHECKING: from physicalai.capture.camera import Camera +_SHARED_TRANSPORT_KEYS = frozenset({ + "zero_copy", + "validate_on_connect", + "overwrite_settings", + "idle_timeout", + "service_name", + "color_mode", +}) -def create_camera(camera_type: str, *, shared: bool = False, **kwargs) -> Camera: # noqa: ANN003 + +def create_camera(camera_type: str, *, shared: bool = False, **kwargs: Any) -> Camera: # noqa: ANN401 """Create a camera by type name. Args: @@ -24,21 +33,59 @@ def create_camera(camera_type: str, *, shared: bool = False, **kwargs) -> Camera Case-insensitive. shared: If True, wrap the camera in a :class:`SharedCamera` (iceoryx2 shared-memory transport). Requires the - ``transport`` extra. - **kwargs: Forwarded to the camera constructor. + ``transport`` extra. Only backends with a real shared registry + entry (``uvc``, ``realsense``, ``basler``) support derived + ``service_name``; stub types must use + :meth:`SharedCamera.from_config` with an explicit + ``service_name`` once a driver exists. + **kwargs: Forwarded to the camera constructor. When *shared* is + True, SharedCamera transport knobs (``zero_copy``, + ``validate_on_connect``, ``overwrite_settings``, + ``idle_timeout``, ``service_name``, ``color_mode``) are peeled + off for the subscriber; remaining kwargs become + ``camera.init_args``. Returns: A new camera instance. Raises: - ValueError: If *camera_type* is not a recognised name. + ValueError: If *camera_type* is not a recognised name, or *shared* + is True for a type without shared service-name derivation. """ camera_type = camera_type.lower() if shared: from physicalai.capture.transport import SharedCamera # noqa: PLC0415 - - return SharedCamera(camera_type, **kwargs) + from physicalai.capture.transport.builtin import ( # noqa: PLC0415 + builtin_class_path_for_type, + builtin_shared_type_tokens, + ) + + class_path = builtin_class_path_for_type(camera_type) + if class_path is None: + if camera_type in {t.value for t in CameraType}: + shareable = ", ".join(sorted(builtin_shared_type_tokens())) + msg = ( + f"camera type {camera_type!r} does not support shared=True " + f"(no shareable driver for service-name derivation); " + f"shareable types: {shareable}. " + "Use SharedCamera.from_config(..., service_name=...) once a " + "driver exists, or create_camera without shared." + ) + raise ValueError(msg) + msg = f"Unknown camera type {camera_type!r}. Expected one of: {', '.join(CameraType)}" + raise ValueError(msg) + + transport: dict[str, Any] = {} + init_args = dict(kwargs) + for key in _SHARED_TRANSPORT_KEYS: + if key in init_args: + transport[key] = init_args.pop(key) + + return SharedCamera( + camera={"class_path": class_path, "init_args": init_args}, + **transport, + ) if camera_type == CameraType.UVC: from physicalai.capture.cameras.uvc import UVCCamera # noqa: PLC0415 diff --git a/src/physicalai/capture/transport/__init__.py b/src/physicalai/capture/transport/__init__.py index 80ed9a3c..503c5f38 100644 --- a/src/physicalai/capture/transport/__init__.py +++ b/src/physicalai/capture/transport/__init__.py @@ -4,9 +4,10 @@ """Shared-memory camera transport via iceoryx2. Provides :class:`SharedCamera` as the public entry point for -multi-process camera sharing. Use the constructor for auto-spawn mode -(``SharedCamera(camera_type, ...)``) or -:meth:`SharedCamera.from_publisher` for subscribe-only mode. +multi-process camera sharing. Prefer :meth:`SharedCamera.from_config` +(or YAML), or use ``SharedCamera(camera=...)``. Use +:meth:`SharedCamera.from_publisher` for subscribe-only mode. The publisher +owns the device exclusively — do not keep a direct camera open while sharing. Requires the ``transport`` extra:: diff --git a/src/physicalai/capture/transport/_publisher.py b/src/physicalai/capture/transport/_publisher.py index e5284f1a..d9642458 100644 --- a/src/physicalai/capture/transport/_publisher.py +++ b/src/physicalai/capture/transport/_publisher.py @@ -14,7 +14,7 @@ from physicalai.capture.errors import CaptureError if TYPE_CHECKING: - from physicalai.capture.transport._spec import CameraSpec + from physicalai.capture.transport._spec import CameraPublisherConfig class CameraPublisher: @@ -25,7 +25,7 @@ class CameraPublisher: self-terminates via idle timeout when zero subscribers remain. Args: - spec: Camera construction specification. + spec: Camera construction specification (``camera: ComponentConfig``). service_name: iceoryx2 service name for the pub-sub channel. idle_timeout: Seconds with zero subscribers before self-exit. max_subscribers: Maximum concurrent subscribers. @@ -33,7 +33,7 @@ class CameraPublisher: def __init__( self, - spec: CameraSpec, + spec: CameraPublisherConfig, service_name: str, *, idle_timeout: float = 5, @@ -62,8 +62,7 @@ def start(self, timeout: float = 10.0) -> None: return config: dict = { - "camera_type": self._spec.camera_type, - "camera_kwargs": self._spec.camera_kwargs, + **self._spec.to_json_dict(), "service_name": self._service_name, "idle_timeout": self._idle_timeout, "max_subscribers": self._max_subscribers, @@ -74,9 +73,9 @@ def start(self, timeout: float = 10.0) -> None: # interpreter) plus a hardcoded internal module path. shell=True is not # used, so there is no shell-injection risk. Configuration is delivered # to the worker via stdin as JSON, not as argv arguments; callers are - # responsible for validating spec fields (camera_type, camera_kwargs, - # service_name, _factory_override) before constructing a - # CameraPublisher. + # responsible for validating spec fields before constructing a + # CameraPublisher. No cwd= override — child inherits parent cwd for + # relative path resolution in camera init_args. self._process = subprocess.Popen( # nosec: B603 [sys.executable, "-m", "physicalai.capture.transport._publisher_worker"], stdin=subprocess.PIPE, diff --git a/src/physicalai/capture/transport/_publisher_worker.py b/src/physicalai/capture/transport/_publisher_worker.py index 862c1471..b4d28524 100644 --- a/src/physicalai/capture/transport/_publisher_worker.py +++ b/src/physicalai/capture/transport/_publisher_worker.py @@ -7,6 +7,7 @@ import contextlib import ctypes +import errno import importlib import json import os @@ -27,9 +28,55 @@ _MAX_CONSECUTIVE_FAILURES = 5 _CONTROL_MAX_SLICE_LEN = 4096 +# Substrings that reliably indicate the device is held by another opener. +_BUSY_MESSAGE_MARKERS = ( + "device or resource busy", + "resource busy", + "already in use", + "device busy", +) + shutdown = threading.Event() +def _looks_like_device_busy(exc: BaseException) -> bool: + """Return True when *exc* (or its cause chain) indicates a busy device. + + Only matches ``errno.EBUSY`` and well-known busy message substrings — + does not guess from generic permission or open failures. + """ + seen: set[int] = set() + current: BaseException | None = exc + while current is not None and id(current) not in seen: + seen.add(id(current)) + if isinstance(current, OSError) and current.errno == errno.EBUSY: + return True + text = str(current).lower() + if any(marker in text for marker in _BUSY_MESSAGE_MARKERS): + return True + current = current.__cause__ or current.__context__ + return False + + +def _format_camera_open_error(exc: BaseException) -> str: + """Format a camera open/connect failure for the READY/ERROR handshake. + + When a busy device is detectable, append exclusive-ownership guidance. + + Returns: + Error string for the publisher ``ERROR:`` handshake line. + """ + base = f"{type(exc).__name__}: {exc}" + if not _looks_like_device_busy(exc): + return base + return ( + f"{base}. The publisher subprocess opens the camera exclusively; " + "another process or a still-connected direct Camera holding the same " + "device will cause open to fail. Disconnect other holders and prefer " + "SharedCamera.from_config without keeping a live direct camera open." + ) + + def sigterm_handler(_signum: int, _frame: FrameType | None) -> None: shutdown.set() @@ -85,36 +132,46 @@ def build_camera(config: dict) -> Camera: """Instantiate a camera from a JSON config dict. Args: - config: Configuration dict with camera_type, camera_kwargs, and - optional _factory_override. + config: Publisher envelope with ``camera: ComponentConfig``, and + optional ``_factory_override`` for tests. Returns: - Connected camera instance. + Camera instance (not yet connected). """ factory_override = config.get("_factory_override") if factory_override: + from physicalai.capture.transport._spec import CameraPublisherConfig # noqa: PLC0415 + + # Validate / reject flat keys even on the factory-override path. + spec = CameraPublisherConfig.from_json_dict(config) module_path, _, attr = factory_override.rpartition(":") mod = importlib.import_module(module_path) factory = getattr(mod, attr) - return factory(**config.get("camera_kwargs", {})) + init_args = spec.camera.get("init_args", {}) + if not isinstance(init_args, dict): + init_args = {} + return factory(**init_args) - from physicalai.capture.transport._spec import CameraSpec # noqa: PLC0415, PLC2701 + from physicalai.capture.transport._spec import CameraPublisherConfig # noqa: PLC0415, PLC2701 - spec = CameraSpec.from_json_dict(config) + spec = CameraPublisherConfig.from_json_dict(config) return spec.build() def _camera_fps_from_config(config: dict[str, object]) -> int: - """Extract fps from camera config when camera_kwargs is mapping-like. + """Extract fps from camera ComponentConfig init_args. Returns: - Requested fps value, or 0 when camera_kwargs is absent or invalid. + Requested fps value, or 0 when absent or invalid. """ - camera_kwargs = config.get("camera_kwargs") - if not isinstance(camera_kwargs, dict): + camera = config.get("camera") + if not isinstance(camera, dict): + return 0 + init_args = camera.get("init_args") + if not isinstance(init_args, dict): return 0 - fps = camera_kwargs.get("fps", 0) + fps = init_args.get("fps", 0) return int(fps) @@ -207,29 +264,32 @@ def _handle_reconfigure(state: _PublisherState, request: dict, service_name: str Args: state: Shared publisher state. - request: Parsed JSON request with ``spec`` key. + request: Parsed JSON request with allowlisted scalar ``settings``. service_name: For logging. Returns: Response dict with ``ok`` and optional ``error``. """ - spec_data = request.get("spec") - if not spec_data or not isinstance(spec_data, dict): - return {"ok": False, "error": "missing or invalid 'spec' in request"} + from ._spec import validate_reconfigure_request # noqa: PLC0415 + + try: + settings = validate_reconfigure_request(request) + except (TypeError, ValueError) as exc: + return {"ok": False, "error": f"invalid reconfigure request: {exc}"} with state.lock: old_config = state.config.copy() old_camera = state.camera old_fps = state.camera_fps - new_config = { - "camera_type": spec_data.get("camera_type", old_config.get("camera_type")), - "camera_kwargs": spec_data.get("camera_kwargs", {}), - "service_name": service_name, + trusted_camera = old_config["camera"] + trusted_init_args = trusted_camera["init_args"] + new_config = old_config.copy() + new_config["camera"] = { + "class_path": trusted_camera["class_path"], + "init_args": {**trusted_init_args, **settings}, } - # Preserve factory override if present in original config - if "_factory_override" in old_config: - new_config["_factory_override"] = old_config["_factory_override"] + new_config["service_name"] = service_name try: old_camera.disconnect() @@ -241,6 +301,7 @@ def _handle_reconfigure(state: _PublisherState, request: dict, service_name: str new_camera.connect() except Exception as exc: # noqa: BLE001 logger.error(f"Reconfigure failed for {service_name}: {exc}. Attempting restore.") + open_err = _format_camera_open_error(exc) try: restored_camera = build_camera(old_config) restored_camera.connect() @@ -252,14 +313,14 @@ def _handle_reconfigure(state: _PublisherState, request: dict, service_name: str f"Cannot restore old config for {service_name} — shutting down.", ) shutdown.set() - return {"ok": False, "error": f"reconfigure failed and restore failed: {exc}"} - return {"ok": False, "error": str(exc)} + return {"ok": False, "error": f"reconfigure failed and restore failed: {open_err}"} + return {"ok": False, "error": open_err} new_fps = _camera_fps_from_config(new_config) state.camera = new_camera state.config = new_config state.camera_fps = new_fps - logger.info(f"Reconfigured publisher for {service_name} with {spec_data}") + logger.info(f"Reconfigured publisher for {service_name} with settings {settings}") return {"ok": True} @@ -344,7 +405,7 @@ def main() -> int: # noqa: C901, PLR0912, PLR0914, PLR0915 if saved_stdout_fd is not None: restore_stdout(saved_stdout_fd) saved_stdout_fd = None - signal_error(f"{type(exc).__name__}: {exc}", tb=traceback.format_exc()) + signal_error(_format_camera_open_error(exc), tb=traceback.format_exc()) if camera is not None: try: camera.disconnect() diff --git a/src/physicalai/capture/transport/_shared_camera.py b/src/physicalai/capture/transport/_shared_camera.py index 09e64850..285bec41 100644 --- a/src/physicalai/capture/transport/_shared_camera.py +++ b/src/physicalai/capture/transport/_shared_camera.py @@ -1,29 +1,41 @@ # Copyright (C) 2026 Intel Corporation # SPDX-License-Identifier: Apache-2.0 -"""Shared-memory camera subscriber transport based on iceoryx2.""" +"""Shared-memory camera subscriber transport based on iceoryx2. + +Construction is :class:`~physicalai.config.ComponentConfig`-only via +``camera=`` or :meth:`SharedCamera.from_config`. ``camera`` is declared as an +``@export_config(config_args=...)`` argument, so a nested +:func:`~physicalai.config.instantiate` hands over the recipe instead of +building a camera this process would immediately discard. Pass ``camera=None`` +with ``service_name`` (or use :meth:`from_publisher`) for attach-only. Flat +``camera_type`` / ``camera_kwargs`` are unsupported. The publisher owns the +device exclusively — do not keep a direct camera open on the same hardware +while sharing. +""" from __future__ import annotations import contextlib import ctypes import time +from collections.abc import Mapping from importlib import import_module -from pathlib import Path from typing import TYPE_CHECKING, Any, cast from loguru import logger -from physicalai.capture.camera import Camera, CameraType, ColorMode +from physicalai.capture.camera import Camera, ColorMode from physicalai.capture.errors import CaptureError, CaptureTimeoutError, NotConnectedError +from physicalai.config import export_config from ._header import FrameHeader, decode_header, decode_rgb +from ._spec import derive_service_name, normalize_camera_config, validate_reconfigure_request if TYPE_CHECKING: - from collections.abc import Mapping - from physicalai.capture.frame import Frame from physicalai.capture.transport._publisher import CameraPublisher + from physicalai.config import ComponentConfig _SERVICE_NAME_EXPECTED_PARTS = 5 @@ -74,16 +86,7 @@ def _probe_with_retry(service_name: str, timeout: float, interval: float = 0.1) time.sleep(interval) -def _derive_service_name(camera_type: str, camera_kwargs: Mapping[str, object]) -> str: - device_id = camera_kwargs.get("serial_number", camera_kwargs.get("device", 0)) - # Resolve symlinks so that /dev/v4l/by-id/... and /dev/videoN produce - # the same service name for the same physical device. - if isinstance(device_id, str) and device_id.startswith("/dev/"): - resolved = Path(device_id).resolve() - device_id = resolved.name - return f"physicalai/camera/{camera_type}/{device_id}/frame" - - +@export_config(class_path="physicalai.capture.SharedCamera", config_args=("camera",)) class SharedCamera(Camera): """Camera subscriber that reads frames from shared memory via iceoryx2. @@ -91,21 +94,39 @@ class SharedCamera(Camera): Multiple SharedCamera instances can subscribe to the same publisher for zero-copy fan-out. + Prefer :meth:`from_config` when sharing; never keep a direct camera open + while sharing. The constructor takes ``camera: ComponentConfig`` to spawn, + or ``camera=None`` + ``service_name`` for attach-only + (:meth:`from_publisher` is the explicit form). + + Opted into :func:`~physicalai.config.export_config` as a **construction + recipe** only (nested ``camera`` ComponentConfig, ``service_name``, + ``color_mode``, transport knobs). Publisher / iceoryx2 session / frame + state is never part of :func:`~physicalai.config.to_config`. + + The publisher subprocess owns the device exclusively. Another connected + holder of the same hardware will cause open to fail; this API does not + hand off an already-open device into the child. + Args: - camera_type: Logical camera type (auto-spawn mode), or ``None`` to - subscribe to an existing publisher only. - color_mode: Pixel format for returned frames. + camera: Trusted camera :class:`~physicalai.config.ComponentConfig` to + spawn if no publisher exists yet for the derived or explicit + ``service_name``. ``None`` means attach-only. Declared as an + ``@export_config`` config arg, so nested + :func:`~physicalai.config.instantiate` passes the recipe through + without constructing the camera here. + color_mode: Pixel format preference for this subscriber. zero_copy: If True, returned frames reference the iceoryx2 SHM buffer directly (read-only). Otherwise, frames are copied. - service_name: Override the iceoryx2 service name. Defaults to one - derived from ``camera_type`` and identifying ``camera_kwargs``. + service_name: iceoryx2 service name. Derived for built-in + ``class_path`` values when omitted; required for third-party + cameras and for attach-only. validate_on_connect: If True and an existing publisher's frame dimensions do not match ``width``/``height`` in - ``camera_kwargs``, :meth:`connect` raises + ``camera.init_args``, :meth:`connect` raises :class:`CaptureError`. If False (default), the mismatch is logged as a warning and the existing publisher's resolution - is used. This validation applies only during - :meth:`connect`. + is used. overwrite_settings: If True, attempt to reconfigure the publisher to match requested settings on config mismatch. Requires the publisher to support the control channel (phase 2+). @@ -117,39 +138,28 @@ class SharedCamera(Camera): def __init__( self, - camera_type: CameraType | str | None, *, - color_mode: ColorMode = ColorMode.RGB, + camera: ComponentConfig | Mapping[str, object] | None = None, + color_mode: ColorMode | str = ColorMode.RGB, zero_copy: bool = False, service_name: str | None = None, validate_on_connect: bool = False, overwrite_settings: bool = False, idle_timeout: float = 5.0, - camera_kwargs: Mapping[str, object] | None = None, - **extra_camera_kwargs: object, ) -> None: - try: - camera_type = CameraType(camera_type) if camera_type is not None else None - except ValueError as exc: - valid = ", ".join(m.value for m in CameraType) - msg = f"unknown camera_type {camera_type!r}; expected one of: {valid}" - raise ValueError(msg) from exc - - if camera_type is None and service_name is None: - msg = "must provide camera_type or service_name" + if camera is None and service_name is None: + msg = "must provide camera ComponentConfig or service_name" raise ValueError(msg) - merged_camera_kwargs: dict[str, object] = {**(camera_kwargs or {}), **extra_camera_kwargs} - - if camera_type is not None: - service_name = _derive_service_name(camera_type, merged_camera_kwargs) + recipe = None if camera is None else normalize_camera_config(camera) + if recipe is not None: + service_name = derive_service_name(recipe, service_name=service_name) elif service_name is None: - msg = "service_name must be provided if camera_type is None" + msg = "service_name must be provided if camera is None" raise ValueError(msg) super().__init__(color_mode=color_mode) - self._camera_type = camera_type - self._camera_kwargs = merged_camera_kwargs + self._camera = recipe self._service_name: str = service_name self._zero_copy = zero_copy self._validate_on_connect = validate_on_connect @@ -165,19 +175,78 @@ def __init__( self._subscriber: Any | None = None self._listener: Any | None = None + @classmethod + def _physicalai_normalize_captured_init_args(cls, supplied: dict[str, object]) -> None: + camera = supplied.get("camera") + if isinstance(camera, Mapping): + supplied["camera"] = normalize_camera_config(camera) + + @classmethod + def from_config( + cls, + config: ComponentConfig | Mapping[str, object], + *, + service_name: str | None = None, + color_mode: ColorMode | str = ColorMode.RGB, + zero_copy: bool = False, + validate_on_connect: bool = False, + overwrite_settings: bool = False, + idle_timeout: float = 5.0, + ) -> SharedCamera: + """Primary API: spawn/attach from a trusted camera ComponentConfig. + + Args: + config: Trusted ``class_path`` + ``init_args`` for the camera. + service_name: Explicit iceoryx2 name; derived for built-ins when omitted. + color_mode: Subscriber pixel-format preference. + zero_copy: Whether frames reference SHM directly. + validate_on_connect: Raise on resolution mismatch at connect. + overwrite_settings: Reconfigure publisher on mismatch. + idle_timeout: Idle self-exit timeout for a spawned publisher. + + Returns: + A ``SharedCamera`` that stores the normalized ComponentConfig and + writes only the new publisher stdin shape on spawn. + """ + return cls( + camera=config, + service_name=service_name, + color_mode=color_mode, + zero_copy=zero_copy, + validate_on_connect=validate_on_connect, + overwrite_settings=overwrite_settings, + idle_timeout=idle_timeout, + ) + @classmethod def from_publisher( cls, service_name: str, *, - color_mode: ColorMode = ColorMode.RGB, + color_mode: ColorMode | str = ColorMode.RGB, zero_copy: bool = False, validate_on_connect: bool = False, overwrite_settings: bool = False, idle_timeout: float = 5.0, ) -> SharedCamera: + """Attach-only construction: subscribe to an existing publisher by name. + + Never spawns a publisher — :meth:`connect` times out if none is + reachable. + + Args: + service_name: iceoryx2 service name of an existing publisher. + color_mode: Subscriber pixel-format preference. + zero_copy: Whether frames reference SHM directly. + validate_on_connect: Raise on resolution mismatch at connect. + overwrite_settings: Reconfigure publisher on mismatch. + idle_timeout: Unused for attach-only (no spawn); kept for API parity. + + Returns: + An attach-only ``SharedCamera``. + """ return cls( - None, + camera=None, color_mode=color_mode, zero_copy=zero_copy, service_name=service_name, @@ -190,11 +259,11 @@ def connect(self, timeout: float = 5.0) -> None: if self._connected: return - if self._camera_type is not None and not _probe_service(self._service_name): + if self._camera is not None and not _probe_service(self._service_name): from ._publisher import CameraPublisher # noqa: PLC0415 - from ._spec import CameraSpec # noqa: PLC0415 + from ._spec import CameraPublisherConfig # noqa: PLC0415 - spec = CameraSpec(self._camera_type, self._camera_kwargs) + spec = CameraPublisherConfig(camera=self._camera) publisher = CameraPublisher(spec, self._service_name, idle_timeout=self._idle_timeout) try: publisher.start() @@ -202,7 +271,11 @@ def connect(self, timeout: float = 5.0) -> None: if _probe_with_retry(self._service_name, timeout=3.0): logger.debug(f"Lost publisher race for {self._service_name} — using existing") else: - msg = f"failed to start camera publisher for {self._service_name}" + msg = ( + f"failed to start camera publisher for {self._service_name}: {exc}. " + "The publisher opens the device exclusively; another connected " + "holder of the same hardware will cause open to fail." + ) raise CaptureError(msg) from exc else: self._publisher = publisher @@ -356,6 +429,32 @@ def _do_disconnect(self) -> None: self._latest = None self._last_header = None + def _build_reconfigure_request(self) -> dict[str, object]: + """Build and validate the settings-only peer control request. + + Returns: + A validated `RECONFIGURE` request. + + Raises: + CaptureError: If this camera has no valid reconfigurable settings. + """ + if self._camera is None: + msg = "reconfigure requires a camera ComponentConfig (attach-only SharedCamera has none)" + raise CaptureError(msg) + + init_args = self._camera["init_args"] + settings = {key: value for key in ("width", "height", "fps") if (value := init_args.get(key)) is not None} + request: dict[str, object] = { + "kind": "RECONFIGURE", + "settings": settings, + } + try: + validate_reconfigure_request(request) + except (TypeError, ValueError) as exc: + msg = f"camera config has no valid reconfigure settings: {exc}" + raise CaptureError(msg) from exc + return request + def _request_reconfigure(self, timeout: float = 5.0) -> dict: """Send a RECONFIGURE request to the publisher's control channel. @@ -375,6 +474,8 @@ def _request_reconfigure(self, timeout: float = 5.0) -> dict: """ import json # noqa: PLC0415 + request = self._build_reconfigure_request() + iox2 = cast("Any", import_module("iceoryx2")) control_name = f"{self._service_name}/control" @@ -391,14 +492,7 @@ def _request_reconfigure(self, timeout: float = 5.0) -> dict: msg = f"publisher does not support reconfigure (no control service at {control_name})" raise CaptureError(msg) from exc - camera_type = self._camera_type.value if self._camera_type is not None else None - request_payload = json.dumps({ - "kind": "RECONFIGURE", - "spec": { - "camera_type": camera_type, - "camera_kwargs": dict(self._camera_kwargs), - }, - }).encode() + request_payload = json.dumps(request).encode() try: sample = client.loan_slice_uninit(len(request_payload)) @@ -507,14 +601,19 @@ def _check_config_match(self, header: FrameHeader) -> None: ) def _detect_mismatch(self, header: FrameHeader) -> tuple[str, str] | None: - """Compare header against requested kwargs. + """Compare header against requested camera init_args. Returns: ``(want_str, actual_str)`` if mismatch found, ``None`` otherwise. """ - want_w = self._camera_kwargs.get("width") - want_h = self._camera_kwargs.get("height") - want_fps = self._camera_kwargs.get("fps") + if self._camera is None: + return None + init_args = self._camera.get("init_args", {}) + if not isinstance(init_args, dict): + return None + want_w = init_args.get("width") + want_h = init_args.get("height") + want_fps = init_args.get("fps") if want_w is None and want_h is None and want_fps is None: return None diff --git a/src/physicalai/capture/transport/_spec.py b/src/physicalai/capture/transport/_spec.py index 69b5f118..6e41594e 100644 --- a/src/physicalai/capture/transport/_spec.py +++ b/src/physicalai/capture/transport/_spec.py @@ -1,68 +1,255 @@ # Copyright (C) 2026 Intel Corporation # SPDX-License-Identifier: Apache-2.0 -"""Serializable camera construction spec for transport endpoints.""" +"""Serializable camera construction spec for transport endpoints. + +Private publisher stdin is ``camera: ComponentConfig`` only. The publisher +envelope is validated schema-positively: required ``camera``, known transport +keys, and rejection of unknown keys (including legacy flat +``camera_type`` / ``camera_kwargs``) before import or hardware access. Public +``SharedCamera`` uses the same ``camera`` / :meth:`~SharedCamera.from_config` +shape. + +Security: ``class_path`` is trusted local application/config input. It must +never originate from network-received data. +""" from __future__ import annotations -from dataclasses import dataclass, field +from collections.abc import Mapping +from dataclasses import dataclass +from pathlib import Path from typing import TYPE_CHECKING, Any +from physicalai.config import ( + ComponentConfig, + instantiate, + normalize_component_config, + validate_envelope, +) + +from .builtin import builtin_type_for_class_path + if TYPE_CHECKING: from physicalai.capture.camera import Camera +# Allowed keys on the trusted publisher stdin handshake. Everything else +# (including legacy flat camera_type / camera_kwargs) is an unknown-key +# schema error. +_PUBLISHER_ENVELOPE_KEYS = frozenset({ + "camera", + "service_name", + "idle_timeout", + "max_subscribers", + "_factory_override", +}) +_RECONFIGURE_REQUEST_KEYS = frozenset({"kind", "settings"}) +_RECONFIGURABLE_SETTINGS = frozenset({"width", "height", "fps"}) + + +def validate_publisher_config(data: Mapping[str, Any]) -> Mapping[str, object]: + """Validate a trusted publisher stdin payload schema-positively. + + Returns: + The validated ``camera`` ComponentConfig mapping (see + :func:`normalize_camera_config` for the JSON-serializability check). + """ + return validate_envelope( + data, + component_key="camera", + allowed_keys=_PUBLISHER_ENVELOPE_KEYS, + envelope_name="publisher", + ) + + +def validate_reconfigure_request(request: Mapping[str, Any]) -> dict[str, int]: + """Validate an untrusted camera reconfigure control request. + + The peer may change only scalar capture settings. The publisher keeps the + trusted startup ``class_path`` and all other constructor arguments. + + Returns: + Validated settings to patch into the trusted camera recipe. + + Raises: + TypeError: If the request or a setting has the wrong type. + ValueError: If required fields are missing, unknown, or out of range. + """ + if not isinstance(request, Mapping): + msg = f"reconfigure request must be a mapping, got {type(request).__name__}" + raise TypeError(msg) + + unknown_request_keys = sorted(set(request) - _RECONFIGURE_REQUEST_KEYS) + if unknown_request_keys: + msg = f"unknown reconfigure request keys {unknown_request_keys}" + raise ValueError(msg) + if request.get("kind") != "RECONFIGURE": + msg = "reconfigure request kind must be 'RECONFIGURE'" + raise ValueError(msg) + + settings = request.get("settings") + if not isinstance(settings, Mapping): + msg = f"reconfigure 'settings' must be a mapping, got {type(settings).__name__}" + raise TypeError(msg) + if not settings: + msg = "reconfigure 'settings' must not be empty" + raise ValueError(msg) + + unknown_settings = sorted(set(settings) - _RECONFIGURABLE_SETTINGS) + if unknown_settings: + msg = f"unknown reconfigure settings {unknown_settings}; allowed: {sorted(_RECONFIGURABLE_SETTINGS)}" + raise ValueError(msg) + + validated: dict[str, int] = {} + for key, value in settings.items(): + if isinstance(value, bool) or not isinstance(value, int): + msg = f"reconfigure setting {key!r} must be an integer, got {type(value).__name__}" + raise TypeError(msg) + if value <= 0: + msg = f"reconfigure setting {key!r} must be greater than zero, got {value}" + raise ValueError(msg) + validated[key] = value + return validated + + +def normalize_camera_config(camera: Mapping[str, object]) -> ComponentConfig: + """Validate a camera ComponentConfig without importing its ``class_path``. + + Returns: + A validated config whose ``class_path`` is a dotted import path. + """ + return normalize_component_config( + camera, + component_key="camera", + class_label="camera class_path", + ) + + +def derive_service_name( + camera: Mapping[str, object], + *, + service_name: str | None = None, +) -> str: + """Resolve iceoryx2 ``service_name`` for a camera ComponentConfig. + + Built-in class paths derive ``physicalai/camera/{token}/{device_id}/frame`` + via the transport class-path → type-token map, which accepts both the + public re-export and the internal spelling of each built-in driver. + Third-party / unknown class paths (including stub types without a shared + registry entry) require an explicit *service_name*. + + Args: + camera: Normalized or raw camera ComponentConfig. + service_name: Explicit override; when set, returned unchanged. + + Returns: + Concrete service name for the publisher envelope. + + Raises: + ValueError: If *service_name* is omitted for a non-built-in class_path. + """ + if service_name is not None: + return service_name + + class_path = str(camera["class_path"]) + token = builtin_type_for_class_path(class_path) + if token is None: + msg = ( + f"camera {class_path!r} requires an explicit service_name; " + "built-in derivation only covers shareable physicalai.capture " + "backends (uvc, realsense, basler)" + ) + raise ValueError(msg) + + init_args = camera.get("init_args", {}) + if not isinstance(init_args, Mapping): + init_args = {} + device_id = init_args.get("serial_number", init_args.get("device", 0)) + # Resolve symlinks so that /dev/v4l/by-id/... and /dev/videoN produce + # the same service name for the same physical device. + if isinstance(device_id, str) and device_id.startswith("/dev/"): + device_id = Path(device_id).resolve().name + return f"physicalai/camera/{token}/{device_id}/frame" + @dataclass(frozen=True) -class CameraSpec: +class CameraPublisherConfig: """Config payload describing how to construct a camera instance. Attributes: - camera_type: Logical camera type passed to :func:`create_camera`. - camera_kwargs: Keyword arguments forwarded to the camera constructor. + camera: Trusted construction config (``class_path`` + ``init_args``). """ - camera_type: str - camera_kwargs: dict[str, Any] = field(default_factory=dict) + camera: Mapping[str, object] - # ------------------------------------------------------------------ - # JSON serialization (used by subprocess publisher protocol) - # ------------------------------------------------------------------ + def __post_init__(self) -> None: + """Normalize ``camera`` to a public ComponentConfig.""" + object.__setattr__(self, "camera", normalize_camera_config(self.camera)) def to_json_dict(self) -> dict[str, Any]: - """Serialize to a JSON-safe dictionary. + """Serialize the construction fragment for the stdin handshake. Returns: - Dictionary with ``camera_type`` and ``camera_kwargs`` keys. + Dictionary with ``camera`` only (transport fields are merged by + the publisher). + + Raises: + TypeError: If ``camera.init_args`` is not a mapping. """ - return {"camera_type": self.camera_type, "camera_kwargs": self.camera_kwargs} + init_args = self.camera["init_args"] + if not isinstance(init_args, dict): + msg = f"camera.init_args must be a mapping, got {type(init_args).__name__}" + raise TypeError(msg) + return { + "camera": { + "class_path": self.camera["class_path"], + "init_args": dict(init_args), + }, + } @classmethod - def from_json_dict(cls, data: dict[str, Any]) -> CameraSpec: - """Deserialize from a JSON dictionary. + def from_json_dict(cls, data: dict[str, Any]) -> CameraPublisherConfig: + """Deserialize from a JSON dictionary (full publisher envelope or fragment). + + Uses :func:`validate_publisher_config` so publisher stdin requires + ``camera``, accepts only known transport keys, and rejects unknown + keys before construction. Args: - data: Dictionary with ``camera_type`` and optional - ``camera_kwargs`` keys. + data: Dictionary produced by :meth:`to_json_dict` or a publisher + stdin envelope containing ``camera``. Returns: - A new :class:`CameraSpec` instance. + A new :class:`CameraPublisherConfig` instance. + + Raises: + TypeError: If ``data`` is not a mapping. """ - return cls( - camera_type=data["camera_type"], - camera_kwargs=data.get("camera_kwargs", {}), - ) + if not isinstance(data, dict): + msg = f"camera spec must be a mapping, got {type(data).__name__}" + raise TypeError(msg) - # ------------------------------------------------------------------ - # Factory - # ------------------------------------------------------------------ + camera = validate_publisher_config(data) + return cls(camera=camera) def build(self) -> Camera: """Instantiate the camera described by this spec. + Uses :func:`physicalai.config.instantiate` on the trusted ``camera`` + ComponentConfig, then verifies the :class:`~physicalai.capture.camera.Camera` + protocol. Does not route through :func:`~physicalai.capture.create_camera`, + so third-party class paths work without a registry entry. + Returns: - A new camera instance configured from ``camera_type`` and - ``camera_kwargs``. + A new, not-yet-connected camera instance. + + Raises: + TypeError: If the instantiated object does not satisfy ``Camera``. """ - from physicalai.capture.factory import create_camera # noqa: PLC0415 + from physicalai.capture.camera import Camera # noqa: PLC0415 - return create_camera(self.camera_type, **self.camera_kwargs) + driver = instantiate(self.camera) # type: ignore[arg-type] + if not isinstance(driver, Camera): + msg = f"{self.camera['class_path']!r} does not satisfy the Camera protocol (got {type(driver).__name__})" + raise TypeError(msg) + return driver diff --git a/src/physicalai/capture/transport/builtin.py b/src/physicalai/capture/transport/builtin.py new file mode 100644 index 00000000..e0fff4e6 --- /dev/null +++ b/src/physicalai/capture/transport/builtin.py @@ -0,0 +1,91 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 + +"""Built-in camera type token ↔ public class_path for SharedCamera. + +Single source of truth for service-name derivation and +``create_camera(..., shared=True)``. Only backends with a real driver +package under ``cameras/`` are listed — IP/Genicam stubs are intentionally +absent so shared spawn cannot claim phantom coverage. + +The table is static on purpose: a subscriber process derives a service name +without the vendor SDK installed, so nothing here may import a driver +module. ``tests/unit/capture/test_factory.py`` asserts the table still +matches each driver's ``@export_config(class_path=...)`` when the optional +camera extras are available. +""" + +from __future__ import annotations + +# token → accepted class paths. Index 0 is the canonical public path declared +# by ``@export_config(class_path=...)`` on the driver; the rest are the +# sub-package re-export and the defining module, so a hand-written config that +# spells a driver the internal way still derives the same service name instead +# of demanding an explicit one. +_BUILTIN_SHARED: dict[str, tuple[str, ...]] = { + "uvc": ( + "physicalai.capture.UVCCamera", + "physicalai.capture.cameras.uvc.UVCCamera", + "physicalai.capture.cameras.uvc._camera.UVCCamera", + ), + "realsense": ( + "physicalai.capture.RealSenseCamera", + "physicalai.capture.cameras.realsense.RealSenseCamera", + "physicalai.capture.cameras.realsense._camera.RealSenseCamera", + ), + "basler": ( + "physicalai.capture.BaslerCamera", + "physicalai.capture.cameras.basler.BaslerCamera", + "physicalai.capture.cameras.basler._camera.BaslerCamera", + ), +} + +_TOKEN_TO_CLASS_PATH: dict[str, str] = {token: paths[0] for token, paths in _BUILTIN_SHARED.items()} +_CLASS_PATH_TO_TOKEN: dict[str, str] = {path: token for token, paths in _BUILTIN_SHARED.items() for path in paths} + + +def builtin_shared_type_tokens() -> frozenset[str]: + """Return CameraType tokens that support shared service-name derivation.""" + return frozenset(_BUILTIN_SHARED) + + +def builtin_class_paths_for_type(token: str) -> tuple[str, ...]: + r"""Return every class_path spelling accepted for a built-in type token. + + Args: + token: Lowercase ``CameraType`` value (e.g. ``\"uvc\"``). + + Returns: + Accepted paths with the canonical public one first, or an empty tuple + if the token is not a shareable built-in. + """ + return _BUILTIN_SHARED.get(token, ()) + + +def builtin_class_path_for_type(token: str) -> str | None: + r"""Return the canonical public class_path for a built-in shared camera type token. + + Args: + token: Lowercase ``CameraType`` value (e.g. ``\"uvc\"``). + + Returns: + Public ``class_path``, or ``None`` if the token is not a shareable built-in. + """ + return _TOKEN_TO_CLASS_PATH.get(token) + + +def builtin_type_for_class_path(class_path: str) -> str | None: + """Return the legacy type token for a built-in camera class_path. + + Accepts the canonical public path plus the internal spellings listed in + :data:`_BUILTIN_SHARED`, so the same physical device derives one service + name however the config names its driver. + + Args: + class_path: Camera ``class_path`` as written in the config. + + Returns: + Type token (``uvc`` / ``realsense`` / ``basler``), or ``None`` for + third-party / stub / unknown paths. + """ + return _CLASS_PATH_TO_TOKEN.get(class_path) diff --git a/tests/unit/capture/conftest.py b/tests/unit/capture/conftest.py index e031109f..86aec9fe 100644 --- a/tests/unit/capture/conftest.py +++ b/tests/unit/capture/conftest.py @@ -6,7 +6,6 @@ from __future__ import annotations import importlib.util -import sys from typing import TYPE_CHECKING import pytest @@ -14,20 +13,24 @@ if TYPE_CHECKING: from collections.abc import Generator - from physicalai.capture.transport._spec import CameraSpec + from physicalai.capture.transport._spec import CameraPublisherConfig HAS_ICEORYX2 = importlib.util.find_spec("iceoryx2") is not None +FAKE_CAMERA_CLASS = "tests.unit.capture.fake.FakeCamera" + @pytest.fixture -def fake_camera_spec() -> CameraSpec: - from physicalai.capture.transport._spec import CameraSpec # noqa: PLC0415 +def fake_camera_spec() -> CameraPublisherConfig: + from physicalai.capture.transport._spec import CameraPublisherConfig # noqa: PLC0415 - return CameraSpec(camera_type="fake", camera_kwargs={}) + return CameraPublisherConfig( + camera={"class_path": FAKE_CAMERA_CLASS, "init_args": {}}, + ) @pytest.fixture -def publisher_service(fake_camera_spec: CameraSpec) -> Generator[str, None, None]: +def publisher_service(fake_camera_spec: CameraPublisherConfig) -> Generator[str, None, None]: from uuid import uuid4 # noqa: PLC0415 from physicalai.capture.transport._publisher import CameraPublisher # noqa: PLC0415 diff --git a/tests/unit/capture/fake.py b/tests/unit/capture/fake.py index 5121b211..58bfa054 100644 --- a/tests/unit/capture/fake.py +++ b/tests/unit/capture/fake.py @@ -17,8 +17,10 @@ from physicalai.capture.camera import Camera, ColorMode from physicalai.capture.errors import NotConnectedError from physicalai.capture.frame import Frame +from physicalai.config import export_config +@export_config class FakeCamera(Camera): """In-memory camera that produces synthetic frames. diff --git a/tests/unit/capture/test_camera_component_config.py b/tests/unit/capture/test_camera_component_config.py new file mode 100644 index 00000000..c8458dab --- /dev/null +++ b/tests/unit/capture/test_camera_component_config.py @@ -0,0 +1,180 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 +# ruff:file-ignore[undocumented-public-module, undocumented-public-class, undocumented-public-method, undocumented-public-init, assert, magic-value-comparison] + +"""Construction round-trips for camera ``@export_config`` wiring.""" + +from __future__ import annotations + +import json +import sys +from collections.abc import Generator +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +from physicalai.capture.camera import Camera, ColorMode +from physicalai.config import ComponentConfig, instantiate, is_config_exportable, to_config + + +def _assert_construction_round_trip(camera: object) -> dict[str, Any]: + assert is_config_exportable(camera) + assert isinstance(camera, Camera) + config = to_config(camera) + wire: dict[str, Any] = json.loads(json.dumps(config)) + restored = instantiate(wire) + assert type(restored) is type(camera) + assert to_config(restored) == wire + assert not restored.is_connected # type: ignore[union-attr] + return wire + + +class TestUVCCameraComponentConfig: + def test_v4l2_backend_round_trip(self) -> None: + from physicalai.capture import UVCCamera + + camera = UVCCamera(device=0, width=640, height=480, fps=30, backend="v4l2") + wire = _assert_construction_round_trip(camera) + assert wire["class_path"] == "physicalai.capture.UVCCamera" + assert wire["init_args"]["device"] == 0 + assert wire["init_args"]["backend"] == "v4l2" + assert wire["init_args"]["width"] == 640 + + def test_color_mode_enum_round_trips_as_string(self) -> None: + from physicalai.capture import UVCCamera + + camera = UVCCamera(device=1, color_mode=ColorMode.BGR, backend="v4l2") + wire = _assert_construction_round_trip(camera) + assert wire["init_args"]["color_mode"] == "bgr" + + +@pytest.fixture +def mock_pyrealsense2() -> Generator[MagicMock, None, None]: + mock_rs = MagicMock() + with patch.dict(sys.modules, {"pyrealsense2": mock_rs}): + # Force reload if already imported with real/missing SDK. + sys.modules.pop("physicalai.capture.cameras.realsense._camera", None) + sys.modules.pop("physicalai.capture.cameras.realsense", None) + yield mock_rs + + +class TestRealSenseCameraComponentConfig: + def test_round_trip(self, mock_pyrealsense2: MagicMock) -> None: + from physicalai.capture.cameras.realsense import RealSenseCamera + + camera = RealSenseCamera(serial_number="SN123", width=640, height=480, fps=30) + wire = _assert_construction_round_trip(camera) + assert wire["class_path"] == "physicalai.capture.RealSenseCamera" + assert wire["init_args"] == { + "serial_number": "SN123", + "width": 640, + "height": 480, + "fps": 30, + } + + +@pytest.fixture +def mock_pypylon() -> Generator[None, None, None]: + mock_pylon = MagicMock() + mock_genicam = MagicMock() + with ( + patch.dict(sys.modules, {"pypylon": MagicMock(), "pypylon.pylon": mock_pylon, "pypylon.genicam": mock_genicam}), + patch.dict(sys.modules, {"cv2": MagicMock()}), + ): + sys.modules.pop("physicalai.capture.cameras.basler._camera", None) + sys.modules.pop("physicalai.capture.cameras.basler", None) + yield + + +class TestBaslerCameraComponentConfig: + def test_round_trip(self, mock_pypylon: None) -> None: + from physicalai.capture.cameras.basler import BaslerCamera + + camera = BaslerCamera(serial_number="BAS-1", fps=15, width=800, height=600) + wire = _assert_construction_round_trip(camera) + assert wire["class_path"] == "physicalai.capture.BaslerCamera" + assert wire["init_args"]["serial_number"] == "BAS-1" + assert wire["init_args"]["fps"] == 15 + assert wire["init_args"]["width"] == 800 + assert wire["init_args"]["height"] == 600 + + +# --------------------------------------------------------------------------- +# SharedCamera (construction recipe only — no publisher / SHM state) +# --------------------------------------------------------------------------- + + +class TestSharedCameraComponentConfig: + def test_spawn_recipe_round_trip(self) -> None: + from physicalai.capture import SharedCamera + + camera = SharedCamera( + camera={ + "class_path": "physicalai.capture.UVCCamera", + "init_args": {"device": "/dev/video0", "width": 640, "height": 480, "fps": 30, "backend": "v4l2"}, + }, + color_mode=ColorMode.BGR, + zero_copy=True, + validate_on_connect=True, + idle_timeout=1.5, + ) + wire = _assert_construction_round_trip(camera) + assert wire["class_path"] == "physicalai.capture.SharedCamera" + assert wire["init_args"]["color_mode"] == "bgr" + assert wire["init_args"]["zero_copy"] is True + assert wire["init_args"]["validate_on_connect"] is True + assert wire["init_args"]["idle_timeout"] == 1.5 + # service_name was derived, not caller-supplied — omitted from recipe. + assert "service_name" not in wire["init_args"] + nested = wire["init_args"]["camera"] + assert isinstance(nested, dict) + assert nested["class_path"] == "physicalai.capture.UVCCamera" + assert nested["init_args"]["device"] == "/dev/video0" + assert "_publisher" not in wire["init_args"] + assert "_connected" not in wire["init_args"] + assert "_node" not in wire["init_args"] + + def test_defining_module_camera_path_round_trips_as_given(self) -> None: + from physicalai.capture import SharedCamera + + defining = "physicalai.capture.cameras.uvc._camera.UVCCamera" + camera = SharedCamera( + camera={ + "class_path": defining, + "init_args": {"device": 0, "backend": "v4l2"}, + }, + ) + wire = _assert_construction_round_trip(camera) + nested = wire["init_args"]["camera"] + assert isinstance(nested, dict) + # Stored as written; the class-path alias table still derives one service name. + assert nested["class_path"] == defining + assert camera._service_name == "physicalai/camera/uvc/0/frame" + + def test_attach_only_round_trip(self) -> None: + from physicalai.capture import SharedCamera + + camera = SharedCamera(camera=None, service_name="physicalai/camera/uvc/0/frame") + wire = _assert_construction_round_trip(camera) + assert wire["class_path"] == "physicalai.capture.SharedCamera" + assert wire["init_args"]["camera"] is None + assert wire["init_args"]["service_name"] == "physicalai/camera/uvc/0/frame" + + def test_nested_recipe_is_not_instantiated(self) -> None: + from physicalai.capture import SharedCamera + + config: ComponentConfig = { + "class_path": "physicalai.capture.SharedCamera", + "init_args": { + "camera": { + "class_path": "physicalai.capture.UVCCamera", + "init_args": {"device": 0, "backend": "v4l2"}, + }, + "service_name": "physicalai/camera/uvc/0/frame", + }, + } + restored = instantiate(config) + assert isinstance(restored, SharedCamera) + # The declared config arg stays a mapping — no UVCCamera is built here. + assert restored._camera == config["init_args"]["camera"] diff --git a/tests/unit/capture/test_factory.py b/tests/unit/capture/test_factory.py index aa329520..5959036f 100644 --- a/tests/unit/capture/test_factory.py +++ b/tests/unit/capture/test_factory.py @@ -6,8 +6,13 @@ import pytest from physicalai.capture.discovery import discover_all -from physicalai.capture.errors import MissingDependencyError from physicalai.capture.factory import create_camera +from physicalai.capture.transport.builtin import ( + builtin_class_path_for_type, + builtin_class_paths_for_type, + builtin_shared_type_tokens, + builtin_type_for_class_path, +) class TestCreateCamera: @@ -24,6 +29,69 @@ def test_case_insensitive(self) -> None: cam = create_camera("UVC", backend="v4l2") assert isinstance(cam, UVCCamera) + def test_shared_stub_types_rejected(self) -> None: + for stub in ("ip", "genicam"): + with pytest.raises(ValueError, match="does not support shared=True"): + create_camera(stub, shared=True, device=0) + + def test_shared_builtin_uses_registry_class_path(self) -> None: + cam = create_camera("uvc", shared=True, device=0, backend="v4l2") + assert cam._camera is not None # type: ignore[attr-defined] + assert cam._camera["class_path"] == builtin_class_path_for_type("uvc") # type: ignore[attr-defined] + assert cam._service_name == "physicalai/camera/uvc/0/frame" # type: ignore[attr-defined] + + +class TestBuiltinSharedRegistry: + """Single source of truth for shareable type ↔ class_path.""" + + def test_shareable_tokens(self) -> None: + assert builtin_shared_type_tokens() == frozenset({"uvc", "realsense", "basler"}) + + def test_round_trip_uvc(self) -> None: + path = builtin_class_path_for_type("uvc") + assert path == "physicalai.capture.UVCCamera" + assert builtin_type_for_class_path(path) == "uvc" + + def test_matches_export_config_when_importable(self) -> None: + from physicalai.capture.cameras.uvc import UVCCamera + from physicalai.config import resolve_public_class_path + + assert builtin_class_path_for_type("uvc") == resolve_public_class_path(UVCCamera) + + @pytest.mark.parametrize("token", ["uvc", "realsense", "basler"]) + def test_static_table_matches_drivers(self, token: str) -> None: + """Every listed spelling must reach the driver the decorator declares.""" + from physicalai.config import import_dotted_path, resolve_public_class_path + + paths = builtin_class_paths_for_type(token) + expected: object = None + try: + expected = import_dotted_path(paths[0]) + except (ImportError, AttributeError) as exc: # optional camera extra absent + pytest.skip(f"{token} driver is not installed: {exc}") + assert isinstance(expected, type) + assert resolve_public_class_path(expected) == paths[0] + + # Compare by defining name rather than identity: other tests reload + # driver modules with a mocked SDK, which replaces the class object. + defining = f"{expected.__module__}.{expected.__qualname__}" + assert defining == paths[-1] + for alias in paths[1:]: + resolved = import_dotted_path(alias) + assert isinstance(resolved, type) + assert f"{resolved.__module__}.{resolved.__qualname__}" == defining + + @pytest.mark.parametrize("token", ["uvc", "realsense", "basler"]) + def test_every_spelling_maps_back_to_one_token(self, token: str) -> None: + for path in builtin_class_paths_for_type(token): + assert builtin_type_for_class_path(path) == token + + def test_no_phantom_ip_genicam(self) -> None: + assert builtin_class_path_for_type("ip") is None + assert builtin_class_path_for_type("genicam") is None + assert builtin_type_for_class_path("physicalai.capture.IPCamera") is None + assert builtin_type_for_class_path("physicalai.capture.GenicamCamera") is None + class TestDiscoverAll: """discover_all() aggregation.""" diff --git a/tests/unit/capture/test_transport.py b/tests/unit/capture/test_transport.py index 500aaa60..a5cf1e17 100644 --- a/tests/unit/capture/test_transport.py +++ b/tests/unit/capture/test_transport.py @@ -6,6 +6,9 @@ import ctypes import importlib.util import pickle +import subprocess +import sys +import textwrap from unittest.mock import MagicMock, patch from uuid import uuid4 @@ -26,7 +29,14 @@ encode_frame, ) from physicalai.capture.transport._shared_camera import SharedCamera -from physicalai.capture.transport._spec import CameraSpec +from physicalai.capture.transport._spec import ( + CameraPublisherConfig, + derive_service_name, + validate_reconfigure_request, +) +from physicalai.config import ComponentConfigError + +from .conftest import FAKE_CAMERA_CLASS HAS_ICEORYX2 = importlib.util.find_spec("iceoryx2") is not None @@ -37,29 +47,194 @@ def _service_name() -> str: return f"physicalai/test/{uuid4().hex[:8]}/frame" -class TestCameraSpec: +class TestCameraPublisherConfig: def test_picklable(self) -> None: - spec = CameraSpec(camera_type="uvc", camera_kwargs={"device": 0, "width": 640}) + spec = CameraPublisherConfig( + camera={ + "class_path": "physicalai.capture.UVCCamera", + "init_args": {"device": 0, "width": 640}, + }, + ) blob = pickle.dumps(spec) restored = pickle.loads(blob) - assert restored.camera_type == spec.camera_type - assert restored.camera_kwargs == spec.camera_kwargs + assert restored.camera == spec.camera - def test_build_delegates_to_factory(self) -> None: - spec = CameraSpec(camera_type="uvc", camera_kwargs={"device": 1, "fps": 30}) + def test_build_uses_instantiate(self) -> None: + spec = CameraPublisherConfig( + camera={ + "class_path": FAKE_CAMERA_CLASS, + "init_args": {"width": 320, "height": 240}, + }, + ) + cam = spec.build() + assert type(cam).__name__ == "FakeCamera" + assert getattr(cam, "_width") == 320 + assert getattr(cam, "_height") == 240 + + def test_from_json_dict_rejects_flat_keys(self) -> None: + with pytest.raises(ValueError, match="unknown publisher config keys"): + CameraPublisherConfig.from_json_dict( + {"camera_type": "uvc", "camera_kwargs": {"device": 0}, "service_name": "x"}, + ) - with patch("physicalai.capture.factory.create_camera") as mock_create: - spec.build() + def test_from_json_dict_rejects_unknown_keys(self) -> None: + with pytest.raises(ValueError, match="unknown publisher config keys"): + CameraPublisherConfig.from_json_dict( + { + "camera": {"class_path": FAKE_CAMERA_CLASS, "init_args": {}}, + "service_name": "x", + "extra_field": 1, + }, + ) + + def test_from_json_dict_requires_camera(self) -> None: + with pytest.raises(ValueError, match="missing required 'camera'"): + CameraPublisherConfig.from_json_dict({"service_name": "x"}) + + def test_validate_publisher_config_shared_helper(self) -> None: + from physicalai.capture.transport._spec import validate_publisher_config + + camera = validate_publisher_config( + { + "camera": {"class_path": FAKE_CAMERA_CLASS, "init_args": {"width": 8}}, + "service_name": "physicalai/test/x/frame", + "idle_timeout": 1.0, + }, + ) + assert camera["class_path"] == FAKE_CAMERA_CLASS + init_args = camera["init_args"] + assert isinstance(init_args, dict) + assert init_args["width"] == 8 + + def test_to_json_dict_camera_shape(self) -> None: + spec = CameraPublisherConfig( + camera={"class_path": FAKE_CAMERA_CLASS, "init_args": {"device_name": "d0"}}, + ) + payload = spec.to_json_dict() + assert set(payload) == {"camera"} + assert payload["camera"]["class_path"] == FAKE_CAMERA_CLASS + assert "service_name" not in payload["camera"]["init_args"] + + +class TestReconfigureRequest: + def test_accepts_partial_positive_integer_settings(self) -> None: + assert validate_reconfigure_request({ + "kind": "RECONFIGURE", + "settings": {"width": 640, "fps": 30}, + }) == {"width": 640, "fps": 30} + + @pytest.mark.parametrize("value", [True, 0, -1, 30.0, "30", None]) + def test_rejects_invalid_setting_values(self, value: object) -> None: + with pytest.raises((TypeError, ValueError), match="reconfigure setting"): + validate_reconfigure_request({ + "kind": "RECONFIGURE", + "settings": {"fps": value}, + }) + + @pytest.mark.parametrize( + "payload", + [ + {"kind": "RECONFIGURE", "settings": {}}, + {"kind": "RECONFIGURE", "settings": {"device": 1}}, + {"kind": "RECONFIGURE", "settings": {"backend": 1}}, + {"kind": "RECONFIGURE", "settings": {"class_path": 1}}, + {"kind": "RECONFIGURE", "spec": {"camera": {}}}, + ], + ) + def test_rejects_empty_unknown_and_legacy_shapes(self, payload: dict[str, object]) -> None: + with pytest.raises((TypeError, ValueError), match="reconfigure"): + validate_reconfigure_request(payload) + + def test_worker_patches_only_trusted_camera_settings(self) -> None: + from physicalai.capture.transport._publisher_worker import _PublisherState, _handle_reconfigure + + old_camera = MagicMock() + old_camera.disconnect.return_value = None + replacement = MagicMock() + state = _PublisherState( + camera=old_camera, + publisher=MagicMock(), + camera_fps=30, + config={ + "camera": { + "class_path": FAKE_CAMERA_CLASS, + "init_args": { + "device_name": "trusted-device", + "backend": "trusted-backend", + "width": 320, + "height": 240, + "fps": 30, + }, + }, + "service_name": "physicalai/test/trusted/frame", + "_factory_override": "tests.unit.capture.fake:FakeCamera", + }, + ) + with patch( + "physicalai.capture.transport._publisher_worker.build_camera", + return_value=replacement, + ) as build: + result = _handle_reconfigure( + state, + {"kind": "RECONFIGURE", "settings": {"width": 640, "fps": 15}}, + "physicalai/test/trusted/frame", + ) - mock_create.assert_called_once_with("uvc", device=1, fps=30) + assert result == {"ok": True} + new_config = build.call_args.args[0] + assert new_config["camera"] == { + "class_path": FAKE_CAMERA_CLASS, + "init_args": { + "device_name": "trusted-device", + "backend": "trusted-backend", + "width": 640, + "height": 240, + "fps": 15, + }, + } + assert new_config["_factory_override"] == "tests.unit.capture.fake:FakeCamera" + old_camera.disconnect.assert_called_once_with() + replacement.connect.assert_called_once_with() + assert state.camera is replacement + assert state.camera_fps == 15 + + def test_worker_rejects_component_payload_before_side_effects(self) -> None: + from physicalai.capture.transport._publisher_worker import _PublisherState, _handle_reconfigure + + camera = MagicMock() + state = _PublisherState( + camera=camera, + publisher=MagicMock(), + camera_fps=30, + config={ + "camera": {"class_path": FAKE_CAMERA_CLASS, "init_args": {"width": 320}}, + "service_name": "physicalai/test/trusted/frame", + }, + ) + with patch("physicalai.capture.transport._publisher_worker.build_camera") as build: + result = _handle_reconfigure( + state, + { + "kind": "RECONFIGURE", + "spec": { + "camera": { + "class_path": "subprocess.Popen", + "init_args": {"args": ["touch", "/tmp/unsafe"]}, + }, + }, + }, + "physicalai/test/trusted/frame", + ) - def test_default_kwargs_empty_dict(self) -> None: - spec = CameraSpec("uvc") - assert spec.camera_kwargs == {} + assert result["ok"] is False + assert "invalid reconfigure request" in result["error"] + camera.disconnect.assert_not_called() + build.assert_not_called() class TestFrameHeader: + def test_sizeof_is_44(self) -> None: assert ctypes.sizeof(FrameHeader) == 44 assert HEADER_SIZE == ctypes.sizeof(FrameHeader) @@ -170,43 +345,168 @@ def test_fps_defaults_to_zero(self) -> None: class TestSharedCameraConstruction: - """Unit tests for SharedCamera constructor and from_publisher.""" + """Unit tests for SharedCamera constructor / from_config.""" - def test_constructor_with_camera_type(self) -> None: - cam = SharedCamera("uvc", device=0) - assert cam._camera_type == "uvc" + def test_from_config_derives_builtin_service_name(self) -> None: + cam = SharedCamera.from_config( + {"class_path": "physicalai.capture.UVCCamera", "init_args": {"device": 0}}, + ) + assert cam._camera is not None + assert cam._camera["class_path"] == "physicalai.capture.UVCCamera" assert cam._service_name == "physicalai/camera/uvc/0/frame" assert cam.device_id == "0" - def test_constructor_with_explicit_service_name(self) -> None: + def test_from_publisher(self) -> None: cam = SharedCamera.from_publisher("custom/name") + assert cam._camera is None assert cam._service_name == "custom/name" - def test_from_publisher(self) -> None: - cam = SharedCamera.from_publisher("physicalai/camera/uvc/0/frame") - assert cam._camera_type is None - assert cam._service_name == "physicalai/camera/uvc/0/frame" - def test_constructor_rejects_no_args(self) -> None: with pytest.raises(ValueError, match="must provide"): - SharedCamera(None) - - def test_constructor_rejects_service_name_as_type(self) -> None: - with pytest.raises(ValueError): - SharedCamera("physicalai/camera/uvc/0/frame") + SharedCamera() def test_default_device_zero(self) -> None: - cam = SharedCamera("uvc") - assert cam._service_name is not None + cam = SharedCamera.from_config( + {"class_path": "physicalai.capture.UVCCamera", "init_args": {}}, + ) assert cam._service_name.endswith("/0/frame") def test_serial_number_in_service_name(self) -> None: - cam = SharedCamera("realsense", serial_number="12345") - assert cam._service_name is not None - assert "12345" in cam._service_name + name = derive_service_name( + { + "class_path": "physicalai.capture.RealSenseCamera", + "init_args": {"serial_number": "12345"}, + }, + ) + assert name == "physicalai/camera/realsense/12345/frame" + + def test_basler_token_derivation(self) -> None: + name = derive_service_name( + { + "class_path": "physicalai.capture.BaslerCamera", + "init_args": {"serial_number": "abc"}, + }, + ) + assert name == "physicalai/camera/basler/abc/frame" + + def test_third_party_requires_explicit_service_name(self) -> None: + with pytest.raises(ValueError, match="requires an explicit service_name"): + SharedCamera.from_config( + {"class_path": FAKE_CAMERA_CLASS, "init_args": {"width": 64}}, + ) + + def test_ip_genicam_class_paths_require_explicit_service_name(self) -> None: + """Stub / phantom paths are not in the shareable builtin map.""" + for class_path in ( + "physicalai.capture.IPCamera", + "physicalai.capture.GenicamCamera", + ): + with pytest.raises(ValueError, match="requires an explicit service_name"): + derive_service_name({"class_path": class_path, "init_args": {"device": 0}}) + + def test_third_party_with_explicit_service_name(self) -> None: + cam = SharedCamera.from_config( + {"class_path": FAKE_CAMERA_CLASS, "init_args": {"width": 64}}, + service_name="physicalai/test/third/frame", + ) + assert cam._service_name == "physicalai/test/third/frame" + assert cam._camera is not None + assert cam._camera["class_path"] == FAKE_CAMERA_CLASS + + def test_envelope_service_name_not_in_init_args(self) -> None: + cam = SharedCamera.from_config( + {"class_path": "physicalai.capture.UVCCamera", "init_args": {"device": 1}}, + ) + assert cam._camera is not None + assert "service_name" not in cam._camera["init_args"] + assert derive_service_name(cam._camera) == cam._service_name + + def test_constructor_requires_component_config_mapping(self) -> None: + from tests.unit.capture.fake import FakeCamera + + driver = FakeCamera(width=32, height=32, device_name="d1") + with pytest.raises(ComponentConfigError, match="camera must be a ComponentConfig mapping"): + SharedCamera(camera=driver, service_name="physicalai/test/x/frame") # type: ignore[arg-type] + + def test_subscriber_never_imports_the_vendor_driver(self) -> None: + """A subscriber derives a service name without the vendor SDK installed. + + Runs in a fresh interpreter so the assertion is not masked by driver + modules other tests already imported. + """ + script = textwrap.dedent(""" + import sys + + from physicalai.capture import SharedCamera + + camera = SharedCamera( + camera={ + "class_path": "physicalai.capture.RealSenseCamera", + "init_args": {"serial_number": "0001"}, + }, + ) + assert camera._service_name == "physicalai/camera/realsense/0001/frame" + leaked = sorted( + name for name in sys.modules + if name.startswith("physicalai.capture.cameras.realsense") or name == "pyrealsense2" + ) + assert not leaked, f"subscriber imported the driver package: {leaked}" + """) + subprocess.run([sys.executable, "-c", script], check=True, timeout=120) + + def test_third_party_build_bypasses_create_camera_registry(self) -> None: + spec = CameraPublisherConfig( + camera={"class_path": FAKE_CAMERA_CLASS, "init_args": {"width": 16, "height": 16}}, + ) + with patch("physicalai.capture.factory.create_camera") as mock_create: + cam = spec.build() + mock_create.assert_not_called() + assert getattr(cam, "_width") == 16 + + def test_publisher_stdin_carries_concrete_service_name(self) -> None: + from physicalai.capture.transport._publisher import CameraPublisher + + captured: dict = {} + + class _FakeProc: + def __init__(self, *args: object, **kwargs: object) -> None: + assert "cwd" not in kwargs + self.stdin = MagicMock() + self.stdout = MagicMock() + self.stdout.readline.return_value = b"READY\n" + def _write(data: bytes) -> None: + captured.update(__import__("json").loads(data.decode())) + self.stdin.write.side_effect = _write + self.poll = MagicMock(return_value=0) + + def terminate(self) -> None: + return None + + def kill(self) -> None: + return None + + def wait(self, timeout: float | None = None) -> int: + return 0 + + spec = CameraPublisherConfig( + camera={"class_path": FAKE_CAMERA_CLASS, "init_args": {"width": 8}}, + ) + publisher = CameraPublisher(spec, "physicalai/test/svc/frame") + with ( + patch("physicalai.capture.transport._publisher.subprocess.Popen", _FakeProc), + patch("physicalai.capture.transport._publisher.select.select", return_value=([object()], [], [])), + ): + publisher.start(timeout=1.0) + + assert captured["service_name"] == "physicalai/test/svc/frame" + assert "camera" in captured + assert "camera_type" not in captured + assert "camera_kwargs" not in captured + assert "service_name" not in captured["camera"]["init_args"] class TestSharedCameraSpawnFlow: + """Unit tests for SharedCamera auto-spawn and race recovery flow.""" @staticmethod @@ -257,7 +557,9 @@ def test_connect_spawns_publisher_when_none_found( mock_publisher = MagicMock() mock_publisher_cls.return_value = mock_publisher - camera = SharedCamera("uvc", device=0) + camera = SharedCamera.from_config( + {"class_path": "physicalai.capture.UVCCamera", "init_args": {"device": 0}}, + ) with patch.object(camera, "_decode_sample", return_value=(MagicMock(), MagicMock())): camera.connect(timeout=0.1) @@ -279,7 +581,9 @@ def test_connect_skips_spawn_when_publisher_found( mock_import_module.return_value = iox2 mock_probe.return_value = True - camera = SharedCamera("uvc", device=0) + camera = SharedCamera.from_config( + {"class_path": "physicalai.capture.UVCCamera", "init_args": {"device": 0}}, + ) with patch.object(camera, "_decode_sample", return_value=(MagicMock(), MagicMock())): camera.connect(timeout=0.1) @@ -304,7 +608,9 @@ def test_connect_race_recovery( mock_publisher.start.side_effect = RuntimeError("publisher already running") mock_publisher_cls.return_value = mock_publisher - camera = SharedCamera("uvc", device=0) + camera = SharedCamera.from_config( + {"class_path": "physicalai.capture.UVCCamera", "init_args": {"device": 0}}, + ) with patch.object(camera, "_decode_sample", return_value=(MagicMock(), MagicMock())): camera.connect(timeout=0.1) @@ -315,7 +621,9 @@ def test_connect_race_recovery( mock_publisher.start.assert_called_once_with() def test_disconnect_stops_spawned_publisher(self) -> None: - camera = SharedCamera("uvc", device=0) + camera = SharedCamera.from_config( + {"class_path": "physicalai.capture.UVCCamera", "init_args": {"device": 0}}, + ) spawned_publisher = MagicMock() camera._publisher = spawned_publisher camera._connected = True @@ -365,7 +673,11 @@ def test_validate_on_connect_raises_on_resolution_mismatch( mock_import_module.return_value = iox2 mock_probe.return_value = True # publisher exists, no spawn - camera = SharedCamera("uvc", device=0, width=640, height=480, validate_on_connect=True) + camera = SharedCamera.from_config( + {"class_path": "physicalai.capture.UVCCamera", + "init_args": {"device": 0, "width": 640, "height": 480}}, + validate_on_connect=True, + ) with patch.object(camera, "_decode_sample", return_value=self._header_frame(1920, 1080)): with pytest.raises(CaptureError, match="does not match"): camera.connect(timeout=0.1) @@ -387,7 +699,11 @@ def test_validate_on_connect_spawned_publisher_mismatch( mock_import_module.return_value = iox2 mock_probe.return_value = False - camera = SharedCamera("uvc", device=0, width=640, height=480, validate_on_connect=True) + camera = SharedCamera.from_config( + {"class_path": "physicalai.capture.UVCCamera", + "init_args": {"device": 0, "width": 640, "height": 480}}, + validate_on_connect=True, + ) with patch.object(camera, "_decode_sample", return_value=self._header_frame(1920, 1080)): with pytest.raises(CaptureError, match="does not match"): camera.connect(timeout=0.1) @@ -407,7 +723,11 @@ def test_no_validate_on_connect_warns_on_mismatch_and_attaches( mock_import_module.return_value = iox2 mock_probe.return_value = True - camera = SharedCamera("uvc", device=0, width=640, height=480, validate_on_connect=False) + camera = SharedCamera.from_config( + {"class_path": "physicalai.capture.UVCCamera", + "init_args": {"device": 0, "width": 640, "height": 480}}, + validate_on_connect=False, + ) with patch.object(camera, "_decode_sample", return_value=self._header_frame(1920, 1080)): camera.connect(timeout=0.1) @@ -425,7 +745,11 @@ def test_validate_on_connect_silent_on_match( mock_import_module.return_value = iox2 mock_probe.return_value = True - camera = SharedCamera("uvc", device=0, width=640, height=480, validate_on_connect=True) + camera = SharedCamera.from_config( + {"class_path": "physicalai.capture.UVCCamera", + "init_args": {"device": 0, "width": 640, "height": 480}}, + validate_on_connect=True, + ) with patch.object(camera, "_decode_sample", return_value=self._header_frame(640, 480)): camera.connect(timeout=0.1) @@ -443,7 +767,10 @@ def test_no_dimensions_requested_skips_check( mock_import_module.return_value = iox2 mock_probe.return_value = True - camera = SharedCamera("uvc", device=0, validate_on_connect=True) + camera = SharedCamera.from_config( + {"class_path": "physicalai.capture.UVCCamera", "init_args": {"device": 0}}, + validate_on_connect=True, + ) with patch.object(camera, "_decode_sample", return_value=self._header_frame(1920, 1080)): camera.connect(timeout=0.1) @@ -461,7 +788,11 @@ def test_fps_mismatch_validate_on_connect_raises( mock_import_module.return_value = iox2 mock_probe.return_value = True - camera = SharedCamera("uvc", device=0, width=640, height=480, fps=30, validate_on_connect=True) + camera = SharedCamera.from_config( + {"class_path": "physicalai.capture.UVCCamera", + "init_args": {"device": 0, "width": 640, "height": 480, "fps": 30}}, + validate_on_connect=True, + ) with patch.object(camera, "_decode_sample", return_value=self._header_frame(640, 480, fps=60)): with pytest.raises(CaptureError, match="does not match"): camera.connect(timeout=0.1) @@ -481,7 +812,11 @@ def test_no_validate_on_connect_warns_once_not_repeatedly( mock_import_module.return_value = iox2 mock_probe.return_value = True - camera = SharedCamera("uvc", device=0, width=640, height=480, validate_on_connect=False) + camera = SharedCamera.from_config( + {"class_path": "physicalai.capture.UVCCamera", + "init_args": {"device": 0, "width": 640, "height": 480}}, + validate_on_connect=False, + ) hf = self._header_frame(1920, 1080) with patch.object(camera, "_decode_sample", return_value=hf): camera.connect(timeout=0.1) @@ -531,11 +866,9 @@ def test_overwrite_reconfigure_success( mock_import_module.return_value = iox2 mock_probe.return_value = True - camera = SharedCamera( - "uvc", - device=0, - width=640, - height=480, + camera = SharedCamera.from_config( + {"class_path": "physicalai.capture.UVCCamera", + "init_args": {"device": 0, "width": 640, "height": 480}}, validate_on_connect=True, overwrite_settings=True, ) @@ -559,11 +892,9 @@ def test_overwrite_validate_on_connect_reconfigure_failure_raises( mock_import_module.return_value = iox2 mock_probe.return_value = True - camera = SharedCamera( - "uvc", - device=0, - width=640, - height=480, + camera = SharedCamera.from_config( + {"class_path": "physicalai.capture.UVCCamera", + "init_args": {"device": 0, "width": 640, "height": 480}}, validate_on_connect=True, overwrite_settings=True, ) @@ -592,11 +923,9 @@ def test_overwrite_no_validate_on_connect_reconfigure_failure_warns( mock_import_module.return_value = iox2 mock_probe.return_value = True - camera = SharedCamera( - "uvc", - device=0, - width=640, - height=480, + camera = SharedCamera.from_config( + {"class_path": "physicalai.capture.UVCCamera", + "init_args": {"device": 0, "width": 640, "height": 480}}, validate_on_connect=False, overwrite_settings=True, ) @@ -624,11 +953,9 @@ def test_no_control_service_validate_on_connect_raises( mock_import_module.return_value = iox2 mock_probe.return_value = True - camera = SharedCamera( - "uvc", - device=0, - width=640, - height=480, + camera = SharedCamera.from_config( + {"class_path": "physicalai.capture.UVCCamera", + "init_args": {"device": 0, "width": 640, "height": 480}}, validate_on_connect=True, overwrite_settings=True, ) @@ -657,11 +984,9 @@ def test_no_control_service_no_validate_on_connect_warns( mock_import_module.return_value = iox2 mock_probe.return_value = True - camera = SharedCamera( - "uvc", - device=0, - width=640, - height=480, + camera = SharedCamera.from_config( + {"class_path": "physicalai.capture.UVCCamera", + "init_args": {"device": 0, "width": 640, "height": 480}}, validate_on_connect=False, overwrite_settings=True, ) @@ -689,11 +1014,9 @@ def test_reconfigure_only_attempted_once( mock_import_module.return_value = iox2 mock_probe.return_value = True - camera = SharedCamera( - "uvc", - device=0, - width=640, - height=480, + camera = SharedCamera.from_config( + {"class_path": "physicalai.capture.UVCCamera", + "init_args": {"device": 0, "width": 640, "height": 480}}, validate_on_connect=False, overwrite_settings=True, ) @@ -714,7 +1037,7 @@ def test_reconfigure_only_attempted_once( @requires_iceoryx2 class TestCameraPublisher: - def test_start_stop_lifecycle(self, fake_camera_spec: CameraSpec) -> None: + def test_start_stop_lifecycle(self, fake_camera_spec: CameraPublisherConfig) -> None: from physicalai.capture.transport._publisher import CameraPublisher publisher = CameraPublisher( @@ -727,7 +1050,7 @@ def test_start_stop_lifecycle(self, fake_camera_spec: CameraSpec) -> None: publisher.stop() assert not publisher.is_alive - def test_context_manager(self, fake_camera_spec: CameraSpec) -> None: + def test_context_manager(self, fake_camera_spec: CameraPublisherConfig) -> None: from physicalai.capture.transport._publisher import CameraPublisher with CameraPublisher( @@ -741,8 +1064,11 @@ def test_context_manager(self, fake_camera_spec: CameraSpec) -> None: def test_start_failure_propagates(self) -> None: from physicalai.capture.transport._publisher import CameraPublisher - bad_spec = CameraSpec(camera_type="does-not-exist", camera_kwargs={}) - publisher = CameraPublisher(bad_spec, _service_name()) + publisher = CameraPublisher( + CameraPublisherConfig(camera={"class_path": FAKE_CAMERA_CLASS, "init_args": {}}), + _service_name(), + _factory_override="tests.unit.capture.fake:DoesNotExist", + ) with pytest.raises(CaptureError, match="failed"): publisher.start(timeout=2.0) @@ -832,7 +1158,9 @@ def test_reconfigure_resolution_change(self) -> None: from physicalai.capture.transport._publisher import CameraPublisher service_name = f"physicalai/test/{uuid4().hex[:8]}/frame" - spec = CameraSpec(camera_type="fake", camera_kwargs={"width": 320, "height": 240}) + spec = CameraPublisherConfig( + camera={"class_path": FAKE_CAMERA_CLASS, "init_args": {"width": 320, "height": 240}}, + ) publisher = CameraPublisher( spec, service_name, @@ -847,8 +1175,10 @@ def test_reconfigure_resolution_change(self) -> None: assert frame.data.shape == (240, 320, 3) camera._overwrite_settings = True - camera._camera_type = None - camera._camera_kwargs = {"width": 640, "height": 480} + camera._camera = { + "class_path": FAKE_CAMERA_CLASS, + "init_args": {"width": 640, "height": 480}, + } result = camera._request_reconfigure(timeout=5.0) assert result["ok"] is True @@ -868,7 +1198,9 @@ def test_reconfigure_failure_restores_old(self) -> None: from physicalai.capture.transport._publisher import CameraPublisher service_name = f"physicalai/test/{uuid4().hex[:8]}/frame" - spec = CameraSpec(camera_type="fake", camera_kwargs={"width": 320, "height": 240}) + spec = CameraPublisherConfig( + camera={"class_path": FAKE_CAMERA_CLASS, "init_args": {"width": 320, "height": 240}}, + ) publisher = CameraPublisher( spec, service_name, @@ -877,11 +1209,14 @@ def test_reconfigure_failure_restores_old(self) -> None: publisher.start(timeout=10.0) try: - camera = SharedCamera.from_publisher(service_name) + camera = SharedCamera.from_config( + {"class_path": FAKE_CAMERA_CLASS, "init_args": {"width": 320, "height": 240}}, + service_name=service_name, + ) camera.connect(timeout=5.0) result = camera._request_reconfigure(timeout=5.0) - assert result["ok"] is False or result["ok"] is True + assert result["ok"] is True frame = camera.read(timeout=5.0) assert frame.data.shape[0] > 0 @@ -890,13 +1225,20 @@ def test_reconfigure_failure_restores_old(self) -> None: publisher.stop() def test_no_control_service_on_v1_publisher(self) -> None: - camera = SharedCamera.from_publisher(f"physicalai/test/{uuid4().hex[:8]}/frame") + camera = SharedCamera.from_config( + {"class_path": FAKE_CAMERA_CLASS, "init_args": {"width": 640}}, + service_name=f"physicalai/test/{uuid4().hex[:8]}/frame", + ) camera._overwrite_settings = True - camera._camera_kwargs = {"width": 640} with pytest.raises(CaptureError, match="does not support reconfigure"): camera._request_reconfigure(timeout=1.0) + def test_attach_only_reconfigure_requires_camera_config(self) -> None: + camera = SharedCamera.from_publisher(f"physicalai/test/{uuid4().hex[:8]}/frame") + with pytest.raises(CaptureError, match="requires a camera ComponentConfig"): + camera._request_reconfigure(timeout=1.0) + def test_end_to_end_overwrite_on_connect(self) -> None: """Full flow: overwrite_settings triggers auto-reconfigure during connect.""" import time @@ -905,7 +1247,9 @@ def test_end_to_end_overwrite_on_connect(self) -> None: service_name = f"physicalai/test/{uuid4().hex[:8]}/frame" # Publisher starts serving 320×240 - spec = CameraSpec(camera_type="fake", camera_kwargs={"width": 320, "height": 240}) + spec = CameraPublisherConfig( + camera={"class_path": FAKE_CAMERA_CLASS, "init_args": {"width": 320, "height": 240}}, + ) publisher = CameraPublisher( spec, service_name, @@ -915,11 +1259,9 @@ def test_end_to_end_overwrite_on_connect(self) -> None: try: # Subscriber requests 640×480 — connect should auto-reconfigure publisher - camera = SharedCamera( - None, + camera = SharedCamera.from_config( + {"class_path": FAKE_CAMERA_CLASS, "init_args": {"width": 640, "height": 480}}, service_name=service_name, - width=640, - height=480, overwrite_settings=True, ) camera.connect(timeout=10.0) @@ -933,3 +1275,30 @@ def test_end_to_end_overwrite_on_connect(self) -> None: camera.disconnect() finally: publisher.stop() + + +class TestPublisherOpenErrorMessaging: + """Busy-device clarification for publisher open failures.""" + + def test_ebusy_is_detectable(self) -> None: + import errno + + from physicalai.capture.transport._publisher_worker import ( + _format_camera_open_error, + _looks_like_device_busy, + ) + + busy = OSError(errno.EBUSY, "Device or resource busy") + assert _looks_like_device_busy(busy) + msg = _format_camera_open_error(busy) + assert "opens the camera exclusively" in msg + + def test_generic_error_unchanged(self) -> None: + from physicalai.capture.transport._publisher_worker import ( + _format_camera_open_error, + _looks_like_device_busy, + ) + + exc = RuntimeError("no such device") + assert not _looks_like_device_busy(exc) + assert _format_camera_open_error(exc) == "RuntimeError: no such device" From 96ca62bfbef3fe93403dd69f9a753ed16b67066e Mon Sep 17 00:00:00 2001 From: Max Xiang Date: Tue, 28 Jul 2026 13:44:57 +0200 Subject: [PATCH 2/5] fix tests --- tests/unit/capture/test_factory.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/tests/unit/capture/test_factory.py b/tests/unit/capture/test_factory.py index 5959036f..0d880655 100644 --- a/tests/unit/capture/test_factory.py +++ b/tests/unit/capture/test_factory.py @@ -77,9 +77,13 @@ def test_static_table_matches_drivers(self, token: str) -> None: defining = f"{expected.__module__}.{expected.__qualname__}" assert defining == paths[-1] for alias in paths[1:]: - resolved = import_dotted_path(alias) - assert isinstance(resolved, type) - assert f"{resolved.__module__}.{resolved.__qualname__}" == defining + try: + resolved = import_dotted_path(alias) + except (ImportError, AttributeError) as exc: # optional camera extra absent + pytest.skip(f"{token} driver is not installed: {exc}") + else: + assert isinstance(resolved, type) + assert f"{resolved.__module__}.{resolved.__qualname__}" == defining @pytest.mark.parametrize("token", ["uvc", "realsense", "basler"]) def test_every_spelling_maps_back_to_one_token(self, token: str) -> None: From f2fc62ae526e09d14fe21d452071c24ee9d0eb5a Mon Sep 17 00:00:00 2001 From: Max Xiang Date: Tue, 28 Jul 2026 13:59:20 +0200 Subject: [PATCH 3/5] fix tests --- tests/unit/capture/test_transport.py | 57 +++++++++++++--------------- 1 file changed, 27 insertions(+), 30 deletions(-) diff --git a/tests/unit/capture/test_transport.py b/tests/unit/capture/test_transport.py index a5cf1e17..e2e40e51 100644 --- a/tests/unit/capture/test_transport.py +++ b/tests/unit/capture/test_transport.py @@ -232,6 +232,33 @@ def test_worker_rejects_component_payload_before_side_effects(self) -> None: camera.disconnect.assert_not_called() build.assert_not_called() + def test_reconfigure_failure_restores_old(self) -> None: + from physicalai.capture.transport._publisher_worker import _PublisherState, _handle_reconfigure + + old_camera = MagicMock() + restored = MagicMock() + config = { + "camera": {"class_path": FAKE_CAMERA_CLASS, "init_args": {"width": 320, "fps": 30}}, + "service_name": "physicalai/test/restore/frame", + } + state = _PublisherState(camera=old_camera, publisher=MagicMock(), camera_fps=30, config=config) + + with patch( + "physicalai.capture.transport._publisher_worker.build_camera", + side_effect=[RuntimeError("simulated open failure"), restored], + ): + result = _handle_reconfigure( + state, + {"kind": "RECONFIGURE", "settings": {"width": 640}}, + "physicalai/test/restore/frame", + ) + + assert result == {"ok": False, "error": "RuntimeError: simulated open failure"} + assert state.camera is restored + assert state.config == config + assert state.camera_fps == 30 + restored.connect.assert_called_once_with() + class TestFrameHeader: @@ -1194,36 +1221,6 @@ def test_reconfigure_resolution_change(self) -> None: finally: publisher.stop() - def test_reconfigure_failure_restores_old(self) -> None: - from physicalai.capture.transport._publisher import CameraPublisher - - service_name = f"physicalai/test/{uuid4().hex[:8]}/frame" - spec = CameraPublisherConfig( - camera={"class_path": FAKE_CAMERA_CLASS, "init_args": {"width": 320, "height": 240}}, - ) - publisher = CameraPublisher( - spec, - service_name, - _factory_override="tests.unit.capture.fake:FakeCamera", - ) - publisher.start(timeout=10.0) - - try: - camera = SharedCamera.from_config( - {"class_path": FAKE_CAMERA_CLASS, "init_args": {"width": 320, "height": 240}}, - service_name=service_name, - ) - camera.connect(timeout=5.0) - - result = camera._request_reconfigure(timeout=5.0) - assert result["ok"] is True - - frame = camera.read(timeout=5.0) - assert frame.data.shape[0] > 0 - camera.disconnect() - finally: - publisher.stop() - def test_no_control_service_on_v1_publisher(self) -> None: camera = SharedCamera.from_config( {"class_path": FAKE_CAMERA_CLASS, "init_args": {"width": 640}}, From 831deeb1fdd485dfaf8e8caf1ebfd07f11e29e2a Mon Sep 17 00:00:00 2001 From: Max Xiang Date: Tue, 28 Jul 2026 14:38:03 +0200 Subject: [PATCH 4/5] remove builtin.py --- docs/explanation/cameras.md | 5 + examples/capture/shared_camera.ipynb | 2 +- src/physicalai/capture/factory.py | 18 ++-- .../capture/transport/_shared_camera.py | 2 +- src/physicalai/capture/transport/_spec.py | 26 ++---- src/physicalai/capture/transport/builtin.py | 91 ------------------- .../capture/test_camera_component_config.py | 10 +- tests/unit/capture/test_factory.py | 71 +++------------ tests/unit/capture/test_transport.py | 39 ++++---- 9 files changed, 66 insertions(+), 198 deletions(-) delete mode 100644 src/physicalai/capture/transport/builtin.py diff --git a/docs/explanation/cameras.md b/docs/explanation/cameras.md index c06fb5af..de185a1c 100644 --- a/docs/explanation/cameras.md +++ b/docs/explanation/cameras.md @@ -61,4 +61,9 @@ frame = shared.read_latest() `create_camera(..., shared=True)` remains a convenience for shareable built-ins (`uvc`, `realsense`, `basler`) that packs a type into `SharedCamera(camera=...)`. +`from_config()` derives `service_name` as +`physicalai/camera///frame` for any `class_path`, +including third-party drivers. Pass `service_name=` explicitly if two camera +classes share a class name and a device id. + `SharedCamera` is the recommended approach for production deployments where multiple consumers need camera frames. It avoids the need for manual synchronization and handles frame distribution efficiently. diff --git a/examples/capture/shared_camera.ipynb b/examples/capture/shared_camera.ipynb index eb16dc55..bcb19b52 100644 --- a/examples/capture/shared_camera.ipynb +++ b/examples/capture/shared_camera.ipynb @@ -219,7 +219,7 @@ "outputs": [], "source": [ "# A second script / process can attach to the same publisher.\n", - "cam = SharedCamera.from_publisher(\"physicalai/camera/uvc/0/frame\", zero_copy=True)\n", + "cam = SharedCamera.from_publisher(\"physicalai/camera/UVCCamera/0/frame\", zero_copy=True)\n", "cam.connect()\n", "frame = cam.read_latest()\n", "print(f\"Attached to existing publisher, seq={frame.sequence} shape={frame.data.shape}\")" diff --git a/src/physicalai/capture/factory.py b/src/physicalai/capture/factory.py index 3a901c32..fdc054fe 100644 --- a/src/physicalai/capture/factory.py +++ b/src/physicalai/capture/factory.py @@ -23,6 +23,16 @@ "color_mode", }) +# CameraType token → public class_path for ``create_camera(..., shared=True)``. +# Static by necessity: subscriber hosts have no vendor SDK, so nothing here may +# import a driver. Stub types (ip, genicam) are absent so shared spawn cannot +# claim phantom coverage; test_factory.py checks values against @export_config. +_SHAREABLE_CLASS_PATHS: dict[str, str] = { + "uvc": "physicalai.capture.UVCCamera", + "realsense": "physicalai.capture.RealSenseCamera", + "basler": "physicalai.capture.BaslerCamera", +} + def create_camera(camera_type: str, *, shared: bool = False, **kwargs: Any) -> Camera: # noqa: ANN401 """Create a camera by type name. @@ -56,15 +66,11 @@ def create_camera(camera_type: str, *, shared: bool = False, **kwargs: Any) -> C if shared: from physicalai.capture.transport import SharedCamera # noqa: PLC0415 - from physicalai.capture.transport.builtin import ( # noqa: PLC0415 - builtin_class_path_for_type, - builtin_shared_type_tokens, - ) - class_path = builtin_class_path_for_type(camera_type) + class_path = _SHAREABLE_CLASS_PATHS.get(camera_type) if class_path is None: if camera_type in {t.value for t in CameraType}: - shareable = ", ".join(sorted(builtin_shared_type_tokens())) + shareable = ", ".join(sorted(_SHAREABLE_CLASS_PATHS)) msg = ( f"camera type {camera_type!r} does not support shared=True " f"(no shareable driver for service-name derivation); " diff --git a/src/physicalai/capture/transport/_shared_camera.py b/src/physicalai/capture/transport/_shared_camera.py index 285bec41..70301a1f 100644 --- a/src/physicalai/capture/transport/_shared_camera.py +++ b/src/physicalai/capture/transport/_shared_camera.py @@ -658,7 +658,7 @@ def service_name(self) -> str: @property def device_id(self) -> str: - # service_name format: physicalai/camera///frame + # service_name format: physicalai/camera///frame parts = self._service_name.split("/") if ( len(parts) >= _SERVICE_NAME_EXPECTED_PARTS diff --git a/src/physicalai/capture/transport/_spec.py b/src/physicalai/capture/transport/_spec.py index 6e41594e..be9d1cd9 100644 --- a/src/physicalai/capture/transport/_spec.py +++ b/src/physicalai/capture/transport/_spec.py @@ -28,8 +28,6 @@ validate_envelope, ) -from .builtin import builtin_type_for_class_path - if TYPE_CHECKING: from physicalai.capture.camera import Camera @@ -132,11 +130,10 @@ def derive_service_name( ) -> str: """Resolve iceoryx2 ``service_name`` for a camera ComponentConfig. - Built-in class paths derive ``physicalai/camera/{token}/{device_id}/frame`` - via the transport class-path → type-token map, which accepts both the - public re-export and the internal spelling of each built-in driver. - Third-party / unknown class paths (including stub types without a shared - registry entry) require an explicit *service_name*. + Derives ``physicalai/camera/{class_name}/{device_id}/frame`` from the + terminal segment of ``class_path``, without importing it. The bare class + name collapses every spelling of one driver onto one publisher; distinct + classes sharing a name and device id need an explicit *service_name*. Args: camera: Normalized or raw camera ComponentConfig. @@ -144,22 +141,11 @@ def derive_service_name( Returns: Concrete service name for the publisher envelope. - - Raises: - ValueError: If *service_name* is omitted for a non-built-in class_path. """ if service_name is not None: return service_name - class_path = str(camera["class_path"]) - token = builtin_type_for_class_path(class_path) - if token is None: - msg = ( - f"camera {class_path!r} requires an explicit service_name; " - "built-in derivation only covers shareable physicalai.capture " - "backends (uvc, realsense, basler)" - ) - raise ValueError(msg) + class_name = str(camera["class_path"]).rsplit(".", 1)[-1] init_args = camera.get("init_args", {}) if not isinstance(init_args, Mapping): @@ -169,7 +155,7 @@ def derive_service_name( # the same service name for the same physical device. if isinstance(device_id, str) and device_id.startswith("/dev/"): device_id = Path(device_id).resolve().name - return f"physicalai/camera/{token}/{device_id}/frame" + return f"physicalai/camera/{class_name}/{device_id}/frame" @dataclass(frozen=True) diff --git a/src/physicalai/capture/transport/builtin.py b/src/physicalai/capture/transport/builtin.py deleted file mode 100644 index e0fff4e6..00000000 --- a/src/physicalai/capture/transport/builtin.py +++ /dev/null @@ -1,91 +0,0 @@ -# Copyright (C) 2026 Intel Corporation -# SPDX-License-Identifier: Apache-2.0 - -"""Built-in camera type token ↔ public class_path for SharedCamera. - -Single source of truth for service-name derivation and -``create_camera(..., shared=True)``. Only backends with a real driver -package under ``cameras/`` are listed — IP/Genicam stubs are intentionally -absent so shared spawn cannot claim phantom coverage. - -The table is static on purpose: a subscriber process derives a service name -without the vendor SDK installed, so nothing here may import a driver -module. ``tests/unit/capture/test_factory.py`` asserts the table still -matches each driver's ``@export_config(class_path=...)`` when the optional -camera extras are available. -""" - -from __future__ import annotations - -# token → accepted class paths. Index 0 is the canonical public path declared -# by ``@export_config(class_path=...)`` on the driver; the rest are the -# sub-package re-export and the defining module, so a hand-written config that -# spells a driver the internal way still derives the same service name instead -# of demanding an explicit one. -_BUILTIN_SHARED: dict[str, tuple[str, ...]] = { - "uvc": ( - "physicalai.capture.UVCCamera", - "physicalai.capture.cameras.uvc.UVCCamera", - "physicalai.capture.cameras.uvc._camera.UVCCamera", - ), - "realsense": ( - "physicalai.capture.RealSenseCamera", - "physicalai.capture.cameras.realsense.RealSenseCamera", - "physicalai.capture.cameras.realsense._camera.RealSenseCamera", - ), - "basler": ( - "physicalai.capture.BaslerCamera", - "physicalai.capture.cameras.basler.BaslerCamera", - "physicalai.capture.cameras.basler._camera.BaslerCamera", - ), -} - -_TOKEN_TO_CLASS_PATH: dict[str, str] = {token: paths[0] for token, paths in _BUILTIN_SHARED.items()} -_CLASS_PATH_TO_TOKEN: dict[str, str] = {path: token for token, paths in _BUILTIN_SHARED.items() for path in paths} - - -def builtin_shared_type_tokens() -> frozenset[str]: - """Return CameraType tokens that support shared service-name derivation.""" - return frozenset(_BUILTIN_SHARED) - - -def builtin_class_paths_for_type(token: str) -> tuple[str, ...]: - r"""Return every class_path spelling accepted for a built-in type token. - - Args: - token: Lowercase ``CameraType`` value (e.g. ``\"uvc\"``). - - Returns: - Accepted paths with the canonical public one first, or an empty tuple - if the token is not a shareable built-in. - """ - return _BUILTIN_SHARED.get(token, ()) - - -def builtin_class_path_for_type(token: str) -> str | None: - r"""Return the canonical public class_path for a built-in shared camera type token. - - Args: - token: Lowercase ``CameraType`` value (e.g. ``\"uvc\"``). - - Returns: - Public ``class_path``, or ``None`` if the token is not a shareable built-in. - """ - return _TOKEN_TO_CLASS_PATH.get(token) - - -def builtin_type_for_class_path(class_path: str) -> str | None: - """Return the legacy type token for a built-in camera class_path. - - Accepts the canonical public path plus the internal spellings listed in - :data:`_BUILTIN_SHARED`, so the same physical device derives one service - name however the config names its driver. - - Args: - class_path: Camera ``class_path`` as written in the config. - - Returns: - Type token (``uvc`` / ``realsense`` / ``basler``), or ``None`` for - third-party / stub / unknown paths. - """ - return _CLASS_PATH_TO_TOKEN.get(class_path) diff --git a/tests/unit/capture/test_camera_component_config.py b/tests/unit/capture/test_camera_component_config.py index c8458dab..054ddccd 100644 --- a/tests/unit/capture/test_camera_component_config.py +++ b/tests/unit/capture/test_camera_component_config.py @@ -148,18 +148,18 @@ def test_defining_module_camera_path_round_trips_as_given(self) -> None: wire = _assert_construction_round_trip(camera) nested = wire["init_args"]["camera"] assert isinstance(nested, dict) - # Stored as written; the class-path alias table still derives one service name. + # Stored as written; the terminal class name still derives one service name. assert nested["class_path"] == defining - assert camera._service_name == "physicalai/camera/uvc/0/frame" + assert camera._service_name == "physicalai/camera/UVCCamera/0/frame" def test_attach_only_round_trip(self) -> None: from physicalai.capture import SharedCamera - camera = SharedCamera(camera=None, service_name="physicalai/camera/uvc/0/frame") + camera = SharedCamera(camera=None, service_name="physicalai/camera/UVCCamera/0/frame") wire = _assert_construction_round_trip(camera) assert wire["class_path"] == "physicalai.capture.SharedCamera" assert wire["init_args"]["camera"] is None - assert wire["init_args"]["service_name"] == "physicalai/camera/uvc/0/frame" + assert wire["init_args"]["service_name"] == "physicalai/camera/UVCCamera/0/frame" def test_nested_recipe_is_not_instantiated(self) -> None: from physicalai.capture import SharedCamera @@ -171,7 +171,7 @@ def test_nested_recipe_is_not_instantiated(self) -> None: "class_path": "physicalai.capture.UVCCamera", "init_args": {"device": 0, "backend": "v4l2"}, }, - "service_name": "physicalai/camera/uvc/0/frame", + "service_name": "physicalai/camera/UVCCamera/0/frame", }, } restored = instantiate(config) diff --git a/tests/unit/capture/test_factory.py b/tests/unit/capture/test_factory.py index 0d880655..42825eb5 100644 --- a/tests/unit/capture/test_factory.py +++ b/tests/unit/capture/test_factory.py @@ -6,13 +6,7 @@ import pytest from physicalai.capture.discovery import discover_all -from physicalai.capture.factory import create_camera -from physicalai.capture.transport.builtin import ( - builtin_class_path_for_type, - builtin_class_paths_for_type, - builtin_shared_type_tokens, - builtin_type_for_class_path, -) +from physicalai.capture.factory import _SHAREABLE_CLASS_PATHS, create_camera class TestCreateCamera: @@ -37,64 +31,27 @@ def test_shared_stub_types_rejected(self) -> None: def test_shared_builtin_uses_registry_class_path(self) -> None: cam = create_camera("uvc", shared=True, device=0, backend="v4l2") assert cam._camera is not None # type: ignore[attr-defined] - assert cam._camera["class_path"] == builtin_class_path_for_type("uvc") # type: ignore[attr-defined] - assert cam._service_name == "physicalai/camera/uvc/0/frame" # type: ignore[attr-defined] + assert cam._camera["class_path"] == _SHAREABLE_CLASS_PATHS["uvc"] # type: ignore[attr-defined] + assert cam._service_name == "physicalai/camera/UVCCamera/0/frame" # type: ignore[attr-defined] -class TestBuiltinSharedRegistry: - """Single source of truth for shareable type ↔ class_path.""" +class TestShareableClassPaths: + """The static token → class_path table must track the drivers it names.""" - def test_shareable_tokens(self) -> None: - assert builtin_shared_type_tokens() == frozenset({"uvc", "realsense", "basler"}) - - def test_round_trip_uvc(self) -> None: - path = builtin_class_path_for_type("uvc") - assert path == "physicalai.capture.UVCCamera" - assert builtin_type_for_class_path(path) == "uvc" - - def test_matches_export_config_when_importable(self) -> None: - from physicalai.capture.cameras.uvc import UVCCamera - from physicalai.config import resolve_public_class_path - - assert builtin_class_path_for_type("uvc") == resolve_public_class_path(UVCCamera) - - @pytest.mark.parametrize("token", ["uvc", "realsense", "basler"]) - def test_static_table_matches_drivers(self, token: str) -> None: - """Every listed spelling must reach the driver the decorator declares.""" + @pytest.mark.parametrize("token", sorted(_SHAREABLE_CLASS_PATHS)) + def test_matches_export_config_when_importable(self, token: str) -> None: + """Keeps the hand-written table honest when the extra is installed.""" from physicalai.config import import_dotted_path, resolve_public_class_path - paths = builtin_class_paths_for_type(token) - expected: object = None + class_path = _SHAREABLE_CLASS_PATHS[token] + driver: object = None try: - expected = import_dotted_path(paths[0]) + driver = import_dotted_path(class_path) except (ImportError, AttributeError) as exc: # optional camera extra absent pytest.skip(f"{token} driver is not installed: {exc}") - assert isinstance(expected, type) - assert resolve_public_class_path(expected) == paths[0] - - # Compare by defining name rather than identity: other tests reload - # driver modules with a mocked SDK, which replaces the class object. - defining = f"{expected.__module__}.{expected.__qualname__}" - assert defining == paths[-1] - for alias in paths[1:]: - try: - resolved = import_dotted_path(alias) - except (ImportError, AttributeError) as exc: # optional camera extra absent - pytest.skip(f"{token} driver is not installed: {exc}") - else: - assert isinstance(resolved, type) - assert f"{resolved.__module__}.{resolved.__qualname__}" == defining - - @pytest.mark.parametrize("token", ["uvc", "realsense", "basler"]) - def test_every_spelling_maps_back_to_one_token(self, token: str) -> None: - for path in builtin_class_paths_for_type(token): - assert builtin_type_for_class_path(path) == token - - def test_no_phantom_ip_genicam(self) -> None: - assert builtin_class_path_for_type("ip") is None - assert builtin_class_path_for_type("genicam") is None - assert builtin_type_for_class_path("physicalai.capture.IPCamera") is None - assert builtin_type_for_class_path("physicalai.capture.GenicamCamera") is None + assert isinstance(driver, type) + assert resolve_public_class_path(driver) == class_path + class TestDiscoverAll: diff --git a/tests/unit/capture/test_transport.py b/tests/unit/capture/test_transport.py index e2e40e51..eb65bef5 100644 --- a/tests/unit/capture/test_transport.py +++ b/tests/unit/capture/test_transport.py @@ -380,7 +380,7 @@ def test_from_config_derives_builtin_service_name(self) -> None: ) assert cam._camera is not None assert cam._camera["class_path"] == "physicalai.capture.UVCCamera" - assert cam._service_name == "physicalai/camera/uvc/0/frame" + assert cam._service_name == "physicalai/camera/UVCCamera/0/frame" assert cam.device_id == "0" def test_from_publisher(self) -> None: @@ -405,7 +405,7 @@ def test_serial_number_in_service_name(self) -> None: "init_args": {"serial_number": "12345"}, }, ) - assert name == "physicalai/camera/realsense/12345/frame" + assert name == "physicalai/camera/RealSenseCamera/12345/frame" def test_basler_token_derivation(self) -> None: name = derive_service_name( @@ -414,22 +414,27 @@ def test_basler_token_derivation(self) -> None: "init_args": {"serial_number": "abc"}, }, ) - assert name == "physicalai/camera/basler/abc/frame" + assert name == "physicalai/camera/BaslerCamera/abc/frame" - def test_third_party_requires_explicit_service_name(self) -> None: - with pytest.raises(ValueError, match="requires an explicit service_name"): - SharedCamera.from_config( - {"class_path": FAKE_CAMERA_CLASS, "init_args": {"width": 64}}, + def test_third_party_derives_service_name(self) -> None: + """Derivation covers any class_path, not just shareable built-ins.""" + cam = SharedCamera.from_config( + {"class_path": FAKE_CAMERA_CLASS, "init_args": {"device": 2, "width": 64}}, + ) + assert cam._service_name == "physicalai/camera/FakeCamera/2/frame" + assert cam.device_id == "2" + + def test_aliased_spellings_derive_one_service_name(self) -> None: + """Two spellings of one driver must not spawn two publishers.""" + names = { + derive_service_name({"class_path": path, "init_args": {"device": 0}}) + for path in ( + "physicalai.capture.UVCCamera", + "physicalai.capture.cameras.uvc.UVCCamera", + "physicalai.capture.cameras.uvc._camera.UVCCamera", ) - - def test_ip_genicam_class_paths_require_explicit_service_name(self) -> None: - """Stub / phantom paths are not in the shareable builtin map.""" - for class_path in ( - "physicalai.capture.IPCamera", - "physicalai.capture.GenicamCamera", - ): - with pytest.raises(ValueError, match="requires an explicit service_name"): - derive_service_name({"class_path": class_path, "init_args": {"device": 0}}) + } + assert names == {"physicalai/camera/UVCCamera/0/frame"} def test_third_party_with_explicit_service_name(self) -> None: cam = SharedCamera.from_config( @@ -472,7 +477,7 @@ def test_subscriber_never_imports_the_vendor_driver(self) -> None: "init_args": {"serial_number": "0001"}, }, ) - assert camera._service_name == "physicalai/camera/realsense/0001/frame" + assert camera._service_name == "physicalai/camera/RealSenseCamera/0001/frame" leaked = sorted( name for name in sys.modules if name.startswith("physicalai.capture.cameras.realsense") or name == "pyrealsense2" From 046e126b1653cdc0a124acfe17142eb1d235dc2c Mon Sep 17 00:00:00 2001 From: Max Xiang Date: Tue, 28 Jul 2026 16:10:20 +0200 Subject: [PATCH 5/5] remove 'trusted' --- .../capture/transport/_publisher_worker.py | 8 ++++---- .../capture/transport/_shared_camera.py | 16 ++++++++-------- src/physicalai/capture/transport/_spec.py | 18 +++++++++--------- tests/unit/capture/test_transport.py | 18 +++++++++--------- 4 files changed, 30 insertions(+), 30 deletions(-) diff --git a/src/physicalai/capture/transport/_publisher_worker.py b/src/physicalai/capture/transport/_publisher_worker.py index b4d28524..c586c979 100644 --- a/src/physicalai/capture/transport/_publisher_worker.py +++ b/src/physicalai/capture/transport/_publisher_worker.py @@ -282,12 +282,12 @@ def _handle_reconfigure(state: _PublisherState, request: dict, service_name: str old_camera = state.camera old_fps = state.camera_fps - trusted_camera = old_config["camera"] - trusted_init_args = trusted_camera["init_args"] + local_camera = old_config["camera"] + local_init_args = local_camera["init_args"] new_config = old_config.copy() new_config["camera"] = { - "class_path": trusted_camera["class_path"], - "init_args": {**trusted_init_args, **settings}, + "class_path": local_camera["class_path"], + "init_args": {**local_init_args, **settings}, } new_config["service_name"] = service_name diff --git a/src/physicalai/capture/transport/_shared_camera.py b/src/physicalai/capture/transport/_shared_camera.py index 70301a1f..feea3f72 100644 --- a/src/physicalai/capture/transport/_shared_camera.py +++ b/src/physicalai/capture/transport/_shared_camera.py @@ -109,12 +109,12 @@ class SharedCamera(Camera): hand off an already-open device into the child. Args: - camera: Trusted camera :class:`~physicalai.config.ComponentConfig` to - spawn if no publisher exists yet for the derived or explicit - ``service_name``. ``None`` means attach-only. Declared as an - ``@export_config`` config arg, so nested - :func:`~physicalai.config.instantiate` passes the recipe through - without constructing the camera here. + camera: Camera :class:`~physicalai.config.ComponentConfig` from local + config input (same boundary as CLI/app args), used to spawn if no + publisher exists yet for the derived or explicit ``service_name``. + ``None`` means attach-only. Declared as an ``@export_config`` + config arg, so nested :func:`~physicalai.config.instantiate` + passes the recipe through without constructing the camera here. color_mode: Pixel format preference for this subscriber. zero_copy: If True, returned frames reference the iceoryx2 SHM buffer directly (read-only). Otherwise, frames are copied. @@ -193,10 +193,10 @@ def from_config( overwrite_settings: bool = False, idle_timeout: float = 5.0, ) -> SharedCamera: - """Primary API: spawn/attach from a trusted camera ComponentConfig. + """Primary API: spawn/attach from a local camera ComponentConfig. Args: - config: Trusted ``class_path`` + ``init_args`` for the camera. + config: Local ``class_path`` + ``init_args`` for the camera. service_name: Explicit iceoryx2 name; derived for built-ins when omitted. color_mode: Subscriber pixel-format preference. zero_copy: Whether frames reference SHM directly. diff --git a/src/physicalai/capture/transport/_spec.py b/src/physicalai/capture/transport/_spec.py index be9d1cd9..bd45b4da 100644 --- a/src/physicalai/capture/transport/_spec.py +++ b/src/physicalai/capture/transport/_spec.py @@ -10,8 +10,8 @@ ``SharedCamera`` uses the same ``camera`` / :meth:`~SharedCamera.from_config` shape. -Security: ``class_path`` is trusted local application/config input. It must -never originate from network-received data. +Security: ``class_path`` is local application/config input. It must never +originate from network-received or control-channel data. """ from __future__ import annotations @@ -31,7 +31,7 @@ if TYPE_CHECKING: from physicalai.capture.camera import Camera -# Allowed keys on the trusted publisher stdin handshake. Everything else +# Allowed keys on the publisher stdin handshake. Everything else # (including legacy flat camera_type / camera_kwargs) is an unknown-key # schema error. _PUBLISHER_ENVELOPE_KEYS = frozenset({ @@ -46,7 +46,7 @@ def validate_publisher_config(data: Mapping[str, Any]) -> Mapping[str, object]: - """Validate a trusted publisher stdin payload schema-positively. + """Validate a publisher stdin payload schema-positively. Returns: The validated ``camera`` ComponentConfig mapping (see @@ -61,13 +61,13 @@ def validate_publisher_config(data: Mapping[str, Any]) -> Mapping[str, object]: def validate_reconfigure_request(request: Mapping[str, Any]) -> dict[str, int]: - """Validate an untrusted camera reconfigure control request. + """Validate a control-channel camera reconfigure request. The peer may change only scalar capture settings. The publisher keeps the - trusted startup ``class_path`` and all other constructor arguments. + local startup ``class_path`` and all other constructor arguments. Returns: - Validated settings to patch into the trusted camera recipe. + Validated settings to patch into the local camera recipe. Raises: TypeError: If the request or a setting has the wrong type. @@ -163,7 +163,7 @@ class CameraPublisherConfig: """Config payload describing how to construct a camera instance. Attributes: - camera: Trusted construction config (``class_path`` + ``init_args``). + camera: Local construction config (``class_path`` + ``init_args``). """ camera: Mapping[str, object] @@ -221,7 +221,7 @@ def from_json_dict(cls, data: dict[str, Any]) -> CameraPublisherConfig: def build(self) -> Camera: """Instantiate the camera described by this spec. - Uses :func:`physicalai.config.instantiate` on the trusted ``camera`` + Uses :func:`physicalai.config.instantiate` on the ``camera`` ComponentConfig, then verifies the :class:`~physicalai.capture.camera.Camera` protocol. Does not route through :func:`~physicalai.capture.create_camera`, so third-party class paths work without a registry entry. diff --git a/tests/unit/capture/test_transport.py b/tests/unit/capture/test_transport.py index eb65bef5..51589a2c 100644 --- a/tests/unit/capture/test_transport.py +++ b/tests/unit/capture/test_transport.py @@ -146,7 +146,7 @@ def test_rejects_empty_unknown_and_legacy_shapes(self, payload: dict[str, object with pytest.raises((TypeError, ValueError), match="reconfigure"): validate_reconfigure_request(payload) - def test_worker_patches_only_trusted_camera_settings(self) -> None: + def test_worker_patches_only_local_camera_settings(self) -> None: from physicalai.capture.transport._publisher_worker import _PublisherState, _handle_reconfigure old_camera = MagicMock() @@ -160,14 +160,14 @@ def test_worker_patches_only_trusted_camera_settings(self) -> None: "camera": { "class_path": FAKE_CAMERA_CLASS, "init_args": { - "device_name": "trusted-device", - "backend": "trusted-backend", + "device_name": "local-device", + "backend": "local-backend", "width": 320, "height": 240, "fps": 30, }, }, - "service_name": "physicalai/test/trusted/frame", + "service_name": "physicalai/test/local/frame", "_factory_override": "tests.unit.capture.fake:FakeCamera", }, ) @@ -178,7 +178,7 @@ def test_worker_patches_only_trusted_camera_settings(self) -> None: result = _handle_reconfigure( state, {"kind": "RECONFIGURE", "settings": {"width": 640, "fps": 15}}, - "physicalai/test/trusted/frame", + "physicalai/test/local/frame", ) assert result == {"ok": True} @@ -186,8 +186,8 @@ def test_worker_patches_only_trusted_camera_settings(self) -> None: assert new_config["camera"] == { "class_path": FAKE_CAMERA_CLASS, "init_args": { - "device_name": "trusted-device", - "backend": "trusted-backend", + "device_name": "local-device", + "backend": "local-backend", "width": 640, "height": 240, "fps": 15, @@ -209,7 +209,7 @@ def test_worker_rejects_component_payload_before_side_effects(self) -> None: camera_fps=30, config={ "camera": {"class_path": FAKE_CAMERA_CLASS, "init_args": {"width": 320}}, - "service_name": "physicalai/test/trusted/frame", + "service_name": "physicalai/test/local/frame", }, ) with patch("physicalai.capture.transport._publisher_worker.build_camera") as build: @@ -224,7 +224,7 @@ def test_worker_rejects_component_payload_before_side_effects(self) -> None: }, }, }, - "physicalai/test/trusted/frame", + "physicalai/test/local/frame", ) assert result["ok"] is False