dexcomm 0.4.2
Accelerated communication library using Zenoh for robotics and AI
Requires Python: >=3.10
Keywords: communication, robotics, AI, zenoh, high-performance

# 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)
```bash
sudo apt-get update
sudo apt-get install ffmpeg libjpeg-turbo8 -y
```

#### macOS
```bash
brew install ffmpeg libjpeg-turbo
```

#### Windows

**Option 1 - Automated (Easiest):**
```powershell
# Run in PowerShell (as Administrator)
irm https://raw.githubusercontent.com/dexmate-ai/dexcomm/main/scripts/setup_windows_dependencies.ps1 | iex
```

**Option 2 - Using Chocolatey:**
```powershell
choco install ffmpeg libjpeg-turbo -y
```

### Install DexComm

After installing system dependencies:

```bash
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)

```python
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)

```python
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)

```python
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

```python
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:

```python
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

```python
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

```python
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:

```bash
# 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: contact@dexmate.ai