Skip to content

Choose a Robot Payload

The catalog is three tiers behind one import path. Choose by the handoff you need to preserve, not by the ontology.

  • Standard spatial (retriever.types.spatial, re-exported by retriever_typing) — registry category spatial, @io-decorated, so they can be Flow ports directly.
  • Applied robot types (retriever_typing.robotics_types) — registry category general, plain dataclasses carried as values and resolved by name.
  • Applied action/status (retriever_typing.core_types) — category general, the lightweight exchange records with built-in Arrow round-trips.
from retriever_typing import get_type, get_type_info, list_types

get_type("WorldState")                 # -> retriever_typing.robotics_types.WorldState
get_type_info("SE3Pose").category      # -> 'spatial'  (schema_name 'spatial/SE3Pose')
get_type_info("WorldState").category   # -> 'general'  (kind 'payload')
len(list_types())                      # spans categories: spatial=10, general=27, data=13, perception=10

Start from the boundary you need to preserve, not the richest type available.

Boundary Start with Use when
Local spatial math Vector3, Quaternion, SE3Pose Frame, unit, and source are already local context.
Graph / process handoff PoseStamped, TwistStamped, WrenchStamped The value crosses a Flow, process, replay, or robot boundary.
Robot state RobotState, JointState A wrapper, monitor, policy, or sim needs current configuration.
Scene and memory WorldState, BeliefGraph Perception, memory, grounding, and planning share a scene.
Planner handoff TaskGoal, Skill, Plan, StructuredPlan A planner hands intent to a controller or monitor.
Motion and progress Trajectory, ExecutionStatus A controller reports timed motion, progress, or terminal state.
Lightweight exchange Action, Command, Status A smoke demo, Hub export, or dataset bridge needs compact records.

These are the @io payloads from retriever.types.spatial. Stamp a value with a Header before it crosses a Flow, process, dataset, or robot boundary.

Type Fields Notes
Vector3 x, y, z Position, direction, velocity, or force component.
Quaternion x, y, z, w norm(), is_unit(tol=1e-3).
SE3Pose position, orientation Object, end-effector, or robot pose.
Twist / Wrench linear, angular / force, torque Spatial velocity / force-torque.
Header stamp_ns, frame_id, source Boundary metadata; attach before handoff.
PoseStamped / TwistStamped / WrenchStamped header + pose/twist/wrench Stamped wrappers for cross-boundary values.
JointState names, positions, velocities, efforts is_aligned() checks index alignment.

Validators live beside them: validate_header, validate_quaternion, validate_pose_stamped, validate_joint_state. Use them at the edge where robot data changes meaning.

from retriever_typing import (
    Header, SE3Pose, Vector3, Quaternion, PoseStamped, JointState,
    validate_pose_stamped, validate_joint_state,
)

pose = PoseStamped(
    header=Header(stamp_ns=1_726_000_000_000_000_000, frame_id="map", source="sim"),
    pose=SE3Pose(position=Vector3(1.0, 2.0, 0.5), orientation=Quaternion(0.0, 0.0, 0.0, 1.0)),
)
joints = JointState(
    names=("j1", "j2", "j3", "j4", "j5", "j6"),
    positions=(0.0, -0.3, 0.8, 0.0, 0.6, 0.0),
    velocities=(0.0,) * 6, efforts=(0.0,) * 6,
)

validate_pose_stamped(pose)     # raises if stamp_ns <= 0, frame_id empty, or quaternion non-unit
validate_joint_state(joints)    # raises unless every array length matches len(names)
pixi run demo-robotics-typing-catalog
Registry lookup: PoseStamped=PoseStamped SE3Pose=SE3Pose
Robotics typing catalog v1 demo
  pose.frame=map pos=Vector3(x=1.0, y=2.0, z=0.5)
  twist.frame=base_link linear=Vector3(x=0.2, y=0.0, z=0.0)
  wrench.frame=wrist force=Vector3(x=0.0, y=0.0, z=3.5)
  joint_count=6 aligned=True

Scene, memory, and planner-handoff payloads. Plain dataclasses — pick the narrowest one that preserves the boundary.

Type Fields (defaults omitted) First use
WorldState object_poses: Dict[str, SE3Pose], robot_pose: SE3Pose, timestamp Shared scene state for perception, memory, planning.
RobotState position, orientation, objects_held, battery_level, last_error Robot wrapper / monitor / policy input.
BeliefGraph nodes: Set[str], edges: Dict[str, Set[str]] Symbolic scene memory and grounding.
Skill name, params, confidence, expected_duration One named capability with parameters.
Plan skills: List[Skill], confidence Ordered skills; total_duration sums known durations.
StructuredPlan steps: List[str], confidence, metadata Typed multi-step plan text.
TaskGoal high_level_description, affordances, success_criteria User intent + available objects/actions.
Trajectory poses: List[SE3Pose], timestamps Timed motion; duration from first/last stamp.
ExecutionStatus status, metadata, progress, error_message Progress in [0,1]; is_complete for terminal states.
from retriever_typing.robotics_types import WorldState, Skill, Plan, ExecutionStatus
from retriever_typing import SE3Pose, Vector3, Quaternion

pose = SE3Pose(position=Vector3(0.1, 0.2, 0.3), orientation=Quaternion(0.0, 0.0, 0.0, 1.0))
world = WorldState(object_poses={"cup": pose}, robot_pose=pose, timestamp=0.0)
plan = Plan(skills=[Skill(name="pick", params={"object": "cup"}, confidence=0.95)])
status = ExecutionStatus(status="IN_PROGRESS", metadata={}, progress=0.4)

assert [s.name for s in plan.skills] == ["pick"]
assert not status.is_complete            # only SUCCESS / FAILURE are terminal

Lightweight exchange records from retriever_typing.core_types, used by smoke demos, Hub exports, and dataset bridges. Action, Command, and Status carry their own Arrow round-trips.

Type Fields Notes
Action type, parameters: dict, timestamp, priority __post_init__ rejects non-dict params; to_arrow / from_arrow.
Command action: Action, robot_id, expected_duration, timeout action_type shortcut.
Status state, message, progress, timestamp, error_code state is one of pending/running/completed/failed/cancelled; is_success.
Timestamp seconds, nanoseconds now(), to_float().
ExecutionTimer start_time, expected_period, actual_period, iteration is_timing_violation at 50% tolerance.

Use direct imports inside example code — readers should not have to ask the registry what a payload means. Use registry lookup (get_type) only at package boundaries: manifests, Hub packs, replay readers, and dataset readers, which resolve a name without importing optional robot integrations.

from retriever_typing import PoseStamped, get_type
assert get_type("PoseStamped") is PoseStamped

Validate at the edge where robot data changes meaning, before it reaches a planner, controller, dataset, or another process.

  • Frames — keep frame_id non-empty when world, camera, robot, or object frames can be confused (validate_header).
  • Time — keep stamp_ns positive and monotonic per source so replay and event-time joins order correctly.
  • Rotations — check Quaternion.is_unit() before a pose crosses into planning or a dataset (validate_quaternion).
  • Arrays — confirm JointState.is_aligned() so names/positions/velocities/efforts share one length (validate_joint_state).
  • Progress — keep ExecutionStatus.progress in [0, 1] so monitors and UI surfaces agree.

Keep the base catalog small. Add covariance-bearing poses, richer frame graphs, or domain-specific schemas as separate Hub packs only once these payloads are not enough.

Detailed field reference for implementers

The base catalog above covers the common handoffs. retriever_typing.robotics_types also carries less-common applied payloads for specific examples:

  • ActionPlanactions: List[Action], total_duration; num_actions property.
  • TaskInstancegoal: TaskGoal, initial_observation, seed.
  • Observation / ObservationHistory — multi-sensor image bundles and a bounded deque.
  • ObjectVariable / ObjectSymbol — symbolic placeholders and grounded object IDs.
  • UnorderedObjectSet / ObjectDescriptionDict — object-id sets and id-to-description maps.

These stay source-only until an example needs them across a stable boundary.

Source: src/retriever_typing/robotics_types.py, src/retriever_typing/core_types.py, retriever.types.spatial (re-exported via src/retriever_typing/v1.py).