Native Modules

Prerequisite for this is to understand dimos Modules and Blueprints. Native modules let you wrap any executable as a first-class DimOS module, given it speaks LCM. Python will handle blueprint wiring, lifecycle, and logging. Native binary handles the actual computation, publishing and subscribing directly on LCM. Python module never touches the pubsub data. It just passes configuration and LCM topic to use via CLI args to your executable. On how to speak LCM with the rest of dimos, you can read our LCM intro

Defining a native module

Python side native module is just a definition of a config dataclass and module class specifying pubsub I/O. Both the config dataclass and pubsub topics get converted to CLI args passed down to your executable once the module is started.

no-result session=nativemodule

from dimos.core.stream import Out
from dimos.core.transport import LCMTransport
from dimos.core.native_module import NativeModule, NativeModuleConfig
from dimos.msgs.sensor_msgs.PointCloud2 import PointCloud2
from dimos.msgs.sensor_msgs.Imu import Imu
import time

class MyLidarConfig(NativeModuleConfig):
    executable: str = "./build/my_lidar"
    host_ip: str = "192.168.1.5"
    frequency: float = 10.0

class MyLidar(NativeModule):
    config: MyLidarConfig
    pointcloud: Out[PointCloud2]
    imu: Out[Imu]

That’s it. MyLidar is a full DimOS module. You can use it with autoconnect, blueprints, transport overrides, and specs. Once this module is started, your ./build/my_lidar will get called with specific CLI args.

How it works

When start() is called, NativeModule:

Builds the executable if it doesn’t exist and build_command is set.
Collects topics from blueprint-assigned transports on each declared port.
Builds the command line: <executable> --<port> <topic> ... --<config_field> <value> ...
Launches the subprocess with Popen, piping stdout/stderr.
Starts a watchdog thread that calls stop() if the process crashes.

For the example above, the launched command would look like:

skip

./build/my_lidar \
    --pointcloud '/pointcloud#sensor_msgs.PointCloud2' \
    --imu '/imu#sensor_msgs.Imu' \
    --host_ip 192.168.1.5 \
    --frequency 10.0

ansi=false session=nativemodule skip

mylidar = MyLidar()
mylidar.pointcloud.transport = LCMTransport("/lidar", PointCloud2)
mylidar.imu.transport = LCMTransport("/imu", Imu)
mylidar.start()

2026-02-14T11:22:12.123963Z [info     ] Starting native process   [dimos/core/native_module.py] cmd='./build/my_lidar --pointcloud /lidar#sensor_msgs.PointCloud2 --imu /imu#sensor_msgs.Imu --host_ip 192.168.1.5 --frequency 10.0' cwd=/home/lesh/coding/dimos/docs/usage/build

Topic strings use the format /<name>#<msg_type>, which is the LCM channel name that Python LCMTransport subscribers use. The native binary publishes on these exact channels. When stop() is called, the process receives SIGTERM. If it doesn’t exit within shutdown_timeout seconds (default 10), it gets SIGKILL.

Config

NativeModuleConfig extends ModuleConfig with subprocess fields:

Field	Type	Default	Description
`executable`	`str`	(required)	Path to the native binary (relative to `cwd` if set)
`build_command`	`str \| None`	`None`	Shell command to run if executable is missing (auto-build)
`cwd`	`str \| None`	`None`	Working directory for build and runtime. Relative paths are resolved against the Python file defining the module
`extra_args`	`list[str]`	`[]`	Additional CLI arguments appended after auto-generated ones
`extra_env`	`dict[str, str]`	`{}`	Extra environment variables for the subprocess
`shutdown_timeout`	`float`	`10.0`	Seconds to wait for SIGTERM before SIGKILL
`log_format`	`LogFormat`	`TEXT`	How to parse subprocess output (`TEXT` or `JSON`)
`cli_exclude`	`frozenset[str]`	`frozenset()`	Config fields to skip when generating CLI args

Auto CLI arg generation

Any field you add to your config subclass automatically becomes a --name value CLI arg. Fields from NativeModuleConfig itself (like executable, extra_args, cwd) are not passed — they’re for Python-side orchestration only.

skip

from pydantic import Field

class LogFormat(enum.Enum):
    TEXT = "text"
    JSON = "json"

class MyConfig(NativeModuleConfig):
    executable: str = "./build/my_module" # relative or absolute path to your executable
    host_ip: str = "192.168.1.5"     # becomes --host_ip 192.168.1.5
    frequency: float = 10.0           # becomes --frequency 10.0
    enable_imu: bool = True           # becomes --enable_imu true
    filters: list[str] = Field(default_factory=lambda: ["a", "b"])  # becomes --filters a,b

None values are skipped.
Booleans are lowercased (true/false).
Lists are comma-joined.

Excluding fields

If a config field shouldn’t be a CLI arg, add it to cli_exclude:

skip

class MyNativeConfig(NativeModuleConfig):
    executable: str = "./build/my_native"
    acc_cov: float = 1.0                                  # rendered into a config file, not a CLI arg
    config_path: str | None = None                        # set at start() to the generated file
    cli_exclude: frozenset[str] = frozenset({"acc_cov"})  # only config_path is passed

Using with blueprints

Native modules work with autoconnect exactly like Python modules:

skip

from dimos.core.coordination.blueprints import autoconnect

class PointCloudConsumer(Module):
    pointcloud: In[PointCloud2]
    imu: In[Imu]

autoconnect(
    MyLidar.blueprint(host_ip="192.168.1.10"),
    PointCloudConsumer.blueprint(),
).build().loop()

autoconnect matches ports by (name, type), assigns LCM topics, and passes them to the native binary as CLI args. You can override transports as usual:

skip

blueprint = autoconnect(
    MyLidar.blueprint(),
    PointCloudConsumer.blueprint(),
).transports({
    ("pointcloud", PointCloud2): LCMTransport("/my/custom/lidar", PointCloud2),
})

Logging

NativeModule pipes subprocess stdout and stderr through structlog:

stdout is logged at info level.
stderr is logged at warning level.

JSON log format

If your native binary outputs structured JSON lines, set log_format=LogFormat.JSON:

skip

class MyConfig(NativeModuleConfig):
    executable: str = "./build/my_module"
    log_format: LogFormat = LogFormat.JSON

The module will parse each line as JSON and feed the key-value pairs into structlog. The event key becomes the log message:

{"event": "sensor initialized", "device": "/dev/ttyUSB0", "baud": 115200}

Malformed lines fall back to plain text logging.

Writing the C++ side

A header-only helper is provided at dimos/hardware/sensors/lidar/common/dimos_native_module.hpp:

#include "dimos_native_module.hpp"
#include "sensor_msgs/PointCloud2.hpp"

int main(int argc, char** argv) {
    dimos::NativeModule mod(argc, argv);

    // Get the LCM channel for a declared port
    std::string pc_topic = mod.topic("pointcloud");

    // Get config values
    float freq = mod.arg_float("frequency", 10.0);
    std::string ip = mod.arg("host_ip", "192.168.1.5");

    // Set up LCM publisher and publish on pc_topic...
}

The helper provides:

Method	Description
`topic(port)`	Get the full LCM channel string (`/topic#msg_type`) for a port
`arg(key, default)`	Get a string config value
`arg_float(key, default)`	Get a float config value
`arg_int(key, default)`	Get an int config value
`has(key)`	Check if a port/arg was provided

It also includes make_header() and time_from_seconds() for building ROS-compatible stamped messages.

Examples

For language interop examples (subscribing to DimOS topics from C++, TypeScript, Lua), see /examples/language-interop/.

Livox Mid-360 Module

The Livox Mid-360 LiDAR driver is a complete example at dimos/hardware/sensors/lidar/livox/module.py:

skip

from dimos.core.stream import Out
from dimos.core.native_module import NativeModule, NativeModuleConfig
from dimos.msgs.sensor_msgs.PointCloud2 import PointCloud2
from dimos.msgs.sensor_msgs.Imu import Imu
from dimos.spec import perception

class Mid360Config(NativeModuleConfig):
    cwd: str | None = "cpp"
    executable: str = "result/bin/mid360_native"
    build_command: str | None = "nix build .#mid360_native"
    host_ip: str = "192.168.1.5"
    lidar_ip: str = "192.168.1.155"
    frequency: float = 10.0
    enable_imu: bool = True
    frame_id: str = "lidar_link"
    # ... SDK port configuration

class Mid360(NativeModule, perception.Lidar, perception.IMU):
    config: Mid360Config
    lidar: Out[PointCloud2]
    imu: Out[Imu]

Usage:

skip

from dimos.hardware.sensors.lidar.livox.module import Mid360

autoconnect(
    Mid360.blueprint(host_ip="192.168.1.5"),
    SomeConsumer.blueprint(),
)

Auto Building

If build_command is set in the module config, and the executable doesn’t exist when start() is called, NativeModule runs the build command automatically. Build output is piped through structlog (stdout at info, stderr at warning).

skip

class MyLidarConfig(NativeModuleConfig):
    cwd: str | None = "cpp"
    executable: str = "result/bin/my_lidar"
    build_command: str | None = "nix build .#my_lidar"

cwd is used for both the build command and the runtime subprocess. Relative paths are resolved against the directory of the Python file that defines the module If the executable already exists, the build step is skipped entirely.

​Native Modules

​Defining a native module

​How it works

​Config

​Auto CLI arg generation

​Excluding fields

​Using with blueprints

​Logging

​JSON log format

​Writing the C++ side

​Examples

​Livox Mid-360 Module

​Auto Building

Native Modules

Defining a native module

How it works

Config

Auto CLI arg generation

Excluding fields

Using with blueprints

Logging

JSON log format

Writing the C++ side

Examples

Livox Mid-360 Module

Auto Building