Mesh

The Mesh object a robot exposes, the session and peer helpers, the bridges that put ROS 2 robots on the mesh.

The mesh puts robots on a shared Zenoh session so agents and peers discover each other, exchange commands and stop together: the Mesh object a robot exposes, the session and peer helpers, and the bridge classes that put ROS 2 and RTPS robots on the same mesh.

Mesh

Core Mesh class - lifecycle, presence, state, cameras, RPC, and subscriptions.

This is the primary component that a Robot or Simulation composes with. Extended sensor loops (pose, IMU, health, etc.) are provided by :class:~strands_robots.mesh.sensors.SensorLoopsMixin.

init_mesh

init_mesh(robot: Any, peer_id: str | None = None, peer_type: str = 'robot', mesh: bool = True) -> Mesh | None

Construct and start a Mesh for the given robot.

Returns None when mesh is disabled. STRANDS_MESH=false is a hard kill switch and an explicit mesh=False both disable mesh; the env var only forces mesh OFF, never ON (so an explicit opt-out is always honoured).

get_local_robots

get_local_robots() -> dict[str, Mesh]

Return a snapshot of in-process mesh-enabled robots.

mesh_disabled_by_env

mesh_disabled_by_env() -> bool

Report whether STRANDS_MESH forces the mesh off.

STRANDS_MESH=false (or 0 / no) is documented in README's Configuration table as "a hard kill switch that also overrides an explicit mesh=True". An operator who sets it is asking for no Zenoh session and no presence on the fleet, so every path that can open one answers this -- not only :func:strands_robots.mesh.core.init_mesh.

The switch is one-directional here: it only ever forces mesh OFF. Opting a bare Robot() on via STRANDS_MESH=true is resolved in the Robot factory, which reads the affirmative spellings instead. A caller asking "may I start a mesh?" wants this predicate; a caller asking "was I asked to start one?" wants that one, and the two are not each other's negation -- an unset variable answers False to both.

Resolved by :func:strands_robots._mesh_switch.mesh_env_request, which holds both halves of the vocabulary. That is what makes an unrecognized value reportable: this predicate alone cannot tell off (a typo) from true (the other reader's business), because both are equally "not a kill" to it.

strands_robots.mesh.core.Mesh

Mesh(robot: Any, peer_id: str, peer_type: str = 'robot')

Bases: SensorLoopsMixin

Peer-to-peer mesh component embedded in a single Robot or Simulation.

Lifecycle: construct via :func:init_mesh, call :meth:stop during cleanup.

Thread safety

:meth:start and :meth:stop are protected by _lifecycle_lock.

alive property

alive: bool

True while this peer is joined to the mesh (between :meth:join and :meth:leave); False once it has left.

peers property

peers: list[dict[str, Any]]

Presence dicts for every other peer currently on the mesh.

Excludes this peer itself. Discovery is asynchronous, so the list grows as presence beacons arrive. Use :attr:peers_by_id for O(1) lookup by peer_id or :meth:get_peer for a None-safe fetch.

peers_by_id property

peers_by_id: dict[str, dict[str, Any]]

Peers keyed by peer_id for dict-style lookup.

Complements :attr:peers (a list[dict]). README pseudo-code used mesh.peers[peer_id] expecting dict access; on the list that raises TypeError (GH #373 friction #8). Use this for O(1) lookup::

info = robot.mesh.peers_by_id[other.peer_id]

or the :meth:get_peer helper for a None-safe single lookup.

lockout_epoch property

lockout_epoch: str | None

The id of the e-stop holding this peer's lockout, or None while clear.

Every peer locked by one fleet e-stop holds the same id (the issuer's estop_id); a signed resume names it, so it clears that lockout and no later one.

start

start() -> None

Acquire a Zenoh session and start all publishing loops.

stop

stop() -> None

Stop all loops and release the session reference.

Waits for the loops :meth:start launched before releasing anything they publish through, so a tick already inside :meth:publish cannot land on the wire after this peer has announced it left. The wait is bounded by :data:LOOP_JOIN_TIMEOUT_S and shared across the loops; a sensor read that blocks past it leaves its loop free to publish once more, and that loop is named at WARNING rather than the stop being reported as complete.

Drops every :meth:subscribe subscription and clears :attr:inbox: the subscribers are undeclared with the session reference, and the (topic, callback) pairs behind them are not retained, so :meth:start re-declares only this peer's built-in topics. A rejoining caller re-declares its own subscriptions; the count dropped is reported at INFO so that is visible rather than inferred.

get_peer

get_peer(peer_id: str, max_age_s: float | None = None) -> dict[str, Any] | None

Return a single peer's info dict by peer_id, or None.

None-safe counterpart to peers_by_id[peer_id] -- prefer this when the peer may not be present yet (discovery is asynchronous).

Parameters:

Name	Type	Description	Default
peer_id	str	The peer to look up. This peer's own id answers None, matching :attr:peers (which lists other peers).	required
max_age_s	float | None	Optional freshness bound in seconds; a record older than this answers None as if unknown. The domain (positive finite) and the reasoning live on :func:strands_robots.mesh.session.get_peer, which this forwards to - two spellings of one bound must not diverge.	None

peer_cert

peer_cert(peer_id: str) -> tuple[str, str] | None

(cert_sha256, cn) of the certificate peer_id last announced itself with, or None.

peer_wire_zid

peer_wire_zid(peer_id: str) -> str | None

The session id peer_id last announced itself from, or None.

A publisher-chosen hint (SourceInfo), not a verified identity: see :mod:~strands_robots.mesh.wire_identity for the signed one.

send

send(target: str, cmd: dict[str, Any], timeout: float = 30.0) -> dict[str, Any]

Send a command to a single peer and return the first response.

Phase-4 / D1 hardening: turn_id is a full 128-bit uuid4 (no truncation), and the expected responder is recorded so :meth:_on_response rejects forged responses from any peer other than target.

explicit guard against passing the :data:BROADCAST_RESPONDER sentinel (or any string containing a NUL byte) as target. init_mesh's peer_id regex already rejects NUL on the receive side, so a real peer can't collide, but a future refactor that loosens that rule must not reopen the response-hijack surface that this method's contract closes.

ping

ping(target: str, timeout: float = 2.0) -> dict[str, Any]

Is target reachable, and how fast: one ping command and its round trip.

The command is {"action": "ping"}, answered by the peer's mesh layer with an empty result (no robot method is reached, and it is answered under an e-stop lockout too). With a direct transport (AWS IoT Core Direct Messaging) the command is delivered to target alone with confirmation and an unreachable peer is reported in one round trip; over Zenoh the command is published and the answer is whoever holds that peer id.

Parameters:

Name	Type	Description	Default
target	str	The peer id.	required
timeout	float	Budget for the whole round trip, seconds.	2.0

Returns:

Type	Description
dict[str, Any]	{"status": "ok", "latency_ms": float, "via": "direct"|"publish", "confirmed": bool}
dict[str, Any]	when the peer answered; ``{"status": "offline", "latency_ms": float,
dict[str, Any]	"via": "direct", "reason": "offline"}`` when the broker knows the peer
dict[str, Any]	is gone; {"status": "timeout", "latency_ms": float, "via": ...}
dict[str, Any]	when nothing answered inside timeout; or send's own
dict[str, Any]	{"status": "error", ...} envelope for a refused precondition.

broadcast

broadcast(cmd: dict[str, Any], timeout: float = 5.0) -> list[dict[str, Any]]

Broadcast a command to every peer and return all responses.

Phase-4 / D1: turn_id is a full 128-bit uuid4 (no truncation). Broadcast turns accept responses from any responder by design, so the expected-target check is bypassed (sentinel BROADCAST_RESPONDER); the wire-source binding and the one-reply-per-session rule in :meth:_on_response still apply.

tell

tell(target: str, instruction: str, **kw: Any) -> dict[str, Any]

Shorthand: ask a peer to run a policy with a natural-language instruction.

Sends {"action": "execute", "instruction": instruction, **kw}. The instruction alone is not a command the peer can act on - it is the text a policy conditions on - so policy_provider= is required: :func:~strands_robots.mesh.security.validate_command refuses an execute without one before it leaves this process ("Silent defaults are not honoured on the security boundary"). Checkpoints travel as Hub ids (pretrained_name_or_path="lerobot/…", an org in STRANDS_MESH_HF_REPO_ALLOW); a local path is refused on the wire. duration defaults to 30 s when omitted.

Example::

mesh.tell(peer, "hold the tray steady", policy_provider="lerobot_local",
          pretrained_name_or_path="lerobot/smolvla_base", duration=10.0)

Parameters:

Name	Type	Description	Default
target	str	The peer id to address.	required
instruction	str	Natural-language instruction the policy conditions on.	required
**kw	Any	The policy - policy_provider (required), its checkpoint or port, duration and any provider keyword the wire allows.	{}

Returns:

Type	Description
dict[str, Any]	The peer's reply for the execute command.

subscribe

subscribe(topic: str, callback: Callable[[str, dict[str, Any]], None] | None = None, name: str | None = None, *, on_sample: Callable[[str, dict[str, Any], str | None], None] | None = None) -> str | None

Subscribe to any Zenoh topic and receive parsed JSON dicts.

callback(key, data) receives the decoded payload. on_sample(key, data, wire_zid) receives it together with the publisher's TLS-bound session id read off the sample (:func:_extract_sample_source_zid, None when the sample carried none), for a subscriber that must bind what it applies to who published it; the teleop input receiver is one. The two are exclusive.

Returns:

Type	Description
str | None	The subscription name (name when given, else topic) once the
str | None	subscriber is declared, or None when it was not. Every
str | None	None says why at WARNING, matching how the rest of this class
str | None	reports a client-side refusal: the peer is not on the mesh, there
str | None	is no session to declare against, or declare_subscriber itself
str | None	failed.

A subscription does not survive :meth:stop. That method drops every subscription this one records and :meth:start re-declares only the peer's own built-in topics, so a caller that rejoins the mesh re-declares its own subscriptions - which is what the WARNING above makes visible when a rejoin has not happened yet.

unsubscribe

unsubscribe(name: str) -> None

Unsubscribe from a topic by name.

publish_step

publish_step(step: int, observation: dict[str, Any], action: dict[str, Any], instruction: str = '', policy: str = '') -> None

Publish one VLA execution step to the mesh.

on_stream

on_stream(peer_id: str, callback: Callable[[str, dict[str, Any]], None] | None = None) -> str | None

Subscribe to another peer's VLA execution stream.

emergency_stop

emergency_stop() -> list[dict[str, Any]]

Stop the local robot, broadcast a stop, and engage the local lockout.

The robot registered in this process is stopped first, through the same :meth:_dispatch path a remote peer runs. broadcast never comes back to the sender -- _on_cmd drops envelopes carrying our own sender_id -- so the one robot an operator is standing next to is the one robot the fanout cannot reach.

After this call the local mesh refuses every action but status, resume and stop until :meth:_resume_lockout admits an assertion signed by the operator's resume key naming this peer and the lockout epoch (the estop_id this call publishes). stop stays admitted because it only ever de-energizes: a second e-stop reaching an already locked-out peer must halt a rollout the first one missed rather than be rejected. The event is also published on strands/safety/estop and recorded in the audit log (see :func:strands_robots.audit.log_safety_event).

Returns the responses collected within the broadcast timeout, the local robot's own answer first (shaped like a peer's, with this peer's id) -- useful for telemetry, and counted in peers_not_stopped exactly as a remote answer is. A peer with no robot registered contributes no local answer: it has nothing to halt.

A response is only an acknowledgement that the peer STOPPED if it says so. A peer whose registered robot exposes no stop_task answers {"ok": False, ...}; such peers are counted separately, logged at CRITICAL, and reported in the safety envelope as peers_not_stopped. Counting them as acknowledgements would tell an operator the fleet had halted while a robot was still moving.

Raises RuntimeError when the mesh is not running: an e-stop that reached no peer must not look like "asked, nobody answered" ([]).

resume

resume(signing_key: Any, *, targets: list[str] | None = None) -> dict[str, Any]

Clear this peer's e-stop lockout, and the fleet's, with the operator's key.

Signs an assertion for this peer's current lockout epoch naming targets (default: this peer and every reachable peer on the roster), admits it here and relays it on strands/safety/resume so each named peer locked by the same e-stop clears too. Only the operator's machine holds signing_key; every peer verifies with its public half.

Parameters:

Name	Type	Description	Default
signing_key	Any	The Ed25519 private key from :func:~strands_robots.mesh.resume_authority.load_signing_key.	required
targets	list[str] | None	Peer ids the resume is for. This peer is always included.	None

Returns:

Type	Description
dict[str, Any]	{"status": "ok"}, or {"status": "error", "error": "resume rejected"}
dict[str, Any]	with the reason in the local audit log.

publish

publish(key: str, payload: dict[str, Any]) -> None

Publish payload on key via the mesh transport.

Wire authentication is owned by the Zenoh transport: outbound bytes ride a TLS link whose cert binds the peer identity, and the ACL gates which key-expressions this peer can publish on. This method simply forwards to put() -- it stays as a single chokepoint so a future hook (audit, telemetry, compression) can land in one place.

Renamed from _put_signed after the application-layer signing envelope was dropped (commit 7113742). The old name was a historical artefact: nothing in the body ever signed anything once Zenoh's mTLS + ACL took over identity and authorization.

Session and peers

Shared Zenoh session and peer registry for the mesh networking layer.

This module provides a single, ref-counted :func:zenoh.open session per process and a thread-safe registry of discovered peers. It is the lowest layer of the mesh stack - higher-level constructs (Mesh, presence, RPC) build on top.

The Zenoh dependency is lazy: import strands_robots.mesh_session does not import zenoh at module level. The first call to :func:get_session triggers the real import. If eclipse-zenoh is not installed the function returns None and all publish helpers become safe no-ops.

Connection strategy (when no explicit endpoint is configured):

1. Try to listen on tcp/127.0.0.1:{STRANDS_MESH_PORT} - this makes the first process the local router.
2. If the port is already bound, fall back to client mode and connect to the same endpoint.
3. Zenoh gossip scouting propagates peers reachable through those endpoints. Multicast scouting is disabled by default (LAN discovery attack surface); operators on a controlled LAN can opt in with STRANDS_MESH_MULTICAST=true. Cross-host peers otherwise need explicit ZENOH_CONNECT endpoints.

Environment variables

ZENOH_CONNECT Comma-separated remote endpoint(s) - e.g. tcp/10.0.0.1:7447. ZENOH_LISTEN Comma-separated listen endpoint(s). STRANDS_MESH_PORT Local auto-mesh port (default 7447). STRANDS_MESH Set to false to disable mesh globally. STRANDS_MESH_MULTICAST true to opt into LAN multicast scouting (logs a warning). Default false.

get_session

get_session() -> Any | None

Acquire the shared mesh transport (lazy, ref-counted).

Backend selection comes from STRANDS_MESH_BACKEND:

• zenoh (default) - open / reuse a zenoh.Session exactly as before. Returned object is the raw session; callers can .declare_subscriber() on it.
• iot / bridge - delegate to :mod:strands_robots.mesh.transport.factory; the returned object is an :class:~strands_robots.mesh.transport.IotMqttTransport or :class:~strands_robots.mesh.transport.BridgeTransport which also exposes put() / declare_subscriber() / close() so existing Mesh code works unchanged.

Returns:

Type	Description
Any | None	Backend-dependent: zenoh.Session, IotMqttTransport,
Any | None	BridgeTransport, or None if the chosen backend is unavailable.

release_session

release_session() -> None

Release one reference to the shared mesh session.

Delegates to the transport factory when the active backend is iot / bridge; otherwise falls back to the legacy Zenoh refcount.

On the Zenoh path the final release closes the session. That close is fail-soft over the surface :func:zenoh_error_types documents - a broker drop or socket teardown race is logged at WARNING and the reference is still dropped, because nothing can retry a close once the only handle to the session is gone. A failure outside that surface (a TypeError or AttributeError, i.e. a bug rather than a transport fault) propagates, matching how :meth:strands_robots.mesh.core.Mesh.stop treats its undeclare calls. The "session closed" INFO line is emitted only when the close actually completed.

current_session

current_session() -> Any | None

Return the existing session/transport without bumping the refcount.

Backend-aware: returns the active transport singleton when STRANDS_MESH_BACKEND is iot / bridge, otherwise the raw Zenoh session (legacy behaviour).

session_alive

session_alive() -> bool

Return True if the current backend's session/transport is open.

put

put(key: str, data: dict[str, Any]) -> None

Publish a JSON payload to the mesh.

Fire-and-forget. No-op when no session/transport is open.

Backend-aware: delegates to the active transport's put() when running under STRANDS_MESH_BACKEND=iot / bridge; otherwise encodes JSON and pushes to the Zenoh session directly (legacy path).

get_peers

get_peers() -> list[dict[str, Any]]

Return all known peers as plain dicts.

update_peer

update_peer(peer_id: str, peer_type: str, hostname: str, caps: dict[str, Any]) -> bool

Insert or update a peer. Returns True when the peer is new.

clear_peers

clear_peers() -> None

Remove all peers. Intended for tests only.

prune_peers

prune_peers(timeout: float = PEER_TIMEOUT) -> list[str]

Remove peers whose silence has outlived what the fleet tolerates.

Out of contact is not gone. A peer past timeout stops being reachable (the field :meth:PeerInfo.to_dict reports) but is deleted only once its silence exceeds max(timeout, STRANDS_MESH_PEER_RETENTION_S). With retention unset (the default 0) that maximum is timeout and behavior is unchanged: silent peers are deleted at the timeout, exactly as before. With retention set, a satellite between ground-station passes or a rover in an RF shadow stays in the registry as an unreachable row a fleet view can render and a dispatcher can decline to fail over - deletion would answer "was it ever here?" with "no", which is the wrong answer for a peer the operator expects back.

Retention is registry policy owned by this process, never something a peer requests about itself, and the :func:update_peer eviction cap still bounds the registry: at the cap the oldest peer goes first, which under retention means out-of-contact peers are the first sacrificed to a flood of new ones - the bound outranks the courtesy.

Returns:

Type	Description
list[str]	List of pruned peer IDs (may be empty).

Input streams

Input device streaming over the mesh - publish and receive teleoperator actions.

Enables remote teleoperation: a leader arm on machine A publishes its joint positions via :class:InputPublisher, and the follower arm on machine B receives and applies them via :class:InputReceiver.

Topic schema for strands/{peer_id}/input/{device_name}::

{
    "peer_id": "<publisher-peer-id>",
    "device": "<device-name>",
    "method": "arm" | "gamepad" | "keyboard" | "phone",
    "t": <unix-timestamp>,
    "seq": <monotonic-frame-counter>,
    "action": {"motor.pos": float, ...},
    "events": {"terminate_episode": bool, ...} | null
}

events is null both when the teleoperator exposes no get_teleop_events() surface and when reading it failed, so the publisher side reports a failed read through InputPublisher.stats (event_read_errors) and a log line rather than only on the wire.

InputPublisher

InputPublisher(mesh: Mesh, teleoperator: Any, device_name: str = 'leader', method: str = 'arm', hz: float = INPUT_HZ_DEFAULT)

Publishes teleoperator actions to the mesh at a fixed rate.

Runs in a background thread, polling the teleoperator and publishing normalized action dicts.

Raises:

Type	Description
ValidationError	device_name is not a valid mesh identifier (see :func:~strands_robots.mesh.security.validate_mesh_identifier); it is interpolated into the published key expression.
ValueError	hz is not a rate this loop can honor, or teleoperator cannot be polled for an action.

Bind a teleoperator to a mesh topic at a fixed publish rate.

Parameters:

Name	Type	Description	Default
mesh	Mesh	Live mesh used as the single publish chokepoint.	required
teleoperator	Any	Any object exposing a callable get_action() -> dict.	required
device_name	str	Input-stream name; becomes the last topic segment.	'leader'
method	str	Input-method label ("arm", "gamepad", "keyboard", "phone").	'arm'
hz	float	Publish rate. Must be a positive finite number - the loop period is 1 / hz, so 0 raises inside the background thread and a negative/nan/inf rate leaves the loop unthrottled, flooding every subscribed peer.	INPUT_HZ_DEFAULT

Raises:

Type

Description

ValueError

If hz is not a positive finite number, or if teleoperator has no callable get_action. Refusing both at construction is what keeps them contracts: :meth:_publish_loop runs on a background thread, where either mistake would surface as a dead publisher that still reports running - the rate as a loop that never completes a tick, the device as a loop that counts an AttributeError per tick and falls silent once its logging budget is spent. The device shares :func:~strands_robots.utils.teleoperator_contract_error with the local attach door and the mesh publish entry point above it.

stats property

stats: dict[str, Any]

Live publishing counters: the target device/method, whether the loop is running, whether its thread_alive, cumulative frames published and errors hit, event_read_errors (frames published with events: null because get_teleop_events() raised, rather than because the teleoperator has no event surface), and the achieved vs. requested rate (hz_actual / hz_target).

running is the session flag, which :meth:stop clears before it joins; thread_alive reads the publish thread itself. The two differ for exactly as long as a loop outlives the stop that asked it to exit - the window in which this publisher can still put a frame on the wire. Without the second reading a caller polling after a stop would be told running=False about a loop that is still publishing.

topic property

topic: str

Mesh key this publisher writes to: strands/{own_peer_id}/input/{device_name}. A remote :class:InputReceiver subscribes to this exact key to mirror the actions locally.

start

start() -> None

Start the input publishing loop.

stop

stop() -> dict[str, Any]

Stop the input publishing loop and return stats.

join() returns None whether or not the thread finished, so the liveness read after it is the only thing that tells a stopped loop from one that outlasted :data:_INPUT_JOIN_TIMEOUT_S. A teleoperator whose get_action() blocks past that budget - a serial read on a wedged bus is the ordinary case - leaves the loop free to publish one more frame after this call returns, so the outcome rides back in stats["thread_alive"] and is logged at WARNING rather than being announced as a stop that happened.

A publisher whose join timed out is still stoppable: the session flag is already clear, so the guard below admits a call that has a live thread left to join. Returning early on the flag alone would make the only handle to that loop unreachable through this surface.

InputReceiver

InputReceiver(mesh: Mesh, robot: Any, source_peer_id: str, device_name: str = 'leader', apply_fn: Callable[[Any, dict[str, float]], Any] | None = None)

Subscribes to a remote peer's input stream and applies actions locally.

Listens on strands/{source_peer_id}/input/{device_name} and calls robot.send_action(action) for each received frame.

Raises:

Type	Description
ValidationError	source_peer_id or device_name is not a valid mesh identifier (see :func:~strands_robots.mesh.security.validate_mesh_identifier). Both are interpolated into the subscribed key expression, where a Zenoh wildcard would widen this stream to every publishing peer.

stats property

stats: dict[str, Any]

Live receive counters: the source peer and device, whether the subscription is running, frames_received, errors, and the loss/back-pressure breakdown - out-of-order drops, rejected frames, and rate_dropped frames (shed to hold the apply-rate cap), and slew_rejected frames (refused for commanding a joint faster than the per-joint slew bound) - plus the achieved hz_actual.

errors counts a frame the host did not apply, in either shape it reports one: an exception out of the apply, and an error envelope returned from it (a hardware follower converts the former into the latter by design, and a simulation follower answers that way for an action key it cannot resolve). Such a frame is still counted in frames_received - it was delivered and attempted - so a follower that refuses everything reads as errors == frames_received, the same signature the local teleop loop reports. That is distinct from a rejected frame, which a guard on this side refused and never applied.

rejected is the total of a breakdown that names which guard refused the frame, so a report does not have to recover the reason from the log: rejected_source (the sample's publisher session is not the one the stream was bound to when it opened), rejected_lockout (arrived during an E-stop lockout), rejected_expired (the stream's lifetime passed; the stream is stopped on that frame), rejected_freshness (the frame's t is missing, non-numeric, stale or too far in the future - the replay defence), and rejected_invalid (validate_input_frame refused the frame's shape or a value: too many keys, an illegal key, or a value that is non-scalar, non-numeric, non-finite or past the magnitude bound). rejected always equals their sum.

topic property

topic: str

Mesh key this receiver subscribes to: strands/{source_peer_id}/input/{device_name} - the stream the remote peer's :class:InputPublisher writes to.

start

start() -> None

Start receiving input actions from the remote peer.

The stream is bound to the session the leader announced its presence from and refused when there is none: the key expression scopes the stream to the leader's NAME, which any admitted peer can publish under, so without the binding an approval for one leader followed whoever reached the topic. The opening, its lifetime and the bound session are written to the safety log; :attr:start_refusal says why a stream did not open.

stop

stop() -> dict[str, Any]

Stop receiving and return stats.

Bridged robots

strands_robots.drivers.ros.ros_bridge.RosBridgedRobot

RosBridgedRobot(node_name: str, cmd_vel_topic: str, odom_topic: str, scan_topic: str | None = None, *, cmd_vel_type: str = _TWIST_TYPE, odom_type: str | None = None, scan_type: str | None = None, publish_rate: float = 10.0, max_linear: float | None = None, max_angular: float | None = None, max_duration: float | None = None, nav_action: str | None = None, nav_action_type: str = _NAV_ACTION_TYPE)

Bases: MobileBaseRobot

A remote ROS 2 robot exposed as a strands-controllable robot.

The bridge owns no ROS 2 state of its own; every method forwards to :func:~strands_robots.ros.ros_action. It is therefore safe to construct without a ROS 2 environment present - errors surface only when a method is actually called and no backend is available.

Parameters:

Name	Type	Description	Default
node_name	str	Human-readable identifier for the remote robot. Used only to name this bridge's agent tools (drive_<node_name> etc.); it does not need to match the ROS 2 node name.	required
cmd_vel_topic	str	Velocity-command topic the robot subscribes to (e.g. /turtle1/cmd_vel or /cmd_vel).	required
odom_topic	str	Topic carrying the robot's pose/odometry (e.g. /turtle1/pose or /odom). Read by :meth:get_pose.	required
scan_topic	str | None	Optional laser-scan topic (e.g. /scan). Read by :meth:get_scan; when omitted, no get_scan tool is exposed.	None
cmd_vel_type	str	Interface type published to cmd_vel_topic. Defaults to geometry_msgs/msg/Twist.	_TWIST_TYPE
odom_type	str | None	Interface type of odom_topic. Optional - when omitted, the transport resolves it from the live graph.	None
scan_type	str | None	Interface type of scan_topic. Optional - resolved from the live graph when omitted.	None
publish_rate	float	Default rate (Hz) for multi-message :meth:drive calls. Must be > 0 and finite: :meth:drive multiplies it by duration to size the message burst and the transport publishes at 1 / rate, so a non-positive rate removes the pacing entirely rather than slowing it. Raises ValueError at construction otherwise.	10.0
max_linear	float | None	Optional linear-velocity clamp (m/s). Unset by default: a generic ROS 2 base declares no speed limit to this bridge, and inventing one would silently cap an existing caller.	None
max_angular	float | None	Optional angular-velocity clamp (rad/s). Unset by default.	None
max_duration	float | None	Optional cap on a single :meth:drive hold, in seconds. Unset by default; when set, a longer request is refused loudly.	None
nav_action	str | None	Optional Nav2-style action server name (e.g. /navigate_to_pose). When set, :meth:navigate_to sends goal-level navigation instead of raw velocity, and a navigate_<node_name> agent tool is exposed.	None
nav_action_type	str	Action interface for nav_action. Defaults to nav2_msgs/action/NavigateToPose.	_NAV_ACTION_TYPE

from_ros classmethod

from_ros(node_name: str, cmd_vel_topic: str, odom_topic: str, scan_topic: str | None = None, **kwargs: Any) -> RosBridgedRobot

Construct a bridge from ROS 2 topic wiring.

Convenience alternate constructor mirroring the keyword style used elsewhere in the library. Equivalent to calling the constructor directly; provided so call sites read as RosBridgedRobot.from_ros( node_name=..., cmd_vel_topic=...).

navigate_to

navigate_to(x: float, y: float, yaw: float = 0.0, frame_id: str = 'map', timeout: float = 120.0, tool_context: ToolContext | None = None) -> dict[str, Any]

Send a goal-level navigation request to the robot's nav_action.

Unlike :meth:drive, which streams raw velocity, this delegates obstacle avoidance, path planning, and recovery to the robot's own navigation stack (Nav2 by default) and blocks until the goal reaches a terminal state or timeout expires - at which point the transport cancels the goal so the robot does not keep navigating unattended.

Parameters:

Name	Type	Description	Default
x	float	Goal position x in frame_id (meters). Must be a finite number; both signs are valid.	required
y	float	Goal position y in frame_id (meters). Must be a finite number; both signs are valid.	required
yaw	float	Goal heading in radians, encoded as a planar quaternion. Must be a finite number; both signs are valid (negative turns the other way).	0.0
frame_id	str	Frame the goal pose is expressed in (default map).	'map'
timeout	float	End-to-end budget in seconds for the navigation goal. Graded here, on the domain :meth:drive grades duration against: a non-positive or non-finite budget is refused and no goal is sent.	120.0
tool_context	ToolContext | None	Operator context forwarded to the command gate, which covers a Nav2-style /navigate_to_pose action goal as well as a cmd_vel publish (see :meth:drive).	None

Returns:

Type	Description
dict[str, Any]	The transport's action result dict (goal status, result, feedback
dict[str, Any]	samples), or an {"status": "error"} result when no
dict[str, Any]	nav_action was configured or when a pose component cannot be
dict[str, Any]	honored - in which case no goal is sent.

strands_robots.drivers.ros.rosbridge_robot.RosbridgeRobot

RosbridgeRobot(node_name: str, cmd_vel_topic: str, odom_topic: str, scan_topic: str | None = None, *, host: str = 'localhost', port: int = 9090, cmd_vel_type: str = _TWIST_TYPE, odom_type: str | None = None, scan_type: str | None = None, max_linear: float = 2.0, max_angular: float = 1.0, max_duration: float = 30.0, publish_rate: float = 10.0)

A rosbridge-reachable mobile robot exposed as a strands-controllable robot.

Parameters:

Name	Type	Description	Default
node_name	str	Identifier used to name this robot's agent tools.	required
cmd_vel_topic	str	Velocity-command topic (geometry_msgs/Twist).	required
odom_topic	str	Odometry/pose topic, read by :meth:get_pose.	required
scan_topic	str | None	Optional laser-scan topic, read by :meth:get_scan.	None
host	str	rosbridge server hostname or IP.	'localhost'
port	int	rosbridge WebSocket port.	9090
cmd_vel_type	str	Interface type of cmd_vel_topic (ROS1 two-segment).	_TWIST_TYPE
odom_type	str | None	Interface type of odom_topic; rosapi-resolved when omitted.	None
scan_type	str | None	Interface type of scan_topic; rosapi-resolved when omitted.	None
max_linear	float	Linear-velocity clamp (m/s).	2.0
max_angular	float	Angular-velocity clamp (rad/s).	1.0
max_duration	float	Longest accepted :meth:drive hold; longer requests are rejected loudly rather than silently truncated.	30.0
publish_rate	float	Command publish rate (Hz) for held :meth:drive calls.	10.0

Raises:

Type	Description
ValueError	When a graph name, the host or the port is malformed, or when any of max_linear, max_angular, max_duration and publish_rate is not a finite number greater than zero. Each bounds every later command, so an unusable one is refused at construction instead of silently reshaping the robot's limits.

tools property

tools: list[AgentTool]

This robot's capabilities as named strands agent tools.

The two tools that carry a command are declared @tool(context=True) and forward the injected context to :meth:drive and :meth:stop, so a publish to a gated cmd_vel prompts the operator instead of failing closed on every call. The read-only tools take no context because a read is never gated.

drive

drive(linear: float = 0.0, angular: float = 0.0, duration: float | None = None, count: int = 1, tool_context: ToolContext | None = None) -> dict[str, Any]

Publish a velocity command over rosbridge.

Fleet-standard across all three mobile-base bridges: inputs are validated against the shared numeric domains before any side effect, a bare single-shot command latches until :meth:stop, like any raw cmd_vel publish, and every timed or multi-message non-zero command is followed by a single zero Twist - even if the main publish failed - so a timed drive does not leave the robot with a live velocity. That zero is a gated command in its own right, so when it is the call that fails the result says so rather than reporting the hold's success. The trailing zero was this bridge's alone until the shared mobile base took over the drive contract; the other two inherit it now, so a timed drive self-stops wherever it is issued.

Not carried by every mobile base: velocities are clamped to max_linear and max_angular, and a hold beyond max_duration is refused. :class:~strands_robots.drivers.ros.ackermann_robot.AckermannRosRobot declares both as well, as max_speed and a max_duration of its own, because it too wraps a platform whose limits are known - so a hold this bridge accepts can be refused on that car, and the reverse. :meth:RosBridgedRobot.drive and :meth:RtpsRobot.drive carry neither: neither knows the ceilings of the third-party robot it drives, so they declare no velocity or duration limit and publish the requested burst unclamped. An unset limit there means "this platform declares no limit", never zero.

Parameters:

Name	Type	Description	Default
linear	float	Forward linear velocity (m/s), mapped to linear.x. Must be a finite number; both signs are valid (negative reverses).	0.0
angular	float	Yaw angular velocity (rad/s), mapped to angular.z. Must be a finite number; both signs are valid (negative turns the other way).	0.0
duration	float | None	When given, hold the command for this many seconds by publishing round(duration * publish_rate) messages (at least one). Takes precedence over count, must be > 0 and finite - a zero or negative hold has no message count that expresses it, and publishing a single velocity command anyway would start the robot moving - and may not exceed max_duration.	None
count	int	Number of messages to publish when duration is omitted. Must be a positive whole number; 0 or a negative count publishes nothing, so reporting a successful drive for it hides a command that never left the process.	1
tool_context	ToolContext | None	Operator context forwarded to the shared operator gate, which prompts before a publish to a safety-critical command surface. Without it the gate fails closed, so a command this bridge could otherwise have carried is refused with no operator ever asked.	None

Returns:

Type	Description
dict[str, Any]	The transport's publish result dict, or an
dict[str, Any]	{"status": "error"} result naming the parameter when a value
dict[str, Any]	cannot be honored - in which case nothing is published.

from_curiosity classmethod

from_curiosity(node_name: str = 'curiosity', host: str = 'localhost', port: int = 9090, **overrides: Any) -> RosbridgeRobot

Wiring for the NASA Curiosity rover Gazebo simulation (ROS1 Noetic).

The rover's ackermann_drive_controller consumes geometry_msgs/Twist directly, so no client-side kinematic model is needed. Limits ported from the strands-robots-ros2 registry entry that first drove this sim.

get_pose

get_pose(timeout: float = 5.0) -> dict[str, Any]

Read one odometry/pose sample from odom_topic.

Refuses a timeout it cannot wait out before the bridge is dialed: on an already-connected bridge a non-positive wait returns at once, so an unchecked value would report success with no sample in it.

get_scan

get_scan(timeout: float = 5.0) -> dict[str, Any]

Read one laser-scan sample (error when no scan_topic configured).

Grades timeout on the same domain as :meth:get_pose.

stop

stop(tool_context: ToolContext | None = None) -> dict[str, Any]

Publish a single zero Twist.

Never gated on this bridge's own state: a halt does not depend on a prior command having succeeded, and there is no enable handshake to satisfy. It is not exempt from the transport's command gate, which is keyed on the surface rather than the payload - zero means "stationary" on a Twist but commands motion to the zero pose on a joint-command topic, so a payload-shaped carve-out could not be written correctly. The halt stays reachable through the same approval path as any other command instead, which is why it forwards the context.

Parameters:

Name	Type	Description	Default
tool_context	ToolContext | None	Operator context forwarded to the operator gate.	None

Returns:

Type	Description
dict[str, Any]	The transport's publish result dict.

strands_robots.drivers.ros.rtps_robot.RtpsRobot

RtpsRobot(node_name: str, cmd_vel_topic: str, *, cmd_vel_type: str = _TWIST_TYPE, publish_rate: float = 10.0, max_linear: float | None = None, max_angular: float | None = None, max_duration: float | None = None)

Bases: MobileBaseRobot

A ROS 2 robot driven over pure RTPS (no rclpy), exposed as a strands robot.

The robot owns no DDS state of its own; every method forwards to :func:use_rtps, which manages the shared participant and cached writers. Safe to construct without cyclonedds present - errors surface only when a method is called and the backend is unavailable.

Parameters:

Name	Type	Description	Default
node_name	str	Identifier used to name this robot's agent tools (drive_<node_name> etc.). Need not match any ROS 2 node name.	required
cmd_vel_topic	str	Velocity-command topic to publish Twist on.	required
cmd_vel_type	str	Interface type for cmd_vel_topic (default geometry_msgs/msg/Twist).	_TWIST_TYPE
publish_rate	float	Default rate (Hz) for multi-message :meth:drive calls. Must be > 0 and finite: :meth:drive multiplies it by duration to size the message burst and use_rtps publishes at 1 / rate, so a non-positive rate removes the pacing entirely rather than slowing it. Raises ValueError at construction otherwise.	10.0
max_linear	float | None	Optional linear-velocity clamp (m/s). Unset by default - an RTPS peer drives arbitrary third-party robots whose limits this class cannot know.	None
max_angular	float | None	Optional angular-velocity clamp (rad/s). Unset by default.	None
max_duration	float | None	Optional cap on a single :meth:drive hold, in seconds.	None

advertise

advertise() -> dict[str, Any]

Create the cmd_vel publisher up front (appear on the ROS 2 graph).

RTPS-only: a DDS participant can announce a writer before it has anything to say, which is what makes an agent visible to ros2 topic list and rviz. No other transport has an equivalent, so this stays on the subclass rather than becoming a base-class capability of one.

from_rtps classmethod

from_rtps(node_name: str, cmd_vel_topic: str, **kwargs: Any) -> RtpsRobot

Construct an RTPS robot from ROS 2 topic wiring.

Keyword-style alternate constructor mirroring :meth:RosBridgedRobot.from_ros. The top-level Robot is a factory function (not a class), so the alternate constructor lives here where it is type-safe and discoverable.

strands_robots.drivers.ros.ackermann_robot.AckermannRosRobot

AckermannRosRobot(node_name: str, servo_topic: str, scan_topic: str | None = None, *, servo_type: str = _SERVO_TYPE, scan_type: str | None = None, wheelbase_m: float = 0.164, max_speed: float = 1.5, max_steering_rad: float = 0.5236, max_duration: float = 10.0, publish_rate: float = 20.0, init_services: list[dict[str, Any]] | None = None)

An Ackermann-steering ROS 2 car exposed as a strands-controllable robot.

The bridge owns no ROS 2 state; every method forwards to :func:~strands_robots.ros.ros_action. Constructing it never needs a ROS 2 environment - errors surface as structured results when a method actually runs.

Parameters:

Name	Type	Description	Default
node_name	str	Identifier used to name this robot's agent tools (drive_<node_name> etc.); it does not need to match a ROS 2 node name.	required
servo_topic	str	Topic the vehicle's servo stack subscribes to (DeepRacer: /webserver_pkg/manual_drive).	required
scan_topic	str | None	Optional laser-scan topic. Read by :meth:get_scan.	None
servo_type	str	Interface type of servo_topic. Defaults to the DeepRacer ServoCtrlMsg (normalized angle/throttle).	_SERVO_TYPE
scan_type	str | None	Interface type of scan_topic. Optional - resolved from the live graph when omitted.	None
wheelbase_m	float	Front-to-rear axle distance for the bicycle model.	0.164
max_speed	float	Linear speed (m/s) mapped to full throttle; commands are clamped to this magnitude.	1.5
max_steering_rad	float	Steering angle mapped to full servo deflection.	0.5236
max_duration	float	Longest single :meth:drive hold accepted; longer requests are rejected loudly rather than silently truncated.	10.0
publish_rate	float	Command publish rate (Hz) for held :meth:drive calls.	20.0
init_services	list[dict[str, Any]] | None	Ordered service calls ({"service", "type", "fields"} dicts) that put the vehicle into a commandable state. Run once, automatically, before the first :meth:drive. The DeepRacer manual-mode handshake in :meth:from_deepracer is the reference use.	None

There is deliberately no get_pose: the stock platform publishes no odometry.

tools property

tools: list[AgentTool]

This robot's capabilities as named strands agent tools.

Tools are bound to this instance and suffixed with node_name so multiple robots coexist in one Agent(tools=[...]) call. The drive tool's description states the Ackermann kinematic limit (minimum turning radius) so the agent can plan paths the platform can follow.

Every tool that carries a command (drive, stop) is declared @tool(context=True) and forwards the injected context into the operator gate, so it prompts rather than failing closed. get_scan takes no context because a read is never gated.

drive

drive(linear: float = 0.0, angular: float = 0.0, duration: float | None = None, count: int = 1, tool_context: ToolContext | None = None) -> dict[str, Any]

Publish a velocity command, converted through the bicycle model.

Same contract as RosBridgedRobot.drive: linear (m/s) and angular (rad/s), an optional duration hold (publishes round(duration * publish_rate) messages, takes precedence over count). Validation runs before any side effect, in order: inputs must be finite, duration (when given) must be a positive finite number within max_duration, the pair must name a motion the steering geometry can execute - a yaw asked for at rest is refused, because this platform cannot turn in place and the conversion would otherwise publish it as the same zero servo pair a stop uses - and only then does the vehicle's init_services handshake run (automatically, before the first command; a failed handshake aborts the drive). Every timed or multi-message non-zero command is followed by a single zero servo message - even if the main publish failed - so it can never leave the car with a live throttle latched, and when that halt is the call that fails the result says so instead of reporting the drive's success. A bare single-shot command (no duration, count=1) latches like a raw servo command until :meth:stop.

tool_context is the operator context forwarded to the command gate, which prompts for approval on a safety-critical surface such as this vehicle's servo topic and mode services. The drive_<node_name> agent tool passes the one the framework injects; a programmatic call has none, and the gate then refuses unless the surface is pre-approved via STRANDS_ROS2_COMMAND_ALLOW / BYPASS_TOOL_CONSENT.

enable

enable(tool_context: ToolContext | None = None) -> dict[str, Any]

Run the init_services handshake once; idempotent on success.

Stops at the first failing call and returns its structured error without latching, so a later attempt retries from the start.

tool_context is the operator context forwarded to the gate: the services that arm a vehicle are gated command surfaces, so a handshake that forwards nothing is refused rather than prompted (see :meth:drive).

from_deepracer classmethod

from_deepracer(node_name: str, **overrides: Any) -> AckermannRosRobot

Construct a bridge wired for the stock AWS DeepRacer software stack.

Servo commands go to the webserver package's manual-drive topic, the RPLIDAR scan topic is preconfigured, and init_services carries the two-step manual-mode handshake (vehicle_state state=1, then enable_state is_active=True) the car requires before it acts on servo messages. Any keyword can be overridden for a modified car.

get_scan

get_scan(timeout: float = 5.0) -> dict[str, Any]

Read one sample from the laser-scan topic (error when unconfigured).

Grades timeout at this seam, on the domain every bridge's read shares, so the refusal names the verb the caller invoked rather than the transport's own echo.

stop

stop(tool_context: ToolContext | None = None) -> dict[str, Any]

Publish a single zero servo command; it does not require :meth:enable.

The halt reaches the servo topic through the same publish as :meth:drive, so it passes the same operator gate - which is why tool_context is forwarded here too (see :meth:drive).

strands_robots.hardware_ros_bridge.HardwareRosBridge

HardwareRosBridge(robot: Robot | None = None, *, domain_id: int = 0, node_name: str | None = None, qos_depth: int = 10, enable_commands: bool = True, command_robot_name: str | None = None, spin_period: float = 0.02, joint_limits: dict[str, tuple[float, float]] | None = None)

Bases: RosTelemetryBridge

Full-duplex ROS 2 bridge for a real robot (node name strands_hardware).

Telemetry (outbound) is byte-identical to its simulation sibling :class:~strands_robots.simulation.ros_bridge.SimRosBridge - see :class:~strands_robots.ros_telemetry.RosTelemetryBridge for the publish API. On top of that, when constructed with a bound robot and enable_commands=True (the default for a hardware robot), this bridge also subscribes to /<robot>/joint_command and forwards inbound messages to robot.send_action over a background spin thread.

Parameters:

Name	Type	Description	Default
robot	Robot | None	The hardware Robot to drive on inbound commands. When None (the pure-publisher construction), no command surface is created and the bridge behaves exactly like the base telemetry bridge - this preserves the sim-symmetry contract and lets callers that only publish (the per-step control-loop path) opt out of the inbound half entirely.	None
domain_id	int	ROS 2 domain (ROS_DOMAIN_ID) to publish/subscribe on. Only an int in [0, 232] names a domain: RTPS derives its discovery ports from it, and 233 lands past the end of the port space.	0
node_name	str | None	Internal rclpy node name (defaults to strands_hardware).	None
qos_depth	int	Depth of the publishers'/subscription's KEEP_LAST history. Only a positive int up to :data:~strands_robots.ros_telemetry.MAX_QOS_HISTORY_DEPTH names a depth the transport can build an endpoint with. A publisher is built on the first publish rather than here, so an unusable depth would otherwise be reported mid-run, from inside rclpy and naming no parameter.	10
enable_commands	bool	When True (default) and a robot is bound, subscribe to /<robot>/joint_command and drive the arm. Set False for a read-only (telemetry-only) bridge. Only a boolean names a posture: the value is checked, not read by truthiness, so "false" cannot select the surface it asks to close.	True
command_robot_name	str | None	Topic namespace for the inbound command topic. Defaults to the bound robot's name (matching the namespace this bridge publishes joint_states under), so a controller can echo our own joint names straight back to drive the arm. Only a string names a topic segment, so only a string (or None for the default) is accepted: this is the one caller-supplied name rendered into a topic, and a non-string one reached the sanitiser's re.sub - raising TypeError naming no parameter when truthy, and, when falsy, being filtered by the default-selecting or so the bridge read commands under the robot's own name instead.	None
spin_period	float	Seconds between spin_once calls on the command thread. Only a positive finite number paces a loop. The value is handed straight to Event.wait on the backoff path, where 0, a negative and nan all return immediately - turning the thread into a busy-spin with no bound - and inf raises OverflowError out of it, killing the loop while the bridge reports a successful construction.	0.02
joint_limits	dict[str, tuple[float, float]] | None	Optional {"<motor>.pos": (min, max)} clamp ranges, keyed by the joint name as it arrives in joint_command - the same <motor>.pos names this bridge publishes in joint_states, so a controller can echo them straight back. A key that names no commanded joint constrains nothing. Each bound must be a finite number - a non-finite one declares a range that admits nothing, so the bridge refuses it at construction rather than dropping every inbound command for that joint mid-run. When set, an inbound joint_command whose ANY commanded joint is outside its declared range is rejected whole (no partial application), so a single out-of-range joint can never drive part of the arm.	None

Raises:

Type

Description

ValueError

If enable_commands is not a boolean, if domain_id is outside [0, 232], if qos_depth is not a positive int the transport can carry, if spin_period is not a positive finite number, if command_robot_name is neither a string nor None, or if joint_limits is not a {"<motor>.pos": (min, max)} mapping of finite numeric pairs with min <= max. Every one of them is answered before the base constructor writes the process-wide ROS_DOMAIN_ID, initializes the rclpy context and creates the node, so a refused bridge leaves the environment as it found it and leaks neither.

shutdown

shutdown() -> None

Stop the command thread, then destroy the node (base). Idempotent.

strands_robots.hardware_rtps_bridge.HardwareRtpsBridge

HardwareRtpsBridge(robot: Robot | None = None, *, domain_id: int = 0, enable_commands: bool = True, command_robot_name: str | None = None, poll_period: float = 0.02, joint_limits: dict[str, tuple[float, float]] | None = None, dds_security_config: dict[str, str] | None = None)

Bases: RosTelemetryBase

Full-duplex hardware ROS 2 bridge over pure RTPS (cyclonedds, no rclpy).

The rclpy-free sibling of :class:~strands_robots.hardware_ros_bridge.HardwareRosBridge. Both derive from :class:~strands_robots.ros_telemetry.RosTelemetryBase, so they share the topic names and the joint_command -> send_action contract and are wire-compatible by construction; they differ only in transport (cyclonedds RTPS vs rclpy) and in type coverage (bounded by the local IDL bundle).

Parameters:

Name	Type	Description	Default
robot	Robot | None	The hardware Robot to drive on inbound commands. When None, no command surface is created (telemetry-only), mirroring the rclpy bridge's pure-publisher mode.	None
domain_id	int	ROS 2 / DDS domain id to publish/subscribe on. Only an int in [0, 232] names a domain: RTPS derives its discovery ports from it, and 233 lands past the end of the port space.	0
enable_commands	bool	When True (default) and a robot is bound, subscribe to /<robot>/joint_command and drive the arm. Only a boolean names a posture: the value is checked, not read by truthiness, so "false" cannot select the surface it asks to close.	True
command_robot_name	str | None	Topic namespace for the command topic; defaults to the bound robot's name (the namespace we publish joint_states under). Only a string names a topic segment, so only a string (or None for the default) is accepted: this is the one caller-supplied name rendered into a topic, and a non-string one reached the sanitiser's re.sub - raising TypeError naming no parameter when truthy, and, when falsy, being filtered by the default-selecting or so the bridge read commands under the robot's own name instead.	None
poll_period	float	Seconds between inbound command reads on the poll thread. Only a positive finite number paces a loop. It is the sole pacing of _poll_loop, handed to Event.wait, where 0, a negative and nan all return immediately - turning the thread into a busy-spin with no bound - and inf raises OverflowError out of it, killing the loop while the bridge reports a successful construction.	0.02
joint_limits	dict[str, tuple[float, float]] | None	Optional {"<motor>.pos": (min, max)} clamp ranges, keyed by the joint name as it arrives in joint_command - the same <motor>.pos names this bridge publishes in joint_states, so a controller can echo them straight back. A key that names no commanded joint constrains nothing. Each bound must be a finite number - a non-finite one declares a range that admits nothing, so the bridge refuses it at construction rather than dropping every inbound command for that joint mid-run. When set, an inbound joint_command whose ANY commanded joint falls outside its declared range is rejected whole (no partial application), so a single out-of-range joint can never drive part of the arm.	None
dds_security_config	dict[str, str] | None	Optional DDS Security credentials. Required keys (identity_ca, certificate, private_key, governance, permissions; permissions_ca optional) wire the participant's DDS Security plugins so the whole graph is authenticated and access-controlled. Each value must be a non-empty string - a path or a file: / data: URI - and a supplied key that is not is refused before any participant exists, because a credential this participant would drop or stringify is not one it can present. When enable_commands is in effect this (or the STRANDS_ROS2_BRIDGE_I_KNOW_THIS_IS_INSECURE=1 opt-out) is REQUIRED - the bridge refuses to expose an arm-driving command surface on an unsecured DDS graph.	None

Raises:

Type	Description
ImportError	If cyclonedds (the [ros2] extra) is not installed.
ValueError	If enable_commands is not a boolean, domain_id is outside [0, 232], poll_period is not a positive finite number, or command_robot_name is neither a string nor None (all four checked before the cyclonedds probe, so the same caller mistake reports identically on an install without the extra), if joint_limits / dds_security_config is malformed, or if commands are enabled with neither a security config nor the explicit insecure opt-out.

publish_image

publish_image(robot: str, camera: str, image: ndarray) -> None

Publish one RGB Image on /<robot>/<camera>/image_raw.

publish_joint_states

publish_joint_states(robot: str, names: list[str], positions: list[float]) -> None

Publish one JointState for robot on /<robot>/joint_states.

Signature matches RosTelemetryBridge.publish_joint_states so the hardware Robot telemetry path is transport-agnostic - including the writer being resolved per topic. The writers are lazy, so a bridge only ever advertises the topics it was actually asked to publish on.

A names/positions pair of differing length is dropped whole with a warning rather than published misaligned - see :meth:RosTelemetryBase._joint_state_arrays_error.

shutdown

shutdown() -> None

Stop the poll thread and drop DDS entities. Idempotent.

Safety audit

strands_robots.audit.log_safety_event

log_safety_event(event_type: str, peer_id: str, payload: dict[str, Any]) -> None

Append a single safety event to the audit log.

Parameters:

Name	Type	Description	Default
event_type	str	Short, lowercase event identifier (e.g. "emergency_stop").	required
peer_id	str	The mesh peer that originated the event.	required
payload	dict[str, Any]	Event-specific fields. Must be JSON-serialisable.	required

Raises:

Type	Description
represent is not dropped either	the record is written with the

Device connect

Device Connect integration for strands-robots.

Provides DeviceDriver adapters that wrap Robot and Simulation instances, exposing them to Device Connect's device registry, RPC routing, and event system.

Usage

from strands_robots.device_connect import init_device_connect

robot = Robot("so100") runtime = await init_device_connect(robot, peer_id="so100-lab-1")

Now discoverable via Device Connect tools:

discover_devices(device_type="strands_robot")

invoke_device("so100-lab-1", "execute", {"instruction": "pick up cube"})

Module-load discipline

Every symbol this package exports is behind __getattr__ (:pep:562). The package imports on a stock pip install strands-robots -- with no extras -- and only reaches device_connect_edge when a caller uses a name that needs it. The four Device Connect drivers, DeviceRuntime and the two init_device_connect entry points all sit behind that gate.

This is required, not a stylistic choice. The sibling module strands_robots.device_connect.reachy_transport is stdlib-only and is imported by the native Reachy driver (strands_robots.drivers.reachy, landing in #2762). Importing that leaf executes this __init__, so a package init that eagerly imports device_connect_edge raises ModuleNotFoundError inside the native driver's first daemon touch on any install without [device-connect] -- escaping the driver's own no-raise refusal contract and breaking three :mod:AGENTS.md conventions at once ("Return error dicts, never raise", require_optional() for optional deps, the module-load discipline the driver's docstring makes explicit).

The public names are unchanged: from strands_robots.device_connect import init_device_connect still works, and only fails when the caller reaches for a name whose implementation genuinely needs device_connect_edge. Static tools (mypy, IDE autocomplete) see the names through the TYPE_CHECKING guard.

RobotDeviceDriver

RobotDeviceDriver(robot)

Bases: DeviceDriver

Device Connect device driver wrapping a strands-robots Robot instance.

identity property

identity: DeviceIdentity

Static Device Connect identity for the wrapped robot.

Returns a :class:~device_connect_edge.types.DeviceIdentity reporting device_type="strands_robot", the strands-robots manufacturer, and the robot's tool_name_str as the model (falling back to "robot").

status property

status: DeviceStatus

Live availability of the wrapped robot.

Returns a :class:~device_connect_edge.types.DeviceStatus that is "busy" (busy_score 1.0) while a task is running and "idle" (busy_score 0.0) otherwise, derived from the robot's task state.

connect async

connect() -> None

No-op - the Robot manages its own hardware connection.

disconnect async

disconnect() -> None

No-op - the Robot manages its own hardware shutdown.

emergencyStop async

emergencyStop(reason: str = '')

Emitted when this device triggers an emergency stop.

Parameters:

Name	Type	Description	Default
reason	str	Why the emergency stop was triggered	''

execute async

execute(instruction: str, policy_provider: str = 'mock', duration: float = 30.0, policy_port: int = 0) -> dict[str, Any]

Execute a VLA task instruction on the robot.

Parameters:

Name	Type	Description	Default
instruction	str	Natural language task instruction	required
policy_provider	str	Policy backend (lerobot_local, mock, remote, ...)	'mock'
duration	float	Maximum task duration in seconds	30.0
policy_port	int	Policy server port (0 for default)	0

getFeatures async

getFeatures() -> dict[str, Any]

Get robot observation and action features.

getState async

getState() -> dict[str, Any]

Get current robot state (joints, task info).

Returns joint positions and task state if a task is running.

The joints are read through the shared motor-bus lock, so this RPC waits its turn behind an in-flight rollout, teleop write or mesh probe instead of colliding with it, and reports the joints even when a camera on the same driver is failing.

The device those joints are read from is resolved by :func:~strands_robots.bus_access.joint_read_source, so a native driver that owns its bus directly answers this RPC as well as a lerobot wrapper does.

getStatus async

getStatus() -> dict[str, Any]

Get current task execution status.

onEmergencyStop async

onEmergencyStop(device_id: str, event_name: str, payload: dict[str, Any]) -> None

React to emergencyStop from an authorized safety controller.

Security hardening: only act on emergency-stop events whose source is in the emergency-stop allowlist, so a spoofed event from an arbitrary device cannot interrupt operations.

The stop's own verdict is read rather than discarded. stop_task is written to report a stop that did not happen -- G1Driver returns status="error" with stopped=False precisely "so the caller cannot read 'success' while the payload's own running=True says the loop is still writing frames" -- and this handler was the caller that read neither field. Mesh.emergency_stop grades that same verdict for every peer it fans out to and logs one that did not stop at CRITICAL; a stop that arrives over Device Connect rather than over the mesh is the same operator request and gets the same accounting.

stateUpdate async

stateUpdate(task_status: str = '', instruction: str = '', step_count: int = 0)

Periodic state update.

Parameters:

Name	Type	Description	Default
task_status	str	Current task status	''
instruction	str	Current task instruction	''
step_count	int	Steps completed so far	0

stop async

stop() -> dict[str, Any]

Stop the currently running task.

streamStep async

streamStep(step: int, observation: dict[str, Any], action: dict[str, Any]) -> None

Emitted for each VLA inference step (high frequency).

Parameters:

Name	Type	Description	Default
step	int	Step number	required
observation	dict[str, Any]	Observation dict (joints only, no camera frames)	required
action	dict[str, Any]	Action dict	required

taskComplete async

taskComplete(instruction: str, steps: int, duration: float)

Emitted when a VLA task finishes.

Parameters:

Name	Type	Description	Default
instruction	str	The task instruction	required
steps	int	Total steps executed	required
duration	float	Total execution time in seconds	required

taskStarted async

taskStarted(instruction: str, policy_provider: str)

Emitted when a VLA task begins execution.

Parameters:

Name	Type	Description	Default
instruction	str	The task instruction	required
policy_provider	str	The policy backend used	required

SimulationDeviceDriver

SimulationDeviceDriver(sim)

Bases: DeviceDriver

Device Connect device driver wrapping a strands-robots Simulation instance.

identity property

identity: DeviceIdentity

Static Device Connect identity for the wrapped simulation.

Returns a :class:~device_connect_edge.types.DeviceIdentity reporting device_type="strands_sim", the strands-robots manufacturer, and the simulation's tool_name_str as the model (falling back to "simulation").

status property

status: DeviceStatus

Live availability of the wrapped simulation.

Returns a :class:~device_connect_edge.types.DeviceStatus that is "busy" (busy_score 1.0) when any robot in the world is running a policy and "idle" (busy_score 0.0) otherwise.

connect async

connect() -> None

No-op - the Simulation manages its own MuJoCo state.

disconnect async

disconnect() -> None

No-op - the Simulation manages its own cleanup.

emergencyStop async

emergencyStop(reason: str = '')

Emitted when this device triggers an emergency stop.

Parameters:

Name	Type	Description	Default
reason	str	Why the emergency stop was triggered	''

execute async

execute(instruction: str, policy_provider: str = 'mock', duration: float = 30.0, robot_name: str = '') -> dict[str, Any]

Execute a policy on a simulated robot.

Parameters:

Name	Type	Description	Default
instruction	str	Natural language task instruction	required
policy_provider	str	Policy backend (mock, lerobot_local, ...)	'mock'
duration	float	Maximum task duration in seconds	30.0
robot_name	str	Target robot name (empty = first robot)	''

getFeatures async

getFeatures() -> dict[str, Any]

Get simulation features (joints, actuators, cameras).

getStatus async

getStatus() -> dict[str, Any]

Get simulation state and running policies.

observationUpdate async

observationUpdate(robot_name: str = '', sim_time: float = 0.0, step_count: int = 0, joints: dict[str, float] | None = None) -> None

Periodic per-robot observation with joint positions.

Parameters:

Name	Type	Description	Default
robot_name	str	Name of the robot	''
sim_time	float	Current simulation time	0.0
step_count	int	Total physics steps	0
joints	dict[str, float] | None	Dict of joint name -> position (radians)	None

onEmergencyStop async

onEmergencyStop(device_id: str, event_name: str, payload: dict[str, Any]) -> None

React to emergencyStop from an authorized safety controller.

Halts EVERY motion source the simulation has, not only its policies. A Simulation mixes in :class:~strands_robots.teleop_mixin.TeleopMixin, so a leader arm can be driving it from a thread the policy_running flag says nothing about; stopping policies alone left that loop polling get_action() and applying the result through send_action after an operator's emergency stop, with the handler reporting the halt.

Both stops are attempted even if one fails, and a source that did not stop is logged at CRITICAL naming it -- the accounting reachy_mini_driver.onEmergencyStop gives its two stop actions, and that robot_driver.onEmergencyStop gives the stop verdict it reads. A stop that arrives here rather than over the mesh is the same operator request, so it gets the same accounting.

Security hardening: only act on emergency-stop events whose source is in the emergency-stop allowlist, so a spoofed event from an arbitrary device cannot interrupt operations.

policyComplete async

policyComplete(robot_name: str, instruction: str, steps: int)

Emitted when a policy finishes.

Parameters:

Name	Type	Description	Default
robot_name	str	The simulated robot	required
instruction	str	The task instruction	required
steps	int	Total steps executed	required

policyStarted async

policyStarted(robot_name: str, instruction: str, policy_provider: str)

Emitted when a policy begins execution.

Parameters:

Name	Type	Description	Default
robot_name	str	The simulated robot running the policy	required
instruction	str	The task instruction	required
policy_provider	str	The policy backend used	required

reset async

reset() -> dict[str, Any]

Reset simulation to initial state.

stateUpdate async

stateUpdate(sim_time: float = 0.0, step_count: int = 0, running_policies: dict[str, Any] | None = None) -> None

Periodic simulation state update.

Parameters:

Name	Type	Description	Default
sim_time	float	Current simulation time	0.0
step_count	int	Total physics steps	0
running_policies	dict[str, Any] | None	Dict of running policy info per robot	None

step async

step(n_steps: int = 1) -> dict[str, Any]

Step simulation physics forward.

Parameters:

Name	Type	Description	Default
n_steps	int	Number of physics steps to take	1

stop async

stop() -> dict[str, Any]

Stop every rollout in this simulation and report which ones halted.

Routes each robot through the simulation's own stop_policy, the verb that owns the question, and reads the verdict it returns. This used to lower policy_running itself and answer a fixed "All policies stopped", which made one sentence out of four different facts: a halted rollout, an idle simulation, a world already torn down, and -- because the loop had no guard -- a scene teardown racing it, which escaped as a RuntimeError past the RPC instead of an envelope. None of the four named a robot, so an operator could not tell what had been halted or cross-check it against list_policies_running.

The mesh fanout (:meth:~strands_robots.mesh.Mesh._dispatch, action stop) is the other remote way into this same simulation and it already answers this way: per-robot stop_policy, each answer graded through :func:~strands_robots.mesh.core._reports_failure_to_stop, the halted robots named. The rule has one owner and this is now its third reader rather than a surface that produced no answer to grade.

A simulation with nothing to halt answers affirmatively-empty rather than erroring, for the reason the mesh branch states in full: nothing to stop makes "did not stop" wrong rather than conservative, and a false negative on the safety path is what trains an operator to ignore the warning.

Returns:

Type	Description
dict[str, Any]	status="success" when no rollout refused, naming the halted ones
dict[str, Any]	in the text and listing them under stopped in the json
dict[str, Any]	block. status="error" when a rollout refused or the world
dict[str, Any]	mutated under the loop, with the refusals under not_stopped and
dict[str, Any]	whatever did halt still under stopped.

ReachyMiniDriver

ReachyMiniDriver(host: str = 'reachy-mini.local', prefix: str = 'reachy_mini', api_port: int = 8000)

Bases: DeviceDriver

Device Connect driver for Pollen Reachy Mini.

Auto-detects Wireless (Zenoh) vs Lite (WebSocket) via the daemon's wireless_version flag. REST API calls work the same for both.

Configure the driver for a Reachy Mini reachable at host.

Parameters:

Name	Type	Description	Default
host	str	Hostname or IP of the Reachy Mini daemon. Must name a host: a bare hostname or IP literal, since it is interpolated into the daemon URL (http://<host>:<api_port>) beside api_port. A URI delimiter there re-cuts that URL and the validated port becomes part of the path. reachy-mini.local and 192.168.1.42 are accepted; 127.0.0.1/foo is not.	'reachy-mini.local'
prefix	str	Zenoh key prefix used by the Wireless variant. Must be a /-joined sequence of mesh identifiers -- a Zenoh wildcard (* / **) would widen the command key to every Mini beneath the pattern. reachy_mini/robot_a is accepted; reachy_mini/* is not.	'reachy_mini'
api_port	int	TCP port the daemon serves its REST API and WebSocket on. Must name a port: an int in [1, 65535].	8000

Raises:

Type	Description
ValueError	If api_port cannot address a TCP port, if host cannot address the host half of the daemon URL (see :func:~strands_robots.utils.dial_host_error), or if prefix cannot address a single robot's key expressions (see :func:_key_prefix_error).

identity property

identity: DeviceIdentity

Static Device Connect identity for the Reachy Mini head.

Returns a :class:~device_connect_edge.types.DeviceIdentity reporting device_type="reachy_mini", the Pollen Robotics manufacturer, and the configured host in the model string.

status property

status: DeviceStatus

Availability of the Reachy Mini; always reports "idle".

The head has no long-running task state, so it advertises itself as available for commands at all times.

antennas async

antennas(left: float = 0, right: float = 0) -> dict[str, Any]

Set antenna angles.

Parameters:

Name	Type	Description	Default
left	float	Left antenna angle in degrees	0
right	float	Right antenna angle in degrees. Both must be finite numbers of either sign.	0

Returns:

Type	Description
dict[str, Any]	{"status": "success", ...}, or a ``{"status": "error",
dict[str, Any]	"reason": ...}`` dict naming the argument that was refused.

body async

body(yaw: float = 0) -> dict[str, Any]

Set body yaw angle.

Parameters:

Name	Type	Description	Default
yaw	float	Body yaw angle in degrees. Must be a finite number of either sign and inside the shared travel envelope.	0

Returns:

Type	Description
dict[str, Any]	{"status": "success", ...}, or a ``{"status": "error",
dict[str, Any]	"reason": ...}`` dict naming the argument that was refused.

connect async

connect() -> None

Connect to the Reachy Mini, auto-detecting Wireless vs Lite.

disableMotors async

disableMotors(motor_ids: str = '') -> dict[str, Any]

Disable motors (torque off).

Parameters:

Name	Type	Description	Default
motor_ids	str	Comma-separated motor IDs (empty = all). A non-empty selector that names no motor is refused.	''

disconnect async

disconnect() -> None

Tear down the hardware link and drop the handle to it.

The handle is dropped before the stop is awaited because :meth:_send_cmd reads it as its "is the link connected?" test, and a link left in _hw keeps that guard unreachable. A movement RPC issued after a disconnect is then not refused: on the Wireless variant :meth:ZenohLink.send_cmd publishes to <prefix>/command and the RPC reports success, actuating the head after the driver was told to let go of it; on the Lite variant it reaches a closed socket instead.

Clearing first also holds when the stop itself fails - the link is being torn down either way, so the driver must stop treating it as connected.

emergencyStop async

emergencyStop(reason: str = '') -> None

Emitted when this device triggers an emergency stop.

Parameters:

Name	Type	Description	Default
reason	str	Why the emergency stop was triggered	''

enableMotors async

enableMotors(motor_ids: str = '') -> dict[str, Any]

Enable motors (torque on).

Parameters:

Name	Type	Description	Default
motor_ids	str	Comma-separated motor IDs (empty = all). A non-empty selector that names no motor is refused.	''

getDaemonStatus async

getDaemonStatus() -> dict[str, Any]

Report the daemon's status payload under this driver's own verdict.

The payload's keys are merged into the envelope so a caller reads motors_on / freq at the top level, and status is re-asserted afterwards. Merged last, a daemon reply carrying a status field of its own replaced this envelope's verdict: a healthy call answered status="idle", and a daemon reporting its own fault answered status="error" for an RPC that reached it and succeeded. The presence path resolves the same collision the same way, by spreading the foreign mapping first so the locally decided keys win - see :meth:strands_robots.mesh.sensors.SensorLoopsMixin._stamp_local_keys.

A daemon that was not reached is reported rather than merged. :func:~strands_robots.drivers.reachy_transport.api answers every HTTP and connection failure with {"error": ...} instead of raising, so the unreachable daemon this RPC exists to detect came back as status="success" with the reason merged in beside it. The native driver reads this same endpoint and refuses that shape - see :meth:strands_robots.drivers.reachy.ReachyDriver.connect_eagerly. The error envelope is used rather than the RuntimeError :meth:_stop_motion_impl raises, because that one guards a stop, where a caller acting on a false success stops nothing; this one answers a question, and its callers already branch on status.

A body that decodes to something other than a JSON object is reported for the same reason: spreading it raises TypeError out of a method whose whole contract is the envelope.

Returns:

Type	Description
dict[str, Any]	{**payload, "status": "success"} when the daemon answered with
dict[str, Any]	a JSON object, otherwise {"status": "error", "reason": ...}
dict[str, Any]	naming the daemon that was not reached or the body that could not
dict[str, Any]	be merged.

getImu async

getImu() -> dict[str, Any]

Get IMU data (accelerometer, gyroscope, quaternion, temperature).

getJoints async

getJoints() -> dict[str, Any]

Get current joint positions (head + antennas).

happy async

happy() -> dict[str, Any]

Happy antenna wiggle expression.

listMoves async

listMoves(library: str = 'emotions') -> dict[str, Any]

List available recorded moves.

Parameters:

Name	Type	Description	Default
library	str	Which library, one of 'emotions' or 'dances'.	'emotions'

Returns:

Type	Description
dict[str, Any]	A success envelope whose moves is the daemon's array of names,
dict[str, Any]	or an error envelope naming what refused - the library, or a
dict[str, Any]	daemon that was not reached, in which case the catalogue is unknown
dict[str, Any]	rather than empty.

look async

look(pitch: float = 0, roll: float = 0, yaw: float = 0, x: float = 0, y: float = 0, z: float = 0) -> dict[str, Any]

Set head pose instantly.

Parameters:

Name	Type	Description	Default
pitch	float	Pitch angle in degrees	0
roll	float	Roll angle in degrees	0
yaw	float	Yaw angle in degrees	0
x	float	X offset in mm	0
y	float	Y offset in mm	0
z	float	Z offset in mm. Every value must be a finite number of either sign, and pitch / roll / yaw must be inside the shared travel envelope. The millimetre offsets carry no envelope bound and are the daemon's to enforce.	0

Returns:

Type	Description
dict[str, Any]	{"status": "success", ...}, or a ``{"status": "error",
dict[str, Any]	"reason": ...}`` dict naming the first argument that cannot be
dict[str, Any]	carried to the robot.

nod async

nod() -> dict[str, Any]

Nod the head (yes gesture).

onEmergencyStop async

onEmergencyStop(device_id: str, event_name: str, payload: dict[str, Any]) -> None

React to emergencyStop from an authorized safety controller.

Security hardening: only act on emergency-stop events whose source is in the emergency-stop allowlist, so a spoofed event from an arbitrary device cannot interrupt operations.

playMove async

playMove(move_name: str, library: str = 'emotions') -> dict[str, Any]

Play a recorded move from the HuggingFace library.

Parameters:

Name	Type	Description	Default
move_name	str	Name of the move to play	required
library	str	Which library, one of 'emotions' or 'dances'.	'emotions'

Returns:

Type	Description
dict[str, Any]	A success envelope naming the move, or an error envelope naming the
dict[str, Any]	gate that refused - the caller, the library, the move_name,
dict[str, Any]	or a daemon that was not reached, in which case no move was played.

shake async

shake() -> dict[str, Any]

Shake the head (no gesture).

sleep async

sleep() -> dict[str, Any]

Put robot to sleep (play sleep animation + disable motors).

Returns:

Type	Description
dict[str, Any]	A success envelope, or an error envelope naming the caller that was
dict[str, Any]	not authorized or the daemon that was not reached - in which case
dict[str, Any]	the robot is still awake.

stopMotion async

stopMotion() -> dict[str, Any]

Stop all current motion.

wakeUp async

wakeUp() -> dict[str, Any]

Wake up the robot (enable motors + play wake animation).

Returns:

Type	Description
dict[str, Any]	A success envelope, or an error envelope naming the caller that was
dict[str, Any]	not authorized or the daemon that was not reached - in which case
dict[str, Any]	the motors were not enabled.

init_device_connect async

init_device_connect(robot, peer_id: str | None = None, peer_type: str = 'robot', messaging_url: str | None = None, messaging_backend: str | None = None, tenant: str = 'default', allow_insecure: bool | None = None) -> DeviceRuntime

Initialize Device Connect for a Robot or Simulation.

Drop-in replacement for init_mesh(). Creates a DeviceDriver adapter and starts a DeviceRuntime in the background.

When messaging_backend="zenoh" and messaging_url is None, the runtime enters D2D mode - devices discover each other directly via Zenoh multicast scouting on the LAN. No broker, no Docker, no env vars.

Parameters:

Name	Type	Description	Default
robot		A Robot or Simulation instance to wrap.	required
peer_id	str | None	Device ID for registration (auto-generated if None).	None
peer_type	str	"robot" or "sim" - selects the appropriate driver.	'robot'
messaging_url	str | None	Explicit messaging URL (overrides env vars).	None
messaging_backend	str | None	Messaging backend - "zenoh" or "nats". None = auto-detect from MESSAGING_BACKEND env var (default "zenoh").	None
tenant	str	Device Connect tenant namespace.	'default'
allow_insecure	bool | None	Allow insecure (unencrypted, unauthenticated) transport. Must be a boolean or None; a string spelling such as "false" is refused here rather than read, because the string vocabulary belongs to DEVICE_CONNECT_ALLOW_INSECURE and a non-empty string is truthy as an argument. None = auto-detect: respects the DEVICE_CONNECT_ALLOW_INSECURE env var if set, otherwise defaults to False (secure). Insecure transport must be explicitly opted into; a prominent warning is logged whenever it is active.	None

Returns:

Type	Description
DeviceRuntime	The running DeviceRuntime instance.

init_device_connect_sync

init_device_connect_sync(robot, peer_id: str | None = None, peer_type: str = 'robot', messaging_url: str | None = None, messaging_backend: str | None = None, tenant: str = 'default', allow_insecure: bool | None = None) -> DeviceRuntime

Non-blocking sync wrapper around init_device_connect().

Starts the DeviceRuntime on a dedicated daemon thread so the caller returns immediately - matching the Zenoh mesh init_mesh() pattern. The runtime stays alive as long as the process (daemon thread).

That outliving is the successful outcome, and it is the only one with something to serve. A bring-up that produced no runtime -- because it raised, or because it finished without returning one -- left the loop and the thread unreachable from the caller: the pair is adopted onto the runtime below, on the success path only. Parking that thread in run_forever anyway left an idle loop -- and the epoll and self-pipe descriptors it holds -- alive for the life of the process, once per failed attempt, so a caller retrying an unreachable broker accumulated one parked thread and three descriptors per try with no way to reach any of them. The bring-up thread therefore closes its own loop and returns whenever the holder is empty, and this call waits :data:~strands_robots.device_connect._LOOP_JOIN_TIMEOUT_S for it, so the failure the caller is handed also means the machinery is gone.

Same parameters as :func:init_device_connect.

Raises:

Type	Description
Exception	Whatever :func:init_device_connect raised on the background thread, re-raised here so a failed bring-up reaches the caller rather than being confined to a thread it cannot see.
TimeoutError	If the bring-up does not finish within the wrapper's budget. The runtime is not returned in that case, so the caller is never handed None in place of a DeviceRuntime.
RuntimeError	If the bring-up finished without returning a runtime. Nothing came up, so this is a failed bring-up like any other, and it is reported as one rather than returned as an empty success.

Edit page