Streams

Typed pub/sub between modules, built on RxPY: what a stream is and which guide covers your use case

A stream is how one module sends data to another in DimOS. A module declares typed ports on its config: In[T] for data it consumes, Out[T] for data it produces (T is a message type, often something from dimos.msgs.*). When you compose modules into a blueprint with autoconnect(), streams are wired together automatically by matching (name, type) pairs across the union of modules - no manual plumbing required.

Under the hood, streams are built on RxPY (reactivex). An Out[T] is observable and broadcasts to every subscriber; an In[T] subscribes through whatever transport connects it to its source. This reactive-streams approach fits robotics well: sensors emit data asynchronously at different rates, and downstream consumers are often slower than the sources feeding them.

This section is a practical guide to working with streams once they exist - composing operators, handling backpressure, aligning timestamps across sensors, filtering for quality, and recording/replaying data. For how streams fit into the module and blueprint model, see How DimOS fits together; for how to declare In[T]/Out[T] on a module, see Modules.

Declaring stream ports

A module declares its streams as typed fields on the class:

skip

from dimos.core.module import Module
from dimos.core.stream import In, Out
from dimos.msgs.sensor_msgs.Image import Image

class CameraModule(Module):
    color_image: Out[Image]

class DetectorModule(Module):
    color_image: In[Image]

Compose the two with autoconnect() and color_image is wired automatically, since the output and input agree on both name and type:

skip

from dimos.core.coordination.blueprints import autoconnect

my_app = autoconnect(CameraModule.blueprint(), DetectorModule.blueprint())

From here, color_image.observable() on the consuming side gives you an RxPY Observable[Image] you can pipe through the operators covered in this section. See Modules for handler patterns (handle_x methods, sync subscriptions, Specs) once a stream is wired.

Guides

Guide	What it covers
ReactiveX Fundamentals	RxPY operators quick reference - observables, subscriptions, and the common .pipe() patterns used in this codebase
Storage & Replay	Recording streams to disk and replaying them later with their original timing
Temporal Alignment	Matching messages from multiple sensors (e.g. camera + lidar) by timestamp so they can be processed together
Quality-Based Filtering	Selecting the best-quality frame in a time window instead of blindly dropping frames when downsampling
Advanced Streams	Backpressure and parallel-subscriber patterns for fast producers and slow consumers

Which page do I need?

• I’m new to RxPY or need an operator refresher -> ReactiveX Fundamentals
• I want to record sensor data and play it back later -> Storage & Replay
• I need to pair up messages from two sensors that arrive at different rates -> Temporal Alignment
• I’m downsampling a stream and don’t want to lose the sharpest/best frame -> Quality-Based Filtering
• My consumer is slower than my producer, or I have multiple subscribers with different needs -> Advanced Streams

Quick example

This sketch combines several of the guides above: filter blurry camera frames, align the result with lidar by timestamp, and apply backpressure so a slow consumer doesn’t stall the pipeline.

skip

from reactivex import operators as ops
from dimos.utils.reactive import backpressure
from dimos.types.timestamped import align_timestamped
from dimos.msgs.sensor_msgs.Image import sharpness_barrier

# Camera at 30fps, lidar at 10Hz
camera_stream = camera.observable()
lidar_stream = lidar.observable()

# Pipeline: filter blurry frames -> align with lidar -> handle slow consumers
processed = (
    camera_stream.pipe(
        sharpness_barrier(10.0),  # Keep sharpest frame per 100ms window (10Hz)
    )
)

aligned = align_timestamped(
    backpressure(processed),     # Camera as primary
    lidar_stream,                # Lidar as secondary
    match_tolerance=0.1,
)

aligned.subscribe(lambda pair: process_frame_with_pointcloud(*pair))

See also

• How DimOS fits together - the mental model for modules, streams, and blueprints
• Modules - declaring In[T]/Out[T] on a module and handling incoming messages
• Transports - how a stream actually moves bytes between modules (LCM, Zenoh, shared memory, etc.)