Robot Data and Event Streams

Camera frames, joint states, commands, and status updates arrive at different times and in different orders. retriever_typing.data gives every value an event time and a source so a stream stays deterministic under replay. The types are immutable dataclasses in registry category data.

from retriever_typing.data import Event, EventBuffer, StreamId

evt = Event(
    stream_id=StreamId("camera"),
    event_time_ns=1_000,     # when the robot observed it
    ingest_time_ns=1_050,    # when the process received it
    seq=0,
    value=0.2,
    type_name="float",
)

Two clocks are always explicit: event_time_ns (what actually happened) and ingest_time_ns (when the callback fired). Event.ordering_key() sorts by (event_time, ingest_time, stream_id, seq), so a merge is reproducible even when two streams share a timestamp.

Deterministic ordering and windows

merge_sorted interleaves buffers by event time; latest and window_agg read from the merged stream by recorded time, not arrival order.

from retriever_typing.data import EventBuffer, merge_sorted, latest, window_agg, WindowPolicy

merged = merge_sorted(camera, joint)          # ordered by ordering_key()
latest(merged)                                # value with the largest event_time
window_agg(merged, now_ns=2_000, policy=WindowPolicy(duration_ns=1_000, agg="mean"))

pixi run demo-robotics-data-eventstream

Note the camera event at event=2000 sorts after the joint event at the same time because its ingest is later — the ordering key is total and stable:

Deterministic merged order:
  stream=camera event=1000 ingest=1050 seq=0 value=0.2
  stream=joint event=1500 ingest=1510 seq=0 value=1.0
  stream=joint event=2000 ingest=2020 seq=1 value=1.4
  stream=camera event=2000 ingest=2100 seq=1 value=0.5

Processing-time profile:
  latest = 0.5
  window mean at now=2_000ns over 1_000ns = 0.7749999999999999

Event-time joins carry lineage

The join operators align two streams by recorded time and stamp each result with a LineageRef naming the source events it came from — so an exported row can still explain where it came from.

from retriever_typing.data import align_exact, align_latest_before, align_window

align_exact(left, right)                                   # same event_time only
align_latest_before(left, right, max_delta_ns=50)          # most recent left <= right, within delta
align_window(left, right, window_ns=20)                    # any left within +/- window of right

pixi run demo-robotics-data-join

Each joined value keeps lineage=[A:<t>, B:<t>], the exact source timestamps that produced it:

Exact join:
  t=180 value=(2, 10) lineage=['A:180', 'B:180']
Latest-before join (max_delta=50):
  t=180 value=(2, 10) lineage=['A:180', 'B:180']
  t=200 value=(2, 20) lineage=['A:180', 'B:200']
Window join (+/-20):
  t=180 value=(2, 10) lineage=['A:180', 'B:180']
  t=200 value=(2, 20) lineage=['A:180', 'B:200']

What each piece is for

Type / helper	Use it for	Check
Event	One timestamped value with source metadata.	Event time, ingest time, source, and lineage are explicit.
EventBuffer	An immutable event history.	sorted(), within(start, end), and latest_event() are deterministic.
StreamId	A stable stream identity.	Non-empty; used in ordering and manifests.
align_*	Aligning two streams by event time.	Uses recorded timestamps, not callback timing.
WindowPolicy	Aggregating over a temporal slice.	Duration and agg (first/last/max/min/mean) are explicit.
LineageRef	Tracking derivation.	Joined/exported values still name their sources.

Core Retriever owns the runtime event-buffer mechanics and scheduler. GoldenRetriever adds this applied data profile: enough timing and provenance on a camera observation, command, or state estimate to debug it after the fact. It feeds directly into the LeRobot export.

Source: src/retriever_typing/data/ (v1.py, buffer.py, join.py); demos examples/advanced/robotics_typing_standard/data_spec_eventstream_demo.py, multi_stream_join_demo.py.