Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 23 additions & 20 deletions docs/how-to/runtime/share-a-robot.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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
Expand All @@ -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

Expand Down Expand Up @@ -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,
)
```
Expand Down
52 changes: 45 additions & 7 deletions src/physicalai/cli/robot.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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).",

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we add an example here? I imagine this would require a json string?

)
parser.add_argument(
"--allow_remote",
Expand Down Expand Up @@ -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``.

Expand Down Expand Up @@ -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

Expand Down
8 changes: 8 additions & 0 deletions src/physicalai/robot/so101/calibration.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
15 changes: 10 additions & 5 deletions src/physicalai/robot/so101/so101.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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).

Expand All @@ -119,20 +121,23 @@ 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).

``calibration`` may be:

* ``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"``.
Expand All @@ -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(...) "
Expand Down Expand Up @@ -202,7 +207,7 @@ def uncalibrated(
baudrate=baudrate,
role=role,
unit=unit,
_allow_uncalibrated=True,
allow_uncalibrated=True,
)

@property
Expand Down
Loading
Loading