diff --git a/docs/development/security.md b/docs/development/security.md index 89f980be..f7e15f67 100644 --- a/docs/development/security.md +++ b/docs/development/security.md @@ -14,9 +14,23 @@ These rules apply when writing, editing, or reviewing code under `src/physicalai raise ValueError(f"Path escapes base directory: {user_path!r}") ``` -4. No arbitrary `class_path` import from untrusted manifests or YAML. `instantiate_component` in `inference/component_factory.py` resolves `class_path` via `ComponentRegistry` and `importlib`. Only register trusted short names; treat manifest `class_path` values as untrusted unless the export directory is trusted. Prefer registered `type` names for built-ins. - -5. Enforce component nesting limits. `_MAX_COMPONENT_DEPTH` in `component_factory.py` caps recursive manifest/YAML instantiation — do not raise or bypass without a security review. +4. No arbitrary `class_path` import from untrusted manifests, YAML, or peer + payloads. `instantiate_component` in `inference/component_factory.py` + resolves `class_path` via `ComponentRegistry` and `importlib`. Only register + trusted short names; treat manifest `class_path` values as untrusted unless + the export directory is trusted. Prefer registered `type` names for + built-ins. `physicalai.config.instantiate` is a separate trusted-local / + parent→child-only construction boundary: never pass robot/camera network + metadata, Zenoh payloads, shared-memory control requests, or other + untrusted peer data into it. Camera reconfigure requests may carry only + explicitly allowlisted scalar settings; the publisher must merge them into + its trusted startup recipe without accepting a peer-selected `class_path`. + +5. Enforce component nesting limits. `_MAX_COMPONENT_DEPTH` in + `component_factory.py` caps recursive manifest/YAML instantiation; + `_MAX_CONFIG_DEPTH` in `physicalai.config` caps recursive + `to_config` / `instantiate` trees — do not raise or bypass either without a + security review. 6. Never use `pickle`, `eval()`, `exec()`, `joblib`, `dill`, or `cloudpickle` on untrusted data. Prefer `json` for structured metadata, `safetensors` for weights, and `numpy.load(..., allow_pickle=False)` for arrays. diff --git a/docs/explanation/configuration.md b/docs/explanation/configuration.md index 23f64d5b..65931aa1 100644 --- a/docs/explanation/configuration.md +++ b/docs/explanation/configuration.md @@ -1,33 +1,11 @@ # Configuration -The config system is intended to make Python, YAML, CLI, and Studio payloads use the same workflow shape. +PhysicalAI uses constructor-shaped data so Python, YAML, the CLI, and Studio +can describe the same runtime graph. -```bash -physicalai run --config runtime.yaml -``` - -## Layers - -```text -Config - typed constructor args for one class - -ComponentSpec - target + args for one instantiable component - -Workflow config - user-authored workflow before execution - -Manifest - exported package metadata after build/export - -Orchestrator - live object that executes the workflow -``` +## Construction Recipes -## ComponentSpec - -Direct class mode: +A `ComponentConfig` names one class and the arguments needed to create it. ```yaml class_path: physicalai.capture.UVCCamera @@ -37,47 +15,40 @@ init_args: height: 480 ``` -> **Tip:** Use stable device paths (`/dev/v4l/by-id/...`) in config files. Index-based paths like `/dev/video0` can change after reboot. - -Registry mode: +Classes opt in to exporting this recipe with `@export_config`. `to_config()` +captures construction intent rather than mutable runtime state: supplied +arguments are preserved, omitted defaults remain omitted, and connections or +open handles are never exported. -```yaml -type: uvc -device: /dev/v4l/by-id/usb-Example_Camera-video-index0 -width: 640 -height: 480 +```text +live opted-in component -> to_config() -> JSON/YAML -> instantiate() -> new component ``` -If both `class_path` and `type` are present, `class_path` takes precedence. +`instantiate()` calls constructors only. The application remains responsible +for lifecycle methods such as `connect()` and `run()`. -## Typed Config +## Workflow Documents -Typed configs are useful when constructor validation and IDE support matter. +A workflow document adds command-level structure around components. For +example, `physicalai run` expects runtime constructor arguments under +`runtime:` and optional run-method arguments under `run:`. A bare exported +`RobotRuntime` ComponentConfig is also accepted and reshaped by the loader. -```python -@dataclass -class Pi05Config(Config): - chunk_size: int = 50 - n_action_steps: int = 50 +Nested components retain the same `class_path` + `init_args` form, so a robot, +action source, model, camera map, and callbacks form one recursive recipe. - def __post_init__(self) -> None: - if self.n_action_steps > self.chunk_size: - raise ValueError("n_action_steps must be <= chunk_size") -``` - -Typed configs do not decide which class to instantiate. They only validate and carry constructor arguments. +## Manifests -```python -cfg = Pi05Config(chunk_size=50) -policy = instantiate_obj(cfg, target_cls=Pi05) -``` +An inference manifest describes an exported policy package: artifacts, +features, processors, runners, and compatibility metadata. Its +`ComponentSpec` supports registry aliases and artifact handling and remains +separate from strict captured `ComponentConfig` data. -## Execution Boundary +Use workflow configuration for deployment intent. Use manifests for package +metadata produced during export. -Configuration objects remain passive data. Orchestrators are responsible for creating live objects and executing workflows. +## Trust -```python -config = RuntimeConfig.load("runtime.yaml") -runtime = RobotRuntime.from_config(config) -runtime.run() -``` +Resolving `class_path` imports and executes local Python code. Only trusted +application or user-authored configuration belongs at the instantiation +boundary; peer metadata and transport control messages do not. diff --git a/docs/getting-started/run-a-policy.md b/docs/getting-started/run-a-policy.md index c35df405..b21b591f 100644 --- a/docs/getting-started/run-a-policy.md +++ b/docs/getting-started/run-a-policy.md @@ -36,9 +36,10 @@ The minimal runtime configuration looks like this. ```yaml runtime: robot: - class_path: physicalai.robot.so101.SO101 + class_path: physicalai.robot.SO101 init_args: port: /dev/ttyACM0 + calibration: ./calibration.json action_source: class_path: physicalai.runtime.PolicySource init_args: diff --git a/docs/how-to/config/instantiate-components.md b/docs/how-to/config/instantiate-components.md index 34e1d19a..58853083 100644 --- a/docs/how-to/config/instantiate-components.md +++ b/docs/how-to/config/instantiate-components.md @@ -1,10 +1,7 @@ # Instantiate Components -> **Preview:** The config system (`physicalai.config`) is a planned API. The examples below document the target design. Currently, `ComponentSpec` lives in `physicalai.inference.manifest`. - -A component spec describes one instantiable object. - -The most explicit form uses a class path. +A `ComponentConfig` is a construction recipe with one importable class and +its supplied constructor arguments. ```yaml class_path: physicalai.capture.UVCCamera @@ -14,52 +11,51 @@ init_args: height: 480 ``` -The shorter form uses a registry name. +Use `instantiate()` to construct trusted local configuration. It calls the +constructor but does not call lifecycle methods such as `connect()`, `run()`, +or `start()`. -```yaml -type: uvc -device: /dev/video0 -width: 640 -height: 480 +```python +from physicalai.config import instantiate + +camera = instantiate({ + "class_path": "physicalai.capture.UVCCamera", + "init_args": {"device": "/dev/video0", "width": 640, "height": 480}, +}) +camera.connect() ``` -You can construct and instantiate the same spec from Python. +## Export a live component -```python -from physicalai.inference.manifest import ComponentSpec -from physicalai.inference.component_factory import instantiate_component +Classes opt in with `@export_config`. The decorator remembers only arguments +the caller supplied, so omitted constructor defaults remain omitted. -spec = ComponentSpec( - class_path="physicalai.capture.UVCCamera", - init_args={"device": "/dev/video0", "width": 640, "height": 480}, -) +```python +from physicalai.capture import UVCCamera +from physicalai.config import to_config -camera = instantiate_component(spec) +camera = UVCCamera(device="/dev/video0", width=640, height=480) +config = to_config(camera) ``` -Nested component specs are instantiated recursively. +Nested opted-in components use the same shape recursively. Configs contain +JSON values only; paths become strings, tuples become lists during export, +and non-finite floats are rejected. -```yaml -class_path: physicalai.runtime.RobotRuntime -init_args: - robot: - class_path: physicalai.robot.so101.SO101 - init_args: - port: /dev/ttyACM0 - action_source: - class_path: physicalai.runtime.PolicySource - init_args: - model: - class_path: physicalai.inference.InferenceModel - init_args: - export_dir: ./exports/act_policy - execution: - class_path: physicalai.runtime.SyncExecution - cameras: - wrist: - class_path: physicalai.capture.UVCCamera - init_args: - device: /dev/video0 +```python +from physicalai.config import save_yaml + +save_yaml(camera, "camera.yaml") ``` -`ComponentSpec` describes what should be built. Instantiation is the separate step that creates the live object. +## Trust boundary + +`class_path` selects Python code to import and execute. Pass only trusted +application or user-authored configuration to `instantiate()`. Do not +instantiate robot metadata, camera metadata, shared-memory messages, or other +peer-controlled payloads. + +Inference manifest `ComponentSpec` is a separate compatibility schema with +registry aliases and artifact handling. Use `physicalai.config` for captured +construction recipes; use the manifest APIs when loading exported policy +metadata. diff --git a/docs/how-to/config/write-inference-config.md b/docs/how-to/config/write-inference-config.md index 7ba5f67f..13fe34c5 100644 --- a/docs/how-to/config/write-inference-config.md +++ b/docs/how-to/config/write-inference-config.md @@ -1,7 +1,5 @@ # Write Inference Config -> **Preview:** The config system (`physicalai.config`) is a planned API. The examples below document the target design. - Use an inference config when you need to author an inference pipeline outside an exported manifest. ```yaml @@ -21,3 +19,8 @@ action = model.select_action(observation) ``` Use workflow config to express user-authored intent. Use a manifest to describe exported package metadata. + +To export a model's supplied constructor arguments from Python, call +`physicalai.config.to_config(model)`. Rebuild that trusted local recipe with +`physicalai.config.instantiate(config)`; construction loads the model package +and can allocate substantial resources. diff --git a/docs/how-to/config/write-runtime-config.md b/docs/how-to/config/write-runtime-config.md index 9cc22b6a..67df0218 100644 --- a/docs/how-to/config/write-runtime-config.md +++ b/docs/how-to/config/write-runtime-config.md @@ -6,9 +6,10 @@ A runtime config describes a robot control workflow before execution starts. # runtime.yaml runtime: robot: - class_path: physicalai.robot.so101.SO101 + class_path: physicalai.robot.SO101 init_args: port: /dev/ttyACM0 + calibration: ./calibration.json action_source: class_path: physicalai.runtime.PolicySource init_args: diff --git a/docs/how-to/runtime/run-policy-on-robot.md b/docs/how-to/runtime/run-policy-on-robot.md index 7eb2dbd9..6b904095 100644 --- a/docs/how-to/runtime/run-policy-on-robot.md +++ b/docs/how-to/runtime/run-policy-on-robot.md @@ -31,9 +31,10 @@ Write a runtime configuration file. # runtime.yaml runtime: robot: - class_path: physicalai.robot.so101.SO101 + class_path: physicalai.robot.SO101 init_args: port: /dev/ttyACM0 + calibration: ./calibration.json action_source: class_path: physicalai.runtime.PolicySource init_args: diff --git a/docs/reference/cli.md b/docs/reference/cli.md index 65af4c6c..37cb870b 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -34,11 +34,13 @@ with runtime: physicalai robot serve --config examples/so101/serve.yaml ``` -The flat YAML fields are `name`, `robot_class`, optional `robot_kwargs`, optional -`allow_remote` (default `false`), and optional positive `rate_hz` (default 100 Hz). -Direct CLI arguments use the same names, including nested constructor values such as -`--robot_kwargs.port /dev/ttyACM0`. Serving stays in the foreground until SIGINT or -SIGTERM and returns nonzero for startup, loop, repeated-read, or disconnect failures. +The YAML fields are `name`, required `robot` (a ComponentConfig with +`class_path` + `init_args`), optional `allow_remote` (default `false`), and +optional positive `rate_hz` (default 100 Hz). Direct CLI arguments use the same +names, including nested robot constructor values such as +`--robot.init_args.port /dev/ttyACM0`. Flat `robot_class` / `robot_kwargs` are +unsupported. Serving stays in the foreground until SIGINT or SIGTERM and returns +nonzero for startup, loop, repeated-read, or disconnect failures. Use `--verbose` to include driver construction, lock acquisition, initial observation, endpoint declaration, and cleanup details. diff --git a/docs/reference/inference-api.md b/docs/reference/inference-api.md index 329be9b1..18fc6316 100644 --- a/docs/reference/inference-api.md +++ b/docs/reference/inference-api.md @@ -16,7 +16,8 @@ InferenceModel( ) ``` -The model can be constructed directly from an export directory or loaded from config. +The model can be constructed directly from an export directory or nested in a +trusted workflow ComponentConfig. ## Constructors @@ -38,8 +39,6 @@ This downloads the policy package snapshot from the Hugging Face Hub and loads it like a local export. Additional keyword arguments are forwarded to the constructor (e.g. `backend`, `device`). -> **Note:** `InferenceModel.from_config()` is a planned API. - ## Methods ### `select_action` diff --git a/examples/runtime/async_inference.py b/examples/runtime/async_inference.py index 6a5eda7f..9b4c4a02 100644 --- a/examples/runtime/async_inference.py +++ b/examples/runtime/async_inference.py @@ -40,8 +40,10 @@ import argparse import signal +from pathlib import Path from physicalai.capture import select_cameras_interactive +from physicalai.config import save_yaml, to_config from physicalai.inference import InferenceModel from physicalai.runtime import ( AsyncExecution, @@ -102,6 +104,11 @@ def _handle_sigint(sig: int, frame: object) -> None: rt_group.add_argument("--shared-camera", action="store_true", help="Use shared memory cameras (iceoryx2) — faster but incompatible with debugger") rt_group.add_argument("--request-threshold", type=float, default=0.75, help="Request new inference when queue drops below this fraction of chunk_size (default: 0.75 = trigger when 75%% of actions remain)") rt_group.add_argument("--lerp-frames", type=int, default=3, help="LerpSmoother blend duration in frames (default: 3)") + rt_group.add_argument( + "--export-config", + type=Path, + help="Save the constructed runtime as YAML and exit before connecting hardware.", + ) # Rerun rr_group = parser.add_argument_group("rerun") @@ -158,6 +165,11 @@ def _handle_sigint(sig: int, frame: object) -> None: callbacks=callbacks, ) + if args.export_config: + save_yaml(to_config(runtime), args.export_config) + print(f"Saved runtime config to {args.export_config}") + return + with runtime: for name, cam in cameras.items(): w = getattr(cam, "actual_width", None) diff --git a/examples/runtime/rtc_inference.py b/examples/runtime/rtc_inference.py index 68c9ddda..6f84510c 100644 --- a/examples/runtime/rtc_inference.py +++ b/examples/runtime/rtc_inference.py @@ -35,8 +35,10 @@ import argparse import signal +from pathlib import Path from physicalai.capture import select_cameras_interactive +from physicalai.config import save_yaml, to_config from physicalai.inference import InferenceModel from physicalai.inference.callbacks import RTCLatencyTracker from physicalai.runtime import ( @@ -94,6 +96,11 @@ def _handle_sigint(sig: int, frame: object) -> None: rt_group.add_argument("--duration-s", type=float, default=None, help="Run duration in seconds (default: run indefinitely)") rt_group.add_argument("--task", type=str, default=None, help="Task string for the model (e.g. 'pick up the can')") rt_group.add_argument("--shared-camera", action="store_true", help="Use shared memory cameras (iceoryx2) — faster but incompatible with debugger") + rt_group.add_argument( + "--export-config", + type=Path, + help="Save the constructed runtime as YAML and exit before connecting hardware.", + ) # RTC parameters rtc_group = parser.add_argument_group("rtc") @@ -176,6 +183,11 @@ def _handle_sigint(sig: int, frame: object) -> None: callbacks=callbacks, ) + if args.export_config: + save_yaml(to_config(runtime), args.export_config) + print(f"Saved runtime config to {args.export_config}") + return + with runtime: for name, cam in cameras.items(): w = getattr(cam, "actual_width", None) diff --git a/examples/runtime/runtime.yaml b/examples/runtime/runtime.yaml index 034bbbf0..097fbea3 100644 --- a/examples/runtime/runtime.yaml +++ b/examples/runtime/runtime.yaml @@ -47,21 +47,23 @@ runtime: overhead: # scene/workspace view class_path: physicalai.capture.SharedCamera init_args: - camera_type: uvc - camera_kwargs: - device: /dev/video0 # CHANGE_ME: use /dev/v4l/by-id/... for stable identity - width: 640 - height: 480 - fps: 30 + camera: + class_path: physicalai.capture.UVCCamera + init_args: + device: /dev/video0 # CHANGE_ME: use /dev/v4l/by-id/... for stable identity + width: 640 + height: 480 + fps: 30 gripper: # wrist-mounted view (rename to ``arm`` if the model expects that key) class_path: physicalai.capture.SharedCamera init_args: - camera_type: uvc - camera_kwargs: - device: /dev/video2 # CHANGE_ME - width: 640 - height: 480 - fps: 30 + camera: + class_path: physicalai.capture.UVCCamera + init_args: + device: /dev/video2 # CHANGE_ME + width: 640 + height: 480 + fps: 30 fps: 30.0 @@ -71,32 +73,10 @@ runtime: # mode: connect # connect_addr: "127.0.0.1:9876" # and start ``rerun`` separately. - # - # Cameras are declared again here so the callback can subscribe to the - # same shared-memory streams as the runtime. SharedCamera matches - # publishers by ``camera_type`` + ``camera_kwargs``, so keep these blocks - # in sync with the runtime ``cameras:`` section above. + # Camera frames come from each tick's ``TickEvent.camera_frames`` (same + # values the action source saw); no separate camera wiring on this callback. - class_path: physicalai.runtime.RerunCallback init_args: - cameras: - overhead: - class_path: physicalai.capture.SharedCamera - init_args: - camera_type: uvc - camera_kwargs: - device: /dev/video0 # CHANGE_ME: same as overhead camera above - width: 640 - height: 480 - fps: 30 - gripper: - class_path: physicalai.capture.SharedCamera - init_args: - camera_type: uvc - camera_kwargs: - device: /dev/video2 # CHANGE_ME: same as gripper camera above - width: 640 - height: 480 - fps: 30 mode: spawn log_images: true image_decimation: 1 # log every Nth frame; raise to reduce viewer load diff --git a/examples/runtime/sync_inference.py b/examples/runtime/sync_inference.py index 613efad9..506ca58c 100644 --- a/examples/runtime/sync_inference.py +++ b/examples/runtime/sync_inference.py @@ -30,8 +30,10 @@ import argparse import signal +from pathlib import Path from physicalai.capture import select_cameras_interactive +from physicalai.config import save_yaml, to_config from physicalai.inference import InferenceModel from physicalai.runtime import ( ChunkedActionQueue, @@ -87,6 +89,11 @@ def _handle_sigint(sig: int, frame: object) -> None: rt_group.add_argument("--duration-s", type=float, default=60.0, help="Duration in seconds") rt_group.add_argument("--task", type=str, default=None, help="Task string for the model (e.g. 'pick up the can')") rt_group.add_argument("--request-threshold", type=float, default=0.5, help="Request new inference when queue drops below this fraction of chunk_size (default: 0.75 = trigger when 75%% of actions remain)") + rt_group.add_argument( + "--export-config", + type=Path, + help="Save the constructed runtime as YAML and exit before connecting hardware.", + ) # Rerun rr_group = parser.add_argument_group("rerun") @@ -136,6 +143,11 @@ def _handle_sigint(sig: int, frame: object) -> None: callbacks=callbacks, ) + if args.export_config: + save_yaml(to_config(runtime), args.export_config) + print(f"Saved runtime config to {args.export_config}") + return + with runtime: print(f"Running SYNC at {args.fps} fps for {args.duration_s}s...") if args.task: diff --git a/examples/so101/serve.yaml b/examples/so101/serve.yaml index 31100d97..2b93aad2 100644 --- a/examples/so101/serve.yaml +++ b/examples/so101/serve.yaml @@ -5,10 +5,11 @@ # host B walkthrough (serve on the host with the physical robot attached, # attach from anywhere else with a `SharedRobot` runtime config). name: follower-arm -robot_class: physicalai.robot.SO101 -robot_kwargs: - port: /dev/ttyACM0 - calibration: /path/to/so101-calibration.json - role: follower +robot: + class_path: physicalai.robot.SO101 + init_args: + port: /dev/ttyACM0 + calibration: /path/to/so101-calibration.json + role: follower allow_remote: false rate_hz: 100.0 diff --git a/src/physicalai/cli/run.py b/src/physicalai/cli/run.py index 264110d9..d4255469 100644 --- a/src/physicalai/cli/run.py +++ b/src/physicalai/cli/run.py @@ -5,9 +5,12 @@ from __future__ import annotations +import json import logging +from pathlib import Path from typing import TYPE_CHECKING +import yaml from jsonargparse import ActionConfigFile, ArgumentParser from physicalai.cli._spec import SubcommandSpec # noqa: PLC2701 @@ -43,11 +46,68 @@ """ +def _reshape_config_value(parser: ArgumentParser, value: object) -> object: + """Return CLI-shaped config content for a bare exported runtime document. + + A ``--config`` value that points to a document with a top-level + ``class_path`` (as produced by ``to_config(runtime)`` / ``save_yaml``) is + loaded and unwrapped to the ``runtime:`` mapping the parser expects, + returned as JSON content (valid YAML that ``ActionConfigFile`` accepts + inline). Any other value — a non-path, an unreadable file, or a + ``runtime:`` CLI document — is returned unchanged. + + Returns: + The original value, or unwrapped config content for a bare document. + """ + if not isinstance(value, str): + return value + try: + if not Path(value).is_file(): + return value + document = yaml.safe_load(Path(value).read_text(encoding="utf-8")) + except (OSError, ValueError, yaml.YAMLError): + return value + if not isinstance(document, dict) or "class_path" not in document: + return value + + from physicalai.config import ComponentConfigError # noqa: PLC0415 + from physicalai.runtime import RobotRuntime # noqa: PLC0415 + from physicalai.runtime.core import _unwrap_runtime_document # noqa: PLC0415, PLC2701 + + try: + return json.dumps(_unwrap_runtime_document(document, target=RobotRuntime)) + except ComponentConfigError as exc: + parser.error(str(exc)) + + +class _RuntimeConfigFile(ActionConfigFile): + """``--config`` action accepting both the CLI document and a bare export. + + Reshapes a bare exported ``RobotRuntime`` ComponentConfig (a document with a + top-level ``class_path``) into the ``runtime:`` document jsonargparse + expects before applying it; a ``runtime:`` / ``run:`` CLI document is passed + through unchanged. This keeps ``to_config(runtime)`` → ``save_yaml`` → + ``physicalai run --config`` a single-shape round-trip. + """ + + def __call__( + self, + parser: ArgumentParser, + cfg: Namespace, + values: object, + option_string: str | None = None, + ) -> None: + """Reshape a bare exported runtime document, then delegate to the base action.""" + super().__call__(parser, cfg, _reshape_config_value(parser, values), option_string) + + def build_parser() -> ArgumentParser: """Build the ``run`` subcommand parser. One path: ``RobotRuntime`` constructor arguments (``action_source:`` - always explicit) plus ``run()`` method arguments. + always explicit) plus ``run()`` method arguments. ``--config`` accepts + both the CLI document (``runtime:`` / ``run:``) and a bare exported + ComponentConfig produced by ``to_config(runtime)``. Returns: Parser for the ``physicalai run`` subcommand. @@ -55,7 +115,7 @@ def build_parser() -> ArgumentParser: from physicalai.runtime import RobotRuntime # noqa: PLC0415 parser = ArgumentParser(prog="physicalai run", description=HELP) - parser.add_argument("--config", action=ActionConfigFile, help="YAML/JSON config file.") + parser.add_argument("--config", action=_RuntimeConfigFile, help="YAML/JSON config file.") parser.add_class_arguments(RobotRuntime, "runtime") parser.add_method_arguments(RobotRuntime, "run", "run") return parser diff --git a/src/physicalai/inference/callbacks/rtc_latency.py b/src/physicalai/inference/callbacks/rtc_latency.py index 3b2d1737..ced36a45 100644 --- a/src/physicalai/inference/callbacks/rtc_latency.py +++ b/src/physicalai/inference/callbacks/rtc_latency.py @@ -17,9 +17,11 @@ import numpy as np from typing_extensions import override +from physicalai.config import export_config from physicalai.inference.callbacks.base import Callback +@export_config(class_path="physicalai.inference.callbacks.RTCLatencyTracker") class RTCLatencyTracker(Callback): """Track inference latency for RTC delay computation. diff --git a/src/physicalai/inference/model.py b/src/physicalai/inference/model.py index e48d2b2f..596545ff 100644 --- a/src/physicalai/inference/model.py +++ b/src/physicalai/inference/model.py @@ -12,6 +12,7 @@ import numpy as np +from physicalai.config import export_config from physicalai.inference.adapters import adapter_registry, get_adapter from physicalai.inference.component_factory import instantiate_component, resolve_artifact from physicalai.inference.constants import ACTION @@ -38,6 +39,7 @@ def _is_safe_policy_name(name: str) -> bool: return _SAFE_POLICY_NAME_RE.fullmatch(name) is not None +@export_config(class_path="physicalai.inference.InferenceModel", scalar_var_kwargs=True) class InferenceModel: """Unified inference interface for exported policies. diff --git a/src/physicalai/runtime/action_sources/policy.py b/src/physicalai/runtime/action_sources/policy.py index b390f05f..75ba923c 100644 --- a/src/physicalai/runtime/action_sources/policy.py +++ b/src/physicalai/runtime/action_sources/policy.py @@ -10,6 +10,7 @@ import numpy as np +from physicalai.config import export_config from physicalai.inference.constants import IMAGES, STATE, TASK from physicalai.runtime.action_sources.base import ActionSource from physicalai.runtime.events import MetricsEvent @@ -30,6 +31,7 @@ _DEFAULT_LERP_FRAMES = 5 +@export_config(class_path="physicalai.runtime.PolicySource") class PolicySource(ActionSource): """Action source adapting a model + execution + action-queue policy pipeline.""" diff --git a/src/physicalai/runtime/action_sources/teleop.py b/src/physicalai/runtime/action_sources/teleop.py index 10685ed1..2652e837 100644 --- a/src/physicalai/runtime/action_sources/teleop.py +++ b/src/physicalai/runtime/action_sources/teleop.py @@ -8,6 +8,7 @@ import contextlib from typing import TYPE_CHECKING +from physicalai.config import export_config from physicalai.runtime.action_sources.base import ActionSource if TYPE_CHECKING: @@ -20,6 +21,7 @@ from physicalai.runtime._callback_bus import _CallbackBus +@export_config(class_path="physicalai.runtime.TeleopSource") class TeleopSource(ActionSource): """Action source that reads a leader arm and writes to the follower. diff --git a/src/physicalai/runtime/callbacks/async_callback.py b/src/physicalai/runtime/callbacks/async_callback.py index b3c3b78b..5afb2eef 100644 --- a/src/physicalai/runtime/callbacks/async_callback.py +++ b/src/physicalai/runtime/callbacks/async_callback.py @@ -12,6 +12,7 @@ from typing import TYPE_CHECKING, Any from physicalai.capture.frame import Frame +from physicalai.config import export_config if TYPE_CHECKING: from physicalai.runtime.events import InferenceEvent, LifecycleEvent, MetricsEvent, TickEvent @@ -19,6 +20,7 @@ logger = logging.getLogger(__name__) +@export_config(class_path="physicalai.runtime.AsyncCallback") class AsyncCallback: """Wraps a callback so all hooks run on a dedicated background thread. diff --git a/src/physicalai/runtime/callbacks/console.py b/src/physicalai/runtime/callbacks/console.py index eddf8834..3a37564a 100644 --- a/src/physicalai/runtime/callbacks/console.py +++ b/src/physicalai/runtime/callbacks/console.py @@ -8,10 +8,13 @@ import time from typing import TYPE_CHECKING +from physicalai.config import export_config + if TYPE_CHECKING: from physicalai.runtime.events import LifecycleEvent, TickEvent +@export_config(class_path="physicalai.runtime.ConsoleCallback") class ConsoleCallback: """Periodic one-line summary to stdout, throttled by step count. diff --git a/src/physicalai/runtime/callbacks/jsonl.py b/src/physicalai/runtime/callbacks/jsonl.py index 6eb3daf3..1de04806 100644 --- a/src/physicalai/runtime/callbacks/jsonl.py +++ b/src/physicalai/runtime/callbacks/jsonl.py @@ -9,12 +9,15 @@ from pathlib import Path from typing import TYPE_CHECKING, Any +from physicalai.config import export_config + if TYPE_CHECKING: import numpy as np from physicalai.runtime.events import InferenceEvent, LifecycleEvent, TickEvent +@export_config(class_path="physicalai.runtime.JsonlCallback") class JsonlCallback: """Append-only JSONL recording. Numpy arrays converted to lists.""" diff --git a/src/physicalai/runtime/callbacks/low_pass.py b/src/physicalai/runtime/callbacks/low_pass.py index c4495cbb..ce76cc7e 100644 --- a/src/physicalai/runtime/callbacks/low_pass.py +++ b/src/physicalai/runtime/callbacks/low_pass.py @@ -7,10 +7,13 @@ from typing import TYPE_CHECKING +from physicalai.config import export_config + if TYPE_CHECKING: import numpy as np +@export_config(class_path="physicalai.runtime.LowPassFilterCallback") class LowPassFilterCallback: """Stateful low-pass filter (Exponential Moving Average) callback for smooth actions. diff --git a/src/physicalai/runtime/callbacks/rerun.py b/src/physicalai/runtime/callbacks/rerun.py index 7aedb7fa..abf53812 100644 --- a/src/physicalai/runtime/callbacks/rerun.py +++ b/src/physicalai/runtime/callbacks/rerun.py @@ -10,6 +10,8 @@ from collections import deque from typing import TYPE_CHECKING, Any, Literal +from physicalai.config import export_config + if TYPE_CHECKING: from collections.abc import Mapping @@ -21,6 +23,7 @@ logger = logging.getLogger(__name__) +@export_config(class_path="physicalai.runtime.RerunCallback") class RerunCallback: """In-process Rerun logging for runtime visualization. diff --git a/src/physicalai/runtime/core.py b/src/physicalai/runtime/core.py index 3183bd14..bb34aefe 100644 --- a/src/physicalai/runtime/core.py +++ b/src/physicalai/runtime/core.py @@ -11,6 +11,7 @@ from typing import TYPE_CHECKING, Protocol, Self from physicalai.capture.errors import CaptureError +from physicalai.config import export_config from physicalai.runtime._callback_bus import _CallbackBus # noqa: PLC2701 from physicalai.runtime.events import LifecycleEvent, TickEvent from physicalai.runtime.execution.base import WorkerDiedError @@ -36,6 +37,42 @@ _GOAL_TIME_TICKS = 3 +def _unwrap_runtime_document(document: dict[str, Any], *, target: type) -> dict[str, Any]: + """Rewrite a bare exported ComponentConfig document to the CLI ``runtime:`` shape. + + A document without a top-level ``class_path`` is returned unchanged (it is + already a CLI document). Otherwise it must be a valid + :class:`~physicalai.config.ComponentConfig` whose ``class_path`` resolves + to *target* (or a subclass), and its ``init_args`` become the ``runtime:`` + section — making ``to_config(runtime)`` → YAML → CLI round-trip. + + Returns: + A CLI-shaped document with constructor args under ``runtime:``. + + Raises: + ComponentConfigError: If ``class_path`` does not resolve to *target*. + """ + if "class_path" not in document: + return document + + from physicalai.config import ( # noqa: PLC0415 + ComponentConfigError, + import_dotted_path, + validate_component_config, + ) + + config = validate_component_config(document) + resolved = import_dotted_path(config["class_path"]) + if not (isinstance(resolved, type) and issubclass(resolved, target)): + msg = ( + f"config class_path {config['class_path']!r} does not resolve to " + f"{target.__module__}.{target.__qualname__} (or a subclass); " + "expected a runtime config exported via to_config(runtime)" + ) + raise ComponentConfigError(msg) + return {"runtime": dict(config["init_args"])} + + class RuntimeCallback(Protocol): """Optional action-transform hooks a callback may implement. @@ -64,6 +101,7 @@ def on_action_sent(self, *, action: np.ndarray, step: int) -> None: ... +@export_config(class_path="physicalai.runtime.RobotRuntime") class RobotRuntime: """Generic robot runtime loop with a required, pluggable action source.""" @@ -177,19 +215,26 @@ def __exit__(self, *exc_info: object) -> None: # noqa: D105 def from_config(cls, config: str | Path) -> Self: """Build runtime from YAML/JSON config file. + Accepts two document shapes: the CLI document (constructor args under + ``runtime:``, optional ``run:``) and a bare exported + :class:`~physicalai.config.ComponentConfig` as produced by + :func:`~physicalai.config.to_config` / :func:`~physicalai.config.save_yaml` + (top-level ``class_path`` resolving to this class). ``action_source:`` is always required and explicit — one schema, no flat/legacy shorthand. Returns: Instantiated runtime object. """ - from jsonargparse import ActionConfigFile, ArgumentParser # noqa: PLC0415 + from jsonargparse import ArgumentParser # noqa: PLC0415 + + from physicalai.config import load_yaml # noqa: PLC0415 + document = _unwrap_runtime_document(load_yaml(config), target=cls) parser = ArgumentParser() - parser.add_argument("--config", action=ActionConfigFile) parser.add_class_arguments(cls, "runtime") parser.add_method_arguments(cls, "run", "run") - ns = parser.parse_args(["--config", str(config)]) + ns = parser.parse_object(document) return parser.instantiate(ns).runtime def run(self, *, duration_s: float | None = None) -> int: diff --git a/src/physicalai/runtime/execution/async_execution.py b/src/physicalai/runtime/execution/async_execution.py index a7685056..78fba4cd 100644 --- a/src/physicalai/runtime/execution/async_execution.py +++ b/src/physicalai/runtime/execution/async_execution.py @@ -12,6 +12,7 @@ import numpy as np +from physicalai.config import export_config from physicalai.runtime.execution.base import NOT_STARTED, Execution, WorkerDiedError if TYPE_CHECKING: @@ -22,6 +23,7 @@ logger = logging.getLogger(__name__) +@export_config(class_path="physicalai.runtime.AsyncExecution") class AsyncExecution(Execution): """Async inference in a background thread with health monitoring.""" diff --git a/src/physicalai/runtime/execution/queue.py b/src/physicalai/runtime/execution/queue.py index 958e3c84..6a7a199e 100644 --- a/src/physicalai/runtime/execution/queue.py +++ b/src/physicalai/runtime/execution/queue.py @@ -11,6 +11,7 @@ import numpy as np +from physicalai.config import export_config from physicalai.runtime.smoothers import ChunkSmoother, ReplaceSmoother @@ -63,6 +64,7 @@ def reset(self) -> None: ... +@export_config(class_path="physicalai.runtime.ChunkedActionQueue") class ChunkedActionQueue: """Thread-safe action queue with chunk smoothing.""" diff --git a/src/physicalai/runtime/execution/rtc.py b/src/physicalai/runtime/execution/rtc.py index 93af3650..bdc504f9 100644 --- a/src/physicalai/runtime/execution/rtc.py +++ b/src/physicalai/runtime/execution/rtc.py @@ -18,6 +18,7 @@ import numpy as np +from physicalai.config import export_config from physicalai.runtime.execution.base import Execution, WorkerDiedError if TYPE_CHECKING: @@ -36,6 +37,7 @@ _JOIN_TIMEOUT_S: float = 5.0 +@export_config(class_path="physicalai.runtime.RTCExecution") class RTCExecution(Execution): """Async RTC execution strategy with background inference thread. diff --git a/src/physicalai/runtime/execution/rtc_queue.py b/src/physicalai/runtime/execution/rtc_queue.py index 373f2eea..689a0d28 100644 --- a/src/physicalai/runtime/execution/rtc_queue.py +++ b/src/physicalai/runtime/execution/rtc_queue.py @@ -13,12 +13,15 @@ import threading from typing import TYPE_CHECKING +from physicalai.config import export_config + if TYPE_CHECKING: import numpy as np logger = logging.getLogger(__name__) +@export_config(class_path="physicalai.runtime.RTCActionQueue") class RTCActionQueue: """Thread-safe dual-track action queue for RTC inference. diff --git a/src/physicalai/runtime/execution/sync.py b/src/physicalai/runtime/execution/sync.py index d1ba6fae..30d30491 100644 --- a/src/physicalai/runtime/execution/sync.py +++ b/src/physicalai/runtime/execution/sync.py @@ -8,6 +8,7 @@ import time from typing import TYPE_CHECKING, cast +from physicalai.config import export_config from physicalai.runtime.execution.base import NOT_STARTED, Execution if TYPE_CHECKING: @@ -18,6 +19,7 @@ from physicalai.runtime.execution.queue import ActionQueue, ChunkedActionQueue +@export_config(class_path="physicalai.runtime.SyncExecution") class SyncExecution(Execution): """Synchronous inference in the control thread.""" diff --git a/src/physicalai/runtime/smoothers.py b/src/physicalai/runtime/smoothers.py index 0181adf7..6ee6e95c 100644 --- a/src/physicalai/runtime/smoothers.py +++ b/src/physicalai/runtime/smoothers.py @@ -10,6 +10,8 @@ import numpy as np from typing_extensions import override +from physicalai.config import export_config + _NDIM_2 = 2 _ERR_2D = "remaining and incoming must be 2D arrays" _ERR_ACTION_DIM = "remaining and incoming must have the same action_dim" @@ -24,9 +26,13 @@ def merge(self, remaining: np.ndarray, incoming: np.ndarray) -> np.ndarray: raise NotImplementedError +@export_config(class_path="physicalai.runtime.ReplaceSmoother") class ReplaceSmoother(ChunkSmoother): """Replace remaining actions with the incoming chunk.""" + def __init__(self) -> None: + """Create a replace smoother (no configuration).""" + @override def merge(self, remaining: np.ndarray, incoming: np.ndarray) -> np.ndarray: """Return the incoming chunk (remaining is discarded).""" @@ -34,6 +40,7 @@ def merge(self, remaining: np.ndarray, incoming: np.ndarray) -> np.ndarray: return incoming +@export_config(class_path="physicalai.runtime.LerpSmoother") class LerpSmoother(ChunkSmoother): """Blend overlapping actions and append the incoming tail.""" diff --git a/tests/unit/cli/test_cli.py b/tests/unit/cli/test_cli.py index 66568461..c4e0b72c 100644 --- a/tests/unit/cli/test_cli.py +++ b/tests/unit/cli/test_cli.py @@ -249,6 +249,41 @@ def test_config_file_accepts_shared_robots(self, tmp_path: Path) -> None: assert cfg.runtime.robot.class_path == "physicalai.robot.SharedRobot" assert cfg.runtime.action_source.init_args.leader.class_path == "physicalai.robot.SharedRobot" + def test_config_file_accepts_bare_component_export(self, tmp_path: Path) -> None: + cfg_file = tmp_path / "export.yaml" + cfg_file.write_text( + "class_path: physicalai.runtime.RobotRuntime\n" + "init_args:\n" + " fps: 30\n" + " robot:\n" + f" class_path: {_FAKE_ROBOT}\n" + " init_args:\n" + " port: /dev/null\n" + " action_source:\n" + f" class_path: {_POLICY_SOURCE}\n" + " init_args:\n" + " model:\n" + f" class_path: {_REAL_MODEL}\n" + " init_args:\n" + " export_dir: /tmp/fake\n" + " execution:\n" + f" class_path: {_REAL_SYNC}\n", + ) + for argv in ([f"--config={cfg_file}"], ["--config", str(cfg_file)]): + parser = run_module.build_parser() + cfg = parser.parse_args(argv) + assert cfg.runtime.fps == 30 + assert cfg.runtime.robot.class_path == _FAKE_ROBOT + assert cfg.runtime.action_source.class_path == _POLICY_SOURCE + + def test_bare_export_foreign_class_path_errors(self, tmp_path: Path) -> None: + cfg_file = tmp_path / "foreign.yaml" + cfg_file.write_text(f"class_path: {_REAL_SYNC}\ninit_args: {{}}\n") + parser = run_module.build_parser() + with pytest.raises(SystemExit) as exc: + parser.parse_args([f"--config={cfg_file}"]) + assert exc.value.code != 0 + def test_cli_overrides_config_file(self, tmp_path: Path) -> None: cfg_file = tmp_path / "runtime.yaml" cfg_file.write_text( diff --git a/tests/unit/inference/test_inference_model_component_config.py b/tests/unit/inference/test_inference_model_component_config.py new file mode 100644 index 00000000..8512900f --- /dev/null +++ b/tests/unit/inference/test_inference_model_component_config.py @@ -0,0 +1,193 @@ +# 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 path-rooted ``InferenceModel`` ``@export_config``.""" + +from __future__ import annotations + +import json +from collections.abc import Generator +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +from physicalai.config import ComponentConfigError, instantiate, is_config_exportable, to_config + + +def _make_export_dir(tmp_path: Path, *, backend: str = "openvino") -> Path: + export_dir = tmp_path / "exports" + export_dir.mkdir(exist_ok=True) + artifact = "act.xml" if backend == "openvino" else f"act.{backend}" + manifest = { + "format": "policy_package", + "version": "1.0", + "policy": { + "name": "act", + "source": {"class_path": "physicalai.policies.act.ACT"}, + }, + "model": { + "artifacts": {backend: artifact}, + "runner": {"class_path": "physicalai.inference.runners.SinglePass", "init_args": {}}, + }, + } + with (export_dir / "manifest.json").open("w") as f: + json.dump(manifest, f) + (export_dir / artifact).touch() + if artifact.endswith(".xml"): + (export_dir / artifact.replace(".xml", ".bin")).touch() + return export_dir + + +def _assert_construction_round_trip(model: object) -> dict[str, Any]: + assert is_config_exportable(model) + config = to_config(model) + wire: dict[str, Any] = json.loads(json.dumps(config)) + restored = instantiate(wire) + assert type(restored) is type(model) + assert to_config(restored) == wire + return wire + + +@pytest.fixture +def mock_adapter() -> MagicMock: + adapter = MagicMock() + adapter.input_names = [] + adapter.output_names = [] + adapter.default_device.return_value = "cpu" + return adapter + + +@pytest.fixture +def _patch_adapter(mock_adapter: MagicMock) -> Generator[MagicMock, None, None]: + with patch("physicalai.inference.model.get_adapter", return_value=mock_adapter): + yield mock_adapter + + +class TestInferenceModelComponentConfig: + def test_path_rooted_round_trip(self, tmp_path: Path, _patch_adapter: MagicMock) -> None: + from physicalai.inference import InferenceModel + + export_dir = _make_export_dir(tmp_path) + model = InferenceModel( + export_dir=export_dir, + policy_name="act", + backend="openvino", + device="cpu", + ) + wire = _assert_construction_round_trip(model) + assert wire["class_path"] == "physicalai.inference.InferenceModel" + assert wire["init_args"] == { + "export_dir": str(export_dir), + "policy_name": "act", + "backend": "openvino", + "device": "cpu", + } + + def test_relative_export_dir_as_given( + self, tmp_path: Path, _patch_adapter: MagicMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + from physicalai.inference import InferenceModel + + monkeypatch.chdir(tmp_path) + _make_export_dir(tmp_path) + rel = Path("exports") + + model = InferenceModel(export_dir=rel, backend="openvino", device="cpu") + wire = _assert_construction_round_trip(model) + assert wire["init_args"]["export_dir"] == "exports" + + def test_scalar_adapter_kwargs_round_trip(self, tmp_path: Path, _patch_adapter: MagicMock) -> None: + from physicalai.inference import InferenceModel + + model = InferenceModel( + export_dir=_make_export_dir(tmp_path), + backend="openvino", + device="cpu", + num_threads=4, + ) + wire = _assert_construction_round_trip(model) + assert wire["init_args"]["num_threads"] == 4 + + def test_omitted_overrides_stay_omitted(self, tmp_path: Path, _patch_adapter: MagicMock) -> None: + from physicalai.inference import InferenceModel + + model = InferenceModel(export_dir=_make_export_dir(tmp_path), backend="openvino", device="cpu") + wire = to_config(model) + for key in ("runner", "preprocessors", "postprocessors", "callbacks"): + assert key not in wire["init_args"] + + def test_live_runner_fails(self, tmp_path: Path, _patch_adapter: MagicMock) -> None: + from physicalai.inference import InferenceModel + from physicalai.inference.runners import SinglePass + + model = InferenceModel( + export_dir=_make_export_dir(tmp_path), + backend="openvino", + device="cpu", + runner=SinglePass(), + ) + with pytest.raises(ComponentConfigError, match=r"init_args\.runner"): + to_config(model) + + def test_live_preprocessors_fail(self, tmp_path: Path, _patch_adapter: MagicMock) -> None: + from physicalai.inference import InferenceModel + + model = InferenceModel( + export_dir=_make_export_dir(tmp_path), + backend="openvino", + device="cpu", + preprocessors=[MagicMock()], + ) + with pytest.raises(ComponentConfigError, match=r"init_args\.preprocessors"): + to_config(model) + + def test_live_postprocessors_fail(self, tmp_path: Path, _patch_adapter: MagicMock) -> None: + from physicalai.inference import InferenceModel + + model = InferenceModel( + export_dir=_make_export_dir(tmp_path), + backend="openvino", + device="cpu", + postprocessors=[MagicMock()], + ) + with pytest.raises(ComponentConfigError, match=r"init_args\.postprocessors"): + to_config(model) + + def test_live_callbacks_fail(self, tmp_path: Path, _patch_adapter: MagicMock) -> None: + from physicalai.inference import InferenceModel + + model = InferenceModel( + export_dir=_make_export_dir(tmp_path), + backend="openvino", + device="cpu", + callbacks=[MagicMock()], + ) + with pytest.raises(ComponentConfigError, match=r"init_args\.callbacks"): + to_config(model) + + def test_non_scalar_dict_adapter_kwarg_fails(self, tmp_path: Path, _patch_adapter: MagicMock) -> None: + from physicalai.inference import InferenceModel + + model = InferenceModel( + export_dir=_make_export_dir(tmp_path), + backend="openvino", + device="cpu", + config_blob={"a": 1}, + ) + with pytest.raises(ComponentConfigError, match=r"init_args\.config_blob"): + to_config(model) + + def test_non_scalar_list_adapter_kwarg_fails(self, tmp_path: Path, _patch_adapter: MagicMock) -> None: + from physicalai.inference import InferenceModel + + model = InferenceModel( + export_dir=_make_export_dir(tmp_path), + backend="openvino", + device="cpu", + tags=["x", "y"], + ) + with pytest.raises(ComponentConfigError, match=r"init_args\.tags"): + to_config(model) diff --git a/tests/unit/runtime/test_runtime_component_config.py b/tests/unit/runtime/test_runtime_component_config.py new file mode 100644 index 00000000..795183f1 --- /dev/null +++ b/tests/unit/runtime/test_runtime_component_config.py @@ -0,0 +1,702 @@ +# 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 runtime ``@export_config`` wiring (steps 6–7).""" + +from __future__ import annotations + +import json +import sys +from collections.abc import Generator +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +from physicalai.config import ComponentConfigError, instantiate, is_config_exportable, to_config + +# SO101 imports scservo_sdk at module load; keep unit tests hardware-free. +# Use setdefault — not patch.dict(sys.modules) — because patch.dict restores by +# clearing sys.modules and drops modules imported while the patch is active +# (e.g. jsonargparse ``build_parser``), which poisons later YAML config loads. +sys.modules.setdefault("scservo_sdk", MagicMock()) + + +def _assert_construction_round_trip(value: object) -> dict[str, Any]: + assert is_config_exportable(value) + config = to_config(value) + wire: dict[str, Any] = json.loads(json.dumps(config)) + restored = instantiate(wire) + assert type(restored) is type(value) + assert to_config(restored) == wire + return wire + + +def _make_export_dir(tmp_path: Path, *, backend: str = "openvino") -> Path: + export_dir = tmp_path / "exports" + export_dir.mkdir(exist_ok=True) + artifact = "act.xml" if backend == "openvino" else f"act.{backend}" + manifest = { + "format": "policy_package", + "version": "1.0", + "policy": { + "name": "act", + "source": {"class_path": "physicalai.policies.act.ACT"}, + }, + "model": { + "artifacts": {backend: artifact}, + "runner": {"class_path": "physicalai.inference.runners.SinglePass", "init_args": {}}, + }, + } + with (export_dir / "manifest.json").open("w") as f: + json.dump(manifest, f) + (export_dir / artifact).touch() + if artifact.endswith(".xml"): + (export_dir / artifact.replace(".xml", ".bin")).touch() + return export_dir + + +@pytest.fixture +def mock_adapter() -> MagicMock: + adapter = MagicMock() + adapter.input_names = [] + adapter.output_names = [] + adapter.default_device.return_value = "cpu" + return adapter + + +@pytest.fixture +def _patch_adapter(mock_adapter: MagicMock) -> Generator[MagicMock, None, None]: + with patch("physicalai.inference.model.get_adapter", return_value=mock_adapter): + yield mock_adapter + + +@pytest.fixture +def inference_model(tmp_path: Path, _patch_adapter: MagicMock) -> Any: + from physicalai.inference import InferenceModel + + return InferenceModel(export_dir=_make_export_dir(tmp_path), backend="openvino", device="cpu") + + +# --------------------------------------------------------------------------- +# Smoothers / queues / execution +# --------------------------------------------------------------------------- + + +class TestSmootherComponentConfig: + def test_lerp_round_trip(self) -> None: + from physicalai.runtime import LerpSmoother + + wire = _assert_construction_round_trip(LerpSmoother(duration_frames=7)) + assert wire["class_path"] == "physicalai.runtime.LerpSmoother" + assert wire["init_args"] == {"duration_frames": 7} + + def test_lerp_default_omitted(self) -> None: + from physicalai.runtime import LerpSmoother + + wire = _assert_construction_round_trip(LerpSmoother()) + assert wire["init_args"] == {} + + def test_replace_round_trip(self) -> None: + from physicalai.runtime import ReplaceSmoother + + wire = _assert_construction_round_trip(ReplaceSmoother()) + assert wire["class_path"] == "physicalai.runtime.ReplaceSmoother" + assert wire["init_args"] == {} + + +class TestActionQueueComponentConfig: + def test_chunked_bare_restores_replace_smoother(self) -> None: + from physicalai.runtime import ChunkedActionQueue, ReplaceSmoother + + queue = ChunkedActionQueue() + wire = _assert_construction_round_trip(queue) + assert wire["class_path"] == "physicalai.runtime.ChunkedActionQueue" + assert wire["init_args"] == {} + restored = instantiate(wire) + assert isinstance(restored._smoother, ReplaceSmoother) # type: ignore[attr-defined] + + def test_chunked_explicit_lerp_nested(self) -> None: + from physicalai.runtime import ChunkedActionQueue, LerpSmoother + + queue = ChunkedActionQueue(smoother=LerpSmoother(duration_frames=3)) + wire = _assert_construction_round_trip(queue) + assert wire["init_args"]["smoother"] == { + "class_path": "physicalai.runtime.LerpSmoother", + "init_args": {"duration_frames": 3}, + } + + def test_chunked_undecorated_smoother_fails(self) -> None: + from physicalai.runtime import ChunkedActionQueue + from physicalai.runtime.smoothers import ChunkSmoother + + class UndecoratedSmoother(ChunkSmoother): + def merge(self, remaining: Any, incoming: Any) -> Any: # noqa: ANN401 + return incoming + + queue = ChunkedActionQueue(smoother=UndecoratedSmoother()) + with pytest.raises(ComponentConfigError, match=r"init_args\.smoother"): + to_config(queue) + + def test_rtc_action_queue_round_trip(self) -> None: + from physicalai.runtime import RTCActionQueue + + wire = _assert_construction_round_trip(RTCActionQueue()) + assert wire["class_path"] == "physicalai.runtime.RTCActionQueue" + assert wire["init_args"] == {} + + +class TestExecutionComponentConfig: + def test_sync_default_omitted(self) -> None: + from physicalai.runtime import SyncExecution + + wire = _assert_construction_round_trip(SyncExecution()) + assert wire["class_path"] == "physicalai.runtime.SyncExecution" + assert wire["init_args"] == {} + + def test_sync_explicit_threshold(self) -> None: + from physicalai.runtime import SyncExecution + + wire = _assert_construction_round_trip(SyncExecution(request_threshold=0.25)) + assert wire["init_args"] == {"request_threshold": 0.25} + + def test_async_round_trip(self) -> None: + from physicalai.runtime import AsyncExecution + + wire = _assert_construction_round_trip( + AsyncExecution(request_threshold=0.1, watchdog_timeout_s=10.0) + ) + assert wire["class_path"] == "physicalai.runtime.AsyncExecution" + assert wire["init_args"] == {"request_threshold": 0.1, "watchdog_timeout_s": 10.0} + + def test_rtc_scalar_args_round_trip(self) -> None: + from physicalai.runtime import RTCExecution + + wire = _assert_construction_round_trip( + RTCExecution(chunk_size=50, execution_horizon=10, fps=30.0, max_guidance_weight=3.0) + ) + assert wire["class_path"] == "physicalai.runtime.RTCExecution" + assert wire["init_args"] == { + "chunk_size": 50, + "execution_horizon": 10, + "fps": 30.0, + "max_guidance_weight": 3.0, + } + + def test_rtc_live_latency_tracker_fails(self) -> None: + from physicalai.runtime import RTCExecution + + execution = RTCExecution(latency_tracker=MagicMock()) + with pytest.raises(ComponentConfigError, match=r"init_args\.latency_tracker"): + to_config(execution) + + def test_rtc_live_postprocessors_fail(self) -> None: + from physicalai.runtime import RTCExecution + + execution = RTCExecution(postprocessors=[MagicMock()]) + with pytest.raises(ComponentConfigError, match=r"init_args\.postprocessors"): + to_config(execution) + + +# --------------------------------------------------------------------------- +# PolicySource +# --------------------------------------------------------------------------- + + +class TestPolicySourceComponentConfig: + def test_model_only_omits_execution_and_queue( + self, inference_model: Any, _patch_adapter: MagicMock + ) -> None: + from physicalai.runtime import ChunkedActionQueue, LerpSmoother, PolicySource, SyncExecution + + source = PolicySource(model=inference_model, task="pick cube") + wire = _assert_construction_round_trip(source) + assert wire["class_path"] == "physicalai.runtime.PolicySource" + assert "execution" not in wire["init_args"] + assert "action_queue" not in wire["init_args"] + assert wire["init_args"]["task"] == "pick cube" + assert wire["init_args"]["model"]["class_path"] == "physicalai.inference.InferenceModel" + + restored = instantiate(wire) + assert isinstance(restored._execution, SyncExecution) # type: ignore[attr-defined] + assert isinstance(restored._action_queue, ChunkedActionQueue) # type: ignore[attr-defined] + assert isinstance(restored._action_queue._smoother, LerpSmoother) # type: ignore[attr-defined] + + def test_explicit_nested_graph(self, inference_model: Any, _patch_adapter: MagicMock) -> None: + from physicalai.runtime import ( + AsyncExecution, + ChunkedActionQueue, + PolicySource, + ReplaceSmoother, + ) + + source = PolicySource( + model=inference_model, + execution=AsyncExecution(request_threshold=0.2), + action_queue=ChunkedActionQueue(smoother=ReplaceSmoother()), + task="stack", + ) + wire = _assert_construction_round_trip(source) + assert wire["init_args"]["execution"]["class_path"] == "physicalai.runtime.AsyncExecution" + assert wire["init_args"]["action_queue"]["init_args"]["smoother"]["class_path"] == ( + "physicalai.runtime.ReplaceSmoother" + ) + assert wire["init_args"]["task"] == "stack" + + def test_rtc_policy_graph(self, inference_model: Any, _patch_adapter: MagicMock) -> None: + from physicalai.runtime import PolicySource, RTCActionQueue, RTCExecution + + source = PolicySource( + model=inference_model, + execution=RTCExecution(chunk_size=40, fps=30.0), + action_queue=RTCActionQueue(), + ) + wire = _assert_construction_round_trip(source) + assert wire["init_args"]["execution"]["class_path"] == "physicalai.runtime.RTCExecution" + assert wire["init_args"]["action_queue"]["class_path"] == "physicalai.runtime.RTCActionQueue" + + +# --------------------------------------------------------------------------- +# TeleopSource +# --------------------------------------------------------------------------- + + +class TestTeleopSourceComponentConfig: + def test_leader_nested_round_trip(self) -> None: + from physicalai.robot import SO101 + from physicalai.runtime import TeleopSource + + calibration = { + "shoulder_pan": {"id": 1, "drive_mode": 0, "homing_offset": 2048, "range_min": 707, "range_max": 3439}, + "shoulder_lift": {"id": 2, "drive_mode": 1, "homing_offset": 1024, "range_min": 669, "range_max": 3292}, + "elbow_flex": {"id": 3, "drive_mode": 0, "homing_offset": 2048, "range_min": 846, "range_max": 3069}, + "wrist_flex": {"id": 4, "drive_mode": 0, "homing_offset": 2048, "range_min": 956, "range_max": 3311}, + "wrist_roll": {"id": 5, "drive_mode": 0, "homing_offset": 2048, "range_min": 59, "range_max": 3946}, + "gripper": {"id": 6, "drive_mode": 0, "homing_offset": 2048, "range_min": 2026, "range_max": 3074}, + } + leader = SO101(port="/dev/ttyUSB0", calibration=calibration, role="leader") + source = TeleopSource(leader=leader) + wire = _assert_construction_round_trip(source) + assert wire["class_path"] == "physicalai.runtime.TeleopSource" + assert "to_action" not in wire["init_args"] + assert wire["init_args"]["leader"]["class_path"] == "physicalai.robot.SO101" + + def test_supplied_to_action_fails(self) -> None: + from physicalai.robot import SO101 + from physicalai.runtime import TeleopSource + + calibration = { + "shoulder_pan": {"id": 1, "drive_mode": 0, "homing_offset": 2048, "range_min": 707, "range_max": 3439}, + "shoulder_lift": {"id": 2, "drive_mode": 1, "homing_offset": 1024, "range_min": 669, "range_max": 3292}, + "elbow_flex": {"id": 3, "drive_mode": 0, "homing_offset": 2048, "range_min": 846, "range_max": 3069}, + "wrist_flex": {"id": 4, "drive_mode": 0, "homing_offset": 2048, "range_min": 956, "range_max": 3311}, + "wrist_roll": {"id": 5, "drive_mode": 0, "homing_offset": 2048, "range_min": 59, "range_max": 3946}, + "gripper": {"id": 6, "drive_mode": 0, "homing_offset": 2048, "range_min": 2026, "range_max": 3074}, + } + leader = SO101(port="/dev/ttyUSB0", calibration=calibration, role="leader") + source = TeleopSource(leader=leader, to_action=lambda obs: obs.joint_positions) + with pytest.raises(ComponentConfigError, match=r"init_args\.to_action"): + to_config(source) + + +# --------------------------------------------------------------------------- +# Callbacks +# --------------------------------------------------------------------------- + + +class TestCallbackComponentConfig: + def test_console_round_trip(self) -> None: + from physicalai.runtime import ConsoleCallback + + wire = _assert_construction_round_trip(ConsoleCallback(throttle_steps=15)) + assert wire["class_path"] == "physicalai.runtime.ConsoleCallback" + assert wire["init_args"] == {"throttle_steps": 15} + + def test_low_pass_round_trip(self) -> None: + from physicalai.runtime import LowPassFilterCallback + + wire = _assert_construction_round_trip(LowPassFilterCallback(alpha=0.3)) + assert wire["class_path"] == "physicalai.runtime.LowPassFilterCallback" + assert wire["init_args"] == {"alpha": 0.3} + + def test_jsonl_temp_path_round_trip(self, tmp_path: Path) -> None: + from physicalai.runtime import JsonlCallback + + path = tmp_path / "events.jsonl" + cb = JsonlCallback(path=path, record_chunks=True) + restored = None + try: + assert is_config_exportable(cb) + config = to_config(cb) + wire: dict[str, Any] = json.loads(json.dumps(config)) + restored = instantiate(wire) + assert type(restored) is type(cb) + assert to_config(restored) == wire + assert wire["class_path"] == "physicalai.runtime.JsonlCallback" + assert wire["init_args"]["path"] == str(path) + assert wire["init_args"]["record_chunks"] is True + finally: + cb.close() + if restored is not None: + restored.close() # type: ignore[attr-defined] + + def test_rerun_scalar_round_trip(self) -> None: + from physicalai.runtime import RerunCallback + + rr = MagicMock() + rr.__name__ = "rerun" + with patch.dict(sys.modules, {"rerun": rr}): + cb = RerunCallback( + mode="connect", + connect_addr="10.0.0.1:9876", + application_id="test-app", + image_decimation=5, + log_images=False, + ) + wire = _assert_construction_round_trip(cb) + assert wire["class_path"] == "physicalai.runtime.RerunCallback" + assert wire["init_args"]["mode"] == "connect" + assert wire["init_args"]["connect_addr"] == "10.0.0.1:9876" + assert wire["init_args"]["application_id"] == "test-app" + assert wire["init_args"]["image_decimation"] == 5 + assert wire["init_args"]["log_images"] is False + + def test_async_nested_console_round_trip(self) -> None: + from physicalai.runtime import AsyncCallback, ConsoleCallback + + cb = AsyncCallback(ConsoleCallback(throttle_steps=10), max_queue=64) + try: + wire = _assert_construction_round_trip(cb) + assert wire["class_path"] == "physicalai.runtime.AsyncCallback" + assert wire["init_args"]["max_queue"] == 64 + assert wire["init_args"]["inner"]["class_path"] == "physicalai.runtime.ConsoleCallback" + restored = instantiate(wire) + try: + assert isinstance(restored._inner, ConsoleCallback) # type: ignore[attr-defined] + finally: + restored.close() # type: ignore[attr-defined] + finally: + cb.close() + + def test_async_undecorated_inner_fails(self) -> None: + from physicalai.runtime import AsyncCallback + + class UndecoratedTickCallback: + def on_tick(self, event: object) -> None: + pass + + cb = AsyncCallback(UndecoratedTickCallback()) + try: + with pytest.raises(ComponentConfigError, match=r"init_args\.inner"): + to_config(cb) + finally: + cb.close() + + def test_async_rejects_action_hook_inner(self) -> None: + from physicalai.runtime import AsyncCallback, LowPassFilterCallback + + with pytest.raises(TypeError, match="action hooks"): + AsyncCallback(LowPassFilterCallback(alpha=0.5)) + + +# --------------------------------------------------------------------------- +# RobotRuntime (step 7) +# --------------------------------------------------------------------------- + + +_SAMPLE_CALIBRATION: dict[str, Any] = { + "shoulder_pan": {"id": 1, "drive_mode": 0, "homing_offset": 2048, "range_min": 707, "range_max": 3439}, + "shoulder_lift": {"id": 2, "drive_mode": 1, "homing_offset": 1024, "range_min": 669, "range_max": 3292}, + "elbow_flex": {"id": 3, "drive_mode": 0, "homing_offset": 2048, "range_min": 846, "range_max": 3069}, + "wrist_flex": {"id": 4, "drive_mode": 0, "homing_offset": 2048, "range_min": 956, "range_max": 3311}, + "wrist_roll": {"id": 5, "drive_mode": 0, "homing_offset": 2048, "range_min": 59, "range_max": 3946}, + "gripper": {"id": 6, "drive_mode": 0, "homing_offset": 2048, "range_min": 2026, "range_max": 3074}, +} + + +def _make_so101(*, port: str = "/dev/ttyACM0") -> Any: + from physicalai.robot import SO101 + + return SO101(port=port, calibration=_SAMPLE_CALIBRATION) + + +def _make_full_runtime(inference_model: Any) -> Any: + from physicalai.capture import UVCCamera + from physicalai.runtime import ConsoleCallback, PolicySource, RobotRuntime, SyncExecution + + return RobotRuntime( + robot=_make_so101(), + action_source=PolicySource( + model=inference_model, + execution=SyncExecution(request_threshold=0.25), + task="pick cube", + ), + fps=30.0, + cameras={"wrist": UVCCamera(device="/dev/video0", width=640, height=480, fps=30, backend="v4l2")}, + callbacks=[ConsoleCallback(throttle_steps=15)], + ) + + +class TestRobotRuntimeComponentConfig: + def test_complete_tree_instantiate_round_trip( + self, inference_model: Any, _patch_adapter: MagicMock + ) -> None: + runtime = _make_full_runtime(inference_model) + wire = _assert_construction_round_trip(runtime) + assert wire["class_path"] == "physicalai.runtime.RobotRuntime" + assert wire["init_args"]["fps"] == 30.0 + assert wire["init_args"]["robot"]["class_path"] == "physicalai.robot.SO101" + assert wire["init_args"]["action_source"]["class_path"] == "physicalai.runtime.PolicySource" + assert wire["init_args"]["cameras"]["wrist"]["class_path"] == "physicalai.capture.UVCCamera" + assert wire["init_args"]["callbacks"][0]["class_path"] == "physicalai.runtime.ConsoleCallback" + # Connection / session state are never part of the recipe. + assert "session_id" not in wire["init_args"] + assert "_connected" not in wire["init_args"] + from physicalai.runtime import RobotRuntime + + restored = instantiate(wire) + assert isinstance(restored, RobotRuntime) + assert restored._connected is False # noqa: SLF001 + assert restored._session_id == "" # noqa: SLF001 + + def test_complete_tree_jsonargparse_under_runtime( + self, + inference_model: Any, + _patch_adapter: MagicMock, + tmp_path: Path, + ) -> None: + from physicalai.capture import UVCCamera + from physicalai.cli.run import build_parser + from physicalai.robot import SO101 + from physicalai.runtime import ConsoleCallback, PolicySource, RobotRuntime + + runtime = _make_full_runtime(inference_model) + wire = to_config(runtime) + # CLI nests constructor args under ``runtime:`` (not the bare ComponentConfig). + # Prefer JSON for the ActionConfigFile payload — avoids PyYAML C-loader + # interaction with ``yaml.safe_dump`` that can poison later YAML parses in + # the same process. + cfg_path = tmp_path / "runtime.json" + cfg_path.write_text(json.dumps({"runtime": wire["init_args"]})) + + parser = build_parser() + ns = parser.parse_args(["--config", str(cfg_path)]) + restored = parser.instantiate(ns).runtime + assert isinstance(restored, RobotRuntime) + assert restored._connected is False # noqa: SLF001 + assert restored._session_id == "" # noqa: SLF001 + assert restored._fps == 30.0 # noqa: SLF001 + assert isinstance(restored.robot, SO101) + assert isinstance(restored.action_source, PolicySource) + assert set(restored.cameras) == {"wrist"} + assert isinstance(restored.cameras["wrist"], UVCCamera) + assert isinstance(restored._bus._callbacks[0], ConsoleCallback) # noqa: SLF001 + # jsonargparse may supply omitted ctor defaults; check public class_path shape + # after a JSON boundary (strips StrEnum identity / applied defaults noise). + restored_wire = json.loads(json.dumps(to_config(restored))) + assert restored_wire["class_path"] == "physicalai.runtime.RobotRuntime" + assert restored_wire["init_args"]["fps"] == 30.0 + assert restored_wire["init_args"]["robot"]["class_path"] == "physicalai.robot.SO101" + assert restored_wire["init_args"]["action_source"]["class_path"] == "physicalai.runtime.PolicySource" + assert restored_wire["init_args"]["cameras"]["wrist"]["class_path"] == "physicalai.capture.UVCCamera" + assert restored_wire["init_args"]["callbacks"][0]["class_path"] == "physicalai.runtime.ConsoleCallback" + + via_from_config = RobotRuntime.from_config(cfg_path) + assert isinstance(via_from_config, RobotRuntime) + assert via_from_config._connected is False # noqa: SLF001 + assert via_from_config._fps == 30.0 # noqa: SLF001 + + def test_bare_component_export_round_trips( + self, + inference_model: Any, + _patch_adapter: MagicMock, + tmp_path: Path, + ) -> None: + from physicalai.cli.run import build_parser + from physicalai.runtime import RobotRuntime + + runtime = _make_full_runtime(inference_model) + # The bare to_config shape (top-level class_path) — no manual rewrite + # to ``runtime:`` needed on either load path. + cfg_path = tmp_path / "runtime_export.json" + cfg_path.write_text(json.dumps(to_config(runtime))) + + parser = build_parser() + ns = parser.parse_args(["--config", str(cfg_path)]) + restored = parser.instantiate(ns).runtime + assert isinstance(restored, RobotRuntime) + assert restored._fps == 30.0 # noqa: SLF001 + assert restored._connected is False # noqa: SLF001 + + via_from_config = RobotRuntime.from_config(cfg_path) + assert isinstance(via_from_config, RobotRuntime) + assert via_from_config._fps == 30.0 # noqa: SLF001 + + def test_bare_export_foreign_class_path_rejected(self, tmp_path: Path) -> None: + from physicalai.config import ComponentConfigError + from physicalai.runtime import RobotRuntime + + cfg_path = tmp_path / "not_runtime.json" + cfg_path.write_text(json.dumps({"class_path": "physicalai.runtime.SyncExecution", "init_args": {}})) + + with pytest.raises(ComponentConfigError, match="does not resolve to"): + RobotRuntime.from_config(cfg_path) + + def test_omitted_cameras_and_callbacks( + self, inference_model: Any, _patch_adapter: MagicMock + ) -> None: + from physicalai.runtime import PolicySource, RobotRuntime + + runtime = RobotRuntime( + robot=_make_so101(), + action_source=PolicySource(model=inference_model), + fps=20.0, + ) + wire = _assert_construction_round_trip(runtime) + assert "cameras" not in wire["init_args"] + assert "callbacks" not in wire["init_args"] + + def test_empty_callbacks_stay_empty( + self, inference_model: Any, _patch_adapter: MagicMock + ) -> None: + from physicalai.runtime import PolicySource, RobotRuntime + + runtime = RobotRuntime( + robot=_make_so101(), + action_source=PolicySource(model=inference_model), + fps=20.0, + callbacks=[], + ) + wire = _assert_construction_round_trip(runtime) + assert wire["init_args"]["callbacks"] == [] + + def test_undecorated_robot_fails( + self, inference_model: Any, _patch_adapter: MagicMock + ) -> None: + from physicalai.runtime import PolicySource, RobotRuntime + + class UndecoratedRobot: + def __init__(self) -> None: + pass + + runtime = RobotRuntime( + robot=UndecoratedRobot(), # type: ignore[arg-type] + action_source=PolicySource(model=inference_model), + fps=10.0, + ) + with pytest.raises(ComponentConfigError, match=r"init_args\.robot"): + to_config(runtime) + + def test_undecorated_action_source_fails(self) -> None: + from physicalai.runtime import RobotRuntime + + class UndecoratedSource: + def connect(self, **_kwargs: object) -> None: ... + def disconnect(self) -> None: ... + def update(self, *_args: object, **_kwargs: object) -> object: + return None + + runtime = RobotRuntime( + robot=_make_so101(), + action_source=UndecoratedSource(), # type: ignore[arg-type] + fps=10.0, + ) + with pytest.raises(ComponentConfigError, match=r"init_args\.action_source"): + to_config(runtime) + + def test_undecorated_camera_fails( + self, inference_model: Any, _patch_adapter: MagicMock + ) -> None: + from physicalai.runtime import PolicySource, RobotRuntime + + class UndecoratedCamera: + def __init__(self) -> None: + pass + + runtime = RobotRuntime( + robot=_make_so101(), + action_source=PolicySource(model=inference_model), + fps=10.0, + cameras={"wrist": UndecoratedCamera()}, # type: ignore[dict-item] + ) + with pytest.raises(ComponentConfigError, match=r"init_args\.cameras\.wrist"): + to_config(runtime) + + def test_undecorated_callback_fails( + self, inference_model: Any, _patch_adapter: MagicMock + ) -> None: + from physicalai.runtime import PolicySource, RobotRuntime + + class UndecoratedCallback: + def on_tick(self, event: object) -> None: + pass + + runtime = RobotRuntime( + robot=_make_so101(), + action_source=PolicySource(model=inference_model), + fps=10.0, + callbacks=[UndecoratedCallback()], + ) + with pytest.raises(ComponentConfigError, match=r"init_args\.callbacks\[0\]"): + to_config(runtime) + + def test_jsonargparse_sees_original_ctor_signature(self) -> None: + import inspect + + from physicalai.runtime import RobotRuntime + + params = list(inspect.signature(RobotRuntime.__init__).parameters) + assert params == ["self", "robot", "action_source", "fps", "cameras", "callbacks"] + + def test_nested_shared_robot_and_shared_camera( + self, inference_model: Any, _patch_adapter: MagicMock + ) -> None: + from physicalai.capture import SharedCamera + from physicalai.robot import SharedRobot + from physicalai.runtime import PolicySource, RobotRuntime, SyncExecution + + runtime = RobotRuntime( + robot=SharedRobot( + "follower", + robot={ + "class_path": "tests.unit.robot.transport.fake.FakeRobot", + "init_args": {"port": "/dev/fake-rt", "device_ids": ["fake:follower"]}, + }, + ), + action_source=PolicySource( + model=inference_model, + execution=SyncExecution(), + ), + fps=15.0, + cameras={ + "wrist": SharedCamera( + camera={ + "class_path": "physicalai.capture.UVCCamera", + "init_args": { + "device": "/dev/video2", + "width": 320, + "height": 240, + "fps": 15, + "backend": "v4l2", + }, + }, + ), + }, + ) + wire = _assert_construction_round_trip(runtime) + assert wire["init_args"]["robot"]["class_path"] == "physicalai.robot.SharedRobot" + assert wire["init_args"]["robot"]["init_args"]["name"] == "follower" + nested_robot = wire["init_args"]["robot"]["init_args"]["robot"] + assert nested_robot["class_path"] == "tests.unit.robot.transport.fake.FakeRobot" + assert wire["init_args"]["cameras"]["wrist"]["class_path"] == "physicalai.capture.SharedCamera" + nested_cam = wire["init_args"]["cameras"]["wrist"]["init_args"]["camera"] + assert nested_cam["class_path"] == "physicalai.capture.UVCCamera" + restored = instantiate(wire) + assert isinstance(restored, RobotRuntime) + assert isinstance(restored.robot, SharedRobot) + assert isinstance(restored.cameras["wrist"], SharedCamera) + assert restored._connected is False # noqa: SLF001 + assert not restored.robot.is_connected() + assert not restored.cameras["wrist"].is_connected