dexcomm 0.4.2

Accelerated communication library using Zenoh for robotics and AI

DexComm - High-Performance Communication Library

A high-performance communication library providing three levels of abstraction for distributed systems. Designed for robotics, AI, and real-time applications with minimal latency and maximum throughput.

Features

• Three Abstraction Levels:
  • Raw API: Direct publisher/subscriber for simple scripts
  • Node Pattern: ROS-like component organization with namespaces
  • Manager Pattern: Dynamic topic management for gateways and monitoring
• Efficient Resource Management: Shared session management for optimal resource usage
• High-Precision Rate Limiting: Nanosecond-precision rate control with adaptive processing compensation
• Flexible Configuration: Comprehensive configuration options with file and environment variable support
• Quality of Service: Configurable reliability, congestion control, and priority
• Service Support: Request-reply pattern with timeout and async calls

Installation

System Requirements

DexComm requires FFmpeg and libjpeg-turbo libraries to be installed on your system before installing the Python package.

Linux (Ubuntu/Debian)

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

macOS

brew install ffmpeg libjpeg-turbo

Windows

Option 1 - Automated (Easiest):

# Run in PowerShell (as Administrator)
irm https://raw.githubusercontent.com/dexmate-ai/dexcomm/main/scripts/setup_windows_dependencies.ps1 | iex

Option 2 - Using Chocolatey:

choco install ffmpeg libjpeg-turbo -y

Install DexComm

After installing system dependencies:

pip install dexcomm

Platform Support

• Python: 3.10, 3.11, 3.12, 3.13
• Operating Systems: Linux, macOS, Windows
• Architectures: x86_64, ARM64 (Apple Silicon), AArch64 (Linux)

Quick Start

Important: All imports are available directly from the dexcomm module, regardless of which pattern you use.

Choosing the Right Pattern

Use Case	Recommended Pattern	Why
Simple script with 1-2 topics	Raw API	Minimal code, direct control
Robot/device controller	Node	Component encapsulation, namespaces
ROS/ROS2 migration	Node	Familiar interface
Data logger/recorder	Manager	Dynamic topic management
Gateway/bridge service	Manager	Runtime topic addition/removal

Pattern 1: Raw API (Simplest)

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

# Create publisher with AutoData encoding
pub = Publisher("sensor/temperature", encoder=PickleDataCodec.encode)
pub.publish(25.5)

# Create subscriber with AutoData decoding
sub = Subscriber("sensor/temperature",
                 decoder=PickleDataCodec.decode,
                 callback=lambda msg: print(f"Temp: {msg}°C"))

# For raw bytes (default), no encoder/decoder needed
raw_pub = Publisher("sensor/raw")
raw_pub.publish(b"raw sensor data")

Pattern 2: Node-based (ROS-like)

from dexcomm import Node
from dexcomm.codecs import PickleDataCodec

class RobotController:
    def __init__(self):
        # Create node with namespace
        self.node = Node("controller", namespace="robot1")

        # Publishers and subscribers are namespaced
        self.cmd_pub = self.node.create_publisher("cmd_vel", encoder=PickleDataCodec.encode)
        self.odom_sub = self.node.create_subscriber("odometry",
                                                     self.on_odometry,
                                                     decoder=PickleDataCodec.decode)

        # Services
        self.node.create_service("reset", self.handle_reset,
                                 request_decoder=PickleDataCodec.decode,
                                 response_encoder=PickleDataCodec.encode)

    def on_odometry(self, msg):
        # Process odometry and publish commands
        self.cmd_pub.publish({"linear": 0.5, "angular": 0.0})

    def handle_reset(self, request):
        # Reset robot state
        return {"success": True}

    def run(self):
        self.node.spin()  # Process callbacks

# Usage
controller = RobotController()
controller.run()

Pattern 3: Manager Pattern (Dynamic Topics)

from dexcomm import SubscriberManager
from dexcomm.codecs import PickleDataCodec

class DataLogger:
    def __init__(self):
        self.subscribers = SubscriberManager()
        self.recording = {}

    def start_recording(self, topics):
        """Dynamically subscribe to topics"""
        for topic in topics:
            self.subscribers.add(
                topic,
                callback=lambda msg, t=topic: self.record(t, msg),
                decoder=PickleDataCodec.decode,
                buffer_size=100
            )

    def record(self, topic, msg):
        if topic not in self.recording:
            self.recording[topic] = []
        self.recording[topic].append(msg)

    def stop_recording(self, topic):
        """Dynamically unsubscribe"""
        self.subscribers.remove(topic)

    def get_stats(self):
        """Get statistics for all subscribers"""
        return self.subscribers.get_stats()

# Usage
logger = DataLogger()
logger.start_recording(["sensors/lidar", "sensors/camera", "robot/pose"])
# Topics can be added/removed at runtime
logger.start_recording(["sensors/imu"])
logger.stop_recording("sensors/camera")

Configuration

Network Configuration

from dexcomm import Publisher, ZenohConfig

# Peer mode (default) - automatic peer discovery
config = ZenohConfig.default_peer()

# Client mode - connect to router
config = ZenohConfig.default_client("tcp/localhost:7447")

# Use config with any component
pub = Publisher("my/topic", config=config)

Quality of Service

Configure reliability, congestion control, and priority:

from dexcomm import Publisher

# Critical: guaranteed delivery, highest priority
pub = Publisher("robot/emergency_stop", encoder=PickleDataCodec.encode,
    qos={"reliability": "reliable", "congestion_control": "block", "priority": "real_time"})

# Sensors: fast delivery, drop old data
pub = Publisher("sensors/lidar", encoder=PickleDataCodec.encode,
    qos={"reliability": "best_effort", "congestion_control": "drop", "priority": "data_high"})

Parameters: reliability (best_effort/reliable), congestion_control (drop/block), priority (background → real_time)

Rate Limiting

from dexcomm.utils import RateLimiter
from dexcomm import RateLimitedPublisher

# High-precision rate limiter
limiter = RateLimiter(rate_hz=30.0)  # 30 Hz
for data in sensor_stream:
    limiter.sleep()  # Maintain 30 Hz
    process(data)

# Adaptive mode - automatically compensates for processing time
limiter = RateLimiter(rate_hz=30.0, adaptive=True)
for data in sensor_stream:
    limiter.sleep()  # Automatically learns and adjusts
    result = heavy_processing(data)  # Variable 10-50ms

# Rate-limited publisher - built-in rate limiting
pub = RateLimitedPublisher("sensor/data", rate=20.0)
for data in stream:
    pub.publish(data)  # Automatically rate-limited to 20 Hz

Services

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

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

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

# For raw bytes (default), no encoder/decoder needed
raw_service = Service("echo", lambda req: req)
raw_client = ServiceClient("echo")
result = raw_client.call(b"hello")

Using Configuration Files

DexComm can load network configuration from JSON files via environment variables:

# using ZENOH_CONFIG (fallback)
export ZENOH_CONFIG=/path/to/config.json
python your_script.py

License

This project is licensed under a Proprietary Software License - see the LICENSE file for details.

Copyright (c) 2025 Dexmate. All rights reserved.

For licensing inquiries, please contact: [e-mail]