Skip to content

ARCHITECTURE_V2.md

Latest commit

 

History

History
504 lines (406 loc) · 14.8 KB

ARCHITECTURE_V2.md

File metadata and controls

504 lines (406 loc) · 14.8 KB

OpenClaw ROS Bridge - Architecture Redesign

Vision: Universal Robot Gateway

Core Philosophy

Transform from a "TCP bridge to ROS" into a Universal Robot Gateway - a multi-protocol, multi-robot, cloud-native connectivity layer that makes any robot accessible to any AI agent, anywhere.

New Architecture: The Three-Layer Model

┌─────────────────────────────────────────────────────────────────────────┐
│                         AI AGENT LAYER                                   │
│  (OpenClaw, Custom Agents, Cloud AI, Web Dashboards, Mobile Apps)      │
└────────────────────────────────┬────────────────────────────────────────┘
                                 │
                    ┌────────────▼────────────┐
                    │   UNIVERSAL GATEWAY      │
                    │   (The Core Bridge)      │
                    └────────────┬────────────┘
                                 │
         ┌───────────────────────┼───────────────────────┐
         │                       │                       │
         ▼                       ▼                       ▼
┌─────────────────┐  ┌──────────────────┐  ┌──────────────────┐
│  TRANSPORT      │  │   ORCHESTRATION  │  │   CONNECTORS     │
│  LAYER          │  │   LAYER          │  │   LAYER          │
├─────────────────┤  ├──────────────────┤  ├──────────────────┤
│ • WebSocket     │  │ • Multi-Robot    │  │ • ROS1/ROS2      │
│ • gRPC          │  │   Management     │  │ • MQTT Broker    │
│ • TCP Socket    │  │ • Load Balancing │  │ • DDS Bridge     │
│ • Unix Socket   │  │ • Rate Limiting  │  │ • CAN Bus        │
│ • MQTT          │  │ • Message Router │  │ • Serial/UART    │
│ • HTTP/REST     │  │ • Plugin Manager │  │ • Modbus         │
│ • QUIC          │  │ • Hot Reload     │  │ • EtherCAT       │
└─────────────────┘  └──────────────────┘  └──────────────────┘
                                 │
                                 ▼
                    ┌────────────────────────┐
                    │    ROBOT FLEET         │
                    │  (Physical + Simulated)│
                    └────────────────────────┘

Layer 1: Transport Layer (Multi-Protocol)

Innovation: Support every major protocol. Let the agent choose.

class TransportManager:
    """Manages multiple concurrent transport endpoints"""
    
    def __init__(self):
        self.transports = {
            'websocket': WebSocketTransport(port=8765),
            'grpc': GRPCServer(port=50051),
            'tcp': TCPSocketTransport(port=9999),
            'unix': UnixSocketTransport(path='/tmp/openclaw.sock'),
            'mqtt': MQTTTransport(broker='localhost:1883'),
            'http': HTTPServer(port=8080),
            'quic': QUICServer(port=4433),
        }

Protocol Characteristics

Protocol Use Case Pros Cons
WebSocket Web dashboards, browsers Bidirectional, firewall-friendly Slightly higher overhead
gRPC Microservices, cloud Strong typing, streaming, auth Requires protobuf
TCP Socket Simple agents, legacy Simple, universal No built-in auth
Unix Socket Local-only, IPC Zero overhead, secure Same-machine only
MQTT IoT, many robots Pub/sub, lightweight, 3G/4G Broker required
HTTP/REST External integrations Standard, debuggable Polling overhead
QUIC Mobile, lossy networks Fast reconnect, multiplexing Newer protocol

Layer 2: Orchestration Layer (The Brain)

Innovation: Treat robots as a fleet, not individual connections.

2.1 Multi-Robot Management

class RobotFleet:
    """Manages multiple robots as a unified fleet"""
    
    def __init__(self):
        self.robots: Dict[str, Robot] = {}
        self.groups: Dict[str, RobotGroup] = {}
        self.topics = TopicRegistry()
        
    async def broadcast(self, command: Command, selector: RobotSelector):
        """Send command to multiple robots matching selector"""
        targets = self.select_robots(selector)
        results = await asyncio.gather(*[
            robot.execute(command) for robot in targets
        ])
        return FleetResult(targets, results)

2.2 Service Discovery

class ServiceDiscovery:
    """Dynamic robot discovery and registration"""
    
    def __init__(self):
        self.discovery_methods = [
            MDNSDiscovery(),      # Local network
            SSDPDiscovery(),      # UPnP devices
            ROSDiscovery(),       # ROS master
            MQTTDiscovery(),      # MQTT presence
            ManualRegistry(),     # Static config
        ]
    
    async def discover(self) -> List[RobotEndpoint]:
        """Continuously discover available robots"""
        pass

2.3 Message Router

class MessageRouter:
    """Intelligent message routing and transformation"""
    
    def route(self, message: Message) -> RouteDecision:
        # Content-based routing
        if message.topic.startswith('/sensor'):
            return RouteDecision(
                targets=['telemetry_service', 'ml_pipeline'],
                qos=QoS.BEST_EFFORT
            )
        elif message.topic.startswith('/cmd'):
            return RouteDecision(
                targets=['robot_actuator'],
                qos=QoS.EXACTLY_ONCE
            )

2.4 Plugin System (Hot Reload)

class PluginManager:
    """Dynamic plugin loading with hot reload"""
    
    def __init__(self):
        self.plugins: Dict[str, Plugin] = {}
        self.watcher = FileWatcher()
        
    async def load_plugin(self, path: str) -> Plugin:
        """Load plugin from filesystem or remote URL"""
        # Support:
        # - Local Python files
        # - Git repos (auto-clone)
        # - Container images
        # - WASM modules (for sandboxed plugins)
        pass
    
    def watch_and_reload(self):
        """Auto-reload plugins on file change (dev mode)"""
        pass

Layer 3: Connector Layer (Multi-Transport)

Innovation: Connect to any robot, any protocol.

class ConnectorRegistry:
    """Registry of robot connectors"""
    
    def __init__(self):
        self.connectors = {
            # ROS ecosystem
            'ros1': ROS1Connector(),
            'ros2': ROS2Connector(),
            
            # Industrial
            'modbus_tcp': ModbusTCPConnector(),
            'modbus_rtu': ModbusRTUConnector(),
            'ethercat': EtherCATConnector(),
            'profinet': ProfinetConnector(),
            
            # IoT
            'mqtt': MQTTConnector(),
            'coap': CoAPConnector(),
            
            # Drones
            'mavlink': MAVLinkConnector(),
            'dji': DJISDKConnector(),
            
            # Arms
            'ur_rtde': URRTEConnector(),  # Universal Robots
            'fanuc': FanucConnector(),
            'kuka': KukaConnector(),
            
            # Custom
            'http_rest': HTTPRestConnector(),
            'websocket': WebSocketConnector(),
            'grpc': GRPCConnector(),
            
            # Simulation
            'gazebo': GazeboConnector(),
            'isaacsim': IsaacSimConnector(),
            'mujoco': MuJoCoConnector(),
        }

Key Innovations

1. Protocol Translation Engine

class ProtocolTranslator:
    """Translates between any robot protocol and standard OpenClaw messages"""
    
    def __init__(self):
        self.schemas = SchemaRegistry()
        
    def translate(self, 
                  source: Message, 
                  source_schema: Schema,
                  target_schema: Schema) -> Message:
        """
        Examples:
        - ROS Twist → Modbus registers
        - MAVLink → gRPC
        - Protobuf → JSON
        """
        pass

2. Unified Message Schema

// Standard message format for all communications
message OpenClawMessage {
  Header header = 1;
  oneof payload {
    Command command = 2;
    Telemetry telemetry = 3;
    Event event = 4;
    Discovery discovery = 5;
  }
  Metadata metadata = 6;
}

message Command {
  string command_id = 1;
  string target_robot = 2;
  string action = 3;
  google.protobuf.Any parameters = 4;
  int64 timeout_ms = 5;
}

message Telemetry {
  string topic = 1;
  google.protobuf.Any data = 2;
  Quality quality = 3;  // Data quality metrics
}

3. Cloud-Native Architecture

# Deployment modes
modes:
  # Edge: Bridge runs on robot/edge device
  edge:
    discovery: mDNS
    storage: local SQLite
    cloud_sync: optional
  
  # Gateway: Bridge runs on dedicated server
  gateway:
    discovery: consul/etcd
    storage: PostgreSQL
    cloud_sync: real-time
  
  # Cloud: Bridge runs in Kubernetes
  cloud:
    discovery: Kubernetes DNS
    storage: cloud-native (RDS, BigQuery)
    multi_tenant: true
    auto_scaling: true

4. Security-First Design

class SecurityManager:
    """Comprehensive security layer"""
    
    def __init__(self):
        self.auth = AuthenticationProvider([
            JWTAuth(),           # Token-based
            mTLSAuth(),          # Mutual TLS
            APIKeyAuth(),        # API keys
            OAuth2Auth(),        # OAuth2/OIDC
            HMACAuth(),          # Request signing
        ])
        
        self.authorization = RBACManager()
        self.encryption = EncryptionLayer()
        
    async def authenticate(self, request) -> Identity:
        """Multi-method authentication"""
        pass

5. AI-Native Observability

class AIObservability:
    """AI-aware telemetry and monitoring"""
    
    def __init__(self):
        self.metrics = PrometheusMetrics()
        self.tracing = OpenTelemetry()
        self.logging = StructuredLogging()
        
        # AI-specific observability
        self.token_usage = TokenCounter()
        self.latency_perception = LatencyTracker()  # Perception → Action
        self.decision_confidence = ConfidenceHistogram()
        self.hallucination_detector = HallucinationMonitor()

Use Cases Coverage

Use Case Architecture Support
Home robot Unix socket, local discovery, simple auth
Industrial fleet gRPC, Modbus, RBAC, audit logs
Drone swarm MQTT, MAVLink, multicast, geofencing
Research lab WebSocket, hot reload, data recording
Cloud robotics QUIC, Kubernetes, multi-tenant
Legacy integration HTTP/REST, Modbus, protocol translation
Mobile robots MQTT, 4G/5G, reconnect handling
Sim-to-real MuJoCo/IsaacSim connectors, identical API

Implementation Phases

Phase 1: Core Refactor (Foundation)

  • Multi-transport abstraction
  • Unified message schema (protobuf)
  • Plugin hot-reload system
  • Basic multi-robot support

Phase 2: Connectors (Breadth)

  • ROS1/ROS2 connectors
  • MQTT connector
  • HTTP/REST connector
  • WebSocket connector
  • Modbus connector

Phase 3: Advanced Features (Depth)

  • Protocol translation engine
  • Service discovery
  • gRPC + QUIC support
  • Cloud deployment modes
  • Security layer

Phase 4: Ecosystem (Scale)

  • MAVLink, DJI connectors
  • Industrial connectors (EtherCAT, Profinet)
  • WASM plugin sandbox
  • AI observability
  • Marketplace for plugins

API Examples

Connecting to Robot

# Method 1: Discovery
async with OpenClawGateway() as gateway:
    robots = await gateway.discover()
    robot = robots[0]
    await robot.connect()

# Method 2: Direct connection
robot = await gateway.connect(
    uri="ros2://192.168.1.100:7411",
    auth=TokenAuth("secret_token")
)

# Method 3: Fleet selection
fleet = gateway.fleet("warehouse_robots")
robots = await fleet.where(status="idle", battery=">20%")

Sending Commands

# Single command
result = await robot.execute({
    "action": "move_to",
    "parameters": {"x": 1.0, "y": 2.0, "theta": 0.0}
})

# Streaming commands (trajectory)
async with robot.streaming() as stream:
    for waypoint in path:
        await stream.send(waypoint)

# Fleet command
results = await fleet.broadcast({
    "action": "return_to_dock"
}, selector=Robots.where(battery="<15%"))

Receiving Data

# Subscribe to topics
async for message in robot.subscribe("/camera/image"):
    await process_image(message.data)

# Multiple topics with filtering
async for message in gateway.subscribe([
    "/robot1/sensors",
    "/robot2/sensors"
], where=lambda m: m.data.temperature > 30):
    await alert_overheating(message)

Configuration

# config/gateway.yaml
gateway:
  name: "warehouse_gateway_01"
  
  transports:
    websocket:
      enabled: true
      port: 8765
      tls: true
      cert: /etc/certs/gateway.crt
    
    grpc:
      enabled: true
      port: 50051
      reflection: true
    
    mqtt:
      enabled: true
      broker: "mqtt.local:1883"
      client_id: "gateway_01"
  
  discovery:
    methods: ["mdns", "ros2"]
    scan_interval: 30
  
  connectors:
    ros2:
      domain_id: 0
    modbus:
      timeout: 5000
  
  security:
    authentication:
      methods: ["jwt", "mtls"]
      jwt:
        secret: ${JWT_SECRET}
    authorization:
      policy: "rbac"
      roles:
        - name: "operator"
          permissions: ["read", "execute"]
        - name: "admin"
          permissions: ["*"]
  
  plugins:
    directory: "./plugins"
    hot_reload: true
    registry:
      - name: "greenhouse_control"
        source: "git+https://github.com/.../greenhouse_plugin.git"
      - name: "safety_monitor"
        source: "./local_plugins/safety.py"

This architecture transforms OpenClaw ROS Bridge from a single-purpose tool into a universal robot connectivity platform capable of serving any AI agent, any robot, any protocol, anywhere.