arcros

Arc ROS - A ZMQ-based messaging library for robot operations with automatic publisher registration and name-based pub/sub.

Developed by EdUHK Robotics Team | Eric Gong Da

Features

• 🚀 Name-based messaging - Publish and subscribe using friendly names instead of ports
• 🔌 Auto port allocation - Publishers automatically find available ports from configured range
• 💾 Persistent registry - SQLite database stores name→port mappings
• 🔍 Smart discovery - Scans only registered publishers for instant status checks
• 📊 Relationship tracking - Automatically tracks and visualizes pub/sub relationships
• 🌳 Visual graph - See publisher-subscriber connections at a glance
• ⚙️ Configurable port range - Define custom port allocation ranges
• 🐍 Pythonic API - Context managers, callbacks, and clean interfaces

Installation

From PyPI (recommended)

pip install arcros

This installs:

• arcros Python package
• zmq-monitor CLI tool

For web dashboard support:

pip install arcros[web]

From Source

git clone https://github.com/yourusername/arcros.git
cd arcros
pip install -e .

For development with all dependencies:

pip install -e ".[web,dev]"

Quick Start

1. Publishing Messages

Create a publisher with a name. The port is automatically allocated and stored in the database.

from arcros import create_publisher
import time

# Create publisher (auto allocates port, stores in database)
with create_publisher('sensor_data', description='IMU and encoders') as pub:
    while True:
        data = {
            'x': 1.0,
            'y': 2.0,
            'theta': 0.5,
            'timestamp': time.time()
        }
        pub.publish(data)
        time.sleep(0.1)

With specific port:

with create_publisher('motors', port=5556) as pub:
    pub.publish({'left': 100, 'right': 100})

2. Subscribing to Messages

Subscribe using the publisher's name. arcros looks up the port from the database and automatically tracks the subscription.

from arcros import create_subscriber

def handle_message(data):
    print(f"Received: {data}")

# Subscribe by name (auto-tracked in database)
with create_subscriber('sensor_data') as sub:
    sub.listen(handle_message)

# Custom subscriber name for better tracking
with create_subscriber('sensor_data', subscriber_name='analytics_service') as sub:
    sub.listen(handle_message)

Alternative receiving methods:

with create_subscriber('sensor_data') as sub:
    # Blocking with timeout
    data = sub.receive(timeout_ms=1000)
    if data:
        print(f"Got: {data}")
    
    # Non-blocking
    data = sub.receive_nonblocking()
    if data:
        print(f"Got: {data}")

Subscribe by port (if you know it):

with create_subscriber('5555') as sub:
    data = sub.receive()

3. Monitoring Publishers

Use the CLI tool to discover, monitor, and visualize publisher-subscriber relationships.

List registered publishers and their status:

zmq-monitor list

Output:

Checking 3 registered publisher(s)...

Name                 Port     Host            Status     Data Types
----------------------------------------------------------------------------------------------------
sensor_data          5555     localhost       ✓ Active   x:float, y:float, theta:float, timestamp:float
motors               5556     localhost       ✗ Inactive -
odom                 5557     localhost       ✓ Active   x:float, y:float, yaw:float

Visualize pub/sub relationships:

zmq-monitor graph

Output:

Publisher → Subscriber Graph:

================================================================================

📡 sensor_data (localhost:5555)
   └─> analytics_service @ server1
   └─> logger_service @ server1
   └─> dashboard_ui @ localhost

📡 motors (localhost:5556)
   (no subscribers)

📡 odom (localhost:5557)
   └─> navigation_node @ robot1

================================================================================

Total: 3 publisher(s), 4 subscription(s)

View subscription details:

zmq-monitor subscriptions

Output:

Publisher            Subscriber                     Host            Last Active
------------------------------------------------------------------------------------------
sensor_data          analytics_service              server1         2025-12-02 14:32:45
sensor_data          logger_service                 server1         2025-12-02 14:32:45
sensor_data          dashboard_ui                   localhost       2025-12-02 14:32:46
odom                 navigation_node                robot1          2025-12-02 14:32:47

Monitor messages in real-time:

# By name
zmq-monitor monitor sensor_data

# By port
zmq-monitor monitor 5555

Output:

Monitoring tcp://localhost:5555
Press Ctrl+C to stop

[14:32:45.123]
  x: 1.0000
  y: 2.0000
  theta: 0.5000
  timestamp: 1732745565.1230

[14:32:45.223]
  x: 1.1000
  y: 2.1000
  theta: 0.5100
  timestamp: 1732745565.2230

View registry:

zmq-monitor registry

Output:

Name                 Port     Host            Description
--------------------------------------------------------------------------------
sensor_data          5555     localhost       IMU and encoders
motors               5556     localhost       Motor commands
odom                 5557     localhost       Robot odometry

Configure port range:

# View current configuration
zmq-monitor config

# Set custom port range (start port + count)
zmq-monitor config --start 5555 --count 200

Output:

Port range configuration:
  Start port: 5555
  Port count: 200
  End port:   5754 (inclusive)

Used ports: 3/200

Manually add/remove from registry:

# Add
zmq-monitor add odom 5557 --desc "Robot odometry"

# Remove
zmq-monitor remove odom

Web Dashboard

Start the web dashboard to visualize and manage publishers, subscribers, and relationships:

arcros-web

Then open http://localhost:8000 in your browser.

Features:

• 📊 Real-time publisher/subscriber status
• 🌳 Interactive relationship graph
• ⚙️ Port range configuration
• 📝 Publisher management (add/remove)
• 🔍 Subscription monitoring

Custom host/port:

arcros-web --host 0.0.0.0 --port 8080

Note: Requires web dependencies: pip install arcros[web]

Complete Example

Terminal 1 - Publisher:

# publisher.py
from arcros import create_publisher
import time
import random

with create_publisher('sensor_data', description='Simulated sensors') as pub:
    print("Publishing...")
    while True:
        data = {
            'e1': random.uniform(40.0, 50.0),
            'e2': random.uniform(-15.0, -10.0),
            'imu': random.uniform(175.0, 185.0),
            'timestamp': time.time()
        }
        pub.publish(data)
        time.sleep(0.1)

python publisher.py
# Output: Publisher 'sensor_data' bound to localhost:5555

Terminal 2 - Subscriber:

# subscriber.py
from arcros import create_subscriber
from datetime import datetime

def handle_message(data):
    timestamp = datetime.now().strftime('%H:%M:%S.%f')[:-3]
    print(f"[{timestamp}] e1={data['e1']:.2f}, e2={data['e2']:.2f}, imu={data['imu']:.2f}")

with create_subscriber('sensor_data') as sub:
    sub.listen(handle_message)

python subscriber.py
# Output: Subscriber connected to 'sensor_data' at localhost:5555
#         [14:32:45.123] e1=45.23, e2=-12.45, imu=180.34

Terminal 3 - Monitor:

# Discover all active publishers
zmq-monitor list

# Monitor specific publisher
zmq-monitor monitor sensor_data

# View registry
zmq-monitor registry

API Reference

Publisher

from arcros import create_publisher

pub = create_publisher(
    name='my_topic',           # Unique publisher name
    port=None,                  # Port (None = auto allocate)
    host='localhost',           # Host address
    description=''              # Optional description
)

pub.publish({'key': 'value'})  # Publish dict (JSON serializable)
pub.close()                     # Clean up

Context manager (recommended):

with create_publisher('my_topic') as pub:
    pub.publish({'data': 123})
# Automatically closed

Subscriber

from arcros import create_subscriber

sub = create_subscriber(
    name_or_port='my_topic',   # Publisher name or port number
    host='localhost',           # Host address (if using port)
    subscriber_name=None        # Optional: custom subscriber ID for tracking
)

# Blocking receive with timeout
data = sub.receive(timeout_ms=1000)

# Non-blocking receive
data = sub.receive_nonblocking()

# Continuous listening with callback
def callback(data):
    print(data)

sub.listen(callback)  # Blocks until Ctrl+C

sub.close()  # Automatically removes subscription from database

Custom subscriber names:

# Auto-generated ID (hostname_pid)
sub = create_subscriber('sensor_data')
# ID: Gongs-MacBook-Air_12345

# Custom name for service identification
sub = create_subscriber('sensor_data', subscriber_name='analytics_service')
# ID: analytics_service

Context manager (recommended):

with create_subscriber('my_topic') as sub:
    data = sub.receive()
# Automatically closed

CLI Commands

# List registered publishers with status
zmq-monitor list
zmq-monitor list --host 192.168.1.10

# Show publisher-subscriber graph
zmq-monitor graph

# Show subscription details
zmq-monitor subscriptions

# Monitor publisher messages
zmq-monitor monitor sensor_data
zmq-monitor monitor 5555

# Registry management
zmq-monitor registry
zmq-monitor add sensors 5555 --desc "Chassis sensors"
zmq-monitor remove sensors

# Port range configuration
zmq-monitor config
zmq-monitor config --start 5555 --count 200

# Help
zmq-monitor --help

How It Works

1. Publisher Registration

• When you create a publisher with a name, arcros checks the database
• If the name exists, it reuses the stored port
• If not, it allocates the first available port from the configured range (default: 5555-5654)
• The name→port→host mapping is stored in SQLite (arcros/database/publishers.db)

2. Sequential Port Allocation

• Publishers get sequential ports: first gets 5555, next gets 5556, etc.
• Database tracks used ports to avoid collisions
• Port range is configurable via zmq-monitor config

3. Subscriber Lookup & Tracking

• When you create a subscriber with a name, arcros queries the database
• It retrieves the port and host for that publisher name
• The subscriber connects to tcp://host:port
• Subscription is automatically recorded with subscriber ID and timestamp
• When subscriber closes, the subscription is removed from database

4. Smart Discovery

• zmq-monitor list scans only registered publishers from database
• Checks each registered port for activity (✓ Active / ✗ Inactive)
• Much faster than scanning entire port ranges
• Shows data types for active publishers

5. Relationship Graph

• Database stores all publisher → subscriber relationships
• zmq-monitor graph visualizes the connections
• zmq-monitor subscriptions shows detailed table view
• Tracks last active timestamp for each subscription

Database

Publisher registry and subscriptions are stored in arcros/database/publishers.db (auto-created).

Schema

Publishers table:

CREATE TABLE publishers (
    id INTEGER PRIMARY KEY,
    name TEXT UNIQUE NOT NULL,
    port INTEGER NOT NULL,
    host TEXT NOT NULL,
    description TEXT,
    created_at TIMESTAMP,
    updated_at TIMESTAMP
);

Subscriptions table:

CREATE TABLE subscriptions (
    id INTEGER PRIMARY KEY,
    subscriber_name TEXT NOT NULL,
    publisher_name TEXT NOT NULL,
    subscriber_host TEXT NOT NULL,
    created_at TIMESTAMP,
    last_active TIMESTAMP,
    UNIQUE(subscriber_name, publisher_name)
);

Config table:

CREATE TABLE config (
    key TEXT PRIMARY KEY,
    value TEXT NOT NULL,
    updated_at TIMESTAMP
);

Location

• Default: arcros/database/publishers.db
• Created automatically on first use
• Stores: publishers, subscriptions, and configuration (port range)

Requirements

• Python 3.8+
• pyzmq >= 25.0.0

Examples

See tests/examples/ for complete working examples:

• publisher.py - Name-based publisher with auto port allocation
• subscriber.py - Name-based subscriber with callback
• README.md - Detailed API usage examples

Troubleshooting

Publisher not found:

ValueError: Publisher 'my_topic' not found in database

→ The publisher hasn't been created yet. Start the publisher first, or use a port number directly.

Port already in use:

RuntimeError: No available ports in configured range

→ All ports in range are used. Increase the port range: zmq-monitor config --start 5555 --count 200

No active publishers found:

zmq-monitor list
# Checking 0 registered publisher(s)...
# No publishers registered in database.

→ No publishers have been created. Use create_publisher() to register publishers.

Subscription not showing in graph: → Make sure the subscriber is still running. Subscriptions are auto-removed when subscribers close.

Old subscriptions still showing: → These are zombie subscriptions from crashed processes. Restart the subscriber to update last_active.

Code Style

• PEP 8 formatting
• 4-space indentation
• snake_case for files and functions
• PascalCase for classes
• Context managers for resource management

License

MIT