How to Add Support for New Robot Hardware to the DimOS Module System

To add support for new robot hardware to the DimOS module system, implement a protocol-compliant adapter in dimos/hardware/<category>/<vendor>/, register it with the AdapterRegistry, wrap it in a Module subclass that exposes typed streams and RPC methods, and compose it into a blueprint listed in dimos/robot/all_blueprints.py.

DimOS (Dimensional OS) enables developers to treat any robot component—sensors, drive-trains, or manipulators—as modular units that expose typed streams (In[T], Out[T]) and RPC methods. Adding support for new robot hardware follows a structured pipeline that keeps hardware-specific code isolated from high-level logic through the adapter pattern implemented in the dimensionalOS/dimos repository.

Core Architecture of the DimOS Hardware Abstraction

Understanding the four-layer architecture is essential before implementing new hardware support.

• ModuleBase / Module (located in dimos/core/module.py): The base class for all runtime modules. It handles stream creation, RPC registration via the @rpc decorator, and graceful shutdown. When you subclass Module, the __init_subclass__ method (lines 90–110) automatically converts class attributes like joint_positions: Out[list[float]] into concrete stream objects at runtime.

• Stream types (In, Out, RemoteOut): Defined in dimos/core/stream.py, these typed data pipes are auto-wired by the blueprint system based on name and type matching.

• Hardware specification protocols: Found in paths like dimos/hardware/manipulators/spec.py, these Protocol classes (decorated with @runtime_checkable) define the uniform API that every concrete driver must implement—methods like connect(), disconnect(), and domain-specific calls such as read_joint_positions().

• Adapter registry: The AdapterRegistry class in dimos/hardware/manipulators/registry.py provides auto-discovery via the discover() method, which scans subpackages for adapter.py files containing a register(registry) function.

• Blueprint composition: The autoconnect helper in dimos/core/blueprints.py wires modules together by matching stream names and types, while dimos/robot/all_blueprints.py serves as the central registry making blueprints addressable via the CLI (dimos run …).

Step 1: Define or Select a Hardware Protocol

If you are adding a completely new category of hardware (not manipulators, cameras, or drive-trains), first define a protocol that enforces structural typing:

# dimos/hardware/<category>/spec.py

from typing import Protocol, runtime_checkable
from enum import Enum

class MyHardwareStatus(Enum):
    DISCONNECTED = "disconnected"
    CONNECTED = "connected"
    ERROR = "error"

@runtime_checkable
class MyHardwareAdapter(Protocol):
    """Protocol that every concrete driver for this hardware must implement."""
    def connect(self) -> bool: ...
    def disconnect(self) -> None: ...
    def get_status(self) -> MyHardwareStatus: ...

If you are extending an existing category (e.g., adding a new manipulator arm), reuse the existing spec such as ManipulatorAdapter in dimos/hardware/manipulators/spec.py.

Step 2: Implement the Concrete Adapter

Create a subpackage under the appropriate hardware category and implement the adapter class. The class must satisfy the protocol through structural typing—it does not need to explicitly inherit from the protocol—and must expose a register function.

# dimos/hardware/manipulators/xarm/adapter.py

from dimos.hardware.manipulators.spec import ManipulatorAdapter, ControlMode
from dimos.hardware.manipulators.registry import AdapterRegistry

class XArmAdapter:
    """Thin wrapper around the XArm SDK."""
    def __init__(self, ip: str, dof: int = 6):
        self.ip = ip
        self.dof = dof

    def connect(self) -> bool:
        return True  # SDK connection logic here

    def disconnect(self) -> None: ...

    def is_connected(self) -> bool: ...

    def get_info(self): ...

    def set_control_mode(self, mode: ControlMode) -> bool: ...

    def read_joint_positions(self) -> list[float]: ...

    def write_joint_positions(self, positions: list[float], velocity: float = 1.0) -> bool: ...
    
def register(registry: AdapterRegistry) -> None:
    """Called by the auto-discovery system in registry.py."""
    registry.register("xarm", XArmAdapter)

Place this file at dimos/hardware/<category>/<vendor>/adapter.py. Ensure the subpackage has an __init__.py so the registry can import it.

Step 3: Verify Auto-Discovery

The AdapterRegistry automatically discovers adapters when you import the registry module. The registry is instantiated at import time in dimos/hardware/manipulators/registry.py:

adapter_registry = AdapterRegistry()
adapter_registry.discover()

After adding your adapter, verify it appears in the available adapters:

from dimos.hardware.manipulators.registry import adapter_registry
print(adapter_registry.available())  # Should include "xarm"

No manual registration in central files is required—the register function in your adapter.py handles everything.

Step 4: Create the Module Wrapper

Modules translate low-level adapter calls into typed streams and RPCs that the rest of the system can consume. Inherit from Module (which extends ModuleBase) and define streams as class attributes.

# dimos/hardware/manipulators/xarm/module.py

import asyncio
from dimos.core.module import Module
from dimos.core.stream import Out
from dimos.core.core import rpc
from dimos.hardware.manipulators.registry import adapter_registry

class XArmModule(Module):
    # Streams exposed to downstream modules

    joint_positions: Out[list[float]]
    joint_efforts: Out[list[float]]

    def __init__(self, ip: str, dof: int = 6):
        super().__init__()
        # Create the low-level adapter using the registry factory

        self.adapter = adapter_registry.create("xarm", ip=ip, dof=dof)

    @rpc
    def start(self) -> None:
        self.adapter.connect()
        
        async def publish_loop():
            while self.adapter.is_connected():
                self.joint_positions.publish(self.adapter.read_joint_positions())
                await asyncio.sleep(0.02)  # 50 Hz

        
        asyncio.create_task(publish_loop())

    @rpc
    def move_to(self, positions: list[float], velocity: float = 1.0) -> str:
        success = self.adapter.write_joint_positions(positions, velocity)
        return "OK" if success else "Failed"

The super().__init__() call wires the RPC transport and creates the asyncio event loop. Streams defined as class attributes are automatically instantiated as concrete Out objects.

Step 5: Compose the Blueprint

Blueprints tie hardware modules together and expose them to the CLI. Use the autoconnect helper to automatically wire streams.

# dimos/robot/manipulators/xarm/blueprints.py

from dimos.core.blueprints import autoconnect
from dimos.hardware.manipulators.xarm.module import XArmModule

def xarm_simple():
    """Expose a single XArm manipulator as a standalone stack."""
    return autoconnect(
        XArmModule(ip="192.168.1.10", dof=6),
    )

Step 6: Register and Launch

Add the blueprint entry to dimos/robot/all_blueprints.py to make it discoverable by the CLI:

# dimos/robot/all_blueprints.py

"demo-xarm": "dimos.robot.manipulators.xarm.blueprints:xarm_simple",

Launch the hardware stack:

dimos run demo-xarm --log-level debug

Inspect the running streams:

dimos lcmspy  # Shows joint_positions topic publishing

Call RPC methods from the agent or other modules:

dimos agent-send "move arm to [0.0, 0.5, -0.3, 0.0, 0.0, 0.0]"

Complete Integration Example

Here is a minimal, copy-pasteable example for a fictional "AwesomeArm" manipulator:

# dimos/hardware/manipulators/awesome/adapter.py

from dimos.hardware.manipulators.spec import ManipulatorAdapter, ControlMode
from dimos.hardware.manipulators.registry import AdapterRegistry

class AwesomeAdapter:
    def __init__(self, serial: str, dof: int = 7):
        self.serial = serial
        self.dof = dof

    def connect(self) -> bool: 
        return True
    
    def disconnect(self) -> None: ...
    
    def is_connected(self) -> bool: ...
    
    def get_info(self): ...
    
    def set_control_mode(self, mode: ControlMode) -> bool: ...
    
    def read_joint_positions(self) -> list[float]: ...
    
    def write_joint_positions(self, positions: list[float], velocity: float = 1.0) -> bool: ...

def register(reg: AdapterRegistry) -> None:
    reg.register("awesome", AwesomeAdapter)

# dimos/hardware/manipulators/awesome/module.py

import asyncio
from dimos.core.module import Module
from dimos.core.stream import Out
from dimos.core.core import rpc
from dimos.hardware.manipulators.registry import adapter_registry

class AwesomeModule(Module):
    joint_positions: Out[list[float]]

    def __init__(self, serial: str):
        super().__init__()
        self.adapter = adapter_registry.create("awesome", serial=serial)

    @rpc
    def start(self) -> None:
        self.adapter.connect()
        
        async def poll():
            while self.adapter.is_connected():
                self.joint_positions.publish(self.adapter.read_joint_positions())
                await asyncio.sleep(0.05)
        
        asyncio.create_task(poll())

    @rpc
    def move_to(self, positions: list[float], vel: float = 1.0) -> str:
        return "OK" if self.adapter.write_joint_positions(positions, vel) else "FAIL"

# dimos/robot/manipulators/awesome/blueprints.py

from dimos.core.blueprints import autoconnect
from dimos.hardware.manipulators.awesome.module import AwesomeModule

def awesome_arm():
    return autoconnect(AwesomeModule(serial="ABC123"))

Summary

• Implement a protocol-compliant adapter in dimos/hardware/<category>/<vendor>/adapter.py that exposes a register(registry) function for auto-discovery.
• Create a Module subclass in dimos/hardware/<category>/<vendor>/module.py that instantiates the adapter via adapter_registry.create(), defines typed streams (Out[T]), and exposes RPCs using the @rpc decorator.
• Compose a blueprint using autoconnect in dimos/robot/<category>/<vendor>/blueprints.py to wire the module into the system.
• Register the blueprint in dimos/robot/all_blueprints.py to enable CLI access via dimos run.
• Verify integration using dimos lcmspy to inspect streams and dimos agent-send to test RPC calls.

Frequently Asked Questions

How does the AdapterRegistry auto-discovery mechanism work?

The AdapterRegistry in dimos/hardware/manipulators/registry.py scans all subpackages of dimos.hardware.<category> for files named adapter.py. When it finds a register(registry) function, it executes it, allowing the adapter to register itself under a string key (like "xarm" or "awesome"). This eliminates the need to manually edit central configuration files when adding new hardware vendors.

Do I need to modify core DimOS files to add new hardware support?

No. You only create new files within the dimos/hardware/ and dimos/robot/ directories. The AdapterRegistry discovers your adapter automatically, and all_blueprints.py only requires a single line addition to expose your blueprint to the CLI. You never need to modify dimos/core/module.py or the stream definitions.

What is the difference between an adapter and a module?

The adapter (XArmAdapter) is a thin wrapper around the vendor SDK that implements the hardware protocol (e.g., ManipulatorAdapter). It handles low-level connections and raw data. The module (XArmModule) is a DimOS runtime component that instantiates the adapter, converts its data into typed streams (Out[list[float]]), and exposes RPC methods (@rpc) that other modules or LLM agents can call. Modules run inside the DimOS event loop; adapters are plain Python classes.

How do I expose hardware data to perception or navigation modules?

Define typed stream attributes on your Module class, such as joint_positions: Out[list[float]] or camera_frames: Out[np.ndarray]. The Module base class converts these annotations into concrete stream objects at runtime. Downstream modules declare compatible In[T] attributes with matching type signatures. When you compose blueprints using autoconnect(), DimOS automatically wires Out streams to In streams based on type and name matching, allowing zero-configuration data flow between hardware and perception pipelines.

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