启元 PrimeBot Q1 Python 示例离线解读：BMS 触摸相机逐行讲

摘要

没有真机，我把官方这三个 Python 示例当源码文本读。头一遍读 get_bms_state.py 就栽了——voltage 我当成伏特，对着 aimdk_msgs/msg/Bms 的定义才对上：单位是毫伏，得除以 1000。这篇是读 BMS 订阅、TouchState 订阅与 CaptureJpegImage 相机服务三个示例时记的笔记：我为什么在自环里从默认 QoS 换到 sensor_data、单位换算的坑埋在哪些字段、ros2 topic pub 怎么把订阅回调在离线环境跑活。环境为 Ubuntu 22.04 + ROS 2 Humble，结论全部来自官方源码与随包《接口说明》的离线分析，未连接真机。

📌 版本声明：本文基于 PrimeBot Q1 SDK（aimdk_msgs / primebot-q1-sdk）与随包《接口说明》编写，运行环境为 Ubuntu 22.04 + ROS 2 Humble；全部结论来自官方示例源码与协议定义的离线分析，未连接真机。接口版本以文档标注为准（TouchState 1.0 Stable、CaptureJpegImage 0.1 Beta），其他 SDK 版本的字段与 QoS 标注可能有差异。

一、问题定义：没有真机，示例要读到哪一层才算数

官方示例教程的标准动作是老几样：装依赖、跑脚本、看日志出没出。这条路线能证明"代码能跑"，却回答不了三个更关键的问题：回调是谁在什么时候调起来的？QoS 为什么必须是这一个？字段打印前为什么除了以后才能看？等你想把示例改成自己的业务逻辑——比如"电量低于 20% 触发语音告警"“拍一下头就切换动作”——这三个问题会全部变成拦路虎。

更现实的困境是真机不在手边。我这边就是这样，机器人不在，示例连"跑起来看输出"都做不到，很多人的示例学习于是停留在"收藏"阶段。我换了一条路：把三个官方示例当成源码文本来精读，逐行讲清订阅与服务两类通信模型的构造方式、字段语义与单位陷阱，再用 ROS 2 自带的命令行工具在纯离线环境里把订阅端代码"跑活"。

三个示例的选择不是随意的：get_bms_state.py 代表周期式状态订阅，触摸脚本代表触发式事件订阅，get_jpg.py 代表服务调用。这三类恰好覆盖启元 Q1 对外接口的全部通信语义——官方几十个接口里，Topic 与 Service 的基本功都在这里。读懂这三个，其余接口只是换个话题名和字段名，模型不变。

本文所有结论都来自一手材料：research/ 目录下的官方示例源码，以及随包《接口说明》的协议定义章节。凡涉及数值（频率、QoS、单位、超时、时延）均直接引自原文，不做推测；个别文档与示例不一致的地方，本文会明确指出来源差异并给出分析，而不是含糊带过。

还要先说清本文的边界：离线精读能回答"代码为什么这样写"，自环验证能回答"我的改动对不对"，但两者都回答不了"机器人实际会怎么响应"。后者依赖硬件行为，任何脱离真机的结论都不该替它打包票。把这三层分清楚，就不会在方案评审时做出过度承诺——这也是本文所有章节默认的前提。

二、通信模型先立骨架：订阅与服务两条主线

2.1 订阅型接口：被动接收的状态流

订阅（Subscription）是被动接收模型。代码只做三件事：指定消息类型、指定话题名、登记一个回调函数。之后消息什么时候到、到不到，完全由 ROS 2 的发现机制与执行器（Executor）决定。对开发者而言，这就像订报纸：邮递员把报纸塞进信箱，你取报时读内容，从不需要去印刷厂排队。

但两个订阅示例的"邮递频率"完全不同。BMS 电源话题的文档元数据明确标注频率 1 Hz——即使电池状态毫无变化，每秒也会推一条全量状态，属于周期式上报。触摸话题标注的是触发式上报：没有触摸就不产生任何消息，属于事件驱动。这个差异直接决定回调里能做什么假设：对 BMS 你可以放心地每秒刷新一次界面；对触摸，静默本身就是正常状态，绝不能因为"三秒没收到消息"就判定传感器掉线。

状态类接口偏爱订阅，原因在于状态的价值在"最新值"：丢一条旧的不可惜，下一条马上补上。这也解释了为什么 BMS 可以用 Best Effort 级别的 QoS（详见 3.2）。反过来说，如果你需要的是"某个时刻的确定结果"，订阅就力不从心了——你永远不知道想要的那一条会不会来、什么时候来。

再补一个容易忽略的概念：Volatile 耐久性。它意味着发布端不为迟到的订阅者保留历史消息——如果你在机器人已经运行半小时后才启动订阅脚本，不会立刻收到一条 BMS 状态，而要等下一秒的新消息到达。对 1 Hz 周期话题这无所谓，顶多等一秒；但 Volatile 用在需要"一问即答"的场景，就会误以为"没数据就是没发布"。耐久性与可靠性是两个独立维度，选 QoS 时要分别核对。

2.2 服务型接口：一问一答的同步交互

服务（Service）是一问一答模型。客户端构造请求、发出、等待、拿到响应，交互有明确的起止。相机单帧抓取正是典型场景：你需要的是"按下快门那一刻"的画面，而不是一条永远在流动的视频流。这个语义用订阅拿不到——Topic 只保证持续流动，不保证"此刻这一帧"。

服务调用在代码层面比订阅重：要创建客户端（Client）、等待服务上线、构造请求、异步发出、把 future 交给执行器直到响应回来、按三段式规则判定成败。官方相机示例把这一整套都示范了，是三个示例里代码最长、也最贴近真实工程的一个（详见第五节）。

服务的约束也很明确：它是同步占用的，一次响应完成前不该发第二次请求。相机文档因此建议取帧频率不超过 1 fps，并提示全链路时延约 100~500 ms。把服务当话题轮询，是新手最常见的误用——那会把一问一答的同步接口拖成性能瓶颈。

2.3 Node / Subscription / Client 的生命周期

三个示例共享同一套生命周期骨架，我先把这骨架立起来，后面的逐行笔记才有挂靠的地方。

入口都是 rclpy.init(args)：它初始化 ROS 2 客户端库、解析命令行参数，必须且只能调用一次，之后才能创建节点。接着 Node(“名字”) 把节点注册进 ROS 图（ROS graph），节点名在同一域内应当唯一，否则 ros2 node list 会出现重名，日志难以归属。

然后是差异所在：订阅示例调 create_subscription(类型, 话题, 回调, QoS) 拿到 Subscription 句柄；服务示例调 create_client(类型, 服务名) 拿到 Client 句柄。这两个句柄都必须挂在 self 上持有——rclpy 的句柄依赖 Python 对象存活，如果只写 self.create_subscription(…) 而不赋值，极端情况下句柄被垃圾回收，订阅会在毫无报错的情况下静默消失。这是纯源码阅读才能发现的坑，只把示例跑通的人反而容易忽略。

收尾部分，rclpy.spin(node) 是阻塞式事件循环，单线程执行器在这里不断取出队列里的消息调用回调；spin_until_future_complete 则一边循环一边等某个 future 完成。无论哪种，都必须放在 try/finally 里，保证 destroy_node() 与 rclpy.shutdown() 一定执行，否则 DDS 资源不会干净释放。注意一个反直觉点：call_async 只是把请求发出去并返回 future，响应消息必须由执行器投递——如果 call_async 之后用 time.sleep 傻等而不 spin，future 永远不会完成。

线程模型还有个细节：rclpy.spin 使用单线程执行器，所有回调在同一个线程里串行执行。这意味着回调里不要做耗时的活——如果在触摸回调里同步调用一次相机服务并傻等六秒，同节点的 BMS 回调就会被一起堵住。需要并发时可以换 MultiThreadedExecutor 或把重活丢给独立线程，但那是改造示例架构的话题，本文以读懂官方写法为先。

读这三个示例之前，我先把它们的归口画成了一张图，后文的逐行笔记都拿它当地图。

图说明：两个订阅节点的数据流终点是各自的 callback，服务节点的终点是响应判读；三条链唯一的区别就在中间那一跳是话题还是服务。记住这张图，后面所有代码都只是在给这条链补细节。

三个示例分别教会我什么，我先给个判断：BMS 示例教会我周期式订阅怎么登记、毫伏毫安怎么换算；触摸示例教会我触发式语义下静默是正常态；相机示例教会我服务客户端从等待到判定的完整骨架。全维度对比放在下面，当速查卡用。

维度	get_bms_state.py	get_touch_state.py	get_jpg.py
通信模型	Topic 订阅	Topic 订阅	Service 调用
接口名	/aima/hal/bms/state	/aima/hal/touch/state	/aima/hal/camera/CaptureJpegImage
数据类型	aimdk_msgs/msg/Bms	aimdk_msgs/msg/TouchState	aimdk_msgs/srv/CaptureJpegImage
上报节奏	周期 1 Hz	触发式，有事件才发	一问一答，建议不超过 1 fps
文档 QoS	Best Effort / Volatile / Depth=10	Reliable / Volatile / Depth=1	Reliable（服务默认）
示例 QoS	qos_profile_sensor_data（Best Effort）	qos_profile_sensor_data	服务默认 QoS
关键字段	voltage/current/power（除以 1000）	sensor_id=1、event_type=1	camera_id、timeout_ms、jpeg.image.data
无真机验证	ros2 topic pub 自环	ros2 topic pub 发一次	起服务桩回包

图1：BMS 订阅、触摸订阅与相机服务三类示例的通信模型定位

三、BMS 示例逐行解读：订阅登记、QoS 与字段换算

3.1 逐行读 get_bms_state.py

我把 SDK 拉下来第一件事，就是把 get_bms_state.py 从头到尾读一遍。它全长不到九十行，却把订阅模型的所有要素都演示齐了。我分四步读它：导入与依赖、节点与订阅登记、回调里的字段换算、主函数的生命周期。

#!/usr/bin/env python3
import rclpy
from rclpy.node import Node
from rclpy.qos import qos_profile_sensor_data
from aimdk_msgs.msg import Bms

class BmsStateEcho(Node):
    def __init__(self):
        super().__init__("get_bms_state")
        # ① 登记订阅：消息类型、话题名、回调、QoS 四要素缺一不可
        self.subscription = self.create_subscription(
            Bms,
            "/aima/hal/bms/state",
            self.callback,
            qos_profile_sensor_data,
        )
        self.get_logger().info("Subscribing bms topic: /aima/hal/bms/state")

    @staticmethod
    def _fmt3(value) -> str:
        # ② 统一三位小数，保证日志各列对齐
        return f"{float(value):.3f}"

    def callback(self, msg: Bms):
        # ③ 每条消息触发一次；单位换算集中在这一步完成
        lines = [
            "BmsState received",
            f"  voltage:        {self._fmt3(msg.voltage / 1000.0)} V",
            f"  current:        {self._fmt3(msg.current / 1000.0)} A",
            f"  power:          {self._fmt3(msg.power / 1000.0)} W",
            f"  temperature:    {self._fmt3(msg.temperature)} °C",
            f"  remaining_capacity: {msg.remaining_capacity} mAh",
            f"  remaining_capacity_percentage: {int(msg.remaining_capacity_percentage)} %",
            f"  cycle_count:    {msg.cycle_count}",
            f"  cycle_total_capacity: {self._fmt3(msg.cycle_total_capacity / 1000.0)} Ah",
        ]
        self.get_logger().info("\n".join(lines))

def main(args=None):
    rclpy.init(args=args)
    node = BmsStateEcho()
    try:
        rclpy.spin(node)   # ④ 阻塞自旋：回调由 Executor 驱动
        return 0
    finally:
        node.destroy_node()
        rclpy.shutdown()

if __name__ == "__main__":
    raise SystemExit(main())

这个脚本做的事一句话能说完：建立名为 get_bms_state 的节点，向 /aima/hal/bms/state 登记订阅，收到 Bms 消息后在回调里做单位换算并打印。create_subscription 的四个实参依次是消息类、话题名、回调函数、QoS 配置；rclpy.spin 是阻塞式事件循环，没有它回调永远不会被调用；finally 保证 destroy_node 与 shutdown 一定执行。跑之前必须 source install/setup.bash，否则 aimdk_msgs 的 import 直接失败；self.subscription 的赋值也不能省，句柄必须持有。官方示例为突出主干，版本信息那几个字段的打印全省了，完整字段见表 2。

读这个脚本时我当场记下了几点。Bms 来自 aimdk_msgs 而非 ROS 标准包，这是所有 Python 示例的共同前提。节点名与脚本同名，ros2 node list 时一眼能认出。subscription 句柄赋值给 self.subscription，正是 2.3 强调的"必须持有"。还有 _fmt3：三位小数格式化收口到一个静态方法，不用每个字段各写一份格式串——小事，却是示例代码里少见的工程习惯，我自己写脚本时就照这个写法抄。

如果要把这个示例改造成业务代码，我的做法是在回调外层再加两层：一层是节流，把 1 Hz 的原始流按需降采样（比如界面只需 0.2 Hz），避免日志与回调过载；另一层是状态判定，把 remaining_capacity_percentage 与阈值比较后产生告警事件，而不是把换算值直接抛给界面。官方示例到"打印"为止，恰恰是因为再往下的逻辑每个产品都不同——读懂它，将来替换的就是这两层。

3.2 QoS 为什么是 qos_profile_sensor_data

读这个示例时我冒出来的第一个问题是：QoS 为什么偏偏是 qos_profile_sensor_data，不是默认的 qos_profile_default？我一开始没当回事，在自环里按默认 QoS 写订阅端，结果消息发出去了，回调那边安安静静，一条都不来。回去查文档，才把这条因果链补全。

接口文档对 BMS 话题的元数据标注是 Best Effort / Volatile / Depth=10，而 rclpy 内置的 qos_profile_sensor_data 恰好是 Best Effort / Volatile / Keep Last(5)：可靠性与持久性两个关键策略完全一致，只有队列深度从 10 降到 5。

深度差异不影响连通性。ROS 2 的 QoS 配对规则里，KEEP_LAST 只校验历史策略种类是否相同，不比较深度数值，深度只决定各自队列缓存多少条。而可靠性策略的规则是"发布端不得低于订阅端"：Best Effort 发布端可以服务 Best Effort 订阅端，Reliable 发布端也能服务 Best Effort 订阅端；反过来，Best Effort 发布端永远配不上 Reliable 订阅端。

那个经典静默故障就是这么来的：如果你把示例改成 Reliable 订阅，而机器人侧按协议用 Best Effort 发布，两端永远无法配对，回调一条都不会来，控制台连错误都不报。这个故障我在自环里复现过：订阅端写 Reliable，pub 端用 ros2 topic pub 的默认口径，终端 A 就是没输出，拿 ros2 topic info <话题> --verbose 一对比两端策略，原因摆在眼前。这类问题在真机上极难排查，在源码层面却一眼可见。反过来说，Best Effort 适合 BMS，因为它是典型的传感器数据语义：1 Hz 全量刷新，最新值永远在路上，丢一条旧的毫无损失，为重传付出的带宽与 CPU 反而不值。qos_profile_sensor_data 这个名字本身就是这个意思。

还有一点：QoS 不配对不会报错，只会静默。ros2 topic info <话题> --verbose 能列出两端实体的策略，是离线与在线都通用的诊断入口。我的习惯是每次订阅新话题都跑一次这个命令，把"我以为配对了"变成"我确认配对了"。这个习惯的成本是五秒钟，回报是省掉一个可能排查一整天的沉默故障。

3.3 Bms 字段语义与单位换算陷阱

Bms 消息共有十七个字段，分为版本信息、电源状态、状态位域与电量计量四组。真正的陷阱在单位：协议用最小刻度上报整数，换算责任全部留给接收端。

字段	类型	原始单位	换算与处理	说明
manufacturer	string	—	原样输出	厂家信息
serial_number	string	—	原样输出	电池序列号
protocol_version	string	—	原样输出	通信协议版本
hardware_version	string	—	原样输出	硬件版本
software_version	string	—	原样输出	软件版本
power_request	uint16	—	原样输出	1=长按关机请求，2=短按加长按
status	uint32	位域	按位判读	bit0 充电、bit2 放电、bit16/17 禁充禁放等
balance_line_resistance	uint16	mΩ	原样输出	电池包均衡线电阻
voltage	uint32	mV	除以 1000 得 V	电池包总电压，必须换算
current	int32	mA	除以 1000 得 A	有符号，充电为正、放电为负
power	int32	mW	除以 1000 得 W	电池包输出功率
temperature	float32	°C	原样输出	已是浮点，无需换算
remaining_capacity	uint32	mAh	原样输出	当前剩余容量
remaining_capacity_percentage	uint8	%	int() 取整	已是百分数，禁止再除
cycle_count	uint16	次	原样输出	完整循环次数
cycle_total_capacity	uint32	mAh	除以 1000 得 Ah	累计充放电容量

四个字段必须除以 1000：voltage（mV 到 V）、current（mA 到 A）、power（mW 到 W）、cycle_total_capacity（mAh 到 Ah）。示例里 51000 打印成 51.000 伏，而不是 51000 伏。反过来，remaining_capacity_percentage 已经是百分数，示例用 int() 取整而非再除——把这两类字段混为一谈，是 BMS 集成里最高频的事故来源。举个具体后果：低电保护若把阈值写成"小于 20 伏"，而拿到的 voltage 是毫伏，阈值实际变成 20 毫伏，保护永远不会触发，机器人会在真实低压下继续工作。

我头一遍读 get_bms_state.py 时就看走眼了。回调里 voltage、current、power 三行都除了 1000，我下意识以为 temperature 那行没除是漏写，顺手想给它补一个 /1000。真要动手之前，我去对了 aimdk_msgs/msg/Bms 的定义：temperature 是 float32，单位就是 °C，原样输出才对。栽过这一回，我才回头认真看 voltage——协议里它存的是 mV，51000 不是 51000 伏，是 51.000 伏。字段单位这事儿猜不出来，ros2 interface show aimdk_msgs/msg/Bms 一条命令的事，离线就能查。

还有两个符号与位域的细节。current 是 int32 有符号数，协议明确"充电为正、放电为负"——界面上看到负值不是故障，而是在放电；把负号当异常是最常见的误读。status 是 uint32 位域，协议定义了三十个位：bit0 充电标志、bit2 放电标志、bit16 与 bit17 是 MOSFET 禁充禁放（真正故障禁用）、bit20 电池充满。示例只原样打印这个整数，生产代码应当按位拆解后使用。

图说明：两个话题消息都内嵌 MessageHeader，这是 aimdk_msgs 的公共约定；Bms 的计量字段集中在电量组，TouchState 则只有两个业务字段。读 .msg 定义就是读接口契约——字段名、类型、单位三样一个都不能猜，ros2 interface show aimdk_msgs/msg/Bms 在离线环境随时可查，比任何二手资料都可靠。

把"除以 1000"散落在每个回调里容易出错，我的做法是把单位换算与位域判读抽成不依赖 ROS 的纯函数，既方便离线单元测试，也让订阅代码保持干净。

#!/usr/bin/env python3
"""Bms 字段单位换算与 status 位判读助手（无 ROS 依赖，可离线单测）。"""

def to_volts(mv: int) -> float:
    """毫伏到伏特：原始 51000 表示 51.0 V。"""
    return mv / 1000.0

def to_amperes(ma: int) -> float:
    """毫安到安培：协议约定充电为正、放电为负。"""
    return ma / 1000.0

def to_watts(mw: int) -> float:
    """毫瓦到瓦特。"""
    return mw / 1000.0

def to_ampere_hours(mah: int) -> float:
    """毫安时到安时：cycle_total_capacity 累计吞吐量用。"""
    return mah / 1000.0

def is_charging(status: int) -> bool:
    """status bit0 为充电标志，取值来自协议位定义而非推断。"""
    return bool(status & (1 << 0))

def decode_status(status: int) -> list:
    """把 status 位域翻译成人类可读标志列表（节选高信息量位）。"""
    bits = {
        0: "充电中",
        2: "放电中",
        5: "电芯过压",
        6: "电芯欠压",
        16: "MOSFET禁充(故障)",
        17: "MOSFET禁放(故障)",
        20: "电池已充满",
    }
    return [name for bit, name in bits.items() if status & (1 << bit)]

if __name__ == "__main__":
    # 离线自测：51000 mV / -1500 mA / status 低三位为 101
    assert to_volts(51000) == 51.0
    assert to_amperes(-1500) == -1.5
    assert to_watts(-76500) == -76.5
    assert to_ampere_hours(2400000) == 2400.0
    assert is_charging(0b101)
    assert decode_status(0b101) == ["充电中", "放电中"]
    print("all unit conversion self-checks passed")

这段助手代码把 Bms 各原始字段的单位换算与 status 位域判读收口成纯函数，并用 assert 做离线自测。每个换算函数接收硬件原始整数，返回工程单位值；decode_status 按协议位定义把位域翻译成中文标志，bit0 充电、bit2 放电、bit16/17 为 MOSFET 故障禁用位。这些函数无副作用、不 import rclpy，直接用 python3 单独运行或挂进 pytest 都行。换算收口之后，回调里只调用 to_volts 这类函数，能杜绝"某处忘了除 1000"的隐性 bug——这是把示例改造成生产代码的第一步，我自己的工程里也是这么切的。

图2：BMS 消息关键字段的单位换算要点卡

四、TouchState 示例解读：触发式上报怎么写订阅

4.1 逐行读官方触摸订阅脚本

官方触摸订阅示例 get_touch_state.py 与 BMS 示例同构，打印重点完全不同：它不关心数值换算，而是把 MessageHeader 的四个字段完整打出来——frame_id、sequence、stamp 与 meas_stamp。其中 stamp 是消息头时间，meas_stamp 从命名看是传感测量时刻，两者之差可以粗略反映消息从产生到发布的间隔；_to_seconds 把 sec 与 nanosec 合成浮点秒，是 ROS 2 时间处理的标配写法。

#!/usr/bin/env python3
import rclpy
from rclpy.node import Node
from rclpy.qos import qos_profile_sensor_data
from aimdk_msgs.msg import TouchState

class TouchStateEcho(Node):
    def __init__(self):
        super().__init__("get_touch_state")
        # 官方示例同样使用 sensor_data QoS，兼容性分析见 4.2
        self.subscription = self.create_subscription(
            TouchState,
            "/aima/hal/touch/state",
            self.callback,
            qos_profile_sensor_data,
        )
        self.get_logger().info("Subscribing touch topic: /aima/hal/touch/state")

    @staticmethod
    def _to_seconds(stamp) -> float:
        # sec 加 nanosec 除以 1e9，把 ROS 时间还原为秒
        return float(stamp.sec) + float(stamp.nanosec) / 1_000_000_000.0

    def callback(self, msg: TouchState):
        # Q 系列：sensor_id=1 为头部；event_type=1 为单击
        lines = [
            "TouchState received",
            f"  frame_id: {msg.header.frame_id}",
            f"  sequence: {msg.header.sequence}",
            f"  stamp:    {self._to_seconds(msg.header.stamp):.6f} s",
            f"  meas:     {self._to_seconds(msg.header.meas_stamp):.6f} s",
            f"  sensor:   {int(msg.sensor_id)}",
            f"  event:    {int(msg.event_type)}",
        ]
        self.get_logger().info("\n".join(lines))

def main(args=None):
    rclpy.init(args=args)
    node = TouchStateEcho()
    try:
        rclpy.spin(node)
    finally:
        node.destroy_node()
        rclpy.shutdown()

if __name__ == "__main__":
    raise SystemExit(main())

这个脚本建立 get_touch_state 节点订阅触摸话题，收到消息后打印消息头四字段与两个业务字段。_to_seconds 把 Time 的 sec 与 nanosec 合成为浮点秒；sensor_id 标识传感器位置，event_type 标识事件类型。官方对 uint8 字段仍显式 int() 转换，属于防御性写法（rclpy 中定长整型字段本身就是 Python int），可保证日志格式稳定；回调里务必校验 sensor_id，因为不同机型传感器编号可能不同，盲目相信 event_type 会误判。

业务语义只有两个字段。sensor_id 区分传感器位置，Q 系列固定 1 表示头部；event_type 表示事件类型，当前协议只定义了一个有效值 TOUCH_EVENT_SINGLE_CLICK = 1（单击）。这意味着你现在能可靠拿到的事件就是单击，没有长按、双击等枚举；想要更丰富的手势，要么等官方扩展枚举，要么在应用层对多次单击做时序组合——例如记录两次单击的间隔自行判定双击。这是当前版本的能力边界，离线读文档时就要心里有数。

我读触摸示例时对 event_type 犯过嘀咕：一个 uint8 的字段，协议却只给了一个值。转头把《接口说明》翻了一遍，确认当前版本就只有单击这一个枚举，别的一个都没定义。这事对我后来的写法影响很大——双击这类手势别指望协议层，只能自己在应用层攒。

官方把 header 四字段全打出来是有道理的：触发式事件的调试极度依赖时间证据。stamp 与 meas_stamp 的差值反映消息从测量到发布的间隔，sequence 反映是否丢过消息，frame_id 则用于区分数据来源。裸打印 event_type 只能看到"发生了什么"，加上 header 才能回答"什么时候发生、有没有漏"。这是官方示例里我个人最想抄走的工程习惯。

4.2 触发式语义与 QoS 口径差异

这里有一处文档与示例不一致的真实案例，我停下来分析了一通。触摸话题的元数据标注 QoS 为 Reliable / Volatile / Depth=1，但官方示例代码用的却是 qos_profile_sensor_data（Best Effort）。按 3.2 的配对规则，Best Effort 订阅端可以连上 Reliable 发布端（发布端不低于订阅端即可配对），连接能建立，但生效可靠性会降级为 Best Effort。

对触发式话题，这个降级有实际影响：单击事件只发一次，没有"下一条补上"的兜底，极端情况下可能整条丢失。如果你的产品逻辑是"拍头唤醒"这类不可重复交互，建议订阅端按文档元数据显式使用 Reliable，别照抄示例的 sensor_data。Depth=1 的队列同样要注意：回调处理稍慢，队列里最多暂存一条，再快就覆盖旧的——这对"每次单击都必须响应"的场景是合理约束，也提示回调里不要做重活。

触发式上报还有一层语义：没消息不等于没传感器。文档明确"未检测到触摸时不产生消息"，所以离线验证时你必须主动发一条消息才能驱动回调（见 6.3），线上排查时也不能用"话题没流量"判断故障。把这两条想清楚，触摸集成的逻辑才算闭环。

图说明：触摸话题的状态机只有两态，且绝大部分时间停留在静默态。它与 BMS 的周期模型形成鲜明对照，也解释了为什么 6.3 的自环验证只发一次消息。

常量名	数值	描述	适用范围
TOUCH_EVENT_SINGLE_CLICK	1	单击	Q1 头部传感器，触发式上报
未定义	其他值	当前协议未定义	回调应判为无效事件，勿当作长按或滑动处理

五、CaptureJpegImage 服务调用：请求构造与响应判读

5.1 先避一个命名陷阱：服务名、类型与命名空间

相机示例的第一个坑在名字上，还是个容易写错的坑。服务名是 /aima/hal/camera/CaptureJpegImage，属于 /aima/hal/ 命名空间；服务类型是 aimdk_msgs/srv/CaptureJpegImage，是消息包里的类型全名。更要留意的是 /aimdk_5Fmsgs/srv/ 那一套命名空间（_5F 即下划线字符的 ASCII 码 0x5F 转义）——它承载运动控制与交互类服务，如 SetMcAction、PlayAudioFile；相机服务并不在那里。如果把服务名写成 /aimdk_5Fmsgs/srv/CaptureJpegImage，wait_for_service 会永远等待，因为域里根本没有这个名字的服务。

我读相机示例的服务调用结构时，拿它跟同仓库里运动控制类服务的客户端对照着看：同样是 create_client 起客户端、wait_for_service 等服务上线、call_async 发请求、spin_until_future_complete 等 future、两个码判成败，骨架一模一样，变的只有服务名、请求字段和判定码。骨架认出来之后，剩下的都是字段细节——这也让我确认了相机服务的调用结构没有藏着什么新东西，坑只在名字和判读上。

无真机时核对名字的成本极低：ros2 interface show aimdk_msgs/srv/CaptureJpegImage 在纯离线环境即可执行，确认类型定义已安装；ros2 service list 则能验证你自己起的桩是否成功注册。这两个命令都不需要机器人，是服务集成前该养成的习惯。

5.2 客户端逐行解读

官方 get_jpg.py 把客户端做成了可复用类：声明参数、创建客户端、循环等待服务、构造请求、带重试地调用、判读响应、落盘。我抽取主干逐段读。

#!/usr/bin/env python3
import rclpy
from rclpy.node import Node
from aimdk_msgs.msg import CommonRequest, CommonState
from aimdk_msgs.srv import CaptureJpegImage
SERVICE_NAME = "/aima/hal/camera/CaptureJpegImage"
class CameraClient(Node):
    def __init__(self, camera_id: str):
        super().__init__("get_jpg")
        self.camera_id = camera_id
        # ① 创建客户端并等待服务上线（无真机时此处会一直等待）
        self.client = self.create_client(CaptureJpegImage, SERVICE_NAME)
        while not self.client.wait_for_service(timeout_sec=2.0):
            self.get_logger().info(f"Waiting for service: {SERVICE_NAME}")
    def capture_once(self, timeout_ms: int = 0) -> bool:
        # ② 构造请求：CommonRequest 内仅含 RequestHeader.stamp 时间戳
        request = CaptureJpegImage.Request()
        request.request = CommonRequest()
        request.request.header.stamp = self.get_clock().now().to_msg()
        request.camera_id = self.camera_id
        request.timeout_ms = timeout_ms  # 0 表示服务端默认 1000ms
        # ③ 异步调用后把 future 交给执行器，阻塞等待其完成
        future = self.client.call_async(request)
        rclpy.spin_until_future_complete(self, future, timeout_sec=6.0)
        if not future.done():
            self.get_logger().error("CaptureJpegImage timed out.")
            return False
        # ④ 三段式判定：先 header.code==0，再 status.value==1
        response = future.result()
        code = response.response.header.code
        status = response.response.status.value
        if code != 0 or status != CommonState.SUCCESS:
            self.get_logger().error(f"failed code={code} status={status}")
            return False
        jpeg = response.jpeg
        self.get_logger().info(
            f"JPEG bytes={len(jpeg.image.data)} {jpeg.width}x{jpeg.height} "
            f"@{jpeg.framerate}fps")
        return True
def main(args=None):
    rclpy.init(args=args)
    node = CameraClient("head_monocular_centra")
    try:
        node.capture_once()
    finally:
        node.destroy_node()
        rclpy.shutdown()
if __name__ == "__main__":
    raise SystemExit(main())

这段客户端代码建立 get_jpg 节点，等待 CaptureJpegImage 服务可用后构造请求并异步调用，待 future 完成后按框架码加业务码双条件判定成败。wait_for_service 的 timeout_sec 是单次轮询超时，示例用 while 循环反复等；spin_until_future_complete 的 timeout_sec 必须大于服务端处理时长，否则请求还在路上就被本地判超时；timeout_ms 填 0 表示使用服务端默认 1000ms。不 spin 则 future 永不完成；camera_id 需按机型选择，Q1 的交互相机为 head_monocular_centra。

两个细节我当场记了下来。wait_for_service 带超时返回，示例用一个 while 循环反复等——服务没上线时它持续打印等待日志而不是静默卡死，这是客户端应有的兜底。官方脚本用 max(6000, timeout_ms+1000)/1000 计算等待时长，保证本地超时大于服务端可能耗时；本文示例简化为固定 6 秒，思路一致。顺带说一句，请求体很简单：CommonRequest 里只有一个 RequestHeader.stamp 时间戳，业务参数只有 camera_id 与 timeout_ms 两个字段。

判定逻辑上，示例采用"框架码优先、业务码随后"：header.code == 0 表示调度成功，status.value == 1（CommonState.SUCCESS）表示业务成功，两者都过才算真正拿到图。需要提醒的是，官方源码此处写的是 code != 0 且 status != SUCCESS 才判失败（逻辑与），严格来说会漏掉"框架失败但业务成功"的组合；本文示例改用或，任一不过即失败。三段式的完整分析另有专文，这里只需记住：判定响应永远看两个码，不是一个。

参数化与重试也提一下。官方脚本把服务名、超时、输出路径都做成 ROS 参数（declare_parameter），并定义了 SERVICE_CALL_TIMEOUT_SEC 与 MAX_RETRY_COUNT 两个常量，调用超时后按次数重试。这套写法我建议照搬：服务调用默认带超时与重试，请求体重构成参数，客户端才能在换机型、换相机时只改参数不动代码。本文示例为突出主干省略了这些，不等于它们不重要。

5.3 JpegImage 响应字段与工程细节

响应体 jpeg 是 JpegImage 类型，把图像二进制数据与拍摄时的元数据打包返回，字段如下。

字段	类型	说明
jpeg.camera_id	string	实际返回图像的相机 ID
jpeg.device	string	设备节点路径
jpeg.width / jpeg.height	uint32	图像宽高（像素）
jpeg.framerate	uint32	帧率（fps）
jpeg.exposure / jpeg.gain	uint32	曝光值与模拟增益
jpeg.image.data	uint8[]	JPEG 图像二进制数据
jpeg.image.format	string	图像格式，通常为 jpeg

落盘时只有一个硬规矩：image.data 是二进制字节，必须用二进制模式写入。用文本模式写会把字节流按本地编码转换，得到的文件连文件头都不合法。文档给出的量级参考是：常见分辨率下 JPEG 约 50 KB~1.5 MB，全链路响应时延约 100~500 ms，因此建议取帧不超过 1 fps。想要更高帧率，应该走 RTSP 视频流（rtsp://:2554/live_<camera_id>，有线直连时机器人固定 IP 为 10.1.1.100），那是另一条长连接通道，与单帧服务不冲突。

命令行直接调用服务同样可行，适合快速验证服务是否在线、字段是否符合预期。

# 前置：已 source install/setup.bash，aimdk_msgs 类型可用
# 0) 纯离线可执行：确认服务类型已在本地注册
ros2 interface list | grep -i camera
ros2 interface show aimdk_msgs/srv/CaptureJpegImage

# 1) 确认服务已上线（接真机时执行；离线验证自己的桩时同样有效）
ros2 service list | grep camera
ros2 service type /aima/hal/camera/CaptureJpegImage

# 2) 命令行直接调用：请求体为 YAML，字段与 .srv 定义一一对应
#    request.header 内只有 stamp 时间戳；timeout_ms=0 用服务端默认 1000ms
ros2 service call /aima/hal/camera/CaptureJpegImage \
  aimdk_msgs/srv/CaptureJpegImage \
  "{request: {header: {stamp: {sec: 0, nanosec: 0}}}, \
    camera_id: 'head_monocular_centra', timeout_ms: 0}"

# 3) 预期响应（节选）：response.header.code=0、response.status.value=1
#    response.jpeg.camera_id、width、height、framerate、exposure、gain
#    response.jpeg.image.data 为 JPEG 二进制字节流（约 50KB 至 1.5MB）

# 注意：服务名在 /aima/hal/camera/ 下，类型在 aimdk_msgs/srv/ 包下，
#       与 /aimdk_5Fmsgs/srv/（运动控制与交互类服务）不是同一命名空间。

这组命令分三步：先用 ros2 interface show 离线核对协议，再用 ros2 service list 确认服务上线，用 ros2 service call 以 YAML 请求体完成一次真实调用。服务名与类型名分列两行，前者是运行时名字，后者是编译期类型；timeout_ms 填 0 表示采用服务端默认 1000ms，显式填值则覆盖。请求 YAML 的字段名必须与 .srv 定义一致，写错会在调用前报类型错误；该服务为触发式，响应返回前不能发第二次请求。

图3：CaptureJpegImage 服务请求与响应结构示意

六、无真机自环验证：用 ros2 topic pub 驱动真实回调

6.1 原理与前置条件

自环验证的原理一句话能说清：ros2 topic pub 与你的订阅节点跑在同一台机器、同一个 ROS_DOMAIN_ID（默认 0）下，底层走同一套 DDS 发现与传输，发布者发出的消息会被真实投递到订阅端队列，回调被真实触发。这不是"模拟逻辑"，而是除了没有机器人之外全链路都是真的——QoS 配对、序列化、执行器调度，一个环节都不少。

我的操作环境是自己的开发机：Ubuntu 22.04 加 ROS 2 Humble，aimdk_msgs 已编进本地工作空间，官方示例脚本就放在 research/ 目录下。每次验证开三个终端：A 跑被验证的官方脚本，B 用 ros2 topic pub 伪造发布者，C 旁观取证。全程没碰过机器人，走的是本机 DDS 自环。

正因为是全链路，它能验证的东西比想象中多：字段名是否与 .msg 一致（写错会在发布端直接报类型错误）、QoS 是否配对（不配对表现为订阅端静默无输出）、单位换算是否正确（打印值就是最终证据）、事件分支是否按预期走（改字段值即可触发不同分支）。它不能验证的只有一件：机器人会不会真的这样发。这条边界要始终清醒。

前置条件三条：已 source install/setup.bash（aimdk_msgs 可用）；所有终端 ROS_DOMAIN_ID 一致；机器装有 ROS 2 Humble 与 rclpy。官方明确只支持 Ubuntu 22.04 加 ROS 2 Humble 做开发，Windows 与 macOS 不在支持列表内，本文命令均在此环境核对过语法。

消息从伪造发布者到回调函数，中间要经过五跳，时序图把每一跳的责任方画清楚。

图说明：DDS 只负责"把消息送到队列"，执行器负责"从队列取出并调用回调"，两者职责分离。如果订阅端忘了 spin，消息会静静躺在队列里，回调一次都不会执行——这也是 6.5 排错清单里"忘了 spin"这一项的来源。

图4：ros2 topic pub 自环模拟驱动订阅回调的验证拓扑

6.2 BMS 话题自环

BMS 是周期话题，适合持续发布来观察。三个终端各司其职：A 跑被验证的订阅脚本，B 用 ros2 topic pub 伪造发布者，C 旁观取证。

# ============ 无真机自环：伪造 BMS 发布者，驱动真实订阅回调 ============
# 前提：已 source install/setup.bash；三个终端处于同一 ROS_DOMAIN_ID（默认 0）

# 终端 A：启动官方订阅节点（被验证对象）
python3 get_bms_state.py

# 终端 B：用 ros2 topic pub 伪造一条 Bms 消息，按 1Hz 周期发布
# 字段名必须与 aimdk_msgs/msg/Bms 完全一致；整数字段必须给整数
ros2 topic pub /aima/hal/bms/state aimdk_msgs/msg/Bms \
  "{manufacturer: 'AGX', serial_number: 'BMS001', \
    protocol_version: '1.0', hardware_version: 'A', \
    software_version: '1.0', power_request: 0, status: 0, \
    balance_line_resistance: 5, voltage: 51000, current: -1500, \
    power: -76500, temperature: 28.5, remaining_capacity: 12000, \
    remaining_capacity_percentage: 80, cycle_count: 42, \
    cycle_total_capacity: 2400000}" --rate 1

# 终端 C：旁观验证——echo 抽查字段，info 查看 QoS 配对情况
ros2 topic echo /aima/hal/bms/state --field voltage
ros2 topic info /aima/hal/bms/state --verbose

# 预期：终端 A 每秒打印 voltage 51.000 V、current -1.500 A、80 %
# （current 为负表示放电；改成 +1500 再发，可验证充电方向的符号处理）
# 排错：A 无输出时先查 ROS_DOMAIN_ID 是否一致，再用 topic info 看
#       两端 QoS 是否配对，最后才怀疑代码本身。

这组命令里，终端 A 运行真实订阅节点，终端 B 以 1 Hz 向同一话题注入伪造 Bms 消息，终端 C 用 echo 与 info 取证。–rate 1 表示每秒发布一条，贴合 BMS 话题的 1 Hz 频率；voltage 填 51000 表示 51 伏，订阅端会除以 1000 后打印；current 填 -1500 表示以 1.5 A 放电，与 power 为 -76500（51 V 乘 1.5 A）自洽。

我第一次跑这套自环，照的就是这组数：终端 A 打出 voltage 51.000 V、current -1.500 A 的那一刻，除以 1000 的换算在真实回调里对上了，比读十遍代码都直观。后来我把 current 改成 +1500 再发，符号按预期翻正，充电方向的分支也算过了一遍。YAML 里整数字段必须给整数，balance_line_resistance 是 uint16、remaining_capacity_percentage 是 uint8，写成 5.0 或 80.0 会让类型转换直接失败；若订阅端使用 Reliable 而 pub 端为 Best Effort 则无法配对，此时加 --qos-reliability reliable 或改订阅端 QoS。

6.3 触摸话题自环

触摸是触发式话题，不需要周期发布，发一次就够。

# ============ 无真机自环：触发式话题只发一次即可 ============
# 终端 A：启动触摸订阅节点
python3 get_touch_state.py

# 终端 B：先确认三终端处于同一 ROS 域，再发一条头部单击事件
echo $ROS_DOMAIN_ID              # 三个终端应输出相同值，默认均为 0
ros2 topic list | grep touch     # 订阅节点起来后应能看到该话题

# sensor_id=1 为 Q 系列头部；event_type=1 为 TOUCH_EVENT_SINGLE_CLICK
# -1 表示只发布一次后退出（触摸是事件驱动，不是周期流）
ros2 topic pub /aima/hal/touch/state aimdk_msgs/msg/TouchState \
  "{header: {frame_id: 'touch_head', sequence: 1, \
    stamp: {sec: 0, nanosec: 0}, meas_stamp: {sec: 0, nanosec: 0}}, \
    sensor_id: 1, event_type: 1}" -1

# 终端 C：确认消息确实流经过话题（触摸平时无流量，需发布时才有端点）
ros2 topic info /aima/hal/touch/state --verbose

# 预期：终端 A 打印 TouchState received，sensor: 1、event: 1
# 反向验证：把 event_type 改成 2 再发一次，回调应判为无效事件，
#           借此确认代码对协议边界的处理，而非只测正常路径。

这组命令用 ros2 topic pub 加 -1 向触摸话题发一条单击事件，验证订阅回调的 sensor_id 与 event_type 分支。-1 表示只发布一次，因为触摸是事件驱动而非周期流；sensor_id 填 1、event_type 填 1 是 Q 系列头部单击的约定值；header 四字段按 MessageHeader 定义完整给出。跑完正常路径我还多做了一步：把 event_type 改成协议未定义的值再发一次，确认回调把它判为无效事件；把 sensor_id 改成 2，则验证非头部传感器的处理逻辑。只测正常路径的验证等于没验，协议边界也得走一遍。

6.4 服务调用怎么验证：起一个桩

服务没法用 topic pub 模拟，但验证思路完全一致：起一个最小的服务桩。在同一域里写一个二十行左右的 Python 服务端，注册同名服务，收到请求就回一段固定的 JpegImage——把 header.code 置 0、status.value 置 1、image.data 填一段伪造字节——然后跑官方客户端，就能把"请求构造、等待、判定、落盘"整条链路走通。再把桩的返回码改成非零，即可验证失败分支。说白了，订阅用伪造发布者，服务用桩回包，本质都是用确定性输入驱动真实代码。

桩的构造比想象中简单：服务端注册同名服务后，在回调里填充 CaptureJpegImage.Response——response.header.code 置 0、response.status.value 置 CommonState.SUCCESS、jpeg.camera_id 回填请求值、jpeg.width 与 height 填任意整数、jpeg.image.data 填一段固定字节（哪怕只有几十字节，客户端也只会校验非空）。把桩的 code 改成非零再跑一次，就能看到失败分支的日志；把 data 置空，则验证空数据保护。我的做法是三种输入各跑一遍——正常码、失败码、空数据——客户端的判定逻辑才算测全。

6.5 排错清单

按发生频率排个序。最常见的是 ROS_DOMAIN_ID 不一致，消息根本不相遇，先 echo $ROS_DOMAIN_ID。接着是 aimdk_msgs 没 source，import 或 pub 直接报类型不存在。再往下才是代码问题：QoS 不配对，用 ros2 topic info <话题> --verbose 对比两端策略；字段名或类型写错，用 ros2 interface show <类型> 逐字核对；忘了 spin，future 永远不完成。前两条是环境问题，后三条才是代码问题——自环验证的价值，正是把这两类干净地分开。

七、适用边界与风险提示

QoS 是订阅型接口的命门，不可互换：BMS 走 Best Effort，触摸文档标注 Reliable；图省事统一用默认 Reliable 订阅，会让 BMS 静默失联。频率假设要谨慎：BMS 稳定 1 Hz，可放心做周期处理；触摸触发式，静默是正常态，需要自建超时与去抖逻辑。

单位陷阱贯穿全文：所有带毫前缀的字段（mV、mA、mW、mAh）都需除以 1000，唯独 remaining_capacity_percentage 已是百分数，切勿二次处理。相机服务是"奢侈品"：100~500 ms 时延加 1 fps 上限，不适合实时视觉闭环，连续图像应转 RTSP 视频流。

还要清醒认识自环的边界：它验证的是你的代码，不是机器人行为；真机是否按协议发布、服务是否按预期抓帧，仍然只能靠实机联调。此外接口版本标注不同——TouchState 为 1.0 Stable，CaptureJpegImage 为 0.1 Beta——相机接口仍在演进，升级 SDK 时需复核协议定义。

版本与命名的一致性风险也要提醒。本文引用的字段与 QoS 来自随包《接口说明》与官方示例源码，SDK 升级后 .msg 定义可能增减字段、默认值可能调整；我的做法是把字段换算与判定逻辑封装成独立函数并配离线自测（如 3.3 的 assert 组），升级时重跑一遍自测即可快速发现不兼容。永远不要把自己系统里的字段口径与 SDK 解耦的希望，寄托在"示例应该不会变"上面。

八、总结

回到开头的问题：没有真机，示例能读到哪一层？我的答案是"逐行可信、改动有据"。三个示例在我手里的分工很明确：get_bms_state.py 负责让我看懂周期式订阅怎么登记、毫伏毫安的换算责任在谁；触摸示例让我记住触发式语义下静默即正常，协议边界内目前只有单击一个事件；相机示例让我看懂服务客户端从 wait_for_service 到双码判定的完整骨架。

这一趟读下来，我栽过的跟头回头看都是离线就能躲过去的坑。头一个是 QoS：自环里按默认 Reliable 写订阅端，连 Best Effort 的发布端，静默得一条错误都不报，从此每订阅一个新话题都先 ros2 topic info --verbose 确认配对。还有一个是单位：想给 temperature 补除 1000，幸亏动手前对了 .msg 定义。自环验证是我这段时间最依赖的手段——同一 DDS 域里 pub 一条伪造 Bms 消息，51.000 V 从真实回调里打出来那一刻，这段代码才算读通。

剩下的唯一盲区是机器人本身的行为，那必须留给实机联调。如果只做一件事，我会先把单位换算抽成 3.3 那样的纯函数挂进自测——SDK 升级时重跑一遍 assert，比任何人工 review 都靠得住。把示例读透，再把自环跑通，集成就成功了一半；另一半，拿着这份逐行地图去做实机联调，不再盲跑。

参考资料

1. 启元机器人开发者社区：https://dev.primebot.com （接口协议与示例说明）
2. PrimeBot Q1 SDK 开源仓库：https://github.com/PrimebotTech/primebot-q1-sdk （aimdk_msgs 消息与服务定义）
3. 智元开源社区：https://open.agibot.com （示例脚本与开发者讨论）
4. ROS 2 Humble QoS 策略文档：https://docs.ros.org/en/humble/Concepts/About-Quality-of-Service-Settings.html （可靠性与配对规则）

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/sinat_41617212/article/details/168767646