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

> Learn to remap Blueprint streams and override transport layers LCM SHM DDS ROS2 in DimOS using Blueprint.remappings() and Blueprint.transports(). Enhance your data communication.

- Repository: [Dimensional/dimos](https://github.com/dimensionalOS/dimos)
- Tags: how-to-guide
- Published: 2026-03-15

---

**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`](https://github.com/dimensionalOS/dimos/blob/main/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`](https://github.com/dimensionalOS/dimos/blob/main/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`](https://github.com/dimensionalOS/dimos/blob/main/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`](https://github.com/dimensionalOS/dimos/blob/main/dimos/core/transport.py) and implement the `PubSubTransport` interface from [`dimos/core/stream.py`](https://github.com/dimensionalOS/dimos/blob/main/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

```python
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

```python
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

```python
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

```python
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`](https://github.com/dimensionalOS/dimos/blob/main/dimos/core/transport.py).
- The wiring logic in [`dimos/core/blueprints.py`](https://github.com/dimensionalOS/dimos/blob/main/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.