动作通信：等刷卡这件事交给 Action

01跑出现象：一步步跑通 RFID 读卡

先别急着啃原理和读代码——按命令卡把"自检 → 服务端 → 客户端 → CLI 观察反馈流"四个现象亲手跑出来，特别是三种结局（成功 / 超时 / 取消）都要亲眼看到一遍。现象对上了，再带着「为什么会这样」的疑问进入后面的原理与 04 实验步骤，每一行你都知道自己在学什么。

# ① 进工作空间，并先加载 ROS2 本体（underlay）
cd ~/ros2_ws
source /opt/ros/humble/setup.bash
# ② 只编译本实验需要的两个包：接口包 + 节点包（顺序由 colcon 按依赖自动排）
colcon build --packages-select ros_sensor ros_sensor_interfaces
# ③ 把工作空间注册进当前终端（新终端必做，不能省）
source install/setup.bash

# 把 SPI 设备交给 dialout 组（iflytek 已在该组内）
sudo chgrp dialout /dev/spidev1.0
# 组内成员可读写该设备（660 = 属主读写 + 组读写）
sudo chmod 660 /dev/spidev1.0

# 先把这两个引脚"导出"到用户空间，才会出现 gpio114/116 目录
sudo sh -c 'echo 114 > /sys/class/gpio/export; echo 116 > /sys/class/gpio/export'
# 放开 114 的方向与电平文件：666 让任何用户都能读写
sudo chmod 666 /sys/class/gpio/gpio114/direction /sys/class/gpio/gpio114/value
# 同样放开 116，复位引脚也是普通用户要拉的
sudo chmod 666 /sys/class/gpio/gpio116/direction /sys/class/gpio/gpio116/value

3脱离 ROS 自检 MFRC522终端 A

SPI 接线、GPIO、供电问题都会在这一步暴露，不会污染 ROS 层的判断。权限配好之后，这条命令不再需要 sudo。

# 进驱动目录，跑脱离 ROS 的裸硬件自检
cd src/xflz_eai_car1/driver/ros_sensor/ros_sensor/ros_sensor
# 自检 MFRC522：读版本寄存器 + 回读验证，PASS 即硬件通路正常
python3 rc522_reader.py --test

$ python3 rc522_reader.py --test RC522 RFID Reader on RK3588 SPI : /dev/spidev1.0 CS GPIO : 114 RST GPIO : 116 Speed : 500000 Hz -------------------------------------- VersionReg(0x37) = 0xB2 TmodeReg write 0x8D -> read 0x8D PASS 初始化完成

✔ VersionReg 读出 0x00 或 0xFF = 芯片没应答：查 CS=114 / RST=116 接线、3.3V 供电、python-periphery 是否安装。若报 Permission denied，说明步骤 2 的权限没配好或重启后失效了。

cd ~/ros2_ws
source install/setup.bash
ros2 run ros_sensor rfid_action_server

5发起读卡目标终端 C

参数 30 是 Goal 里的 timeout_sec。启动后把 MIFARE 白卡贴上天线，盯住反馈流的变化。

cd ~/ros2_ws
source install/setup.bash
ros2 run ros_sensor rfid_action_client 30

[INFO] Goal 已接受 Feedback: SCANNING Feedback: SCANNING Feedback: CARD_FOUND Feedback: READING_UID Feedback: VERIFYING 结果: success=True uid=7E1A3B02 type=MIFARE Classic msg=读卡成功

✔ 反馈流从 SCANNING 一路推进到 VERIFYING——这就是 03 节图 8-3 状态机的实况。服务端终端同时会打印收到的 Goal 与读卡结果。

ros2 action list
ros2 action info /rfid_scan

ros2 action send_goal /rfid_scan ros_sensor_interfaces/action/RfidScan "{timeout_sec: 30}" --feedback

# 结局 2 · 超时 abort：故意不放卡，等 timeout_sec 秒
# 与上一步唯一区别：把 30 改成 10，这样不用干等半分钟
ros2 action send_goal /rfid_scan ros_sensor_interfaces/action/RfidScan "{timeout_sec: 10}" --feedback
# → SCANNING 持续滚动 10s → Result: success=False msg=读卡超时

# 结局 3 · 取消 canceled：上一条跑起来后按 Ctrl+C（CLI 会发 cancel）
# 这次 timeout_sec 写大一点（如 60），才有时间在中途按 Ctrl+C
# → Feedback 中断 → Result: success=False msg=已取消

02问题引入：刷卡要等 30 秒，为什么服务不够用

你刚跑通了指纹验证（EXP-07）——那是"一问一答、几秒出结果"的典型。现在换个真实场景：给实验室装一套 RFID 门禁考勤。

这正是机器人的常态：导航到目标点、机械臂抓取、读卡等待——这些任务都耗时长、进度不确定、用户可能中途改主意。它们和"传感器每秒发一次数据"（话题）不一样，也和"问一句答一句"（服务）不一样。

那用上一课的服务来做"等刷卡"行不行？直接撞上三堵墙：

03核心原理：五条隐藏通道 · 时序 · 三回调与状态机

你在代码里写的 ActionClient/ActionServer 各只是一个端点，但 ROS2 在底下悄悄建了 5 条线：3 条服务线（发 Goal、发 Cancel、取 Result）+ 2 条话题线（Feedback 流、Status 流）。也就是说——你已经把前面三课全用上了：动作 = 服务的配对语义 + 话题的流式能力，封装成一个好用的抽象。把两个节点拆开看全貌（图 8-2）：

三通道各管什么

时序：一次完整 Action 生命周期

上面讲的是"线路怎么铺"，这里讲"一次任务怎么走"。下面这张时序图 5.5 秒循环一次：①Goal 出发 → ②Feedback 反复流回 → ③（可选）Cancel 插队 → ④Result 终态。对照图 8-2 认线路：①走 Goal 服务线，②走 Feedback 话题线，④走 Result 服务线。

三回调：服务端为什么拆成三个函数

服务端只有一个回调（收到请求就处理），动作服务端却拆成三个——因为一个长任务要回答三个问题：这个目标接不接？取消同不同意？活怎么干？

反馈状态机：Feedback 里流的是什么

04实验步骤：源码在哪、怎么串起来、关键几行在干什么

这一节不逐行讲全部代码，只解决三件事：源码文件在哪个目录 → 三份文件怎么串成一条链 → 哪几行是必须看懂的。看完你能自己打开源码定位到对应位置，比记住语法有用得多。

第 1 步 · 先找源码：三份文件，一个目录

本实验的全部源码都在同一个目录里，录入门禁脚本、动作服务端、动作客户端共用这一套驱动：

# 源码目录（用一条命令即可跳到）
cd ~/ros2_ws/src/xflz_eai_car1/driver/ros_sensor/ros_sensor/ros_sensor
ls *.py

第 2 步 · 再看框架：三份文件怎么串成一条链

先记住分工，再看代码就不会迷路。

第 3 步 · 再读关键代码：三处非讲不可的地方

下面只挑最能说明问题的几段。每段都标了文件 + 行号，你可以直接打开源码跳过去对照。

① 驱动层：REQA 寻卡 → 防冲突 → 拿 UID

MFRC522 是寄存器驱动的 13.56MHz 读卡芯片，主控通过 SPI 读写它的寄存器。引脚在文件头定义：CS→GPIO114、SCK→112、MOSI→113、MISO→115、RST→116。动作服务端实际只调一个方法 read_uid()：

# 来源：rc522_reader.py L296-305（完整读卡流程：寻卡 + 防冲突两步串成一步）
def read_uid(self) -> tuple:
    ok, atqa = self.request_a()          # ① REQA 寻卡：卡在不在射频场里？
    if not ok:
        return False, "", b''            #    不在 → SCANNING 下一轮再来
    ok, uid = self.anticoll()            # ② AntiCollision 防冲突读 UID
    if not ok:
        return False, f"ATQA=0x{atqa:04X}", b''
    uid_str = "%02X%02X%02X%02X" % (uid[0], uid[1], uid[2], uid[3])
    return True, uid_str, uid            # (成功, 卡号十六进制串, 原始字节)

② 服务端：三回调 + 执行循环

# 来源：rfid_action_server.py L41-48（三个回调一次注册）
self._action_server = ActionServer(
    self, RfidScan, 'rfid_scan',
    execute_callback=self.execute_cb,    # 干活的主体（协程）
    goal_callback=self.goal_cb,          # 收/拒目标
    cancel_callback=self.cancel_cb)      # 同意/拒绝取消

# 来源：rfid_action_server.py L100-141（执行循环：查取消 → 发反馈 → 读卡）
async def execute_cb(self, goal_handle):
    deadline = time.time() + (goal_handle.request.timeout_sec or 30)
    feedback = RfidScan.Feedback()
    while time.time() < deadline:
        if goal_handle.is_cancel_requested:      # ← 必须在循环体内！
            goal_handle.canceled()               #    同意取消，就地收场
            result.success = False
            result.message = '已取消'
            return result
        feedback.status = 'SCANNING'            # 每轮都发反馈
        goal_handle.publish_feedback(feedback)
        ok, atqa = rc.request_a()                # 寻卡
        if not ok:
            time.sleep(0.1); continue            # 没卡：0.1s 后再来
        feedback.status = 'CARD_FOUND'
        goal_handle.publish_feedback(feedback)
        ok, uid = rc.anticoll()                  # 读 UID
        ...  # READING_UID → VERIFYING 反馈 → classify_card 判型
    goal_handle.abort()                          # deadline 耗尽仍无卡
    result.message = '读卡超时'
    return result

# 来源：rfid_action_server.py L19-29（卡类型判定：ATQA + UID 长度）
def classify_card(atqa, uid):
    if atqa == 0x0044 and len(uid) == 7: return 'MIFARE Ultralight'
    if atqa == 0x4400: return 'MIFARE Ultralight C'
    if atqa == 0x0344: return 'MIFARE DESFire'
    if atqa in (0x0002, 0x0003, 0x0004): return 'MIFARE Classic'
    return 'Unknown'

③ 客户端：异步回调链

# 来源：rfid_action_client.py L51-66（发送 Goal：三层回调接力）
def send(self, timeout=30):
    if not self._client.wait_for_server(timeout_sec=3.0):
        self.get_logger().error('RFID Action 服务不可用')
        return False
    goal = RfidScan.Goal()
    goal.timeout_sec = timeout                    # Goal 参数从这里进
    future = self._client.send_goal_async(
        goal, feedback_callback=self.feedback_cb) # 反馈回调随 Goal 一起挂
    future.add_done_callback(self.goal_response_cb) # 第 1 环：接受了吗？
    while rclpy.ok() and not self._done:            # 必须轮转执行器！
        rclpy.spin_once(self, timeout_sec=0.1)      # 回调全靠它驱动
    return self._ok

# 回调链的接力顺序（示意）
goal_response_cb(goal_future)      # 第 1 环：ACCEPTED？
    └─ goal_handle = goal_future.result()
    └─ result_future = goal_handle.get_result_async()
    └─ result_future.add_done_callback(self.result_cb)   # 第 2 环：终态
feedback_cb(feedback_msg)          # 插播：执行期间随时被调
    └─ self.get_logger().info(f'Feedback: {feedback_msg.feedback.status}')
result_cb(result_future)           # 第 2 环：读 Result，按状态判定
    └─ status == GoalStatus.STATUS_SUCCEEDED ? '成功' : '超时或被取消'
    └─ self._done = True          # 放行 send() 里的 while 循环

05动手任务：按操作验证本课的四个目标

01 节是"跟着看一遍"，这一节是"你自己操作，验证目标达成了没有"。四个任务分别对应课程目标里的四项能力——做完这四题，你就能说自己掌握动作通讯了。每题都要求先写下预期，再动手验证。

06注意事项与验收清单

07本课总结与思考题