From 05e1f107ff2cf55e272ad7da2b314dfc17037b7c Mon Sep 17 00:00:00 2001 From: Max Xiang Date: Mon, 27 Jul 2026 12:00:36 +0200 Subject: [PATCH 1/5] feat(robot): export config and cut over SharedRobot transport Decorate robot drivers with @export_config, switch robot owner stdin envelopes to robot: ComponentConfig, and update SharedRobot.from_config() and physicalai robot serve CLI. Co-authored-by: Cursor --- docs/how-to/runtime/share-a-robot.md | 43 +- src/physicalai/cli/robot.py | 52 ++- src/physicalai/robot/so101/calibration.py | 8 + src/physicalai/robot/so101/so101.py | 15 +- .../robot/transport/_owner_config.py | 164 +++++--- .../robot/transport/_owner_worker.py | 16 +- .../robot/transport/_shared_robot.py | 122 ++++-- .../robot/trossen/bimanual_widowxai.py | 2 + src/physicalai/robot/trossen/widowxai.py | 2 + tests/unit/cli/test_robot_cli.py | 75 +++- .../unit/robot/test_robot_component_config.py | 374 ++++++++++++++++++ tests/unit/robot/transport/fake.py | 3 + tests/unit/robot/transport/test_latency.py | 6 +- .../unit/robot/transport/test_owner_config.py | 246 +++++++++--- .../robot/transport/test_owner_handshake.py | 16 +- .../unit/robot/transport/test_owner_worker.py | 38 +- .../unit/robot/transport/test_shared_robot.py | 54 ++- 17 files changed, 1013 insertions(+), 223 deletions(-) create mode 100644 tests/unit/robot/test_robot_component_config.py diff --git a/docs/how-to/runtime/share-a-robot.md b/docs/how-to/runtime/share-a-robot.md index fa731fb..bf6788f 100644 --- a/docs/how-to/runtime/share-a-robot.md +++ b/docs/how-to/runtime/share-a-robot.md @@ -4,7 +4,7 @@ while any number of other processes read its state and send actions over [Zenoh](https://zenoh.io/). It satisfies the same `Robot` protocol as a direct driver, so it is a drop-in replacement anywhere a robot is expected -(including `PolicyRuntime`). +(including `RobotRuntime`). ## Install @@ -17,21 +17,23 @@ pip install "physicalai[transport]" Every `SharedRobot` has a required, caller-chosen logical `name` — it keys the Zenoh topics directly. The first `SharedRobot` constructed for a given `name` that finds no existing owner spawns one (in a detached subprocess); -later instances (same or different process, same `name`) attach to it: +later instances (same or different process, same `name`) attach to it. + +Construction uses `robot=` or `from_config()`. Prefer `from_config()` when you +already have a recipe. A disconnected `@export_config` driver can be passed +directly to the constructor or exported explicitly: ```python import numpy as np -from physicalai.robot import SharedRobot -from physicalai.robot.so101 import SO101 - -robot = SharedRobot( - "left-arm", - robot_class=SO101, - robot_kwargs={ - "port": "/dev/ttyUSB0", - "calibration": "~/.cache/calibration/so101.json", # a path — kwargs must be serializable - }, +from physicalai.config import to_config +from physicalai.robot import SO101, SharedRobot + +driver = SO101( + port="/dev/ttyUSB0", + calibration="~/.cache/calibration/so101.json", # path stays relative/as given ) +robot = SharedRobot.from_config(to_config(driver), name="left-arm") +# or: SharedRobot("left-arm", robot={"class_path": "physicalai.robot.SO101", "init_args": {...}}) robot.connect() obs = robot.get_observation() # pull latest state, non-blocking @@ -40,10 +42,9 @@ robot.send_action(np.asarray(obs.joint_positions), goal_time=0.1) robot.disconnect() # detaches; the owner keeps running ``` -`robot_class` can be the class object (normalized to its dotted import path) -or the path itself, e.g. `robot_class="physicalai.robot.so101.SO101"` — any -importable class works, including third-party plugin robots, with no -registry to update. +Any importable `@export_config` robot class works (including third-party +plugins) — pass its public `class_path` + `init_args`; there is no flat +`robot_class` / `robot_kwargs` API. ## Serve a robot in the foreground @@ -145,10 +146,12 @@ loopback port from `name`. Opt into cross-host reachability explicitly when you need it: ```python -robot = SharedRobot( - "left-arm", - robot_class=SO101, - robot_kwargs={"port": "/dev/ttyUSB0", "calibration": "calibration.json"}, +robot = SharedRobot.from_config( + { + "class_path": "physicalai.robot.SO101", + "init_args": {"port": "/dev/ttyUSB0", "calibration": "calibration.json"}, + }, + name="left-arm", allow_remote=True, ) ``` diff --git a/src/physicalai/cli/robot.py b/src/physicalai/cli/robot.py index b07ba97..58220c4 100644 --- a/src/physicalai/cli/robot.py +++ b/src/physicalai/cli/robot.py @@ -17,6 +17,7 @@ from loguru import logger from physicalai.cli._spec import SubcommandSpec # noqa: PLC2701 +from physicalai.config import ComponentConfigError from physicalai.robot.errors import RobotTransportError from physicalai.robot.transport import ( DEFAULT_RATE_HZ, @@ -55,12 +56,11 @@ def _build_serve_parser() -> ArgumentParser: parser = ArgumentParser(description=_SERVE_HELP) parser.add_argument("--config", action=ActionConfigFile, help="YAML/JSON config file.") parser.add_argument("--name", type=str, required=True, help="Robot logical name.") - parser.add_argument("--robot_class", type=str, required=True, help="Trusted dotted path to the driver class.") parser.add_argument( - "--robot_kwargs", + "--robot", type=dict, - default=None, - help="JSON-serializable driver constructor arguments.", + required=True, + help="Trusted robot ComponentConfig (class_path + init_args).", ) parser.add_argument( "--allow_remote", @@ -133,6 +133,45 @@ def _log_serve_start(config: RobotOwnerConfig) -> None: logger.info(f"Starting robot {config.name!r} using {config.robot_class} [{mode_tag}]") +def _mapping_from_cfg(value: object) -> dict[str, object]: + """Convert a jsonargparse dict/Namespace value to a plain dict. + + Returns: + A shallow ``dict`` copy of *value*. + + Raises: + TypeError: If *value* is not ``None``, a ``dict``, or a Namespace-like + object with ``as_dict()``. + """ + if value is None: + return {} + if isinstance(value, dict): + return dict(value) + as_dict = getattr(value, "as_dict", None) + if callable(as_dict): + converted = as_dict() + if isinstance(converted, dict): + return dict(converted) + msg = f"expected a mapping, got {type(value).__name__}" + raise TypeError(msg) + + +def _robot_config_from_serve_cfg(cfg: Namespace) -> dict[str, object]: + """Require ``--robot`` ComponentConfig for foreground serve. + + Returns: + A ComponentConfig mapping for :class:`RobotOwnerConfig`. + + Raises: + ValueError: If ``--robot`` is missing. + """ + robot = getattr(cfg, "robot", None) + if robot is None: + msg = "robot serve requires --robot ComponentConfig (class_path + init_args)" + raise ValueError(msg) + return _mapping_from_cfg(robot) + + def _format_duration(seconds: float) -> str: """Format elapsed seconds as ``HH:MM:SS``. @@ -174,13 +213,12 @@ def serve(cfg: Namespace) -> int: try: config = RobotOwnerConfig( name=cfg.name, - robot_class=cfg.robot_class, - robot_kwargs=dict(cfg.robot_kwargs or {}), + robot=_robot_config_from_serve_cfg(cfg), allow_remote=cfg.allow_remote, rate_hz=cfg.rate_hz, idle_timeout=None, ) - except (TypeError, ValueError) as exc: + except (TypeError, ValueError, ComponentConfigError) as exc: logger.error(f"Invalid robot configuration: {exc}") return 1 diff --git a/src/physicalai/robot/so101/calibration.py b/src/physicalai/robot/so101/calibration.py index 4765df6..1f0cbad 100644 --- a/src/physicalai/robot/so101/calibration.py +++ b/src/physicalai/robot/so101/calibration.py @@ -71,6 +71,14 @@ def to_dict(self) -> dict[str, dict[str, int]]: """ return {name: joint.to_dict() for name, joint in self.joints.items()} + def to_config_value(self) -> dict[str, dict[str, int]]: + """Encode as a constructor-compatible dict for :func:`~physicalai.config.to_config`. + + Returns: + The LeRobot calibration mapping produced by :meth:`to_dict`. + """ + return self.to_dict() + @classmethod def from_dict(cls, data: object) -> SO101Calibration: """Build a calibration object from parsed JSON data. diff --git a/src/physicalai/robot/so101/so101.py b/src/physicalai/robot/so101/so101.py index 1f916f7..67f0716 100644 --- a/src/physicalai/robot/so101/so101.py +++ b/src/physicalai/robot/so101/so101.py @@ -41,6 +41,7 @@ PortHandler, ) +from physicalai.config import export_config from physicalai.robot import Robot from physicalai.robot.device_ids import device_id_from_serial_port from physicalai.robot.so101.calibration import SO101Calibration @@ -93,6 +94,7 @@ def state(self) -> np.ndarray: return self.joint_positions +@export_config(class_path="physicalai.robot.SO101") class SO101(Robot): """Driver for the SO-101 robot arm (6-DOF, Feetech STS3215 servos). @@ -119,12 +121,14 @@ class SO101(Robot): def __init__( self, port: str, - calibration: SO101Calibration | str | Path | dict | None, + # Defaulted so the required-arg is jsonargparse-loadable when None + # (uncalibrated export); the guard below still enforces explicit intent. + calibration: SO101Calibration | str | Path | dict | None = None, baudrate: int = 1_000_000, role: Literal["leader", "follower"] = "follower", unit: SO101Unit = "normalized", *, - _allow_uncalibrated: bool = False, # must be passed by keyword + allow_uncalibrated: bool = False, ) -> None: """Initialize the SO-101 driver (does not open the connection). @@ -132,7 +136,8 @@ def __init__( * ``SO101Calibration`` — use an already loaded calibration object. * ``str | Path`` — load LeRobot calibration JSON from disk. - * ``None`` — only allowed via :meth:`SO101.uncalibrated` for raw ticks. + * ``None`` — raw-ticks mode; requires ``allow_uncalibrated=True`` + (see :meth:`SO101.uncalibrated`). Raises: ValueError: If ``role`` is not ``"leader"`` or ``"follower"``. @@ -148,7 +153,7 @@ def __init__( self._unit: SO101Unit = unit # Calibration ------------------------------------------------------- - if calibration is None and not _allow_uncalibrated: + if calibration is None and not allow_uncalibrated: msg = ( "calibration is required for SO101. " "Pass a calibration object/path, or use SO101.uncalibrated(...) " @@ -202,7 +207,7 @@ def uncalibrated( baudrate=baudrate, role=role, unit=unit, - _allow_uncalibrated=True, + allow_uncalibrated=True, ) @property diff --git a/src/physicalai/robot/transport/_owner_config.py b/src/physicalai/robot/transport/_owner_config.py index 88643d1..9fe6bf8 100644 --- a/src/physicalai/robot/transport/_owner_config.py +++ b/src/physicalai/robot/transport/_owner_config.py @@ -9,12 +9,18 @@ subprocess stdin handshake. The owner must construct the robot driver itself: a live serial/socket -handle cannot cross a process boundary (D15). Only an importable -``robot_class`` plus JSON-serializable ``robot_kwargs`` survive that -boundary — arbitrary robot types, including third-party plugins, work -without any registry lookup here. - -Security: *robot_class* is trusted local application/config input, exactly +handle cannot cross a process boundary (D15). Only a trusted +:class:`~physicalai.config.ComponentConfig` (``class_path`` + ``init_args``) +survives that boundary — arbitrary robot types, including third-party +plugins, work without any registry lookup here. + +Private stdin is ``robot: ComponentConfig`` only. The owner envelope is +validated schema-positively: required ``robot``, known transport keys, and +rejection of unknown keys (including legacy flat ``robot_class`` / +``robot_kwargs``) before import or hardware access. Public ``SharedRobot`` +and ``physicalai robot serve`` use the same ``robot`` / ``--robot`` shape. + +Security: ``class_path`` is trusted local application/config input, exactly like a jsonargparse ``class_path`` (``docs/development/security.md`` rules 4, 9, 11). It must never originate from network-received data (e.g. a ``/metadata`` payload) — that would let an untrusted peer choose an @@ -23,14 +29,20 @@ from __future__ import annotations -import json import math -from dataclasses import dataclass, field +from dataclasses import dataclass from typing import TYPE_CHECKING, Any -from ._importing import import_dotted_path +from physicalai.config import ( + ComponentConfig, + instantiate, + normalize_component_config, + validate_envelope, +) if TYPE_CHECKING: + from collections.abc import Mapping + from physicalai.robot.interface import Robot DEFAULT_RATE_HZ = 100.0 @@ -42,38 +54,46 @@ justify a different value for a specific robot class. """ +# Allowed keys on owner stdin handshake payloads. Everything else (including +# legacy flat robot_class / robot_kwargs) is an unknown-key schema error. +# Keep this the single allowlist — :meth:`RobotOwnerConfig.from_json_dict` +# and any future reconfigure path should share :func:`validate_owner_config`. +_OWNER_ENVELOPE_KEYS = frozenset({ + "name", + "robot", + "allow_remote", + "rate_hz", + "idle_timeout", +}) -def normalize_robot_class(robot_class: type | str) -> str: - """Normalize a robot class reference to an importable dotted path. - - An explicit string path is trusted and returned unchanged. A class - object is converted via ``cls.__module__ + "." + cls.__qualname__``, - because Python does not retain the re-export path through which a - class happened to be imported by the caller. - Args: - robot_class: A class object, or its dotted import path as a string. +def validate_owner_config(data: Mapping[str, Any]) -> Mapping[str, object]: + """Validate an owner stdin payload schema-positively. Returns: - The normalized dotted path. - - Raises: - TypeError: If *robot_class* is neither a string nor a class. - ValueError: If a class object is a local class (defined inside a - function) — those have no stable import path and cannot be - reconstructed in the owner subprocess. + The validated ``robot`` ComponentConfig mapping (see + :func:`normalize_robot_config` for the JSON-serializability check). """ - if isinstance(robot_class, str): - return robot_class - if not isinstance(robot_class, type): - msg = f"robot_class must be a class or a dotted path string, got {type(robot_class).__name__}" - raise TypeError(msg) + return validate_envelope( + data, + component_key="robot", + allowed_keys=_OWNER_ENVELOPE_KEYS, + envelope_name="owner", + ) + - path = f"{robot_class.__module__}.{robot_class.__qualname__}" - if "" in robot_class.__qualname__: - msg = f"robot_class {path!r} is a local class and cannot be auto-spawned; give it a module-level definition" - raise ValueError(msg) - return path +def normalize_robot_config(robot: Mapping[str, object]) -> ComponentConfig: + """Validate a robot ComponentConfig without importing its ``class_path``. + + Returns: + A validated config whose ``class_path`` is a dotted import path. + """ + return normalize_component_config( + robot, + component_key="robot", + class_label="robot_class", + json_hint=" (e.g. paths as str, not objects)", + ) @dataclass(frozen=True) @@ -88,9 +108,7 @@ class RobotOwnerConfig: Attributes: name: The robot's logical name (keys the Zenoh topics). - robot_class: Normalized importable dotted path to the driver class. - robot_kwargs: JSON-serializable keyword arguments forwarded to the - driver constructor (e.g. ``calibration`` as a file path). + robot: Trusted construction config (``class_path`` + ``init_args``). allow_remote: Whether the owner's Zenoh session is reachable beyond localhost. Fixed for the owner's lifetime once spawned. ``True`` exposes an unauthenticated physical ``/action`` endpoint @@ -101,18 +119,17 @@ class RobotOwnerConfig: """ name: str - robot_class: str - robot_kwargs: dict[str, Any] = field(default_factory=dict) + robot: Mapping[str, object] allow_remote: bool = False rate_hz: float = DEFAULT_RATE_HZ idle_timeout: float | None = 10.0 def __post_init__(self) -> None: - """Validate ``rate_hz`` and that ``robot_kwargs`` is JSON-serializable. + """Validate transport fields and normalize ``robot`` to a public ComponentConfig. Raises: - ValueError: If ``rate_hz`` is not finite and positive, or - ``robot_kwargs`` contains non-JSON-serializable values. + ValueError: If ``rate_hz`` / ``idle_timeout`` are invalid, or + ``robot`` is not a JSON-serializable ComponentConfig. """ if ( isinstance(self.rate_hz, bool) @@ -133,25 +150,32 @@ def __post_init__(self) -> None: from ._ids import validate_name # noqa: PLC0415 validate_name(self.name) - if not isinstance(self.robot_class, str) or not self.robot_class.strip() or "." not in self.robot_class: - msg = f"robot_class must be a nonempty dotted path, got {self.robot_class!r}" - raise ValueError(msg) - try: - json.dumps(self.robot_kwargs) - except TypeError as exc: - msg = f"robot_kwargs must be JSON-serializable (e.g. paths as str, not objects): {exc}" - raise ValueError(msg) from exc + object.__setattr__(self, "robot", normalize_robot_config(self.robot)) + + @property + def robot_class(self) -> str: + """Public ``class_path`` advertised on network metadata as ``robot_class``.""" + return str(self.robot["class_path"]) def to_json_dict(self) -> dict[str, Any]: """Serialize to a JSON-safe dictionary for the stdin handshake. Returns: - Dictionary with every dataclass field. + Dictionary with every dataclass field (new ``robot:`` shape only). + + Raises: + TypeError: If ``robot.init_args`` is not a mapping. """ + init_args = self.robot["init_args"] + if not isinstance(init_args, dict): + msg = f"robot.init_args must be a mapping, got {type(init_args).__name__}" + raise TypeError(msg) return { "name": self.name, - "robot_class": self.robot_class, - "robot_kwargs": self.robot_kwargs, + "robot": { + "class_path": self.robot["class_path"], + "init_args": dict(init_args), + }, "allow_remote": self.allow_remote, "rate_hz": self.rate_hz, "idle_timeout": self.idle_timeout, @@ -161,16 +185,27 @@ def to_json_dict(self) -> dict[str, Any]: def from_json_dict(cls, data: dict[str, Any]) -> RobotOwnerConfig: """Deserialize from a JSON dictionary. + Uses :func:`validate_owner_config` so the owner envelope is validated + schema-positively (required ``robot``, known transport keys, unknown + keys rejected) before any import or hardware access. + Args: data: Dictionary produced by :meth:`to_json_dict`. Returns: A new :class:`RobotOwnerConfig` instance. + + Raises: + TypeError: If ``data`` is not a mapping. """ + if not isinstance(data, dict): + msg = f"owner config must be a mapping, got {type(data).__name__}" + raise TypeError(msg) + + robot = validate_owner_config(data) return cls( name=data["name"], - robot_class=data["robot_class"], - robot_kwargs=data.get("robot_kwargs", {}), + robot=robot, allow_remote=data.get("allow_remote", False), rate_hz=data.get("rate_hz", DEFAULT_RATE_HZ), idle_timeout=data.get("idle_timeout", 10.0), @@ -179,14 +214,21 @@ def from_json_dict(cls, data: dict[str, Any]) -> RobotOwnerConfig: def build(self) -> Robot: """Instantiate the robot driver described by this config. + Owner stdin is a trusted parent→child handshake only — never pass + network metadata to this path. Uses :func:`physicalai.config.instantiate` + on the trusted ``robot`` ComponentConfig, then verifies the + :class:`~physicalai.robot.Robot` protocol. + Returns: A new, not-yet-connected driver instance. Raises: - TypeError: If ``robot_class`` does not resolve to a class. + TypeError: If the instantiated object does not satisfy ``Robot``. """ - cls = import_dotted_path(self.robot_class) - if not isinstance(cls, type): - msg = f"robot_class {self.robot_class!r} does not resolve to a class (got {type(cls).__name__})" + from physicalai.robot.interface import Robot # noqa: PLC0415 + + driver = instantiate(self.robot) # type: ignore[arg-type] + if not isinstance(driver, Robot): + msg = f"{self.robot_class!r} does not satisfy the Robot protocol (got {type(driver).__name__})" raise TypeError(msg) - return cls(**self.robot_kwargs) + return driver diff --git a/src/physicalai/robot/transport/_owner_worker.py b/src/physicalai/robot/transport/_owner_worker.py index fed690c..69ff653 100644 --- a/src/physicalai/robot/transport/_owner_worker.py +++ b/src/physicalai/robot/transport/_owner_worker.py @@ -34,6 +34,7 @@ from loguru import logger +from physicalai.config import ComponentConfigError from physicalai.robot.errors import RobotTransportError from physicalai.robot.interface import Robot from physicalai.robot.transport._codec import ( # noqa: PLC2701 @@ -168,7 +169,7 @@ def _build_metadata( """Assemble the ``/metadata`` record advertised by the queryable. Args: - config: Worker config (for name / robot_class). + config: Worker config (for name / public ``robot.class_path``). driver: Connected driver (for joint names). device_ids: This owner's sorted, deduplicated device ids. Omitted from advertised metadata when remote transport is enabled. @@ -183,7 +184,9 @@ def _build_metadata( metadata: dict[str, Any] = { "protocol_version": ROBOT_TRANSPORT_PROTOCOL_VERSION, "name": config.name, - "robot_class": config.robot_class, + # Network key stays ``robot_class`` (protocol version unchanged); value is + # the normalized public ComponentConfig class_path. + "robot_class": config.robot["class_path"], "host": default_host(), "joint_names": joint_names, "num_joints": len(joint_names), @@ -436,15 +439,16 @@ def _startup(config: RobotOwnerConfig) -> _Endpoints: _StartupError: Naming the failure phase, for the caller to report via :func:`signal_error`. """ + class_path = config.robot["class_path"] try: - logger.trace(f"Constructing robot driver {config.robot_class!r}") + logger.trace(f"Constructing robot driver {class_path!r}") driver = config.build() except Exception as exc: - msg = f"failed to construct {config.robot_class!r}: {exc}" + msg = f"failed to construct {class_path!r}: {exc}" raise _StartupError(msg, phase="construction_failed") from exc if not isinstance(driver, Robot): - msg = f"{config.robot_class!r} does not satisfy the Robot protocol" + msg = f"{class_path!r} does not satisfy the Robot protocol" raise _StartupError(msg, phase="construction_failed") device_ids = tuple(sorted(set(driver.device_ids))) @@ -558,7 +562,7 @@ def main() -> int: sys.stdin.close() try: config = RobotOwnerConfig.from_json_dict(json.loads(raw)) - except (json.JSONDecodeError, ValueError, KeyError, TypeError) as exc: + except (json.JSONDecodeError, ValueError, KeyError, TypeError, ComponentConfigError) as exc: signal_error(f"invalid worker config: {exc}", phase="invalid_config") return 1 diff --git a/src/physicalai/robot/transport/_shared_robot.py b/src/physicalai/robot/transport/_shared_robot.py index 50c3e39..c943e8d 100644 --- a/src/physicalai/robot/transport/_shared_robot.py +++ b/src/physicalai/robot/transport/_shared_robot.py @@ -9,6 +9,13 @@ that finds no existing owner spawns one; later instances (for the same *name*, anywhere reachable) attach. +Construction is :class:`~physicalai.config.ComponentConfig`-only via +``robot=`` or :meth:`from_config`. ``robot`` is declared as an +``@export_config(config_args=...)`` argument, so a nested +:func:`~physicalai.config.instantiate` hands over the recipe instead of +building a driver this process would immediately discard. Pass ``robot=None`` +(or use :meth:`attach`) for attach-only. + Unlike the superseded connection-derived ``robot_id``, *name* is a required, caller-chosen logical identifier — routing never needs a live driver instance to resolve. Physical device identity @@ -25,10 +32,12 @@ from __future__ import annotations import time +from collections.abc import Mapping from typing import TYPE_CHECKING, Any from loguru import logger +from physicalai.config import export_config from physicalai.robot.errors import ( RobotDeviceAlreadyOwned, RobotNameConflict, @@ -40,16 +49,16 @@ from ._codec import ROBOT_TRANSPORT_PROTOCOL_VERSION, TransportObservation, decode_metadata, decode_state, encode_action from ._ids import KEY_PREFIX, METADATA_WILDCARD, action_key, metadata_key, state_key, validate_name from ._lock import active_owner_device_ids, registered_owner_names -from ._owner_config import DEFAULT_RATE_HZ, normalize_robot_class +from ._owner_config import DEFAULT_RATE_HZ, normalize_robot_config from ._session import open_session if TYPE_CHECKING: - from collections.abc import Mapping - import numpy as np + from physicalai.config import ComponentConfig from physicalai.robot import RobotObservation + _PROBE_TIMEOUT = 1.0 _RACE_RETRY_TIMEOUT = 5.0 _RETRY_INTERVAL = 0.2 @@ -169,6 +178,7 @@ def _append_replies(query_session: Any, query_timeout: float) -> None: # noqa: return list({metadata.get("name"): metadata for metadata in robots}.values()) +@export_config(class_path="physicalai.robot.SharedRobot", config_args=("robot",)) class SharedRobot: """Robot subscriber that attaches to (or spawns) a shared owner process. @@ -177,17 +187,25 @@ class SharedRobot: Zenoh. Satisfies the :class:`~physicalai.robot.Robot` protocol, so it is a drop-in replacement for a direct driver. + Prefer :meth:`from_config`. The constructor takes + ``robot: ComponentConfig`` to spawn, or ``robot=None`` (attach-only; + :meth:`attach` is the explicit form). + + Opted into :func:`~physicalai.config.export_config` as a **construction + recipe** only (name, nested ``robot`` ComponentConfig, transport knobs). + Connection / Zenoh session / publisher state is never part of + :func:`~physicalai.config.to_config`. + Args: name: Required logical name — keys the Zenoh topics directly. Two instances constructed with the same *name* (anywhere reachable under the chosen transport scope) share one owner. - robot_class: Driver class (or its dotted import path) to spawn if - no owner exists yet for *name*. ``None`` means attach-only — - use :meth:`attach` for that case instead of passing this - directly. - robot_kwargs: JSON-serializable driver constructor kwargs (e.g. - ``port``, ``calibration`` as a path, ``role`` as a str), used - only when this instance spawns the owner. + robot: Trusted driver :class:`~physicalai.config.ComponentConfig` to + spawn if no owner exists yet for *name*. ``None`` means + attach-only — use :meth:`attach` for that case. Declared as an + ``@export_config`` config arg, so nested + :func:`~physicalai.config.instantiate` passes the recipe through + without constructing the driver here. allow_remote: Whether this instance's own session — and, if it spawns the owner, the owner's session for its whole lifetime — is reachable beyond localhost. Defaults to the secure, @@ -203,8 +221,7 @@ def __init__( self, name: str, *, - robot_class: type | str | None = None, - robot_kwargs: Mapping[str, object] | None = None, + robot: ComponentConfig | Mapping[str, object] | None = None, allow_remote: bool = False, rate_hz: float = DEFAULT_RATE_HZ, idle_timeout: float = 10.0, @@ -212,8 +229,7 @@ def __init__( _session: object | None = None, ) -> None: self._name = validate_name(name) - self._robot_class = normalize_robot_class(robot_class) if robot_class is not None else None - self._robot_kwargs: dict[str, object] = dict(robot_kwargs or {}) + self._robot = None if robot is None else normalize_robot_config(robot) self._allow_remote = allow_remote self._rate_hz = rate_hz self._idle_timeout = idle_timeout @@ -229,6 +245,49 @@ def __init__( self._latest: TransportObservation | None = None self._connected = False + @classmethod + def _physicalai_normalize_captured_init_args(cls, supplied: dict[str, object]) -> None: + robot = supplied.get("robot") + if isinstance(robot, Mapping): + supplied["robot"] = normalize_robot_config(robot) + + @classmethod + def from_config( + cls, + robot_config: ComponentConfig | Mapping[str, object], + *, + name: str, + allow_remote: bool = False, + rate_hz: float = DEFAULT_RATE_HZ, + idle_timeout: float = 10.0, + connect_timeout: float = 10.0, + _session: object | None = None, + ) -> SharedRobot: + """Primary API: spawn/attach from a trusted robot ComponentConfig. + + Args: + robot_config: Trusted ``class_path`` + ``init_args`` for the driver. + name: Logical owner name (Zenoh topic key). + allow_remote: Whether this session / spawned owner may leave localhost. + rate_hz: Owner loop rate when this instance spawns the owner. + idle_timeout: Idle self-exit timeout for a spawned owner. + connect_timeout: Overall budget for :meth:`connect`. + + Returns: + A ``SharedRobot`` that stores the normalized ComponentConfig and + writes only the new owner stdin shape on spawn. + """ + # Omit default None so @export_config does not capture "_session": null. + return cls( + name, + robot=robot_config, + allow_remote=allow_remote, + rate_hz=rate_hz, + idle_timeout=idle_timeout, + connect_timeout=connect_timeout, + **({} if _session is None else {"_session": _session}), + ) + @classmethod def attach( cls, @@ -253,7 +312,15 @@ def attach( Returns: A ``SharedRobot`` that never spawns an owner. """ - return cls(name, allow_remote=allow_remote, connect_timeout=connect_timeout, _session=_session) + # Omit default None so @export_config does not capture "_session": null. + if _session is not None: + return cls(name, allow_remote=allow_remote, connect_timeout=connect_timeout, _session=_session) + return cls(name, allow_remote=allow_remote, connect_timeout=connect_timeout) + + @property + def _robot_class(self) -> str | None: + """Public ``class_path`` used for metadata conflict diagnostics.""" + return None if self._robot is None else self._robot["class_path"] @property def name(self) -> str: @@ -336,8 +403,8 @@ def _spawn_or_reprobe(self, timeout: float) -> dict[str, Any]: The owner's metadata record. Raises: - RobotTransportError: If ``robot_class`` was not given (attach- - only) or spawning failed for a reason other than a benign + RobotTransportError: If no robot config was given (attach-only) + or spawning failed for a reason other than a benign same-device race. RobotNameConflict: If the race was against a *different*-device owner under the same name. @@ -345,8 +412,8 @@ def _spawn_or_reprobe(self, timeout: float) -> dict[str, Any]: locked under another name — propagated as-is, no re-probe needed. """ - if self._robot_class is None: - msg = f"no owner found for {self._name!r} (attach-only mode: robot_class not provided)" + if self._robot is None: + msg = f"no owner found for {self._name!r} (attach-only mode: robot config not provided)" raise RobotTransportError(msg) from ._owner import RobotOwner # noqa: PLC0415 @@ -354,8 +421,7 @@ def _spawn_or_reprobe(self, timeout: float) -> dict[str, Any]: config = RobotOwnerConfig( name=self._name, - robot_class=self._robot_class, - robot_kwargs=self._robot_kwargs, + robot=self._robot, allow_remote=self._allow_remote, rate_hz=self._rate_hz, idle_timeout=self._idle_timeout, @@ -439,14 +505,16 @@ def _validate_metadata(self, metadata: dict[str, Any]) -> None: msg = f"owner of {self._name!r} published malformed /metadata: {metadata!r}" raise RobotTransportError(msg) - if self._robot_class is not None: + if self._robot is not None: + expected = self._robot["class_path"] advertised = metadata.get("robot_class") - if advertised != self._robot_class: + if advertised != expected: logger.warning( - f"SharedRobot(name={self._name!r}) was constructed with robot_class={self._robot_class!r} " - f"but the existing owner advertises robot_class={advertised!r}. Not fatal — public " - "re-exports, wrappers, or subclasses can preserve the wire contract — but double-check " - "this is the robot you expect.", + f"SharedRobot(name={self._name!r}) was constructed with " + f"robot.class_path={expected!r} but the existing owner advertises " + f"robot_class={advertised!r}. Not fatal — public re-exports, wrappers, or " + "subclasses can preserve the wire contract — but double-check this is the " + "robot you expect.", ) def _attach(self) -> None: diff --git a/src/physicalai/robot/trossen/bimanual_widowxai.py b/src/physicalai/robot/trossen/bimanual_widowxai.py index 1db88d0..d1e286e 100644 --- a/src/physicalai/robot/trossen/bimanual_widowxai.py +++ b/src/physicalai/robot/trossen/bimanual_widowxai.py @@ -23,6 +23,7 @@ import numpy as np +from physicalai.config import export_config from physicalai.robot import Robot if TYPE_CHECKING: @@ -58,6 +59,7 @@ def state(self) -> np.ndarray: return self.joint_positions +@export_config(class_path="physicalai.robot.BimanualWidowXAI") class BimanualWidowXAI(Robot): """Two-arm WidowX AI driver composing a left and right :class:`WidowXAI`. diff --git a/src/physicalai/robot/trossen/widowxai.py b/src/physicalai/robot/trossen/widowxai.py index ccb5e5c..75e14eb 100644 --- a/src/physicalai/robot/trossen/widowxai.py +++ b/src/physicalai/robot/trossen/widowxai.py @@ -42,6 +42,7 @@ ) raise ImportError(msg) from e +from physicalai.config import export_config from physicalai.robot import Robot from physicalai.robot.trossen.constants import HOME_POSITION, VALID_ROLES, WIDOWXAI_JOINT_ORDER @@ -77,6 +78,7 @@ def state(self) -> np.ndarray: return self.joint_positions +@export_config(class_path="physicalai.robot.WidowXAI") class WidowXAI(Robot): """Driver for the Trossen WidowX AI robot arm (7-DOF). diff --git a/tests/unit/cli/test_robot_cli.py b/tests/unit/cli/test_robot_cli.py index ab19564..cec5a7c 100644 --- a/tests/unit/cli/test_robot_cli.py +++ b/tests/unit/cli/test_robot_cli.py @@ -13,6 +13,7 @@ import uuid from pathlib import Path from types import SimpleNamespace +from typing import TYPE_CHECKING, cast from unittest.mock import patch from loguru import logger @@ -24,18 +25,28 @@ from tests.unit.robot.transport.conftest import requires_zenoh +if TYPE_CHECKING: + from jsonargparse import Namespace -def _serve_cfg(**overrides: object) -> SimpleNamespace: + +def _ns(**values: object) -> Namespace: + # Duck-typed stand-in for the jsonargparse Namespace the CLI handlers accept. + return cast("Namespace", SimpleNamespace(**values)) + + +def _serve_cfg(**overrides: object) -> Namespace: values: dict[str, object] = { "name": "left-arm", - "robot_class": "tests.unit.robot.transport.fake.FakeRobot", - "robot_kwargs": {"port": "/dev/fake0"}, + "robot": { + "class_path": "tests.unit.robot.transport.fake.FakeRobot", + "init_args": {"port": "/dev/fake0"}, + }, "allow_remote": False, "rate_hz": 100.0, "verbose": False, } values.update(overrides) - return SimpleNamespace(**values) + return _ns(**values) def test_serve_runs_owner_in_foreground_with_persistent_timeout(capsys: object) -> None: @@ -57,11 +68,22 @@ def _run_owner(config: object, shutdown: threading.Event, *, ready: object, on_e config = captured["config"] assert config.idle_timeout is None # type: ignore[attr-defined] assert config.name == "left-arm" # type: ignore[attr-defined] + assert config.robot == { # type: ignore[attr-defined] + "class_path": "tests.unit.robot.transport.fake.FakeRobot", + "init_args": {"port": "/dev/fake0"}, + } stderr = capsys.readouterr().err # type: ignore[attr-defined] assert "unauthenticated" not in stderr assert "[local-only]" in stderr +def test_serve_requires_robot_component_config(capsys: object) -> None: + with patch.object(robot_module, "run_owner") as run: + assert robot_module.serve(_serve_cfg(robot=None)) == 1 + run.assert_not_called() + assert "requires --robot" in capsys.readouterr().err # type: ignore[attr-defined] + + def test_serve_allow_remote_warns_and_tags_mode(capsys: object) -> None: def _run_owner( # noqa: ARG001 _config: object, @@ -122,7 +144,7 @@ def test_discovery_json_is_sorted_and_clean(capsys: object) -> None: {"name": "z-arm", "host": "b", "robot_class": "untrusted.Z", "num_joints": 7}, {"name": "a-arm", "host": "a", "robot_class": "untrusted.A", "num_joints": 6}, ] - cfg = SimpleNamespace(timeout=1.0, allow_remote=False, json=True) + cfg = _ns(timeout=1.0, allow_remote=False, json=True) with patch.object(robot_module, "discover_robots", return_value=records): assert robot_module.discover(cfg) == 0 @@ -136,7 +158,7 @@ def test_discovery_human_output_is_table(capsys: object) -> None: {"name": "z-arm", "host": "b", "robot_class": "untrusted.Z", "num_joints": 7}, {"name": "a-arm", "host": "a", "robot_class": "untrusted.A", "num_joints": 6}, ] - cfg = SimpleNamespace(timeout=1.0, allow_remote=False, json=False) + cfg = _ns(timeout=1.0, allow_remote=False, json=False) with patch.object(robot_module, "discover_robots", return_value=records): assert robot_module.discover(cfg) == 0 @@ -194,7 +216,7 @@ def _run_owner( # noqa: ARG001 def test_empty_discovery_json_is_array(capsys: object) -> None: - cfg = SimpleNamespace(timeout=1.0, allow_remote=False, json=True) + cfg = _ns(timeout=1.0, allow_remote=False, json=True) with patch.object(robot_module, "discover_robots", return_value=[]): assert robot_module.discover(cfg) == 0 assert capsys.readouterr().out == "[]\n" # type: ignore[attr-defined] @@ -203,13 +225,40 @@ def test_empty_discovery_json_is_array(capsys: object) -> None: def test_parser_does_not_expose_idle_timeout() -> None: parser = robot_module.build_parser() cfg = parser.parse_args( - ["serve", "--name", "left-arm", "--robot_class", "pkg.mod.Robot"], + [ + "serve", + "--name", + "left-arm", + "--robot.class_path", + "pkg.mod.Robot", + "--robot.init_args", + "{}", + ], ) assert not hasattr(cfg.serve, "idle_timeout") + assert not hasattr(cfg.serve, "robot_class") + assert not hasattr(cfg.serve, "robot_kwargs") + + +def test_parser_accepts_robot_component_config() -> None: + parser = robot_module.build_parser() + cfg = parser.parse_args( + [ + "serve", + "--name", + "left-arm", + "--robot.class_path", + "tests.unit.robot.transport.fake.FakeRobot", + "--robot.init_args", + '{"port": "/dev/fake0"}', + ], + ) + assert cfg.serve.robot["class_path"] == "tests.unit.robot.transport.fake.FakeRobot" + assert cfg.serve.robot["init_args"]["port"] == "/dev/fake0" def test_discover_rejects_invalid_timeout(capsys: object) -> None: - cfg = SimpleNamespace(timeout=float("nan"), allow_remote=False, json=False) + cfg = _ns(timeout=float("nan"), allow_remote=False, json=False) with patch.object(robot_module, "discover_robots") as discover: assert robot_module.discover(cfg) == 1 discover.assert_not_called() @@ -231,12 +280,10 @@ def test_serve_process_sigterm_disconnects_and_releases_locks(tmp_path: Path) -> "serve", "--name", name, - "--robot_class", + "--robot.class_path", "tests.unit.robot.transport.fake.FakeRobot", - "--robot_kwargs.port", - port, - "--robot_kwargs.disconnect_marker", - str(marker), + "--robot.init_args", + json.dumps({"port": port, "disconnect_marker": str(marker)}), ], stderr=subprocess.PIPE, text=True, diff --git a/tests/unit/robot/test_robot_component_config.py b/tests/unit/robot/test_robot_component_config.py new file mode 100644 index 0000000..4c1b81c --- /dev/null +++ b/tests/unit/robot/test_robot_component_config.py @@ -0,0 +1,374 @@ +# 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 robot ``@export_config`` wiring.""" + +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 ComponentConfig, instantiate, is_config_exportable, to_config +from physicalai.robot import Robot + +# SO101 imports scservo_sdk at module load; keep unit tests hardware-free. +sys.modules.setdefault("scservo_sdk", MagicMock()) + +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 _assert_construction_round_trip(robot: object) -> dict[str, Any]: + assert is_config_exportable(robot) + assert isinstance(robot, Robot) + config = to_config(robot) + wire: dict[str, Any] = json.loads(json.dumps(config)) + restored = instantiate(wire) + assert type(restored) is type(robot) + assert to_config(restored) == wire + assert not restored.is_connected() # type: ignore[union-attr] + return wire + + +# --------------------------------------------------------------------------- +# SO101 +# --------------------------------------------------------------------------- + + +@pytest.fixture +def mock_scservo_sdk() -> Generator[MagicMock, None, None]: + sdk = MagicMock() + with patch.dict(sys.modules, {"scservo_sdk": sdk}): + # Ensure SO101 can be imported under the mock (idempotent if already loaded). + yield sdk + + +class TestSO101ComponentConfig: + def test_dict_calibration_round_trip(self, mock_scservo_sdk: MagicMock) -> None: + from physicalai.robot import SO101 + + robot = SO101(port="/dev/ttyUSB0", calibration=SAMPLE_CALIBRATION) + wire = _assert_construction_round_trip(robot) + assert wire["class_path"] == "physicalai.robot.SO101" + assert wire["init_args"] == { + "port": "/dev/ttyUSB0", + "calibration": SAMPLE_CALIBRATION, + } + + def test_calibration_object_encodes_to_dict(self, mock_scservo_sdk: MagicMock) -> None: + from physicalai.robot import SO101 + from physicalai.robot.so101 import SO101Calibration + + calibration = SO101Calibration.from_dict(SAMPLE_CALIBRATION) + robot = SO101(port="/dev/ttyACM0", calibration=calibration) + wire = _assert_construction_round_trip(robot) + assert wire["init_args"]["calibration"] == SAMPLE_CALIBRATION + + def test_path_calibration_keeps_relative_string( + self, mock_scservo_sdk: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + from physicalai.robot import SO101 + + calib_path = tmp_path / "calibration.json" + calib_path.write_text(json.dumps(SAMPLE_CALIBRATION), encoding="utf-8") + monkeypatch.chdir(tmp_path) + + robot = SO101(port="/dev/ttyUSB0", calibration="calibration.json") + wire = _assert_construction_round_trip(robot) + assert wire["init_args"]["calibration"] == "calibration.json" + + def test_path_object_keeps_relative_as_given( + self, mock_scservo_sdk: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + from physicalai.robot import SO101 + + calib_path = tmp_path / "calib.json" + calib_path.write_text(json.dumps(SAMPLE_CALIBRATION), encoding="utf-8") + monkeypatch.chdir(tmp_path) + + robot = SO101(port="/dev/ttyUSB0", calibration=Path("calib.json")) + wire = _assert_construction_round_trip(robot) + assert wire["init_args"]["calibration"] == "calib.json" + + def test_absolute_path_calibration_stays_absolute( + self, mock_scservo_sdk: MagicMock, tmp_path: Path + ) -> None: + from physicalai.robot import SO101 + + calib_path = (tmp_path / "absolute-calib.json").resolve() + calib_path.write_text(json.dumps(SAMPLE_CALIBRATION), encoding="utf-8") + + robot = SO101(port="/dev/ttyUSB0", calibration=calib_path) + wire = _assert_construction_round_trip(robot) + assert wire["init_args"]["calibration"] == str(calib_path) + assert Path(str(wire["init_args"]["calibration"])).is_absolute() + + def test_uncalibrated_export_round_trips(self, mock_scservo_sdk: MagicMock) -> None: + from jsonargparse import ArgumentParser + + from physicalai.robot import SO101 + + robot = SO101.uncalibrated(port="/dev/ttyUSB0") + wire = _assert_construction_round_trip(robot) + assert wire["init_args"]["calibration"] is None + assert wire["init_args"]["allow_uncalibrated"] is True + assert wire["init_args"]["unit"] == "ticks" + + # Public allow_uncalibrated + defaulted calibration make the export + # loadable by jsonargparse (the physicalai run --config path); a + # required calibration with no default rejects calibration: null. + parser = ArgumentParser() + parser.add_class_arguments(SO101, "robot") + ns = parser.parse_object({"robot": wire["init_args"]}) + assert ns.robot.calibration is None + assert ns.robot.allow_uncalibrated is True + + def test_protocol_still_holds(self, mock_scservo_sdk: MagicMock) -> None: + from physicalai.robot import SO101 + + robot = SO101(port="/dev/ttyUSB0", calibration=SAMPLE_CALIBRATION) + assert isinstance(robot, Robot) + assert robot.to_config() == to_config(robot) # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# WidowXAI / BimanualWidowXAI +# --------------------------------------------------------------------------- + + +def _make_mock_trossen_arm() -> MagicMock: + module = MagicMock() + driver = MagicMock() + driver.get_is_configured.return_value = True + module.TrossenArmDriver.return_value = driver + module.Model.wxai_v0 = MagicMock(name="Model.wxai_v0") + module.StandardEndEffector.wxai_v0_follower = MagicMock(name="wxai_v0_follower") + module.StandardEndEffector.wxai_v0_leader = MagicMock(name="wxai_v0_leader") + module.Mode.position = MagicMock(name="Mode.position") + module.Mode.external_effort = MagicMock(name="Mode.external_effort") + return module + + +@pytest.fixture +def mock_trossen_arm() -> Generator[MagicMock, None, None]: + """Inject a mock ``trossen_arm`` so WidowX drivers import without the SDK.""" + mock_module = _make_mock_trossen_arm() + # Drop cached modules so import picks up the mock even if a prior import failed. + for name in ( + "physicalai.robot.trossen.widowxai", + "physicalai.robot.trossen.bimanual_widowxai", + "physicalai.robot.trossen", + ): + sys.modules.pop(name, None) + with patch.dict(sys.modules, {"trossen_arm": mock_module}): + yield mock_module + + +class TestWidowXAIComponentConfig: + def test_default_role_omitted(self, mock_trossen_arm: MagicMock) -> None: + from physicalai.robot import WidowXAI + + robot = WidowXAI(ip="192.168.1.2") + wire = _assert_construction_round_trip(robot) + assert wire["class_path"] == "physicalai.robot.WidowXAI" + assert wire["init_args"] == {"ip": "192.168.1.2"} + + def test_explicit_role_round_trip(self, mock_trossen_arm: MagicMock) -> None: + from physicalai.robot import WidowXAI + + robot = WidowXAI(ip="10.0.0.1", role="leader") + wire = _assert_construction_round_trip(robot) + assert wire["init_args"] == {"ip": "10.0.0.1", "role": "leader"} + + def test_protocol_still_holds(self, mock_trossen_arm: MagicMock) -> None: + from physicalai.robot import WidowXAI + + robot = WidowXAI(ip="192.168.1.2") + assert isinstance(robot, Robot) + assert robot.to_config() == to_config(robot) # type: ignore[attr-defined] + + +class TestBimanualWidowXAIComponentConfig: + def test_nested_arms_round_trip(self, mock_trossen_arm: MagicMock) -> None: + from physicalai.robot import BimanualWidowXAI, WidowXAI + + left = WidowXAI(ip="192.168.1.10", role="follower") + right = WidowXAI(ip="192.168.1.11", role="follower") + robot = BimanualWidowXAI(left=left, right=right) + wire = _assert_construction_round_trip(robot) + assert wire["class_path"] == "physicalai.robot.BimanualWidowXAI" + left_cfg = wire["init_args"]["left"] + right_cfg = wire["init_args"]["right"] + assert isinstance(left_cfg, dict) + assert isinstance(right_cfg, dict) + assert left_cfg["class_path"] == "physicalai.robot.WidowXAI" + assert right_cfg["class_path"] == "physicalai.robot.WidowXAI" + assert left_cfg["init_args"] == {"ip": "192.168.1.10", "role": "follower"} + assert right_cfg["init_args"] == {"ip": "192.168.1.11", "role": "follower"} + + +# --------------------------------------------------------------------------- +# SharedRobot (construction recipe only — no session / connection state) +# --------------------------------------------------------------------------- + + +class TestSharedRobotComponentConfig: + def test_spawn_recipe_round_trip(self) -> None: + from physicalai.robot import SharedRobot + + robot = SharedRobot( + "follower-arm", + robot={ + "class_path": "tests.unit.robot.transport.fake.FakeRobot", + "init_args": {"port": "/dev/fake-shared", "device_ids": ["fake:follower"]}, + }, + allow_remote=True, + rate_hz=50.0, + idle_timeout=2.5, + connect_timeout=3.0, + ) + wire = _assert_construction_round_trip(robot) + assert wire["class_path"] == "physicalai.robot.SharedRobot" + assert wire["init_args"]["name"] == "follower-arm" + assert wire["init_args"]["allow_remote"] is True + assert wire["init_args"]["rate_hz"] == 50.0 + assert wire["init_args"]["idle_timeout"] == 2.5 + assert wire["init_args"]["connect_timeout"] == 3.0 + nested = wire["init_args"]["robot"] + assert isinstance(nested, dict) + assert nested["class_path"] == "tests.unit.robot.transport.fake.FakeRobot" + assert nested["init_args"]["port"] == "/dev/fake-shared" + assert nested["init_args"]["device_ids"] == ["fake:follower"] + # Run-state / transport handles are never part of the recipe. + assert "_session" not in wire["init_args"] + assert "_connected" not in wire["init_args"] + assert "session" not in wire["init_args"] + + def test_defining_module_robot_path_round_trips_as_given(self) -> None: + from physicalai.robot import SharedRobot + + defining = "physicalai.robot.so101.so101.SO101" + robot = SharedRobot( + "canonical-arm", + robot={ + "class_path": defining, + "init_args": { + "port": "/dev/ttyUSB0", + "calibration": SAMPLE_CALIBRATION, + }, + }, + ) + wire = _assert_construction_round_trip(robot) + nested = wire["init_args"]["robot"] + assert isinstance(nested, dict) + # Stored as written: the subscriber never imports the driver to canonicalize it. + assert nested["class_path"] == defining + + def test_attach_only_round_trip(self) -> None: + from physicalai.robot import SharedRobot + + robot = SharedRobot("leader-arm") + wire = _assert_construction_round_trip(robot) + assert wire["class_path"] == "physicalai.robot.SharedRobot" + assert wire["init_args"] == {"name": "leader-arm"} + assert "robot" not in wire["init_args"] + + def test_explicit_robot_none_round_trips(self) -> None: + from physicalai.robot import SharedRobot + + robot = SharedRobot("attach-null", robot=None) + wire = _assert_construction_round_trip(robot) + assert wire["init_args"]["name"] == "attach-null" + assert wire["init_args"]["robot"] is None + + def test_live_session_arg_fails_at_to_config(self) -> None: + from physicalai.config import ComponentConfigError + from physicalai.robot import SharedRobot + + robot = SharedRobot("sess", _session=object()) + with pytest.raises(ComponentConfigError, match=r"init_args\._session"): + to_config(robot) + + def test_from_config_omits_null_session(self) -> None: + from physicalai.robot import SharedRobot + + robot = SharedRobot.from_config( + { + "class_path": "tests.unit.robot.transport.fake.FakeRobot", + "init_args": {"port": "/dev/fake-from-config", "device_ids": ["fake:follower"]}, + }, + name="from-config-arm", + ) + wire = _assert_construction_round_trip(robot) + assert "_session" not in wire["init_args"] + + def test_constructor_recipe_omits_null_session(self) -> None: + from physicalai.robot import SharedRobot + + robot = SharedRobot( + "from-robot-arm", + robot={ + "class_path": "tests.unit.robot.transport.fake.FakeRobot", + "init_args": {"port": "/dev/fake-from-robot", "device_ids": ["fake:follower"]}, + }, + ) + wire = _assert_construction_round_trip(robot) + assert "_session" not in wire["init_args"] + + def test_attach_omits_null_session(self) -> None: + from physicalai.robot import SharedRobot + + robot = SharedRobot.attach("attach-arm") + wire = _assert_construction_round_trip(robot) + assert "_session" not in wire["init_args"] + + def test_classmethod_live_session_still_fails_at_to_config(self) -> None: + from physicalai.config import ComponentConfigError + from physicalai.robot import SharedRobot + + session = object() + for robot in ( + SharedRobot.from_config( + { + "class_path": "tests.unit.robot.transport.fake.FakeRobot", + "init_args": {"port": "/dev/fake-live", "device_ids": ["fake:follower"]}, + }, + name="live-from-config", + _session=session, + ), + SharedRobot.attach("live-attach", _session=session), + ): + with pytest.raises(ComponentConfigError, match=r"init_args\._session"): + to_config(robot) + + def test_nested_recipe_is_not_instantiated(self) -> None: + from physicalai.config import instantiate + from physicalai.robot import SharedRobot + + config: ComponentConfig = { + "class_path": "physicalai.robot.SharedRobot", + "init_args": { + "name": "nested-arm", + "robot": { + "class_path": "tests.unit.robot.transport.fake.FakeRobot", + "init_args": {"port": "/dev/fake-nested", "device_ids": ["fake:follower"]}, + }, + }, + } + restored = instantiate(config) + assert isinstance(restored, SharedRobot) + # The declared config arg stays a mapping — no driver is built here. + assert restored._robot == config["init_args"]["robot"] diff --git a/tests/unit/robot/transport/fake.py b/tests/unit/robot/transport/fake.py index ecda1a8..581e4d8 100644 --- a/tests/unit/robot/transport/fake.py +++ b/tests/unit/robot/transport/fake.py @@ -17,6 +17,8 @@ import numpy as np +from physicalai.config import export_config + NUM_JOINTS = 6 @@ -35,6 +37,7 @@ def state(self) -> np.ndarray: return self.joint_positions +@export_config class FakeRobot: """In-memory robot producing synthetic observations. diff --git a/tests/unit/robot/transport/test_latency.py b/tests/unit/robot/transport/test_latency.py index c9e621a..9475ba4 100644 --- a/tests/unit/robot/transport/test_latency.py +++ b/tests/unit/robot/transport/test_latency.py @@ -40,8 +40,10 @@ class TestActionLatency: def robot(self, unique_id: str) -> Generator[SharedRobot, None, None]: robot = SharedRobot( unique_id.replace("/", "-"), - robot_class=FAKE_ROBOT_CLASS, - robot_kwargs={"device_ids": [f"fake:{unique_id}"]}, + robot={ + "class_path": FAKE_ROBOT_CLASS, + "init_args": {"device_ids": [f"fake:{unique_id}"]}, + }, rate_hz=_RATE_HZ, idle_timeout=3.0, ) diff --git a/tests/unit/robot/transport/test_owner_config.py b/tests/unit/robot/transport/test_owner_config.py index 0873020..53236b8 100644 --- a/tests/unit/robot/transport/test_owner_config.py +++ b/tests/unit/robot/transport/test_owner_config.py @@ -5,133 +5,259 @@ import json import pickle +import subprocess import sys from typing import Any, cast -from unittest.mock import MagicMock, patch +from unittest.mock import MagicMock import pytest -from physicalai.robot.transport._owner_config import RobotOwnerConfig, normalize_robot_class +from physicalai.config import ComponentConfigError, ComponentImportError +from physicalai.robot.transport._owner import RobotOwner +from physicalai.robot.transport._owner_config import ( + RobotOwnerConfig, + normalize_robot_config, + validate_owner_config, +) +from .conftest import FAKE_ROBOT_CLASS from .fake import FakeRobot +# SO101 imports scservo_sdk at module load; keep unit tests hardware-free. +sys.modules.setdefault("scservo_sdk", MagicMock()) -class _Outer: - class Inner: ... +def _fake_robot(**init_args: object) -> dict[str, object]: + return {"class_path": FAKE_ROBOT_CLASS, "init_args": dict(init_args)} -class _RecordingRobot: - """Stands in for a vendor driver class resolved via a stubbed module.""" - def __init__(self, **kwargs: object) -> None: - self.kwargs = kwargs +class TestNormalizeRobotConfigClassPath: + def test_string_path_stored_as_given_without_importing(self) -> None: + # No scservo_sdk mock: a string path is trusted and never imported here, + # so the owner envelope can be built where the driver is not installed. + defining = "physicalai.robot.so101.so101.SO101" + assert normalize_robot_config({"class_path": defining})["class_path"] == defining + def test_public_path_stored_as_given(self) -> None: + public = "physicalai.robot.SO101" + assert normalize_robot_config({"class_path": public})["class_path"] == public -class TestNormalizeRobotClass: - def test_string_passthrough(self) -> None: - assert normalize_robot_class("pkg.mod.Cls") == "pkg.mod.Cls" + @pytest.mark.parametrize("class_path", [" ", "NotDotted"]) + def test_non_dotted_path_raises(self, class_path: str) -> None: + with pytest.raises(ValueError, match="must be a nonempty dotted path"): + normalize_robot_config({"class_path": class_path}) - def test_class_object(self) -> None: - assert normalize_robot_class(FakeRobot) == "tests.unit.robot.transport.fake.FakeRobot" - - def test_nested_qualname(self) -> None: - path = normalize_robot_class(_Outer.Inner) - assert path.endswith(".test_owner_config._Outer.Inner") - - def test_local_class_raises(self) -> None: - class _Local: ... - - with pytest.raises(ValueError, match="local class"): - normalize_robot_class(_Local) - - def test_non_class_non_string_raises(self) -> None: - with pytest.raises(TypeError, match="must be a class or a dotted path string"): - normalize_robot_class(123) # type: ignore[arg-type] + def test_empty_path_raises(self) -> None: + with pytest.raises(ComponentConfigError, match="must be a non-empty string"): + normalize_robot_config({"class_path": ""}) class TestRobotOwnerConfig: def test_picklable(self) -> None: - config = RobotOwnerConfig(name="left-arm", robot_class="pkg.mod.Cls", robot_kwargs={"port": "/dev/ttyUSB0"}) + config = RobotOwnerConfig(name="left-arm", robot=_fake_robot(port="/dev/ttyUSB0")) restored = pickle.loads(pickle.dumps(config)) assert restored == config def test_json_roundtrip(self) -> None: - config = RobotOwnerConfig(name="left-arm", robot_class="pkg.mod.Cls", robot_kwargs={"ip": "10.0.0.2"}) + config = RobotOwnerConfig(name="left-arm", robot=_fake_robot(ip="10.0.0.2")) assert RobotOwnerConfig.from_json_dict(json.loads(json.dumps(config.to_json_dict()))) == config def test_defaults(self) -> None: - config = RobotOwnerConfig(name="left-arm", robot_class="pkg.mod.Cls") - assert config.robot_kwargs == {} + config = RobotOwnerConfig(name="left-arm", robot=_fake_robot()) + assert config.robot == {"class_path": FAKE_ROBOT_CLASS, "init_args": {}} + assert config.robot_class == FAKE_ROBOT_CLASS assert config.allow_remote is False assert config.rate_hz == 100.0 assert config.idle_timeout == 10.0 def test_persistent_idle_timeout_json_roundtrip(self) -> None: - config = RobotOwnerConfig(name="left-arm", robot_class="pkg.mod.Cls", idle_timeout=None) + config = RobotOwnerConfig(name="left-arm", robot=_fake_robot(), idle_timeout=None) restored = RobotOwnerConfig.from_json_dict(json.loads(json.dumps(config.to_json_dict()))) assert restored == config assert restored.idle_timeout is None def test_invalid_name_raises(self) -> None: with pytest.raises(ValueError, match="invalid robot name"): - RobotOwnerConfig(name="left/arm", robot_class="pkg.mod.Cls") + RobotOwnerConfig(name="left/arm", robot=_fake_robot()) @pytest.mark.parametrize("rate_hz", [float("nan"), True, "100"]) def test_invalid_rate_types_raise(self, rate_hz: object) -> None: with pytest.raises(ValueError, match="rate_hz must be finite"): - RobotOwnerConfig(name="left-arm", robot_class="pkg.mod.Cls", rate_hz=rate_hz) # type: ignore[arg-type] + RobotOwnerConfig(name="left-arm", robot=_fake_robot(), rate_hz=rate_hz) # type: ignore[arg-type] @pytest.mark.parametrize("idle_timeout", [0.0, -1.0, float("inf"), float("nan"), True, "10"]) def test_invalid_idle_timeout_raises(self, idle_timeout: object) -> None: with pytest.raises(ValueError, match="idle_timeout must be finite"): RobotOwnerConfig( name="left-arm", - robot_class="pkg.mod.Cls", + robot=_fake_robot(), idle_timeout=cast(Any, idle_timeout), ) def test_zero_rate_raises(self) -> None: with pytest.raises(ValueError, match="rate_hz must be finite"): - RobotOwnerConfig(name="left-arm", robot_class="pkg.mod.Cls", rate_hz=0.0) + RobotOwnerConfig(name="left-arm", robot=_fake_robot(), rate_hz=0.0) def test_negative_rate_raises(self) -> None: with pytest.raises(ValueError, match="rate_hz must be finite"): - RobotOwnerConfig(name="left-arm", robot_class="pkg.mod.Cls", rate_hz=-1.0) + RobotOwnerConfig(name="left-arm", robot=_fake_robot(), rate_hz=-1.0) def test_infinite_rate_raises(self) -> None: with pytest.raises(ValueError, match="rate_hz must be finite"): - RobotOwnerConfig(name="left-arm", robot_class="pkg.mod.Cls", rate_hz=float("inf")) + RobotOwnerConfig(name="left-arm", robot=_fake_robot(), rate_hz=float("inf")) - def test_non_serializable_kwargs_raises(self) -> None: + def test_non_serializable_init_args_raises(self) -> None: with pytest.raises(ValueError, match="JSON-serializable"): - RobotOwnerConfig(name="left-arm", robot_class="pkg.mod.Cls", robot_kwargs={"calibration": object()}) + RobotOwnerConfig( + name="left-arm", + robot={"class_path": FAKE_ROBOT_CLASS, "init_args": {"calibration": object()}}, + ) def test_build_known_class(self) -> None: - config = RobotOwnerConfig( - name="left-arm", - robot_class="tests.unit.robot.transport.fake.FakeRobot", - robot_kwargs={"port": "/dev/ttyUSB0"}, - ) + config = RobotOwnerConfig(name="left-arm", robot=_fake_robot(port="/dev/ttyUSB0")) robot = config.build() assert isinstance(robot, FakeRobot) - def test_build_arbitrary_module_level_class(self) -> None: - # No registry to update: any importable class works, mirroring how - # the vendor SDK is stubbed for so101 elsewhere in this test suite. - mock_module = MagicMock() - mock_module.SO101 = _RecordingRobot - config = RobotOwnerConfig(name="left-arm", robot_class="physicalai.robot.so101.SO101", robot_kwargs={"port": "x"}) - with patch.dict(sys.modules, {"physicalai.robot.so101": mock_module}): - robot = config.build() - assert isinstance(robot, _RecordingRobot) - assert robot.kwargs == {"port": "x"} - def test_build_non_class_path_raises(self) -> None: - config = RobotOwnerConfig(name="left-arm", robot_class="physicalai.robot.transport._ids.KEY_PREFIX") - with pytest.raises(TypeError, match="does not resolve to a class"): + config = RobotOwnerConfig( + name="left-arm", + robot={"class_path": "physicalai.robot.transport._ids.KEY_PREFIX", "init_args": {}}, + ) + with pytest.raises(ComponentImportError, match="does not resolve to a class"): config.build() def test_build_unknown_path_raises(self) -> None: - config = RobotOwnerConfig(name="left-arm", robot_class="totally.unknown.module.Cls") - with pytest.raises(ValueError, match="could not import"): + config = RobotOwnerConfig( + name="left-arm", + robot={"class_path": "totally.unknown.module.Cls", "init_args": {}}, + ) + with pytest.raises(ComponentImportError, match="cannot import class_path"): config.build() + + def test_flat_stdin_rejected_before_import(self) -> None: + flat = { + "name": "left-arm", + "robot_class": "totally.unknown.module.Cls", + "robot_kwargs": {"port": "/dev/ttyUSB0"}, + } + with pytest.raises(ValueError, match="unknown owner config keys") as exc_info: + RobotOwnerConfig.from_json_dict(flat) + assert "robot_class" in str(exc_info.value) + assert "robot_kwargs" in str(exc_info.value) + + def test_flat_robot_kwargs_alone_rejected(self) -> None: + with pytest.raises(ValueError, match="unknown owner config keys"): + RobotOwnerConfig.from_json_dict( + { + "name": "left-arm", + "robot": _fake_robot(), + "robot_kwargs": {}, + }, + ) + + def test_unknown_keys_rejected(self) -> None: + with pytest.raises(ValueError, match="unknown owner config keys") as exc_info: + RobotOwnerConfig.from_json_dict( + { + "name": "left-arm", + "robot": _fake_robot(), + "extra_field": 1, + }, + ) + assert "extra_field" in str(exc_info.value) + + def test_missing_robot_rejected(self) -> None: + with pytest.raises(ValueError, match="missing required 'robot'"): + RobotOwnerConfig.from_json_dict({"name": "left-arm"}) + + def test_validate_owner_config_shared_helper(self) -> None: + robot = validate_owner_config( + { + "name": "left-arm", + "robot": _fake_robot(port="/dev/ttyUSB0"), + "allow_remote": True, + "rate_hz": 50.0, + "idle_timeout": 2.5, + }, + ) + assert robot["class_path"] == FAKE_ROBOT_CLASS + init_args = robot["init_args"] + assert isinstance(init_args, dict) + assert init_args["port"] == "/dev/ttyUSB0" + + def test_malformed_robot_rejected_before_import(self) -> None: + with pytest.raises(ComponentConfigError): + RobotOwnerConfig.from_json_dict( + { + "name": "left-arm", + "robot": {"class_path": "totally.unknown.module.Cls", "extra": 1}, + }, + ) + + def test_defining_module_path_stored_as_given(self) -> None: + # No scservo_sdk mock: storing the recipe must not import the driver. + defining = "physicalai.robot.so101.so101.SO101" + config = RobotOwnerConfig( + name="left-arm", + robot={ + "class_path": defining, + "init_args": {"port": "/dev/ttyUSB0", "calibration": "./cal.json"}, + }, + ) + assert config.robot["class_path"] == defining + assert config.robot_class == defining + assert config.to_json_dict()["robot"]["class_path"] == defining + + def test_relative_path_survives_json_and_popen_inherits_cwd(self, monkeypatch: pytest.MonkeyPatch) -> None: + config = RobotOwnerConfig( + name="left-arm", + robot=_fake_robot(calibration="./calibration.json"), + ) + payload = config.to_json_dict() + assert payload["robot"]["init_args"]["calibration"] == "./calibration.json" + + captured: dict[str, object] = {} + + class _FakeProc: + stdin = None + stdout = None + + def __init__(self, *_args: object, **kwargs: object) -> None: + captured.update(kwargs) + self.stdin = _FakeStdin() + self.stdout = _FakeStdout() + + def poll(self) -> int | None: + return 0 + + def terminate(self) -> None: + return None + + def kill(self) -> None: + return None + + def wait(self, timeout: float | None = None) -> int: # noqa: ARG002 + return 0 + + class _FakeStdin: + def write(self, _data: bytes) -> None: + return None + + def close(self) -> None: + return None + + class _FakeStdout: + def readline(self) -> bytes: + return b"READY\n" + + monkeypatch.setattr(subprocess, "Popen", _FakeProc) + monkeypatch.setattr(RobotOwner, "_read_stdout_line", lambda _self, _timeout: "READY") + owner = RobotOwner(config) + owner.start(timeout=1.0) + assert "cwd" not in captured + + +class TestNormalizeRobotConfig: + def test_normalize_robot_config_rejects_malformed(self) -> None: + with pytest.raises(ComponentConfigError): + normalize_robot_config({"class_path": FAKE_ROBOT_CLASS, "extra": True}) diff --git a/tests/unit/robot/transport/test_owner_handshake.py b/tests/unit/robot/transport/test_owner_handshake.py index 9938c61..590737a 100644 --- a/tests/unit/robot/transport/test_owner_handshake.py +++ b/tests/unit/robot/transport/test_owner_handshake.py @@ -16,11 +16,14 @@ from .fake import FakeRobot -def _owner(unique_id: str, **robot_kwargs: object) -> RobotOwner: +def _fake_robot(**init_args: object) -> dict[str, object]: + return {"class_path": FAKE_ROBOT_CLASS, "init_args": dict(init_args)} + + +def _owner(unique_id: str, **robot_init_args: object) -> RobotOwner: config = RobotOwnerConfig( name=unique_id.replace("/", "-"), - robot_class=FAKE_ROBOT_CLASS, - robot_kwargs={"device_ids": [f"fake:{unique_id}"], **robot_kwargs}, + robot=_fake_robot(device_ids=[f"fake:{unique_id}"], **robot_init_args), idle_timeout=2.0, ) return RobotOwner(config) @@ -33,7 +36,7 @@ def test_startup_failure_after_connect_disconnects_and_releases_locks( name = unique_id.replace("/", "-") device_id = f"fake:{unique_id}" driver = FakeRobot(device_ids=(device_id,), fail_observation=True) - config = RobotOwnerConfig(name=name, robot_class=FAKE_ROBOT_CLASS) + config = RobotOwnerConfig(name=name, robot=_fake_robot()) monkeypatch.setattr(RobotOwnerConfig, "build", lambda _self: driver) with pytest.raises(_StartupError, match="fake observation failure") as exc_info: @@ -47,7 +50,7 @@ def test_startup_failure_after_connect_disconnects_and_releases_locks( @pytest.mark.parametrize("allow_remote", [False, True]) def test_metadata_redacts_device_ids_for_remote_owner(allow_remote: bool) -> None: - config = RobotOwnerConfig(name="left-arm", robot_class=FAKE_ROBOT_CLASS, allow_remote=allow_remote) + config = RobotOwnerConfig(name="left-arm", robot=_fake_robot(), allow_remote=allow_remote) metadata = _build_metadata( config, FakeRobot(device_ids=("serial:ttyUSB0",)), @@ -56,6 +59,7 @@ def test_metadata_redacts_device_ids_for_remote_owner(allow_remote: bool) -> Non ) assert metadata["host"] + assert metadata["robot_class"] == FAKE_ROBOT_CLASS if allow_remote: assert "device_ids" not in metadata else: @@ -72,7 +76,7 @@ def test_endpoint_collision_error_identifies_endpoint_and_remediation( bind_host: str, ) -> None: name = unique_id.replace("/", "-") - config = RobotOwnerConfig(name=name, robot_class=FAKE_ROBOT_CLASS, allow_remote=allow_remote) + config = RobotOwnerConfig(name=name, robot=_fake_robot(), allow_remote=allow_remote) def _fail_open_session(*_args: object, **_kwargs: object) -> None: raise OSError("address already in use") diff --git a/tests/unit/robot/transport/test_owner_worker.py b/tests/unit/robot/transport/test_owner_worker.py index b22fbf8..f85ffc3 100644 --- a/tests/unit/robot/transport/test_owner_worker.py +++ b/tests/unit/robot/transport/test_owner_worker.py @@ -3,6 +3,8 @@ from __future__ import annotations +import io +import json import threading import time from dataclasses import dataclass, field @@ -11,7 +13,7 @@ from physicalai.robot.transport import _owner_worker from physicalai.robot.transport._owner_config import RobotOwnerConfig -from physicalai.robot.transport._owner_worker import OwnerEvent, OwnerExitReason, _run_loop, run_owner +from physicalai.robot.transport._owner_worker import OwnerEvent, OwnerExitReason, _run_loop, main, run_owner from .fake import FakeRobot @@ -137,7 +139,11 @@ def test_disconnect_failure_upgrades_clean_shutdown(monkeypatch: Any) -> None: locks=SimpleNamespace(release_all=lambda: None), ) monkeypatch.setattr(_owner_worker, "_startup", lambda _config: endpoints) - config = RobotOwnerConfig(name="left-arm", robot_class="pkg.mod.Robot", idle_timeout=None) + config = RobotOwnerConfig( + name="left-arm", + robot={"class_path": "tests.unit.robot.transport.fake.FakeRobot", "init_args": {}}, + idle_timeout=None, + ) result = run_owner(config, shutdown) @@ -157,7 +163,11 @@ def test_ready_failure_still_disconnects(monkeypatch: Any) -> None: # noqa: ANN locks=SimpleNamespace(release_all=lambda: None), ) monkeypatch.setattr(_owner_worker, "_startup", lambda _config: endpoints) - config = RobotOwnerConfig(name="left-arm", robot_class="pkg.mod.Robot", idle_timeout=None) + config = RobotOwnerConfig( + name="left-arm", + robot={"class_path": "tests.unit.robot.transport.fake.FakeRobot", "init_args": {}}, + idle_timeout=None, + ) def _fail_ready() -> None: raise RuntimeError("readiness output failed") @@ -167,3 +177,25 @@ def _fail_ready() -> None: assert result.reason is OwnerExitReason.LOOP_FAILURE assert result.exit_code == 1 assert driver.disconnect_called + + +def test_main_malformed_robot_stdin_signals_invalid_config(monkeypatch: Any) -> None: # noqa: ANN401 + """Malformed new-shape robot: must ERROR with invalid_config, not crash.""" + payload = json.dumps( + { + "name": "left-arm", + "robot": {"class_path": "tests.unit.robot.transport.fake.FakeRobot", "extra": 1}, + }, + ) + errors: list[tuple[str, str | None]] = [] + + def _capture_error(msg: str, tb: str | None = None, *, phase: str | None = None, **_kwargs: object) -> None: + errors.append((msg, phase)) + + monkeypatch.setattr(_owner_worker, "signal_error", _capture_error) + monkeypatch.setattr(_owner_worker.sys, "stdin", io.StringIO(payload)) + + assert main() == 1 + assert len(errors) == 1 + assert errors[0][1] == "invalid_config" + assert "invalid worker config" in errors[0][0] diff --git a/tests/unit/robot/transport/test_shared_robot.py b/tests/unit/robot/transport/test_shared_robot.py index 8dfb6cb..e046dbb 100644 --- a/tests/unit/robot/transport/test_shared_robot.py +++ b/tests/unit/robot/transport/test_shared_robot.py @@ -11,6 +11,7 @@ import numpy as np import pytest +from physicalai.config import ComponentConfigError from physicalai.robot.errors import ( RobotNameConflict, RobotNotConnectedError, @@ -30,11 +31,13 @@ _STATE_DIM = 12 # fake ships positions + velocities -def _shared_robot(name: str, *, allow_remote: bool = False, **robot_kwargs: object) -> SharedRobot: +def _shared_robot(name: str, *, allow_remote: bool = False, **robot_init_args: object) -> SharedRobot: return SharedRobot( name, - robot_class=FAKE_ROBOT_CLASS, - robot_kwargs={"device_ids": [f"fake:{name}"], **robot_kwargs}, + robot={ + "class_path": FAKE_ROBOT_CLASS, + "init_args": {"device_ids": [f"fake:{name}"], **robot_init_args}, + }, idle_timeout=0.5, allow_remote=allow_remote, ) @@ -75,14 +78,33 @@ def test_invalid_name_raises(self) -> None: with pytest.raises(ValueError, match="invalid robot name"): SharedRobot("bad/name") - def test_attach_only_has_no_robot_class(self) -> None: + def test_attach_only_has_no_robot_config(self) -> None: robot = SharedRobot.attach("left-arm") assert robot.name == "left-arm" + assert robot._robot is None assert robot.device_ids == () - def test_class_object_normalized(self) -> None: - robot = SharedRobot("left-arm", robot_class=FakeRobot, robot_kwargs={"port": "/dev/ttyUSB0"}) - assert robot._robot_class == "tests.unit.robot.transport.fake.FakeRobot" + def test_robot_component_config_normalized(self) -> None: + robot = SharedRobot( + "left-arm", + robot={"class_path": FAKE_ROBOT_CLASS, "init_args": {"port": "/dev/ttyUSB0"}}, + ) + assert robot._robot == { + "class_path": FAKE_ROBOT_CLASS, + "init_args": {"port": "/dev/ttyUSB0"}, + } + + def test_from_config_stores_component_config(self) -> None: + robot = SharedRobot.from_config( + {"class_path": FAKE_ROBOT_CLASS, "init_args": {"port": "/dev/fake0"}}, + name="left-arm", + ) + assert robot._robot == {"class_path": FAKE_ROBOT_CLASS, "init_args": {"port": "/dev/fake0"}} + + def test_constructor_requires_component_config_mapping(self) -> None: + driver = FakeRobot(port="/dev/fake0", device_ids=("fake:/dev/fake0",)) + with pytest.raises(ComponentConfigError, match="robot must be a ComponentConfig mapping"): + SharedRobot("left-arm", robot=driver) # type: ignore[arg-type] def test_satisfies_robot_protocol(self) -> None: from physicalai.robot import Robot @@ -231,15 +253,21 @@ def test_class_mismatch_warns_but_attaches(self, module_owner: SharedRobot, capl """robot_class mismatch on an existing owner is diagnostic, not fatal.""" import logging + from physicalai.robot.transport._owner_config import RobotOwnerConfig + impostor = SharedRobot( module_owner.name, - robot_class="physicalai.robot.transport._session.open_session", - robot_kwargs={"device_ids": [f"fake:{module_owner.name}"]}, + robot={ + "class_path": f"{RobotOwnerConfig.__module__}.{RobotOwnerConfig.__qualname__}", + "init_args": {}, + }, ) with caplog.at_level(logging.WARNING): impostor.connect() # attaches to the same owner; must not raise try: assert impostor.is_connected() + # loguru writes to stderr; message shape is asserted via construction above + assert impostor._robot_class != module_owner._robot_class finally: impostor.disconnect() @@ -260,7 +288,10 @@ def test_device_already_owned_under_another_name(self, module_owner: SharedRobot assert metadata is not None shared_device = metadata["device_ids"][0] other_name = unique_id.replace("/", "-") - second = SharedRobot(other_name, robot_class=FAKE_ROBOT_CLASS, robot_kwargs={"device_ids": [shared_device]}) + second = SharedRobot( + other_name, + robot={"class_path": FAKE_ROBOT_CLASS, "init_args": {"device_ids": [shared_device]}}, + ) with pytest.raises(RobotDeviceAlreadyOwned): second.connect() @@ -294,8 +325,7 @@ def test_differing_device_race_raises_name_conflict(self, unique_id: str, *, all def _run(key: str, device_id: str) -> None: robot = SharedRobot( name, - robot_class=FAKE_ROBOT_CLASS, - robot_kwargs={"device_ids": [device_id]}, + robot={"class_path": FAKE_ROBOT_CLASS, "init_args": {"device_ids": [device_id]}}, allow_remote=allow_remote, ) barrier.wait() From c84bcddfe8455e48afc68c23e2e6c63b8ab8c17c Mon Sep 17 00:00:00 2001 From: Max Xiang Date: Tue, 28 Jul 2026 11:26:44 +0200 Subject: [PATCH 2/5] address comments --- src/physicalai/config/_envelope.py | 4 +-- src/physicalai/robot/transport/_importing.py | 2 +- .../robot/transport/_owner_config.py | 28 +++++++++---------- .../robot/transport/_shared_robot.py | 17 +++++------ 4 files changed, 26 insertions(+), 25 deletions(-) diff --git a/src/physicalai/config/_envelope.py b/src/physicalai/config/_envelope.py index 3e93fd3..e4bab43 100644 --- a/src/physicalai/config/_envelope.py +++ b/src/physicalai/config/_envelope.py @@ -36,8 +36,8 @@ def validate_envelope( """Validate a transport stdin envelope schema-positively. Requires *component_key* with a valid ComponentConfig shape and allows - only *allowed_keys*. Unknown keys (including legacy flat shapes) raise a - clear schema error before any import or hardware access. + only *allowed_keys*. Unknown keys raise a clear schema error before any + import or hardware access. Args: data: Full stdin envelope dict. diff --git a/src/physicalai/robot/transport/_importing.py b/src/physicalai/robot/transport/_importing.py index 021561f..00c8b45 100644 --- a/src/physicalai/robot/transport/_importing.py +++ b/src/physicalai/robot/transport/_importing.py @@ -1,7 +1,7 @@ # Copyright (C) 2026 Intel Corporation # SPDX-License-Identifier: Apache-2.0 -"""Dotted-path imports for trusted robot owner configuration. +"""Dotted-path imports for robot owner configuration. Canonical implementation lives in :mod:`physicalai.config.importing`. This module re-exports for backward-compatible imports without loading export or diff --git a/src/physicalai/robot/transport/_owner_config.py b/src/physicalai/robot/transport/_owner_config.py index 9fe6bf8..8a2913d 100644 --- a/src/physicalai/robot/transport/_owner_config.py +++ b/src/physicalai/robot/transport/_owner_config.py @@ -9,22 +9,22 @@ subprocess stdin handshake. The owner must construct the robot driver itself: a live serial/socket -handle cannot cross a process boundary (D15). Only a trusted +handle cannot cross a process boundary (D15). Only a local :class:`~physicalai.config.ComponentConfig` (``class_path`` + ``init_args``) survives that boundary — arbitrary robot types, including third-party plugins, work without any registry lookup here. Private stdin is ``robot: ComponentConfig`` only. The owner envelope is validated schema-positively: required ``robot``, known transport keys, and -rejection of unknown keys (including legacy flat ``robot_class`` / -``robot_kwargs``) before import or hardware access. Public ``SharedRobot`` -and ``physicalai robot serve`` use the same ``robot`` / ``--robot`` shape. - -Security: ``class_path`` is trusted local application/config input, exactly -like a jsonargparse ``class_path`` (``docs/development/security.md`` rules -4, 9, 11). It must never originate from network-received data (e.g. a -``/metadata`` payload) — that would let an untrusted peer choose an -arbitrary module to import. +rejection of unknown keys before import or hardware access. Public +``SharedRobot`` and ``physicalai robot serve`` use the same +``robot`` / ``--robot`` shape. + +Security: ``class_path`` is local application/config input, exactly like a +jsonargparse ``class_path`` (``docs/development/security.md`` rules 4, 9, +11). It must never originate from network-received data (e.g. a +``/metadata`` payload) — that would let a peer choose an arbitrary module to +import. """ from __future__ import annotations @@ -54,8 +54,8 @@ justify a different value for a specific robot class. """ -# Allowed keys on owner stdin handshake payloads. Everything else (including -# legacy flat robot_class / robot_kwargs) is an unknown-key schema error. +# Allowed keys on owner stdin handshake payloads. Everything else is an +# unknown-key schema error. # Keep this the single allowlist — :meth:`RobotOwnerConfig.from_json_dict` # and any future reconfigure path should share :func:`validate_owner_config`. _OWNER_ENVELOPE_KEYS = frozenset({ @@ -214,9 +214,9 @@ def from_json_dict(cls, data: dict[str, Any]) -> RobotOwnerConfig: def build(self) -> Robot: """Instantiate the robot driver described by this config. - Owner stdin is a trusted parent→child handshake only — never pass + Owner stdin is a parent→child local handshake only — never pass network metadata to this path. Uses :func:`physicalai.config.instantiate` - on the trusted ``robot`` ComponentConfig, then verifies the + on the ``robot`` ComponentConfig, then verifies the :class:`~physicalai.robot.Robot` protocol. Returns: diff --git a/src/physicalai/robot/transport/_shared_robot.py b/src/physicalai/robot/transport/_shared_robot.py index c943e8d..3fb41fa 100644 --- a/src/physicalai/robot/transport/_shared_robot.py +++ b/src/physicalai/robot/transport/_shared_robot.py @@ -200,8 +200,9 @@ class SharedRobot: name: Required logical name — keys the Zenoh topics directly. Two instances constructed with the same *name* (anywhere reachable under the chosen transport scope) share one owner. - robot: Trusted driver :class:`~physicalai.config.ComponentConfig` to - spawn if no owner exists yet for *name*. ``None`` means + robot: Driver :class:`~physicalai.config.ComponentConfig` from local + config input (same boundary as CLI/app args), used + to spawn if no owner exists yet for *name*. ``None`` means attach-only — use :meth:`attach` for that case. Declared as an ``@export_config`` config arg, so nested :func:`~physicalai.config.instantiate` passes the recipe through @@ -263,10 +264,10 @@ def from_config( connect_timeout: float = 10.0, _session: object | None = None, ) -> SharedRobot: - """Primary API: spawn/attach from a trusted robot ComponentConfig. + """Primary API: spawn/attach from a local robot ComponentConfig. Args: - robot_config: Trusted ``class_path`` + ``init_args`` for the driver. + robot_config: Local ``class_path`` + ``init_args`` for the driver. name: Logical owner name (Zenoh topic key). allow_remote: Whether this session / spawned owner may leave localhost. rate_hz: Owner loop rate when this instance spawns the owner. @@ -285,7 +286,7 @@ def from_config( rate_hz=rate_hz, idle_timeout=idle_timeout, connect_timeout=connect_timeout, - **({} if _session is None else {"_session": _session}), + _session=_session, ) @classmethod @@ -313,9 +314,9 @@ def attach( A ``SharedRobot`` that never spawns an owner. """ # Omit default None so @export_config does not capture "_session": null. - if _session is not None: - return cls(name, allow_remote=allow_remote, connect_timeout=connect_timeout, _session=_session) - return cls(name, allow_remote=allow_remote, connect_timeout=connect_timeout) + if _session is None: + return cls(name, allow_remote=allow_remote, connect_timeout=connect_timeout) + return cls(name, allow_remote=allow_remote, connect_timeout=connect_timeout, _session=_session) @property def _robot_class(self) -> str | None: From 82671fd3c4e0f688d967732b61aee4c60d48698b Mon Sep 17 00:00:00 2001 From: Max Xiang Date: Tue, 28 Jul 2026 11:33:43 +0200 Subject: [PATCH 3/5] add example --- src/physicalai/cli/robot.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/physicalai/cli/robot.py b/src/physicalai/cli/robot.py index 58220c4..df4b401 100644 --- a/src/physicalai/cli/robot.py +++ b/src/physicalai/cli/robot.py @@ -60,7 +60,13 @@ def _build_serve_parser() -> ArgumentParser: "--robot", type=dict, required=True, - help="Trusted robot ComponentConfig (class_path + init_args).", + help=( + "Robot ComponentConfig (class_path + init_args). " + "Example: --robot='{\"class_path\":\"physicalai.robot.SO101\"," + "\"init_args\":{\"port\":\"/dev/ttyACM0\"}}'. " + "Or nested flags: --robot.class_path=physicalai.robot.SO101 " + "--robot.init_args.port=/dev/ttyACM0" + ), ) parser.add_argument( "--allow_remote", From 8528917153c669187bd5b00a5e69f90b533d4596 Mon Sep 17 00:00:00 2001 From: Max Xiang Date: Tue, 28 Jul 2026 11:50:01 +0200 Subject: [PATCH 4/5] address copilot --- src/physicalai/cli/robot.py | 4 ++-- src/physicalai/robot/transport/_owner_config.py | 7 +++++++ tests/unit/robot/transport/test_owner_config.py | 9 +++++++++ 3 files changed, 18 insertions(+), 2 deletions(-) diff --git a/src/physicalai/cli/robot.py b/src/physicalai/cli/robot.py index df4b401..fd9076a 100644 --- a/src/physicalai/cli/robot.py +++ b/src/physicalai/cli/robot.py @@ -62,8 +62,8 @@ def _build_serve_parser() -> ArgumentParser: required=True, help=( "Robot ComponentConfig (class_path + init_args). " - "Example: --robot='{\"class_path\":\"physicalai.robot.SO101\"," - "\"init_args\":{\"port\":\"/dev/ttyACM0\"}}'. " + 'Example: --robot=\'{"class_path":"physicalai.robot.SO101",' + '"init_args":{"port":"/dev/ttyACM0"}}\'. ' "Or nested flags: --robot.class_path=physicalai.robot.SO101 " "--robot.init_args.port=/dev/ttyACM0" ), diff --git a/src/physicalai/robot/transport/_owner_config.py b/src/physicalai/robot/transport/_owner_config.py index 8a2913d..8af4a1e 100644 --- a/src/physicalai/robot/transport/_owner_config.py +++ b/src/physicalai/robot/transport/_owner_config.py @@ -197,10 +197,17 @@ def from_json_dict(cls, data: dict[str, Any]) -> RobotOwnerConfig: Raises: TypeError: If ``data`` is not a mapping. + ValueError: If required ``name`` is missing or not a string. """ if not isinstance(data, dict): msg = f"owner config must be a mapping, got {type(data).__name__}" raise TypeError(msg) + if "name" not in data: + msg = "owner config missing required 'name'" + raise ValueError(msg) + if not isinstance(data["name"], str): + msg = f"owner config 'name' must be a string, got {type(data['name']).__name__}" + raise TypeError(msg) robot = validate_owner_config(data) return cls( diff --git a/tests/unit/robot/transport/test_owner_config.py b/tests/unit/robot/transport/test_owner_config.py index 53236b8..7fe7573 100644 --- a/tests/unit/robot/transport/test_owner_config.py +++ b/tests/unit/robot/transport/test_owner_config.py @@ -170,6 +170,15 @@ def test_missing_robot_rejected(self) -> None: with pytest.raises(ValueError, match="missing required 'robot'"): RobotOwnerConfig.from_json_dict({"name": "left-arm"}) + def test_missing_name_rejected(self) -> None: + with pytest.raises(ValueError, match="missing required 'name'"): + RobotOwnerConfig.from_json_dict({"robot": _fake_robot()}) + + @pytest.mark.parametrize("bad_name", [None, 1, True]) + def test_non_string_name_rejected(self, bad_name: object) -> None: + with pytest.raises(TypeError, match="owner config 'name' must be a string"): + RobotOwnerConfig.from_json_dict({"name": bad_name, "robot": _fake_robot()}) + def test_validate_owner_config_shared_helper(self) -> None: robot = validate_owner_config( { From 647c5459124c68f711cafe857941bbfcaf540a79 Mon Sep 17 00:00:00 2001 From: Max Xiang Date: Tue, 28 Jul 2026 12:48:51 +0200 Subject: [PATCH 5/5] fix test --- .../robot/transport/_shared_robot.py | 19 +++++++++++++++---- .../unit/robot/transport/test_shared_robot.py | 14 +++++++++++--- 2 files changed, 26 insertions(+), 7 deletions(-) diff --git a/src/physicalai/robot/transport/_shared_robot.py b/src/physicalai/robot/transport/_shared_robot.py index 3fb41fa..81f3108 100644 --- a/src/physicalai/robot/transport/_shared_robot.py +++ b/src/physicalai/robot/transport/_shared_robot.py @@ -214,7 +214,8 @@ class SharedRobot: narrows an already-running owner's reachability. rate_hz: Owner loop rate when this instance spawns the owner. idle_timeout: Seconds with zero subscribers before a spawned owner - self-exits (and homes/holds the robot). + self-exits (and homes/holds the robot). ``None`` disables idle + exit (owner stays up until stopped). connect_timeout: Default overall budget for :meth:`connect`. """ @@ -225,7 +226,7 @@ def __init__( robot: ComponentConfig | Mapping[str, object] | None = None, allow_remote: bool = False, rate_hz: float = DEFAULT_RATE_HZ, - idle_timeout: float = 10.0, + idle_timeout: float | None = 10.0, connect_timeout: float = 10.0, _session: object | None = None, ) -> None: @@ -260,7 +261,7 @@ def from_config( name: str, allow_remote: bool = False, rate_hz: float = DEFAULT_RATE_HZ, - idle_timeout: float = 10.0, + idle_timeout: float | None = 10.0, connect_timeout: float = 10.0, _session: object | None = None, ) -> SharedRobot: @@ -271,7 +272,8 @@ def from_config( name: Logical owner name (Zenoh topic key). allow_remote: Whether this session / spawned owner may leave localhost. rate_hz: Owner loop rate when this instance spawns the owner. - idle_timeout: Idle self-exit timeout for a spawned owner. + idle_timeout: Idle self-exit timeout for a spawned owner. ``None`` + disables idle exit. connect_timeout: Overall budget for :meth:`connect`. Returns: @@ -279,6 +281,15 @@ def from_config( writes only the new owner stdin shape on spawn. """ # Omit default None so @export_config does not capture "_session": null. + if _session is None: + return cls( + name, + robot=robot_config, + allow_remote=allow_remote, + rate_hz=rate_hz, + idle_timeout=idle_timeout, + connect_timeout=connect_timeout, + ) return cls( name, robot=robot_config, diff --git a/tests/unit/robot/transport/test_shared_robot.py b/tests/unit/robot/transport/test_shared_robot.py index e046dbb..e5b78d1 100644 --- a/tests/unit/robot/transport/test_shared_robot.py +++ b/tests/unit/robot/transport/test_shared_robot.py @@ -31,14 +31,20 @@ _STATE_DIM = 12 # fake ships positions + velocities -def _shared_robot(name: str, *, allow_remote: bool = False, **robot_init_args: object) -> SharedRobot: +def _shared_robot( + name: str, + *, + allow_remote: bool = False, + idle_timeout: float | None = 0.5, + **robot_init_args: object, +) -> SharedRobot: return SharedRobot( name, robot={ "class_path": FAKE_ROBOT_CLASS, "init_args": {"device_ids": [f"fake:{name}"], **robot_init_args}, }, - idle_timeout=0.5, + idle_timeout=idle_timeout, allow_remote=allow_remote, ) @@ -238,7 +244,9 @@ def test_metadata_exposed(self, module_owner: SharedRobot) -> None: assert robot.metadata["device_ids"] == [f"fake:{robot.name}"] def test_remote_owner_metadata_redacts_device_ids(self, unique_id: str) -> None: - robot = _shared_robot(unique_id.replace("/", "-"), allow_remote=True) + # Remote scouting can take longer than the default 0.5s idle timeout; + # keep the owner alive until this test attaches and stops it explicitly. + robot = _shared_robot(unique_id.replace("/", "-"), allow_remote=True, idle_timeout=None) robot.connect() try: assert robot.metadata is not None