本文是专栏《机器人人机交互与控制实践》第 2 篇。上一篇讲透了 Xbox 手柄,这一篇把视野扩展到机器人开发中常见的其他输入设备,并给出一套能同时接入它们的统一输入抽象层。

Xbox 手柄能解决 80% 的遥控需求,剩下的 20% 往往才是项目里真正头疼的部分:

  • 团队里有人习惯 PS 手柄,结果上一篇的 XInput 代码根本读不到它;
  • 机械臂末端需要 6 个自由度同时控制,手柄的两个摇杆不够用;
  • 户外机器人要遥控几百米到几公里,蓝牙和 2.4G 手柄都够不着,只能上航模遥控器;
  • 设备一多,业务代码里到处是 if device == "ps4": ... elif device == "rc": ...。

本文依次拆解四类设备的协议和坑,最后用一个统一输入抽象层把它们收拢起来。文中所有解析代码都附带离线自测,没有硬件也能运行验证。


1. 设备全景

设备自由度报文分辨率是否自回中典型接口适合场景
Xbox 手柄4 轴 + 2 扳机摇杆 16 位是USB / 蓝牙 / 2.4G通用遥控
PS 手柄(DS4 / DualSense)4 轴 + 2 扳机 + IMU + 触摸板摇杆 8 位是USB / 蓝牙需要体感、触摸板的交互
飞行摇杆 / HOTAS3~4 轴 + 油门 + 多个苦力帽10~16 位摇杆是,油门否USB HID大量按键、精细操控
3D 鼠标6 自由度有效约 ±350是USB HID / 无线接收器机械臂末端、相机位姿
航模遥控器4~16 通道11 位油门通常否SBUS / CRSF / PPM远距离、户外、带失控保护

协议

物理接口

设备

接收机

PS 手柄

飞行摇杆

3D 鼠标

航模遥控器

USB / 蓝牙 HID

串口 UART

HID 报文
各厂商私有布局

SBUS
100k 8E2 反相

CRSF
420k 8N1 双向

统一输入抽象层

语义控制指令


2. PlayStation 手柄:DualShock 4 与 DualSense

2.1 第一个坑:它不是 XInput 设备

PS 手柄在 Windows 上是标准 HID 设备,不走 XInput。上一篇的 XInput 代码对它完全无感,XInputGetState 会一直返回“未连接”。

能直接读 PS 手柄的方案:

方案说明
SDL2 GameController内置 PS 手柄驱动(HIDAPI 后端),按键按位置映射成 Xbox 布局,还能读陀螺仪。首选
直接读 HID 报文hidapi + 自己解析,完全可控,适合需要 IMU、触摸板原始数据的场景
DS4Windows / Steam Input把 PS 手柄虚拟成 Xbox 手柄,再走 XInput。适合临时调试,不建议在机器人上位机里依赖
Linux evdev较新内核由 hid-playstation 驱动接管,摇杆、IMU、触摸板分别是三个独立的 evdev 设备

2.2 按键命名:按位置对应,而不是按字母

位置PlayStationXboxLinux evdevSDL GameController
下✕ CrossABTN_SOUTHBUTTON_A
右○ CircleBBTN_EASTBUTTON_B
左□ SquareXBTN_WESTBUTTON_X
上△ TriangleYBTN_NORTHBUTTON_Y

业务代码里建议用位置语义(南/东/西/北)或功能语义(确认/取消)命名,而不是 A/B/X/Y。否则换成 PS 手柄后,“按 A 确认”对用户来说是“按 ✕ 确认”,还算直观;但如果业务代码写的是“按 X 急停”,PS 用户会去按 ✕,那是 Xbox 的 A。

2.3 USB 报文:DS4 和 DualSense 长得像,但不一样

两代手柄的 USB 输入报文 ID 都是 0x01、长度都是 64 字节,前几个字节也都是摇杆,很容易让人以为布局相同。实际上 DualSense 把扳机挪到了按键前面:

字节DualShock 4DualSense
0报文 ID 0x01报文 ID 0x01
1–4LX, LY, RX, RY(uint8,中位 128)LX, LY, RX, RY(uint8,中位 128)
5低 4 位十字键 + 高 4 位 □✕○△L2 模拟量
6L1 R1 L2 R2 Share Options L3 R3R2 模拟量
7低 2 位 PS / 触摸板按下,高 6 位计数器报文序号
8L2 模拟量低 4 位十字键 + 高 4 位 □✕○△
9R2 模拟量L1 R1 L2 R2 Create Options L3 R3
10时间戳PS / 触摸板按下 / 静音键
13–18陀螺仪 3 × int16—
16–21—陀螺仪 3 × int16
19–24加速度计 3 × int16—
22–27—加速度计 3 × int16

十字键是 Hat 编码:0 上、1 右上、2 右……7 左上、8 松开,不是 4 个独立的位。另外注意摇杆 Y 轴原始值是下为正(上推为 0),和 Xbox 的 XInput 相反。

以上布局以 Linux hid-playstation 驱动的实现为参考。解析代码与离线自测:

"""DualShock 4 / DualSense USB 输入报文解析(报文 ID 0x01)。底部为离线自测。"""
import struct

HAT = {0: (0, 1), 1: (1, 1), 2: (1, 0), 3: (1, -1), 4: (0, -1),
       5: (-1, -1), 6: (-1, 0), 7: (-1, 1), 8: (0, 0)}  # (x, y),上/右为正


def _u8(v):
    """uint8 摇杆 → [-1, 1]。中位 128,0 比 255 多一格,需要截断。"""
    return max(-1.0, min(1.0, (v - 128) / 127))


def _common(sticks, l2, r2, b0, b1, b2):
    lx, ly, rx, ry = sticks
    return {
        # 原始 Y 轴下为正 → 统一为上为正
        "lx": _u8(lx), "ly": -_u8(ly), "rx": _u8(rx), "ry": -_u8(ry),
        "l2": l2 / 255, "r2": r2 / 255,
        "hat": HAT.get(b0 & 0x0F, (0, 0)),
        "square": bool(b0 & 0x10), "cross": bool(b0 & 0x20),
        "circle": bool(b0 & 0x40), "triangle": bool(b0 & 0x80),
        "l1": bool(b1 & 0x01), "r1": bool(b1 & 0x02),
        "share": bool(b1 & 0x10), "options": bool(b1 & 0x20),
        "l3": bool(b1 & 0x40), "r3": bool(b1 & 0x80),
        "ps": bool(b2 & 0x01), "touch_click": bool(b2 & 0x02),
    }


def parse_ds4_usb(r: bytes) -> dict:
    """DS4 USB 报文:摇杆 1-4,按键 5-7,扳机 8-9,陀螺仪 13-18,加速度计 19-24"""
    assert r[0] == 0x01 and len(r) >= 25
    d = _common(r[1:5], r[8], r[9], r[5], r[6], r[7])
    d["gyro"] = struct.unpack_from("<3h", r, 13)
    d["accel"] = struct.unpack_from("<3h", r, 19)
    return d


def parse_dualsense_usb(r: bytes) -> dict:
    """DualSense USB 报文:摇杆 1-4,扳机 5-6(前移了!),序号 7,按键 8-10,陀螺仪 16-21,加速度计 22-27"""
    assert r[0] == 0x01 and len(r) >= 28
    d = _common(r[1:5], r[5], r[6], r[8], r[9], r[10])
    d["mute"] = bool(r[10] & 0x04)
    d["gyro"] = struct.unpack_from("<3h", r, 16)
    d["accel"] = struct.unpack_from("<3h", r, 22)
    return d


if __name__ == "__main__":
    # 构造:左摇杆推到最上、按下 Cross 与 L1、L2 半按
    ds4 = bytearray(64); ds4[0] = 0x01
    ds4[1:5] = bytes([128, 0, 128, 128]); ds4[5] = 0x08 | 0x20; ds4[6] = 0x01
    ds4[8] = 128; struct.pack_into("<3h", ds4, 13, 10, -20, 30)
    ds5 = bytearray(64); ds5[0] = 0x01
    ds5[1:5] = bytes([128, 0, 128, 128]); ds5[5] = 128; ds5[8] = 0x08 | 0x20; ds5[9] = 0x01

    a, b = parse_ds4_usb(ds4), parse_dualsense_usb(ds5)
    for name, d in (("DS4", a), ("DualSense", b)):
        print(f"{name:9s} ly={d['ly']:+.2f} l2={d['l2']:.2f} cross={d['cross']} "
              f"l1={d['l1']} hat={d['hat']} gyro={d['gyro']}")
    # 反面示例:用 DS4 的偏移去解 DualSense 报文
    wrong = parse_ds4_usb(ds5)
    print(f"错用DS4解析DS5: l2={wrong['l2']:.2f} cross={wrong['cross']} hat={wrong['hat']}  ← 全错")

运行输出:

DS4       ly=+1.00 l2=0.50 cross=True l1=True hat=(0, 0) gyro=(10, -20, 30)
DualSense ly=+1.00 l2=0.50 cross=True l1=True hat=(0, 0) gyro=(0, 0, 0)
错用DS4解析DS5: l2=0.16 cross=False hat=(0, 1)  ← 全错

最后一行演示了用 DS4 的偏移去解 DualSense 报文的后果:扳机值错、按键错、十字键莫名其妙地“按住了上”。这类 bug 不会报错,只会让机器人行为异常,所以按 PID 选择解析器是必须的:

# 读取循环(USB 连接),pip install hidapi
import hid
from ps_report import parse_ds4_usb, parse_dualsense_usb

PARSERS = {0x05C4: parse_ds4_usb, 0x09CC: parse_ds4_usb,      # DS4 一代 / 二代
           0x0CE6: parse_dualsense_usb, 0x0DF2: parse_dualsense_usb}  # DualSense / Edge

dev = next(d for d in hid.enumerate(0x054C, 0) if d["product_id"] in PARSERS)
parse = PARSERS[dev["product_id"]]
h = hid.device()
h.open_path(dev["path"])
while True:
    r = bytes(h.read(64, 100))
    if r and r[0] == 0x01:
        s = parse(r)
        print(f"\rly={s['ly']:+.2f} l2={s['l2']:.2f} cross={s['cross']} gyro={s['gyro']}   ", end="")

2.4 蓝牙的两个坑

坑一:默认只发“精简报文”。 蓝牙连接后,两代手柄默认都只发送一个只含摇杆和按键的简化报文,没有 IMU 数据。主机需要先读取一次校准特性报告(Feature Report),手柄才会切换到完整报文:DS4 切换为报文 ID 0x11,数据整体后移 2 字节;DualSense 切换为 0x31,数据后移 1 字节。

坑二:输出报文要带 CRC32。 通过蓝牙控制震动、灯条、DualSense 自适应扳机时,输出报文末尾必须附加 CRC32,否则手柄直接忽略,而且不会有任何错误提示。USB 下则不需要。

如果不需要 IMU,最省事的办法是交给 SDL 处理,这两个坑它都已经处理好了。

2.5 PS 手柄的额外价值:IMU

PS 手柄内置的陀螺仪和加速度计,是它相对 Xbox 手柄最大的优势。在机器人上的典型用法是倾斜控制:倾斜手柄控制云台或机械臂末端姿态,比用摇杆推更直觉。

但要注意:手柄 IMU 是消费级器件,陀螺仪零偏会漂移。用来做“相对手柄按下某个键那一刻的姿态变化”是可行的,用来做绝对姿态就不行了。


3. 飞行摇杆与 HOTAS

3.1 协议:标准 HID,靠 Usage 区分轴

飞行摇杆是最“标准”的 HID 设备,每个轴的含义由 HID 报告描述符中的 Usage 定义(Generic Desktop 页):

Usage ID名称飞行摇杆上的常见用途
0x30X摇杆左右
0x31Y摇杆前后
0x32Z油门(部分设备)
0x33 / 0x34Rx / Ry辅助旋钮
0x35Rz摇杆扭转(偏航)
0x36Slider油门 / 滑块
0x37Dial旋钮
0x39Hat Switch苦力帽(8 方向)

问题在于:同样是油门,有的厂商用 Z,有的用 Slider,有的用 Rz。所以飞行摇杆一定要做可配置的映射,不能写死。

3.2 坑一:多设备枚举顺序不稳定

HOTAS 套装的摇杆和油门通常是两个独立的 USB 设备。很多代码写成“Joystick(0) 是摇杆、Joystick(1) 是油门”,重启电脑、换个 USB 口,顺序就可能对调,摇杆前推变成了加油门。

正确做法是按稳定标识绑定:名称 + GUID,同型号多台时再加序列号或物理路径。Linux 下 /dev/input/by-id/ 和 /dev/input/by-path/ 提供了稳定的符号链接。

# joystick_list.py —— 枚举所有 HID 摇杆类设备,按稳定标识绑定(pip install pygame)
import os
os.environ["SDL_JOYSTICK_ALLOW_BACKGROUND_EVENTS"] = "1"
import pygame

pygame.init()
pygame.joystick.init()

devices = []
for i in range(pygame.joystick.get_count()):
    js = pygame.joystick.Joystick(i)
    info = {"index": i, "name": js.get_name(), "guid": js.get_guid(),
            "axes": js.get_numaxes(), "buttons": js.get_numbuttons(), "hats": js.get_numhats()}
    devices.append((info, js))
    print(info)

# 反面做法:Joystick(0) 是摇杆、Joystick(1) 是油门 —— 重启或换 USB 口后顺序可能对调
# 正确做法:按名称 + GUID 绑定(同型号多台设备时再加序列号 / 物理路径区分)
BIND = {"stick": "T.16000M", "throttle": "TWCS Throttle"}   # 按你的设备名修改


def find(keyword):
    for info, js in devices:
        if keyword.lower() in info["name"].lower():
            return js
    return None


stick, throttle = find(BIND["stick"]), find(BIND["throttle"])
print("摇杆:", stick and stick.get_name(), " 油门:", throttle and throttle.get_name())

3.3 坑二:Windows 下的轴数上限

Windows 传统的 DirectInput 接口每个设备最多暴露 8 个轴、128 个按键,系统自带的“游戏控制器”设置面板只显示 32 个按键。部分高端 HOTAS 的按键和轴超过这个数量,超出部分在某些软件里就“消失”了。遇到这种情况,改用 Raw Input 或 hidapi 直接读报文。

3.4 坑三:不回中的油门

这是飞行摇杆(以及航模遥控器)和手柄的本质区别:油门杆松手后停在原位。

对手柄来说,“松手 = 归零”是天然的安全保障。油门杆没有这个保障,带来两个风险:

  1. 机器人上电或解锁的瞬间,如果油门停在 60% 的位置,会直接以 60% 的输出启动;
  2. 失控保护恢复后,如果直接回到控制状态,油门依然停在失联前的位置。

航模领域的标准做法是油门归零才允许解锁,第 6 节的抽象层会把这条规则系统化。


4. 3D 鼠标(SpaceMouse)

4.1 为什么值得用

3D 鼠标的帽子可以同时往 6 个方向推拉扭转,天然对应刚体的 6 个自由度(3 平移 + 3 旋转)。用它控制机械臂末端在笛卡尔空间的速度,比用两个手柄摇杆加按键切换轴直观得多。

4.2 报文格式

3D 鼠标是 USB HID 设备,报文按 Report ID 区分:

报文 ID老款设备新款设备
0x01平移 x, y, z(3 × int16)平移 + 旋转(6 × int16)
0x02旋转 rx, ry, rz(3 × int16)—
0x03按键位掩码按键位掩码

老款设备使用罗技的 VID 0x046D(3Dconnexion 曾是罗技子公司),新款使用自己的 VID 0x256F。

4.3 两个反直觉的特性

数值范围很小。 虽然报文是 int16,但典型满量程只有约 ±350,只用了 int16 范围的 1% 左右。按 32767 归一化的话,推到底也只有 0.01,看起来就像“设备没反应”。

各轴串扰严重。 人手很难只往一个方向推,想纯平移时往往带着几度的旋转。控制机械臂时,这些串扰会让末端在平移的同时轻微转动。解决办法是主轴锁定(Dominant Axis):只保留幅值最大的自由度,其余清零。精细操作时开启,自由操作时关闭,做成一个可切换的按键。

4.4 驱动冲突

官方驱动(Windows / macOS 的 3DxWare,Linux 的 spacenavd)会独占设备。自己用 hidapi 读取之前需要先停掉官方驱动,否则读不到数据或数据被截走。反过来,如果已经装了官方驱动,也可以用 libspnav 等库通过驱动读取,但不同驱动的坐标轴约定不同(部分驱动会交换 Y/Z 轴),接入时务必逐轴实测。

# spacemouse.py —— 3D 鼠标原始 HID 读取与处理(pip install hidapi)
import struct
import hid

VIDS = (0x046D, 0x256F)   # 老款设备用罗技 VID,新款用 3Dconnexion 自己的 VID
FULL_SCALE = 350.0        # 典型满量程约 ±350,个体有差异,建议实测后修改


class SpaceMouseState:
    def __init__(self):
        self.t = [0, 0, 0]   # 平移 x y z
        self.r = [0, 0, 0]   # 旋转 rx ry rz
        self.buttons = 0

    def feed(self, data: list):
        rid, b = data[0], bytes(data[1:])
        if rid == 1 and len(b) >= 12:          # 新款:平移 + 旋转在同一个报文
            self.t = list(struct.unpack_from("<3h", b, 0))
            self.r = list(struct.unpack_from("<3h", b, 6))
        elif rid == 1 and len(b) >= 6:         # 老款:报文 1 只有平移
            self.t = list(struct.unpack_from("<3h", b, 0))
        elif rid == 2 and len(b) >= 6:         # 老款:报文 2 是旋转
            self.r = list(struct.unpack_from("<3h", b, 0))
        elif rid == 3:
            self.buttons = int.from_bytes(b[:4], "little")

    def twist(self, deadzone=0.08, dominant=False):
        """→ 6 维归一化向量。dominant=True 时只保留幅值最大的一个自由度(主轴锁定)"""
        v = [max(-1.0, min(1.0, x / FULL_SCALE)) for x in self.t + self.r]
        v = [0.0 if abs(x) < deadzone else x for x in v]
        if dominant and any(v):
            k = max(range(6), key=lambda i: abs(v[i]))
            v = [x if i == k else 0.0 for i, x in enumerate(v)]
        return v


def open_spacemouse():
    for d in hid.enumerate():
        if d["vendor_id"] in VIDS and "space" in (d["product_string"] or "").lower():
            h = hid.device()
            h.open_path(d["path"])
            return h, d["product_string"]
    raise RuntimeError("未找到 3D 鼠标(Linux 下需先停止 spacenavd,或配置 udev 权限)")


if __name__ == "__main__":
    h, name = open_spacemouse()
    print("设备:", name)
    s = SpaceMouseState()
    while True:
        data = h.read(64, 100)
        if data:
            s.feed(data)
            v = s.twist(dominant=True)
            print("\r平移", [f"{x:+.2f}" for x in v[:3]], "旋转", [f"{x:+.2f}" for x in v[3:]],
                  "键", bin(s.buttons), " " * 8, end="")

5. 航模遥控器:SBUS 与 CRSF

航模遥控器是户外远距离遥控的首选:成熟、便宜、距离可达数公里,而且失控保护是协议层面的一等公民。遥控器通过无线链路发给接收机,接收机再通过串口协议输出给机器人主控,最常见的两种协议是 SBUS 和 CRSF。

5.1 SBUS

电气参数:100000 bps,8 数据位,偶校验,2 停止位(8E2),信号反相。

两个细节经常卡住新手:

  • 波特率 100000 不是常见标准值,部分 USB 转串口芯片不支持或误差很大,选型时要确认;
  • 信号反相:空闲电平为低。很多 MCU 的 UART 不支持硬件 RX 反相,需要加一个三极管反相电路;PC 上用 USB 转串口读取时同样要反相(部分芯片可通过配置工具设置反相)。

帧格式:每帧 25 字节。

字节内容
0帧头 0x0F
1–2216 个通道 × 11 位 = 176 位,紧密打包
23标志位:bit0 CH17,bit1 CH18,bit2 丢帧,bit3 失控保护
24帧尾 0x00(SBUS2 变体为 0x04/0x14/0x24/0x34)

11 位打包的解析方法:把第 1~22 字节按小端拼成一个 176 位整数 BBB,则第 iii 个通道(从 0 开始)为

chi=⌊B/211i⌋ mod 211 \text{ch}_i = \left\lfloor B / 2^{11i} \right\rfloor \bmod 2^{11} chi​=⌊B/211i⌋mod211

网上常见的解法是 16 行手写移位,Python 里用大整数一行就够了,而且不容易写错。

通道值的有效范围一般是 172~1811,中位 992,与舵机脉宽的换算关系:

tμs=58(ch−992)+1500 t_{\mu s} = \frac{5}{8}\left(\text{ch} - 992\right) + 1500 tμs​=85​(ch−992)+1500

即 172 → 988 µs,992 → 1500 µs,1811 → 2012 µs。

时序:每字节 1 起始位 + 8 数据位 + 1 校验位 + 2 停止位 = 12 位,一帧传输时间为

25×12100000=3 ms \frac{25 \times 12}{100000} = 3\ \text{ms} 10000025×12​=3 ms

帧间隔一般为 14 ms(普通模式)或 7 ms(高速模式)。

5.2 SBUS 的失控保护:最危险的默认值

接收机失去遥控器信号后的行为,常见有三种模式:

模式行为风险
保持(Hold)继续输出失联前的最后一组通道值机器人按最后的指令一直运动
预设值(Custom)输出事先设定的一组通道值预设值设错就是灾难
无输出(No Pulses)停止输出帧主控需要自己做超时检测

不少接收机出厂默认是“保持”模式。对航模来说这有其道理(短暂丢信号时保持姿态),对地面机器人则可能是灾难。

“预设值”模式还有一个非常隐蔽的陷阱:很多人把所有通道的预设值都设为中位(992),认为这样“最安全”。如果你的业务代码把油门通道映射成 0~1 的单极性值(172 → 0,1811 → 1),那 992 对应的就是 50% 油门。失控保护一触发,机器人以半速前进。

建议的配置:

  1. 接收机失控模式设为“无输出”,或者“预设值 + 置失控标志位”;
  2. 预设值中,油门类通道设为最小值,而不是中位;
  3. 主控软件同时检查 失控标志位 和 自身的帧超时,任一触发即停车。

另外,丢帧标志(bit2)偶尔出现是正常的,单次丢帧就停车会导致频繁误触发;而失控标志(bit3)一般要在持续丢帧一段时间后才会被接收机置位,这个延迟由接收机决定,可能长达数百毫秒甚至更久。所以更稳妥的做法是在主控里自己统计连续丢帧次数,超过约 100 ms 对应的帧数就进入保护,而不是完全依赖接收机的失控标志。

5.3 CRSF

CRSF 最初由 TBS Crossfire 定义,现在被开源的 ExpressLRS 广泛采用,是当前航模领域的主流协议。和 SBUS 相比,它的几个优势对机器人非常有用:

特性SBUSCRSF
波特率100000,8E2,反相420000(常见),8N1,不反相
方向单向双向(可回传遥测到遥控器屏幕)
校验无CRC8
链路质量只有丢帧 / 失控标志RSSI、LQ、SNR、发射功率等完整链路统计
帧长固定 25 字节可变,最长 64 字节

帧格式:

字段长度说明
地址 / 同步字节1通常为 0xC8,也可能是 0xEA、0xEE
长度1后续字节数 = 类型 + 负载 + CRC
类型10x16 通道数据,0x14 链路统计,……
负载N通道帧为 22 字节,与 SBUS 相同的 16 × 11 位打包
CRC81覆盖“类型 + 负载”,多项式 0xD5

CRC8 使用 DVB-S2 多项式:

G(x)=x8+x7+x6+x4+x2+1(0xD5) G(x) = x^8 + x^7 + x^6 + x^4 + x^2 + 1 \quad (\texttt{0xD5}) G(x)=x8+x7+x6+x4+x2+1(0xD5)

一个完整的通道帧是 26 字节,在 420000 bps 下传输时间约 0.62 ms,远快于 SBUS。

链路统计帧(0x14)是 CRSF 对机器人最有价值的部分。 它包含上行 RSSI(两根天线)、上行链路质量 LQ、SNR、当前天线、射频模式、发射功率,以及下行 RSSI / LQ / SNR。有了它,就可以在信号变差但还没完全断开时提前减速或返航,而不是等到失控才被动停车。

一个经验:看 LQ,而不是 RSSI。RSSI 反映的是信号强度,但在强干扰环境下,RSSI 很高而丢包严重的情况并不少见;LQ(成功接收的包占比)直接反映的是“指令能不能送到”。

失控行为方面,以 ExpressLRS 为例,接收机失联后默认停止输出通道帧,由下游主控靠超时判断。所以用 CRSF 时,主控的帧超时检测是必须的。

5.4 解析代码

SBUS 和 CRSF 共用同一种 11 位打包格式,放在一个文件里实现。两个解析器都是流式的:可以按任意长度分块喂入数据,自动处理半包、粘包和噪声字节。

"""SBUS / CRSF 协议解析(纯 Python,无依赖)。底部 __main__ 为离线自测。"""

# ---------------- 通用:16 通道 × 11 bit 打包/解包 ----------------
def unpack_11bit(payload: bytes, n: int = 16) -> list:
    """22 字节 → 16 个 11 位通道值(LSB first,小端位序)。SBUS 与 CRSF 共用。"""
    bits = int.from_bytes(payload[:22], "little")
    return [(bits >> (11 * i)) & 0x7FF for i in range(n)]


def pack_11bit(ch: list) -> bytes:
    bits = 0
    for i, v in enumerate(ch[:16]):
        bits |= (v & 0x7FF) << (11 * i)
    return bits.to_bytes(22, "little")


# 两种协议通道值的常用有效范围:172 ~ 1811,中位 992
CH_MIN, CH_MID, CH_MAX = 172, 992, 1811


def ch_to_us(v: int) -> float:
    """通道值 → 舵机脉宽(µs):172→988,992→1500,1811→2012"""
    return (v - CH_MID) * 5 / 8 + 1500


def ch_to_norm(v: int) -> float:
    """通道值 → [-1, 1]"""
    return max(-1.0, min(1.0, (v - CH_MID) / (CH_MAX - CH_MID)))


# ---------------- SBUS ----------------
SBUS_HEADER, SBUS_LEN = 0x0F, 25


def sbus_encode(ch, frame_lost=False, failsafe=False) -> bytes:
    flags = (0x04 if frame_lost else 0) | (0x08 if failsafe else 0)
    return bytes([SBUS_HEADER]) + pack_11bit(ch) + bytes([flags, 0x00])


class SbusParser:
    """流式解析:自动在字节流中寻找帧边界。"""

    def __init__(self):
        self.buf = bytearray()

    def feed(self, data: bytes):
        self.buf += data
        frames = []
        while len(self.buf) >= SBUS_LEN:
            if self.buf[0] != SBUS_HEADER:
                del self.buf[0]
                continue
            # SBUS 没有校验和,只能靠帧头 + 帧尾双重确认来降低误同步概率
            if self.buf[24] not in (0x00, 0x04, 0x14, 0x24, 0x34):
                del self.buf[0]
                continue
            f = bytes(self.buf[:SBUS_LEN])
            del self.buf[:SBUS_LEN]
            frames.append({
                "channels": unpack_11bit(f[1:23]),
                "ch17": bool(f[23] & 0x01),
                "ch18": bool(f[23] & 0x02),
                "frame_lost": bool(f[23] & 0x04),
                "failsafe": bool(f[23] & 0x08),
            })
        return frames


# ---------------- CRSF ----------------
CRSF_SYNC = (0xC8, 0xEA, 0xEE)
CRSF_RC_CHANNELS = 0x16
CRSF_LINK_STATS = 0x14


def crc8_d5(data: bytes) -> int:
    """CRSF 使用的 CRC-8,多项式 0xD5(DVB-S2),初值 0。"""
    crc = 0
    for b in data:
        crc ^= b
        for _ in range(8):
            crc = ((crc << 1) ^ 0xD5) & 0xFF if crc & 0x80 else (crc << 1) & 0xFF
    return crc


def crsf_encode(frame_type: int, payload: bytes, addr: int = 0xC8) -> bytes:
    body = bytes([frame_type]) + payload
    return bytes([addr, len(body) + 1]) + body + bytes([crc8_d5(body)])


class CrsfParser:
    def __init__(self):
        self.buf = bytearray()
        self.crc_errors = 0

    def feed(self, data: bytes):
        self.buf += data
        out = []
        while len(self.buf) >= 4:
            if self.buf[0] not in CRSF_SYNC:
                del self.buf[0]
                continue
            n = self.buf[1]                     # n = type + payload + crc 的字节数
            if n < 2 or n > 62:                 # CRSF 整帧不超过 64 字节
                del self.buf[0]
                continue
            if len(self.buf) < n + 2:
                break                           # 半包:等待更多数据
            frame = bytes(self.buf[:n + 2])
            body, crc = frame[2:-1], frame[-1]
            if crc8_d5(body) != crc:
                self.crc_errors += 1
                del self.buf[0]                 # 只丢 1 字节重新找同步,而不是丢整帧
                continue
            del self.buf[:n + 2]
            ftype, payload = body[0], body[1:]
            if ftype == CRSF_RC_CHANNELS and len(payload) == 22:
                out.append(("rc", unpack_11bit(payload)))
            elif ftype == CRSF_LINK_STATS and len(payload) >= 10:
                out.append(("link", {
                    "up_rssi1_dbm": -payload[0], "up_rssi2_dbm": -payload[1],
                    "up_lq": payload[2],
                    "up_snr": int.from_bytes(payload[3:4], "little", signed=True),
                    "active_antenna": payload[4], "rf_mode": payload[5],
                    "tx_power_idx": payload[6],
                    "down_rssi_dbm": -payload[7], "down_lq": payload[8],
                    "down_snr": int.from_bytes(payload[9:10], "little", signed=True),
                }))
            else:
                out.append(("other", ftype, payload))
        return out


if __name__ == "__main__":
    import random
    ch = [CH_MID] * 16
    ch[0], ch[1], ch[2], ch[3] = CH_MIN, CH_MAX, 1200, 600

    # --- SBUS:带噪声字节的流,分块喂入 ---
    stream = bytes([0x55, 0x0F, 0xAA]) + sbus_encode(ch) + sbus_encode(ch, frame_lost=True) \
        + sbus_encode(ch, failsafe=True)
    p, frames = SbusParser(), []
    for i in range(0, len(stream), 7):
        frames += p.feed(stream[i:i + 7])
    print(f"SBUS 解出 {len(frames)} 帧")
    for f in frames:
        print("  前4通道", f["channels"][:4],
              "µs", [round(ch_to_us(v)) for v in f["channels"][:4]],
              "lost", f["frame_lost"], "failsafe", f["failsafe"])

    # --- CRSF:通道帧 + 链路统计帧 + 一个 CRC 损坏的帧 ---
    rc = crsf_encode(CRSF_RC_CHANNELS, pack_11bit(ch))
    link = crsf_encode(CRSF_LINK_STATS, bytes([62, 70, 100, 9, 0, 5, 2, 58, 100, 8]))
    bad = bytearray(rc); bad[10] ^= 0xFF
    stream = bytes(random.Random(1).randbytes(5)) + rc + bytes(bad) + link + rc
    cp, got = CrsfParser(), []
    for i in range(0, len(stream), 9):
        got += cp.feed(stream[i:i + 9])
    print(f"\nCRSF 解出 {len(got)} 帧, CRC 错误 {cp.crc_errors} 次")
    for g in got:
        if g[0] == "rc":
            print("  RC 前4通道", g[1][:4], "归一化", [round(ch_to_norm(v), 3) for v in g[1][:4]])
        elif g[0] == "link":
            print("  LINK", {k: g[1][k] for k in ("up_rssi1_dbm", "up_lq", "up_snr", "down_lq")})
    print("\nCRC8 校验向量 crc8_d5(b'123456789') =", hex(crc8_d5(b"123456789")))

运行输出:

SBUS 解出 3 帧
  前4通道 [172, 1811, 1200, 600] µs [988, 2012, 1630, 1255] lost False failsafe False
  前4通道 [172, 1811, 1200, 600] µs [988, 2012, 1630, 1255] lost True failsafe False
  前4通道 [172, 1811, 1200, 600] µs [988, 2012, 1630, 1255] lost False failsafe True

CRSF 解出 3 帧, CRC 错误 1 次
  RC 前4通道 [172, 1811, 1200, 600] 归一化 [-1.0, 1.0, 0.254, -0.479]
  LINK {'up_rssi1_dbm': -62, 'up_lq': 100, 'up_snr': 9, 'down_lq': 100}
  RC 前4通道 [172, 1811, 1200, 600] 归一化 [-1.0, 1.0, 0.254, -0.479]

CRC8 校验向量 crc8_d5(b'123456789') = 0xbc

几个设计细节:

  • CRC8 实现通过了标准校验向量("123456789" → 0xBC),这是验证 CRC 实现的最快方法;
  • CRSF 校验失败时只丢弃 1 字节重新找同步,而不是跳过整帧长度。因为校验失败时,“长度”字段本身就可能是错的,按它跳会丢掉后面的有效帧;
  • SBUS 没有校验和,只能用帧头 + 帧尾双重判断降低误同步概率。这也是 SBUS 在强干扰环境下不如 CRSF 可靠的原因之一。

串口读取:

# serial_rc.py —— 从串口读取 SBUS / CRSF(pip install pyserial)
import sys
import serial
from rc_protocols import SbusParser, CrsfParser, ch_to_norm

PORT = sys.argv[1] if len(sys.argv) > 1 else "/dev/ttyUSB0"
MODE = sys.argv[2] if len(sys.argv) > 2 else "crsf"

if MODE == "sbus":
    # 100000 bps, 8E2;信号是反相的,USB 转串口前需要硬件反相(或芯片支持 RX 反相)
    ser = serial.Serial(PORT, 100000, parity=serial.PARITY_EVEN,
                        stopbits=serial.STOPBITS_TWO, timeout=0.02)
    parser = SbusParser()
else:
    # 420000 bps, 8N1,不反相
    ser = serial.Serial(PORT, 420000, timeout=0.02)
    parser = CrsfParser()

lost_run = 0
while True:
    for f in parser.feed(ser.read(256)):
        if MODE == "sbus":
            lost_run = lost_run + 1 if f["frame_lost"] else 0
            ch = f["channels"]
            print(f"\rCH1-4 {[round(ch_to_norm(v), 2) for v in ch[:4]]} "
                  f"连续丢帧={lost_run:3d} failsafe={f['failsafe']}   ", end="")
        elif f[0] == "rc":
            print(f"\rCH1-4 {[round(ch_to_norm(v), 2) for v in f[1][:4]]}   ", end="")
        elif f[0] == "link":
            print(f"\n链路: RSSI {f[1]['up_rssi1_dbm']} dBm, LQ {f[1]['up_lq']}%")

6. 统一输入抽象层

6.1 设计原则:抽象“意图”,而不是抽象“设备”

最常见的错误设计是定义一个“通用手柄”接口:把所有设备都转成“左摇杆 X/Y、右摇杆 X/Y、若干按键”。这样做会丢失大量信息:3D 鼠标的 6 个自由度被砍掉 2 个,航模遥控器的链路质量没地方放,油门杆不回中的特性也被抹平了。

更好的做法是分两层:

  1. 设备层:每个驱动输出统一的 InputFrame,包含归一化后的轴、按键、是否在线、链路质量;
  2. 语义层:通过配置文件(Profile)把设备的通道映射到业务语义,例如“前进”“转向”“解锁”。业务代码只认语义名,换设备只换 Profile。

轴需要区分三种类型,因为它们的安全逻辑不同:

类型例子值域松手后
CENTERED手柄摇杆、飞行摇杆、3D 鼠标[-1, 1]回到 0
TRIGGER扳机[0, 1]回到 0
ABSOLUTE油门杆、滑块、航模油门[0, 1]停在原位

语义层

设备层

Xbox 驱动

InputFrame

PS 驱动

InputFrame

RC 驱动
SBUS / CRSF

InputFrame
含链路质量

3D 鼠标驱动

InputFrame

Profile 映射
反向 / 死区 / 类型

控制权仲裁
按优先级

解锁 / 失控保护状态机

语义指令
forward / turn / ...

6.2 解锁与失控保护状态机

按下解锁键
且所有轴归零
且链路正常

按下上锁键

断连 / 超时 / 链路质量过低

链路恢复

备用设备满足解锁条件

DISARMED

ARMED

FAILSAFE

链路恢复后不会自动回到 ARMED
必须重新解锁

这里有三条规则,每条都对应一个真实的事故场景:

  1. 解锁时所有轴必须归零。 对 ABSOLUTE 轴这是硬性要求;对 CENTERED 轴,可以防止摇杆被压住或卡住时解锁即运动。
  2. 失控恢复后不自动回到控制状态。 链路时断时续时,自动恢复意味着机器人会“一顿一顿”地按失联前的油门位置冲出去。恢复后先降为 DISARMED,操作员确认状态、把油门归零、重新解锁。
  3. 同一时刻只有一个设备拥有控制权。 其他设备的输入被忽略,避免两个人同时操作时指令叠加。拥有者失联后,备用设备可以按同样的解锁规则接管。

6.3 实现

"""统一输入抽象层:设备 → 标准帧 → 语义通道 → 解锁/失控保护状态机 → 控制指令"""
import time
from dataclasses import dataclass, field
from enum import Enum


class AxisKind(Enum):
    CENTERED = "centered"   # 自回中双极性:手柄摇杆、飞行摇杆 X/Y、3D 鼠标
    TRIGGER = "trigger"     # 自回零单极性:扳机
    ABSOLUTE = "absolute"   # 不回中:油门杆、滑块、航模遥控器的油门通道


@dataclass
class InputFrame:
    """所有设备驱动的统一输出。axes 已归一化:CENTERED ∈[-1,1],其余 ∈[0,1]"""
    axes: dict = field(default_factory=dict)
    buttons: dict = field(default_factory=dict)
    connected: bool = False
    quality: float = 1.0     # 链路质量 0~1;有线设备恒为 1,RC 可用 LQ/100
    stamp: float = 0.0       # 最近一次确认设备在线的时间


@dataclass
class AxisBinding:
    source: str
    kind: AxisKind = AxisKind.CENTERED
    invert: bool = False
    deadzone: float = 0.05
    scale: float = 1.0


@dataclass
class Profile:
    """一台设备到语义通道的映射。换设备只换 Profile,业务代码不动。"""
    device: str
    priority: int
    axes: dict                 # 语义名 → AxisBinding
    arm_button: str
    disarm_button: str
    zero_tol: float = 0.05     # ABSOLUTE 轴解锁时允许的最大偏离
    min_quality: float = 0.3
    timeout: float = 0.3


class State(Enum):
    DISARMED = "DISARMED"
    ARMED = "ARMED"
    FAILSAFE = "FAILSAFE"


class InputHub:
    def __init__(self, profiles):
        self.profiles = {p.device: p for p in profiles}
        self.state = State.DISARMED
        self.owner = None          # 当前拥有控制权的设备
        self.log = []

    def _alive(self, p: Profile, f: InputFrame, now: float) -> bool:
        return f.connected and (now - f.stamp) < p.timeout and f.quality >= p.min_quality

    def _map(self, p: Profile, f: InputFrame) -> dict:
        out = {}
        for name, b in p.axes.items():
            v = f.axes.get(b.source, 0.0)
            v = -v if b.invert else v
            if b.kind == AxisKind.CENTERED:
                v = 0.0 if abs(v) < b.deadzone else (v - b.deadzone * (1 if v > 0 else -1)) / (1 - b.deadzone)
            else:
                v = 0.0 if v < b.deadzone else (v - b.deadzone) / (1 - b.deadzone)
            out[name] = v * b.scale
        return out

    def _neutral(self, p: Profile, f: InputFrame) -> bool:
        """解锁条件:所有映射后的轴都在零位附近。
        对 ABSOLUTE 轴这是硬性要求(它不会自己回去);对 CENTERED 轴可以拦住卡住/被压住的摇杆。"""
        return all(abs(v) <= p.zero_tol for v in self._map(p, f).values())

    def _emit(self, msg):
        self.log.append(msg)

    def update(self, frames: dict, now: float) -> dict:
        zero = {k: 0.0 for p in self.profiles.values() for k in p.axes}

        # 1) 当前拥有者失联 → FAILSAFE
        if self.state == State.ARMED:
            p, f = self.profiles[self.owner], frames.get(self.owner, InputFrame())
            if not self._alive(p, f, now):
                self.state = State.FAILSAFE
                self._emit(f"{now:.2f}s {self.owner} 失联 → FAILSAFE")
                return zero
            if f.buttons.get(p.disarm_button):
                self.state, self.owner = State.DISARMED, None
                self._emit(f"{now:.2f}s 手动上锁 → DISARMED")
                return zero
            return {**zero, **self._map(p, f)}

        # 2) FAILSAFE:链路恢复后也不自动回到 ARMED,只降为 DISARMED,必须重新解锁
        if self.state == State.FAILSAFE:
            p = self.profiles[self.owner]
            if self._alive(p, frames.get(self.owner, InputFrame()), now):
                self._emit(f"{now:.2f}s {self.owner} 链路恢复 → DISARMED(需重新解锁)")
                self.state, self.owner = State.DISARMED, None

        # 3) DISARMED / FAILSAFE:按优先级检查谁在请求解锁(原拥有者失联时,备用设备可以接管)
        for p in sorted(self.profiles.values(), key=lambda x: -x.priority):
            f = frames.get(p.device, InputFrame())
            if not (self._alive(p, f, now) and f.buttons.get(p.arm_button)):
                continue
            if not self._neutral(p, f):
                self._emit(f"{now:.2f}s {p.device} 请求解锁被拒:存在未归零的轴")
                continue
            self.state, self.owner = State.ARMED, p.device
            self._emit(f"{now:.2f}s {p.device} 解锁成功 → ARMED")
            return {**zero, **self._map(p, f)}
        return zero


if __name__ == "__main__":
    rc = Profile("rc", priority=2, arm_button="sw_arm", disarm_button="sw_disarm",
                 axes={"forward": AxisBinding("ch_throttle", AxisKind.ABSOLUTE, deadzone=0.02),
                       "turn": AxisBinding("ch_yaw", invert=True)})
    pad = Profile("pad", priority=1, arm_button="start", disarm_button="back",
                  axes={"forward": AxisBinding("ly", deadzone=0.1),
                        "turn": AxisBinding("rx", invert=True, deadzone=0.1)})
    hub = InputHub([rc, pad])

    def rc_f(thr, yaw=0.0, arm=False, ok=True, q=1.0, t=0.0):
        return InputFrame({"ch_throttle": thr, "ch_yaw": yaw}, {"sw_arm": arm}, ok, q, t)

    def pad_f(ly=0.0, rx=0.0, start=False, t=0.0):
        return InputFrame({"ly": ly, "rx": rx}, {"start": start}, True, 1.0, t)

    script = [  # (时间, rc 帧, pad 帧, 说明)
        (0.0, rc_f(0.6, arm=True), pad_f(), "油门没归零就解锁"),
        (0.1, rc_f(0.0, arm=True), pad_f(), "油门归零后解锁"),
        (0.2, rc_f(0.5, 0.3), pad_f(ly=1.0), "RC 控制中,手柄输入应被忽略"),
        (0.3, rc_f(0.5, 0.3, q=0.1), pad_f(), "链路质量掉到 10%"),
        (0.4, rc_f(0.5, 0.3, q=0.1), pad_f(ly=0.8, start=True), "手柄推着摇杆请求接管"),
        (0.5, rc_f(0.5, 0.3, q=1.0), pad_f(), "RC 链路恢复,油门仍在 0.5"),
        (0.6, rc_f(0.5, arm=True), pad_f(), "RC 不归零直接再解锁"),
        (0.7, rc_f(0.5), pad_f(start=True), "手柄摇杆回中后接管"),
        (0.8, rc_f(0.5), pad_f(ly=0.8, rx=-0.5), "手柄控制中"),
    ]
    for t, fr, fp, note in script:
        fr.stamp = fp.stamp = t
        cmd = hub.update({"rc": fr, "pad": fp}, t)
        print(f"{t:.1f}s {hub.state.value:9s} owner={str(hub.owner):4s} "
              f"forward={cmd['forward']:+.2f} turn={cmd['turn']:+.2f}  | {note}")
    print("\n事件日志:")
    for m in hub.log:
        print("  " + m)

运行输出:

0.0s DISARMED  owner=None forward=+0.00 turn=+0.00  | 油门没归零就解锁
0.1s ARMED     owner=rc   forward=+0.00 turn=+0.00  | 油门归零后解锁
0.2s ARMED     owner=rc   forward=+0.49 turn=-0.26  | RC 控制中,手柄输入应被忽略
0.3s FAILSAFE  owner=rc   forward=+0.00 turn=+0.00  | 链路质量掉到 10%
0.4s FAILSAFE  owner=rc   forward=+0.00 turn=+0.00  | 手柄推着摇杆请求接管
0.5s DISARMED  owner=None forward=+0.00 turn=+0.00  | RC 链路恢复,油门仍在 0.5
0.6s DISARMED  owner=None forward=+0.00 turn=+0.00  | RC 不归零直接再解锁
0.7s ARMED     owner=pad  forward=+0.00 turn=+0.00  | 手柄摇杆回中后接管
0.8s ARMED     owner=pad  forward=+0.78 turn=+0.44  | 手柄控制中

事件日志:
  0.00s rc 请求解锁被拒:存在未归零的轴
  0.10s rc 解锁成功 → ARMED
  0.30s rc 失联 → FAILSAFE
  0.40s pad 请求解锁被拒:存在未归零的轴
  0.50s rc 链路恢复 → DISARMED(需重新解锁)
  0.60s rc 请求解锁被拒:存在未归零的轴
  0.70s pad 解锁成功 → ARMED

这个仿真覆盖了几个关键场景:油门未归零时拒绝解锁;RC 控制期间手柄输入被忽略;链路质量掉到 10% 立即进入失控保护;手柄在摇杆推着的状态下请求接管被拒;RC 链路恢复后没有自动回到控制状态;手柄摇杆回中后成功接管。

实际项目中,把第 1 篇和本篇的各个驱动包装成输出 InputFrame 的类,Profile 用 YAML 或 JSON 配置文件管理即可。


7. 工程检查清单

PS 手柄

  • 确认读取方案不是 XInput(SDL / hidapi / evdev)
  • 按 PID 选择解析器,DS4 与 DualSense 布局不同
  • 摇杆 Y 轴原始值下为正,已统一方向
  • 蓝牙下如需 IMU,确认已切换到完整报文;输出报文带 CRC32

飞行摇杆

  • 按名称 + GUID(或 /dev/input/by-id)绑定设备,不依赖枚举顺序
  • 油门对应的 Usage(Z / Slider / Rz)已实测确认
  • 油门轴配置为 ABSOLUTE 类型,解锁前要求归零

3D 鼠标

  • 满量程按实测值归一化(典型约 ±350),不是 32767
  • 提供主轴锁定开关
  • 确认没有官方驱动独占设备;坐标轴方向逐一实测

航模遥控器

  • SBUS:确认串口支持 100000 bps,信号已反相
  • 接收机失控模式不是“保持”
  • 预设值模式下,油门类通道预设为最小值而非中位
  • 主控自行统计连续丢帧 / 帧超时,不单纯依赖失控标志位
  • CRSF:监控 LQ,设置降速和停车两级阈值

抽象层

  • 业务代码只使用语义通道名
  • 解锁要求所有轴归零;失控恢复后需重新解锁
  • 实测:推着油门关掉遥控器,机器人立即停止;重新打开遥控器,机器人不会自行恢复运动

总结

需求推荐设备关键注意点
通用遥控Xbox 手柄见第 1 篇
需要体感控制PS 手柄不是 XInput;两代布局不同;IMU 只适合相对姿态
按键多、需要精细油门飞行摇杆枚举顺序、油门不回中
6 自由度末端控制3D 鼠标量程小、串扰大、驱动独占
远距离户外航模遥控器(优先 CRSF)失控保护配置、链路质量监控

设备的协议差异可以靠驱动层消化,真正决定系统是否可靠的,是抽象层里那几条安全规则:解锁要归零、失联要停车、恢复要重新解锁、同时只有一个人说了算。

下一篇预告:手柄震动与触觉反馈——把机器人的状态“传到手上”。从 Xbox 的双马达、Impulse Triggers 到 DualSense 的自适应扳机,看看怎样把碰撞、负载、链路质量变成操作员能直接感知的触觉信号。

Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐