How to Remap Blueprint Streams and Override Transport Layers (LCM, SHM, DDS, ROS 2) in DimOS

Use Blueprint.remappings() to rename streams or inject concrete module implementations, and Blueprint.transports() to supply custom PubSubTransport instances like SHMTransport or ROSTransport for specific stream name and type pairs.

DimOS (dimensionalOS/dimos) composes robot programs through Blueprints defined in dimos/core/blueprints.py. You can remap Blueprint streams and override transport layers—including LCM, shared memory (SHM), DDS, and ROS 2—declaratively at construction time without modifying the underlying module source code.

How Stream Remapping Works

The remappings() API accepts a list of tuples in the form (module_class, old_stream_name, new_target) where new_target is either a string name or a concrete Module/Spec type.

According to the DimOS source code, calling .remappings() merges entries into the remapping_map (type Mapping[tuple[type[Module], str], str | type[Module] | type[Spec]]) as implemented in lines 42-48 of dimos/core/blueprints.py. During stream discovery, the _all_name_types() method consults this map (lines 99-102) to determine the effective stream name. Because _verify_no_name_conflicts() also uses these resolved names, renaming a stream automatically eliminates name collisions between modules.

How Transport Overrides Work

The transports() API accepts a dictionary mapping (stream_name, stream_type) tuples to ready-made PubSubTransport instances.

In dimos/core/blueprints.py, the .transports() call populates transport_map (lines 33-35). When the Blueprint wires connections via _connect_streams(), it retrieves the transport through _get_transport_for(name, type) (lines 182-190). This method checks transport_map first (lines 183-186); if a user-provided transport exists, it returns that instance immediately, bypassing default construction. Otherwise, it creates a default transport based on whether the stream type defines lcm_encode, selecting either LCMTransport (typed) or pLCMTransport (pickled).

Available Transport Layers

All transports reside in dimos/core/transport.py and implement the PubSubTransport interface from dimos/core/stream.py:

• LCMTransport: Typed LCM encoding for messages defining lcm_encode (lines 12-42).
• pLCMTransport: Pickled LCM for arbitrary Python objects, used when lcm_encode is absent.
• SHMTransport: High-bandwidth shared-memory transport for binary blobs (lines 90-106).
• ROSTransport: ROS 2 bridge via DimosROS (lines 251-274).
• DDSTransport: CycloneDDS implementation, available when DDS_AVAILABLE is true (lines 86-108).

Code Examples

Rename a Stream to Resolve Conflicts

from dimos.core.blueprints import autoconnect
from dimos.robot.unitree.go2.modules import CameraModule, ControlModule

bp = (
    autoconnect(
        CameraModule(),
        ControlModule(),
    )
    .remappings([
        (CameraModule, "color_image", "front_camera"),
    ])
)

ControlModule now receives data on its front_camera input instead of color_image. The remapping is stored in remapping_map and applied during _connect_streams().

Override Transport with Shared Memory

from dimos.core.blueprints import autoconnect
from dimos.core.transport import SHMTransport
from dimos.robot.unitree.go2.modules import CameraModule

bp = (
    autoconnect(CameraModule())
    .transports({
        ("color_image", Image): SHMTransport("/cam/shm", quality=95),
    })
)

The color_image stream uses SHMTransport instead of the default LCM transport. The mapping is stored in transport_map and retrieved by _get_transport_for() during stream connection.

Mix Transports: SHM for Images, ROS 2 for Joint States

from dimos.core.blueprints import autoconnect
from dimos.core.transport import SHMTransport, ROSTransport
from dimos.robot.unitree.go2.modules import CameraModule, JointModule

bp = (
    autoconnect(
        CameraModule(),
        JointModule(),
    )
    .transports({
        ("color_image", Image): SHMTransport("/cam/shm", quality=90),
        ("joint_states", JointState): ROSTransport(
            "/joint_states", JointState
        ),
    })
)

Each stream uses the transport best suited to its data rate and consumer ecosystem.

Combine Remapping and Transport Override

from dimos.core.blueprints import autoconnect
from dimos.core.transport import SHMTransport
from dimos.robot.unitree.go2.modules import CameraModule, ControlModule

bp = (
    autoconnect(
        CameraModule(),
        ControlModule(),
    )
    .remappings([
        (CameraModule, "color_image", "front_camera"),
    ])
    .transports({
        ("front_camera", Image): SHMTransport("/front_cam/shm", quality=85),
    })
)

ControlModule receives the image on the remapped front_camera port, with data traveling over shared memory via SHMTransport.

Summary

• Use Blueprint.remappings() to rename streams or replace abstract RPC targets with concrete implementations, resolving naming conflicts automatically via _verify_no_name_conflicts().
• Use Blueprint.transports() to inject custom transport instances for specific (name, type) pairs, overriding the default selection logic in _get_transport_for().
• Available transports include LCMTransport, pLCMTransport, SHMTransport, ROSTransport, and DDSTransport, all residing in dimos/core/transport.py.
• The wiring logic in dimos/core/blueprints.py processes remapping_map during stream discovery and transport_map during connection setup, enabling declarative reconfiguration without module modifications.

Frequently Asked Questions

How do I resolve a "stream name collision" error between two modules?

Call .remappings() on your Blueprint to rename one of the conflicting streams. For example, if two modules both output color_image, remap one to front_camera using the tuple (CameraModule, "color_image", "front_camera"). The _verify_no_name_conflicts() method uses resolved names from remapping_map, so the collision is eliminated automatically.

Can I use different transports for different streams in the same Blueprint?

Yes. Pass a dictionary to .transports() specifying the transport for each (stream_name, stream_type) tuple you want to override. Streams not present in the dictionary continue to use the default transport logic in _get_transport_for(), which selects LCMTransport or pLCMTransport based on the presence of lcm_encode.

What is the difference between LCMTransport and pLCMTransport?

LCMTransport uses typed LCM encoding and requires the message type to define an lcm_encode method. pLCMTransport pickles arbitrary Python objects and is selected automatically when lcm_encode is absent. You can override either by providing a specific transport instance via .transports().

How do I replace an abstract RPC interface with a concrete module implementation?

Use the remapping API with a (ModuleClass, "stream_name", ConcreteModule) tuple. For example, (PlannerModule, "navigator", MyNavModule) replaces the abstract navigator stream with MyNavModule. The remapping system accepts type[Module] or type[Spec] as the replacement target, allowing you to inject specific implementations without changing consumer module code.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}