cloud-robotics

name	cloud-robotics
description	云机器人技能 - 边缘云协同、云端规划、远程操控、数字孪生、ROS2 云桥接
argument-hint	云机器人 OR cloud robotics OR 边缘计算 OR digital twin OR 云边协同
user-invocable	true

云机器人技能

用于实现云机器人架构，涵盖边缘云协同、云端计算、远程操控、数字孪生和 ROS2 云桥接

---

何时使用

当需要以下帮助时使用此技能：

• 云端机器人计算卸载（Computation Offloading）
• 远程机器人操控
• 数字孪生系统构建
• 云端 SLAM/导航计算
• 低延迟云边通信架构

---

快速参考

云机器人架构

机器人端 (Edge)          云端 (Cloud)
    ┌──────────────┐         ┌──────────────┐
    │ 传感器采集   │ ──5G─── │ 密集计算     │
    │ 实时控制     │  WiFi   │ SLAM/导航    │
    │ 安全监控     │         │ AI 推理      │
    └──────────────┘         └──────────────┘
          │                         │
          └────── 数字孪生 ─────────┘

关键指标

指标	要求
控制延迟	< 20ms（本地闭环）
感知延迟	< 100ms（云端处理可接受）
带宽需求	压缩后 10-50 Mbps
可用性	> 99.9%

---

ROS2 云边通信

ZeroMQ 桥接

#!/usr/bin/env python3
"""ROS2-云端 ZeroMQ 桥接节点"""

import rclpy
from rclpy.node import Node
from sensor_msgs.msg import Image, PointCloud2, LaserScan
from geometry_msgs.msg import Twist, PoseArray
import zmq
import pickle
import numpy as np
import threading
import asyncio


class ZMQBridge(Node):
    """
    ROS2 ↔ 云端 ZeroMQ 桥接

    功能:
    - 机器人端：将 ROS2 消息压缩后发送到云端
    - 云端：将控制命令发送回机器人端
    """

    def __init__(self, mode: str = "robot"):
        """
        Args:
            mode: "robot" 或 "cloud"
        """
        super().__init__(f'zmq_bridge_{mode}')
        self.mode = mode

        # ZeroMQ 配置
        self.cloud_address = "tcp://cloud.example.com:5555"
        self.robot_address = "tcp://*:5556"

        # 消息压缩器
        self.compressor = MessageCompressor()

        # 带宽限制
        self.max_bandwidth_mbps = 50
        self.current_bandwidth = 0

        if mode == "robot":
            self._init_robot_mode()
        else:
            self._init_cloud_mode()

        # 统计
        self.msg_sent = 0
        self.msg_received = 0
        self.bytes_sent = 0

    def _init_robot_mode(self):
        """机器人端：发布 ROS2 消息到云端"""
        self.ctx = zmq.Context()
        self.socket = self.ctx.socket(zmq.PUB)
        self.socket.connect(self.cloud_address)

        # 订阅的 ROS2 话题（需要发送到云端）
        self.subscribers = {}
        self._create_subscriber(Image, '/camera/image_compressed', self._send_image)
        self._create_subscriber(PointCloud2, '/velodyne_points', self._send_pointcloud)
        self._create_subscriber(LaserScan, '/scan', self._send_laserscan)

        # 接收云端命令
        self.cmd_sub = self.create_subscription(
            Twist, '/cloud_cmd_vel', self._cmd_callback, 10
        )

        self.get_logger().info(f'Robot ZMQ bridge → {self.cloud_address}')

    def _init_cloud_mode(self):
        """云端：接收机器人数据，发送命令"""
        self.ctx = zmq.Context()
        self.socket = self.ctx.socket(zmq.SUB)
        self.socket.bind(self.robot_address)
        self.socket.setsockopt(zmq.SUBSCRIBE, b'')  # 接收所有

        # 启动接收线程
        self.recv_thread = threading.Thread(target=self._recv_loop)
        self.recv_thread.start()

        # 云端控制发布
        self.cloud_cmd_pub = self.create_publisher(Twist, '/cloud_cmd_vel', 10)

        self.get_logger().info(f'Cloud ZMQ bridge ← {self.robot_address}')

    def _create_subscriber(self, msg_type, topic: str, callback):
        sub = self.create_subscription(msg_type, topic, callback, 10)
        self.subscribers[topic] = sub

    def _send_image(self, msg: Image):
        """压缩并发送图像"""
        # 简化：直接 pickle（实际应使用 JPEG/PNG 压缩）
        data = self.compressor.compress_image(msg)
        self._send('image', topic, data, msg.header.stamp)

    def _send_laserscan(self, msg: LaserScan):
        """发送激光扫描"""
        data = self.compressor.compress_laserscan(msg)
        self._send('laserscan', topic, data, msg.header.stamp)

    def _send(self, msg_type: str, topic: str, data: bytes, stamp):
        """发送数据到云端"""
        envelope = {
            'type': msg_type,
            'topic': topic,
            'data': data,
            'stamp': stamp.sec + stamp.nanosec * 1e-9,
            'robot_id': 'robot_001',
        }
        try:
            self.socket.send(pickle.dumps(envelope), flags=zmq.NOBLOCK)
            self.msg_sent += 1
            self.bytes_sent += len(data)
        except zmq.Again:
            pass  # 带宽已满，丢弃

    def _recv_loop(self):
        """云端接收循环"""
        while True:
            try:
                msg = self.socket.recv()
                envelope = pickle.loads(msg)
                self._handle_cloud_message(envelope)
            except Exception as e:
                self.get_logger().error(f'Recv error: {e}')

    def _handle_cloud_message(self, envelope: dict):
        """处理来自机器人的消息"""
        self.msg_received += 1

        if envelope['type'] == 'laserscan':
            # 云端 SLAM 处理
            cloud_result = self.process_slam(envelope['data'])
            # 发布云端处理结果
            # ...

    def _cmd_callback(self, msg: Twist):
        """接收云端命令并转发到 /cmd_vel"""
        # 实际应用中转发到本地控制器
        pass


class MessageCompressor:
    """消息压缩器"""

    @staticmethod
    def compress_image(msg: Image, quality: int = 85) -> bytes:
        """JPEG 压缩图像"""
        import cv2
        from cv_bridge import CvBridge
        bridge = CvBridge()
        img = bridge.imgmsg_to_cv2(msg, desired_encoding='bgr8')
        encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), quality]
        _, buffer = cv2.imencode('.jpg', img, encode_param)
        return buffer.tobytes()

    @staticmethod
    def compress_laserscan(msg: LaserScan) -> bytes:
        """压缩激光扫描"""
        import struct
        data = struct.pack(f'{len(msg.ranges)}f', *msg.ranges)
        return data


class AdaptiveBandwidthController:
    """自适应带宽控制器"""

    def __init__(self, target_mbps: float = 30.0):
        self.target_mbps = target_mbps
        self.current_compression_quality = 85

    def adjust(self, actual_mbps: float):
        """根据实际带宽调整压缩质量"""
        if actual_mbps > self.target_mbps * 1.1:
            self.current_compression_quality = max(50, self.current_compression_quality - 5)
        elif actual_mbps < self.target_mbps * 0.9:
            self.current_compression_quality = min(95, self.current_compression_quality + 5)
        return self.current_compression_quality

---

数字孪生

ROS2 数字孪生节点

#!/usr/bin/env python3
"""数字孪生节点"""

import rclpy
from rclpy.node import Node
from geometry_msgs.msg import TransformStamped, Pose
from sensor_msgs.msg import JointState
from nav_msgs.msg import Odometry
import numpy as np
from dataclasses import dataclass, field
from typing import Dict, List
import json


@dataclass
class TwinState:
    """数字孪生状态"""
    timestamp: float
    robot_id: str
    position: np.ndarray = field(default_factory=lambda: np.zeros(3))
    orientation: np.ndarray = field(default_factory=lambda: np.zeros(4))  # quaternion
    joint_positions: np.ndarray = None
    velocities: np.ndarray = None
    battery_level: float = 1.0
    task_state: str = "idle"


class DigitalTwin(Node):
    """数字孪生系统"""

    def __init__(self):
        super().__init__('digital_twin')

        self.twins: Dict[, TwinState] = {}
        .sync_period =   

        
        .create_subscription(
            Odometry,
            ,
             msg: ._update_twin(, msg),
            
        )
        .create_subscription(
            JointState,
            ,
             msg: ._update_joints(, msg),
            
        )

        
        .twin_state_pub = .create_publisher(
            Pose,
            ,
            
        )

        
        .create_timer(.sync_period, ._sync_to_cloud)

        
        .cloud_client = CloudSyncClient()

        .get_logger().info()

     ():
        
         robot_id   .twins:
            .twins[robot_id] = TwinState(
                timestamp=.get_clock().now().seconds_nanoseconds()[] * ,
                robot_id=robot_id
            )

        twin = .twins[robot_id]
        twin.position = np.array([
            odom.pose.pose.position.x,
            odom.pose.pose.position.y,
            odom.pose.pose.position.z,
        ])
        twin.orientation = np.array([
            odom.pose.pose.orientation.x,
            odom.pose.pose.orientation.y,
            odom.pose.pose.orientation.z,
            odom.pose.pose.orientation.w,
        ])
        twin.timestamp = odom.header.stamp.sec + odom.header.stamp.nanosec * 

     ():
         robot_id  .twins:
            .twins[robot_id].joint_positions = np.array(joint_state.position)

     ():
        
         robot_id, twin  .twins.items():
            state_json = {
                : twin.robot_id,
                : twin.timestamp,
                : twin.position.tolist(),
                : twin.orientation.tolist(),
                : twin.task_state,
            }
            .cloud_client.publish(, json.dumps(state_json))

     () -> TwinState:
        
         .twins.get(robot_id)

---

云端计算卸载

卸载决策器

import numpy as np
from typing import Tuple, Optional


class ComputationOffloader:
    """
    计算卸载决策器

    决定哪些计算在本地执行，哪些卸载到云端
    """

    def __init__(self):
        # 本地计算能力 (MIPS)
        self.local_mips = 10000

        # 云端计算能力 (相对值)
        self.cloud_mips_ratio = 10.0  # 云端是本地的 10 倍

        # 网络状况
        self.bandwidth_mbps = 50.0
        self.latency_ms = 20.0

        # 任务阈值
        self.compute_threshold = 1000.0  # MIPS
        self.latency_threshold = 50.0  # ms

    def should_offload(
        self,
        task_compute_mips: float,
        data_size_mb: float,
        latency_budget_ms: float
    ) -> Tuple[bool, str]:
        """
        决策是否卸载

        Args:
            task_compute_mips: 任务计算量 (MIPS)
            data_size_mb: 数据大小 (MB)
            latency_budget_ms: 延迟预算 (ms)

        Returns:
            (should_offload, reason)
        """
        # 估计本地执行时间
        local_time = task_compute_mips / self.local_mips  # ms

        
        transfer_time = (data_size_mb * ) / .bandwidth_mbps  

        
        cloud_time = task_compute_mips / (.local_mips * .cloud_mips_ratio)

        
        total_offload_time = transfer_time + cloud_time + .latency_ms

        
         total_offload_time > latency_budget_ms:
            
             , 

        
         task_compute_mips < .compute_threshold:
             , 

        
        speedup = local_time / total_offload_time
         speedup > :
             , 
        :
             , 


 :
    

     ():
        .offloader = ComputationOffloader()

        
        .scan_compute_mips =     
        .scan_data_mb =           
        .loop_close_compute_mips =   
        .loop_close_data_mb =     

     ():
        
         is_loop_closure:
            should_offload, reason = .offloader.should_offload(
                .loop_close_compute_mips,
                .loop_close_data_mb,
                latency_budget_ms=
            )
             should_offload:
                 ._cloud_loop_closure(scan_data), 
            :
                 ._local_loop_closure(scan_data), 
        :
            should_offload, reason = .offloader.should_offload(
                .scan_compute_mips,
                .scan_data_mb,
                latency_budget_ms=
            )
             should_offload:
                 ._cloud_scan_process(scan_data), 
            :
                 ._local_scan_process(scan_data), 

     ():
        
        cloud_result = ._send_to_cloud(, scan_data)
         cloud_result

     ():
        
         scan_data  

     ():
        cloud_result = ._send_to_cloud(, scan_data)
         cloud_result

     ():
         scan_data

     ():
        
        
         data

---

云端导航规划

class CloudNavigationPlanner:
    """
    云端导航规划器

    机器人在本地做感知和局部控制，
    云端做全局规划和长期路径优化
    """

    def __init__(self):
        self.planner_type = "hybrid_astar"  # 混合 A*

    def plan_cloud_path(
        self,
        start: Tuple[float, float],
        goal: Tuple[float, float],
        costmap: np.ndarray,
        environment_model: dict = None
    ) -> List[Tuple[float, float]]:
        """
        云端全局路径规划

        优势:
        - 可使用更大的地图
        - 可用更多计算资源做优化
        - 可整合多机器人信息
        """
        if self.planner_type == "hybrid_astar":
            return self._hybrid_astar(start, goal, costmap)
        elif self.planner_type == "topomap":
            return self._topological_planning(start, goal, environment_model)
        else:
            return self._astar(start, goal, costmap)

    def _hybrid_astar(self, start, goal, costmap):
        """
        混合 A* 算法（适合车辆动力学约束）

        Returns:
            路径点列表 [(x, y), ...]
        """
        # 简化实现
        import heapq

         :
             ():
                .x, .y = x, y
                .g, .h = g, h
                .f = g + h
                .parent = parent

             ():
                 .f < other.f

        open_set = [State(start[], start[], h=._heuristic(start, goal))]
        came_from = {}
        visited = ()

         open_set:
            current = heapq.heappop(open_set)

             ._is_goal(current, goal):
                 ._reconstruct_path(came_from, current)

            visited.add((current.x, current.y))

             dx, dy  [(,), (,), (,-), (-,), (,)]:
                nx, ny = current.x + dx, current.y + dy
                 (nx, ny)   visited  ._in_bounds(nx, ny, costmap):
                    g = current.g + np.sqrt(dx** + dy**)
                    h = ._heuristic((nx, ny), goal)
                    neighbor = State(nx, ny, g, h, current)
                     costmap[ny, nx] < :  
                        heapq.heappush(open_set, neighbor)

         [start, goal]  

     ():
         np.sqrt((a[]-b[])** + (a[]-b[])**)

     ():
         (state.x - goal[]) <   (state.y - goal[]) < 

     ():
          <= x < costmap.shape[]   <= y < costmap.shape[]

     ():
        path = []
         current:
            path.append((current.x, current.y))
            current = came_from.get((current.x, current.y))
         path[::-]

---

远程操控

class TeleoperationBridge(Node):
    """远程操控桥接"""

    def __init__(self):
        super().__init__('teleop_bridge')

        self.declare_parameter('cloud_url', 'wss://cloud.example.com/teleop')
        self.declare_parameter('video_quality', 70)
        self.declare_parameter('command_rate', 50)  # Hz

        # 视频压缩
        self.video_compressor = VideoCompressor(
            quality=self.get_parameter('video_quality').value
        )

        # 云端 WebSocket
        self.ws_client = WebSocketClient(
            self.get_parameter('cloud_url').value
        )
        self.ws_client.on_message = self._on_command_received

        # 视频发布（机器人端）
        self.video_pub = self.create_publisher(
            Image,
            '/teleop/video',
            10
        )

        # 命令订阅（云端）
        self.cmd_pub = self.create_publisher(
            Twist,
            '/teleop/cmd_vel',
            10
        )

        # 命令速率限制
        self.last_cmd_time = 0
        .cmd_interval =  / .get_parameter().value

        .get_logger().info()

     ():
        
        now = .get_clock().now().seconds_nanoseconds()[] * 
         now - .last_cmd_time < .cmd_interval:
            

        cmd = Twist()
        cmd.linear.x = message.get(, )
        cmd.linear.y = message.get(, )
        cmd.angular.z = message.get(, )
        .cmd_pub.publish(cmd)
        .last_cmd_time = now

     ():
        
        compressed = .video_compressor.compress(frame)
        .ws_client.send({: , : compressed})

---

故障排查

问题	原因	解决方案
控制延迟过高	网络带宽不足	降低视频质量，增加本地预测
数字孪生不同步	网络中断	添加离线缓冲，重连后补发
云端 SLAM 失败	数据包丢失	添加 FEC 前向纠错
卸载决策不准	模型过时	实时测量网络状况更新模型
远程操控卡顿	视频延迟 > 控制延迟	视频降帧率，优先保证控制

调试命令

# 测量网络延迟
ping cloud.example.com

# 测量可用带宽
iperf3 -c cloud.example.com

# 查看 ZMQ 桥接状态
ros2 topic list | grep zmq

# 录制云边通信数据
ros2 bag record /twin/state /cloud_cmd_vel -o cloud_robotics_data

---

相关技能

• system-integration/ros2-communication — ROS2 通信基础
• navigation/nav2-integration — Nav2 导航集成
• perception/edge-inference — 边缘推理
• edge-platforms/edge-deployment — 边缘部署