dexcomm 0.4.3

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

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}

License

AGPL-3.0 - see the LICENSE file for details.

Copyright (c) 2026 Dexmate. All rights reserved.

For licensing inquiries: [e-mail]