dexcomm 0.6.0

Accelerated communication library using Zenoh for robotics and AI

DexComm - High-Performance Communication Library

High-efficiency pub/sub and RPC for robotics and AI. Zero configuration required.

Installation

System Dependencies

Linux (Ubuntu/Debian):

sudo apt-get update
sudo apt-get install ffmpeg libjpeg-turbo8 -y

macOS:

brew install ffmpeg libjpeg-turbo

Install

pip install dexcomm

dexcomm-video has been merged into dexcomm; current users should install only dexcomm.

Platform Support

• Python: 3.10+
• Linux: x86_64, aarch64
• macOS: x86_64, arm64 (Apple Silicon)

Quick Start

Raw API (Simplest)

from dexcomm import Publisher, Subscriber
from dexcomm.codecs import PickleDataCodec

# Publisher
pub = Publisher("sensor/temperature", encoder=PickleDataCodec.encode)
pub.publish({"value": 25.5, "unit": "celsius"})

# Subscriber
sub = Subscriber("sensor/temperature",
                 decoder=PickleDataCodec.decode,
                 callback=lambda msg: print(f"Received: {msg}"))

Node Pattern (ROS-like)

from dexcomm import Node
from dexcomm.codecs import PickleDataCodec

class RobotController:
    def __init__(self):
        self.node = Node("controller", namespace="robot1")
        self.cmd_pub = self.node.create_publisher("cmd_vel", encoder=PickleDataCodec.encode)
        self.node.create_subscriber("odometry", self.on_odometry, decoder=PickleDataCodec.decode)

    def on_odometry(self, msg):
        self.cmd_pub.publish({"linear": 0.5, "angular": 0.0})

robot = RobotController()
input("Press Enter to stop...\n")

Manager Pattern (Dynamic Topics)

from dexcomm import SubscriberManager
from dexcomm.codecs import PickleDataCodec

manager = SubscriberManager()

# Add/remove topics at runtime
manager.add("sensors/lidar", callback=lambda msg: print(msg), decoder=PickleDataCodec.decode)
manager.add("sensors/imu", callback=lambda msg: print(msg), decoder=PickleDataCodec.decode)
manager.remove("sensors/lidar")

Services (RPC)

from dexcomm import Service, ServiceClient
from dexcomm.codecs import PickleDataCodec

# Server
service = Service("math/add",
                  lambda req: {"sum": req["a"] + req["b"]},
                  request_decoder=PickleDataCodec.decode,
                  response_encoder=PickleDataCodec.encode)

# Client
client = ServiceClient("math/add",
                       request_encoder=PickleDataCodec.encode,
                       response_decoder=PickleDataCodec.decode)
result = client.call({"a": 5, "b": 3})
print(result)  # {"sum": 8}

High-Throughput Data

For video frames, point clouds, and other large payloads, use raw/buffer APIs and latest-only polling to avoid extra Python serialization work:

from dexcomm import LocalFrameBuffer, Publisher, Service, ServiceClient, Subscriber
import numpy as np

frame = np.zeros((1200, 1920, 3), dtype=np.uint8)

pub = Publisher("camera/raw")
pub.publish_buffer(frame)

sub = Subscriber("camera/raw")
latest = sub.get_latest_raw()

service = Service("vision/score", handler=lambda req: b"ok")
client = ServiceClient("vision/score")
client.wait_for_service(timeout=1.0)
response = client.call_buffer(frame, timeout=1.0)

frames = LocalFrameBuffer.shared_memory([1200, 1920, 3], dtype="uint8")
frames.write_from(frame)
target = np.empty_like(frame)
header = frames.read_into(target)

Remote Robot Connection

Connect directly to a remote robot with a single environment variable:

# Port defaults to 7447 if omitted:
export ROBOT_IP=192.168.50.20
# Or with explicit port:
export ROBOT_IP=192.168.50.20:7447

When ROBOT_IP is set, dexcomm switches to client mode and connects directly to the specified address. Port 7447 is used by default if not specified.

Release hardening

Released wheels are built through scripts/build_wheel.py which applies path remapping, symbol stripping, extra ELF-note scrubbing, SBOM removal from the wheel tree, and a pinned Rust toolchain (rust-toolchain.toml) for reproducible output. Debug/trace tracing events are compiled out of release builds via release_max_level_info; INFO/WARN/ERROR logs are preserved for operations.

Verify a wheel locally:

python scripts/verify_wheel_security.py path/to/dexcomm-*.whl

See docs/PACKAGING_HARDENING_PLAN.md for the full roadmap and implementation status.

Tests and CI

pytest tests/unit/ -v
pytest tests/integration/ -v
DEXCOMM_RUN_STRESS=1 pytest tests/stress/ -v
DEXCOMM_RUN_STRESS=1 DEXCOMM_RUN_SOAK=1 DEXCOMM_SOAK_SECONDS=1800 pytest tests/stress/ -v

CI runs Rust checks, Python quality checks, wheel builds, cross-platform unit tests, Linux integration tests, and Linux x86_64/ARM64 stress tests for pushes to main, pull requests targeting main, and merge-queue checks. The separate production validation workflow runs router-backed distributed tests and long-running soak tests on main, nightly, and on demand.

License

Proprietary — Copyright © 2026 Dexmate. All rights reserved.

For licensing inquiries: [e-mail]