Skip to content

native_modules.md

Latest commit

 

History

History
300 lines (220 loc) · 11.7 KB

native_modules.md

File metadata and controls

300 lines (220 loc) · 11.7 KB
Raw
title 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.

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:

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

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

./build/my_lidar \
    --pointcloud '/pointcloud#sensor_msgs.PointCloud2' \
    --imu '/imu#sensor_msgs.Imu' \
    --host_ip 192.168.1.5 \
    --frequency 10.0
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.

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:

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:

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:

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:

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

The header-only C++ SDK lives at native/cpp/. Set stdin_config: bool = True in the Python config. Topics and config then arrive as one JSON line on stdin instead of CLI args. A module includes dimos/native.hpp, subclasses Module, and calls run_with_transport<M>() from main():

#include "dimos/native.hpp"
#include "geometry_msgs/Twist.hpp"

using dimos::native::Builder;
using dimos::native::Config;
using dimos::native::Module;
using dimos::native::Output;
using geometry_msgs::Twist;

struct PongConfig {
    std::int64_t sample_config;
};

class Pong : public Module {
public:
    void build(Builder& builder, Config& config) override {
        config_ = config.parse<PongConfig>();
        confirm_ = builder.output<Twist>("confirm");
        builder.input<Twist>("data", &Pong::on_data, this);
    }

private:
    void on_data(const Twist& msg) {
        Twist reply = msg;
        reply.angular.z = static_cast<double>(config_.sample_config);
        confirm_.publish(reply);
    }

    Output<Twist> confirm_;
    PongConfig config_;
};

int main() {
    dimos::native::run_with_transport<Pong>();
    return 0;
}

The config is a plain aggregate struct. config.parse<PongConfig>() reflects over its fields (via PFR, C++20), so the struct declaration is the whole contract: every field is required, unknown fields are rejected, and there is no limit on field count. Python owns all defaults and always sends every field. Add a void validate() const method for range checks. It runs automatically after parsing. Input handlers run serialized on the dispatch thread. Each output publishes through its own worker, so a slow channel only stalls itself. A source-style module with no inputs (a sensor driver) overrides handle() with its own loop and setup()/teardown() for device lifecycle.

A complete ping-pong pair lives at /examples/native-modules/cpp/, and dimos/hardware/sensors/lidar/livox/cpp/main.cpp is a real driver example.

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:

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:

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 streamed line by line through structlog at info, with stderr merged into stdout. nix build prints no build logs unless -L is passed, so the built-in modules all include it.

class MyLidarConfig(NativeModuleConfig):
    cwd: str | None = "cpp"
    executable: str = "result/bin/my_lidar"
    build_command: str | None = "nix build -L .#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.

Faster builds via the Cachix substituter

Nix-built native modules can be substituted from the dimensionalos Cachix cache (the same substituter CI uses) instead of compiled from source. Opt in locally to skip cold compiles when the cache has them:

# ~/.config/nix/nix.conf  (single-user)  or  /etc/nix/nix.conf  (multi-user)
extra-substituters = https://dimensionalos.cachix.org
extra-trusted-public-keys = dimensionalos.cachix.org-1:20ynj6TjpoD3qTxkdNoeHtgs2G2pNvgAq1EQYLTHJXI=