简介:《UR机器人/优遨机器人实时数据交换手册》面向需要将外部应用与优遨(UR)控制器通过标准TCP/IP同步的自动化工程师、机器人集成商与开发者,重点解决实时数据交换(RTDE)接口的协议理解与落地配置问题。资源仅含1份PDF文档,约226KB,篇幅精炼,便于随查随用。文档系统梳理RTDE的关键特性、输入/输出字段及数据类型,并依次说明协议头、协议版本请求、控制器版本查询、文本消息、数据包以及输出/输入设置、启动与暂停等控制包,同时交代setup与同步循环两阶段流程、recipe配方和2048字节字段名限制等要点。已有926人学习,适合用于现场总线驱动集成、机器人I/O操作与轨迹状态绘制的开发参考,也能帮助读者快速定位报文结构与排错方向。
1. RTDE 的现场价值:30004 端口上跑着的实时数据通道
产线上想把协作机器人的关节角实时喂给外部视觉程序做补偿,很多人的第一反应是加一块现场总线板卡,其实控制器里一直开着一个 30004 端口的 RTDE(Real-Time Data Exchange,实时数据交换)服务,走标准 TCP/IP,不用装驱动、不用改配置,上电就在跑。这份手册把 RTDE 的字段表、数据类型、协议包结构一次摊开:哪些输入能写、哪些输出能读、每个字段从哪一版控制器软件开始才有、包怎么拼、字节序是什么、125 Hz 是怎么来的又为什么会跳包。它适合两类人——一类要在外部 PC 上写客户端做数据采集或 I/O 联动,一类要把机器人状态接进上位机画轨迹。它解决的不是「机器人怎么动」,而是「机器人动的时候,数据怎么稳定地进出控制器,还不破坏实时性」。这也是我在现场判断该不该用它的一条硬标准:只要你能接受 125 Hz 这个量级、能接受偶尔跳包,它就比任何外挂采集方案都省事。
2. RTDE 的 TCP/IP 帧结构:三个字节的长度头与八种包类型
2.1 包头只有四字节,但它决定了整条解析路径
RTDE 所有包共用同一套头:前 3 个字节是一个大端无符号整数,表示整包长度,包含包头自身那 4 个字节;第 4 个字节是包类型码;载荷紧跟其后。这个「长度含头」的细节是新手最容易翻车的地方——如果按「长度 = 载荷长度」去算,第一个包就会多读或少读四个字节,之后整条流全部错位,而且不会立刻抛异常,只会让字段值变成一堆莫名其妙的小数,查起来极其折磨。
包类型码是单字节的 ASCII 字母,手册里把请求和返回分开列,但码值是一致的。我把它整理成一张表,接到新现场时第一件事就是对着这张表把流量抓一遍,确认每个包的方向和载荷长度对得上:
| 包类型 | 码值 | 方向 | 典型载荷 |
|---|---|---|---|
| RTDE_REQUEST_PROTOCOL_VERSION | 86('V') | 客户端 → 控制器 | 2 字节协议版本 |
| 同上返回 | 86 | 控制器 → 客户端 | 1 字节是否接受(0/1) |
| RTDE_GET_URCONTROL_VERSION | 118('v') | 客户端 → 控制器 | 空 |
| 同上返回 | 118 | 控制器 → 客户端 | 4 个 UINT32:主版本、次版本、修订、构建号 |
| RTDE_CONTROL_PACKAGE_SETUP_OUTPUTS | 79('O') | 客户端 → 控制器 | 8 字节 double 频率 + 字段名列表 |
| 同上返回 | 79 | 控制器 → 客户端 | 1 字节 recipe id + N 个类型码 |
| RTDE_CONTROL_PACKAGE_SETUP_INPUTS | 73('I') | 客户端 → 控制器 | 字段名列表 |
| 同上返回 | 73 | 控制器 → 客户端 | 1 字节 recipe id |
| RTDE_CONTROL_PACKAGE_START | 83('S') | 客户端 → 控制器 | 空 |
| 同上返回 | 83 | 控制器 → 客户端 | 1 字节是否成功 |
| RTDE_CONTROL_PACKAGE_PAUSE | 80('P') | 客户端 → 控制器 | 空 |
| 同上返回 | 80 | 控制器 → 客户端 | 1 字节是否成功 |
| RTDE_DATA_PACKAGE | 85('U') | 双向 | 1 字节 recipe id + 序列化数据 |
| RTDE_TEXT_MESSAGE | 77('M') | 双向 | 1 字节级别 + 长度 + ASCII 文本 |
有一处必须在写代码前就定死:setup 阶段的包有返回,同步循环里的 DATA_PACKAGE 没有返回。也就是说,进入循环之后如果你还按「发一个收一个」的节奏写,程序会直接卡死在 recv 上。我见过不止一个客户端是这么挂住的——现象是 CPU 占用掉到 0、日志停在「已启动同步」,其实就是在等一个永远不会来的应答。
提示:长度字段是 3 字节而不是 2 字节,意味着单个包理论上可以到 16 MB,但实际 DATA_PACKAGE 通常只有几百字节。解析时不要假设「一个 recv 就是一整包」,TCP 是字节流,粘包和半包都会发生。
2.2 载荷里的每个标量都是大端,且没有对齐填充
手册在数据类型那节末尾专门写了一句话:数据始终以网络字节序发送,也就是大端。这句话的分量比它看起来重得多。VECTOR6D 就是 6 个 IEEE 754 双精度浮点连续排下来,48 字节,中间没有分隔符、没有 4 字节对齐、没有类型标签。你在 Python 里用struct一次解出来,格式串必须以>开头,写成<或者不加前缀(默认本机序,x86 是小端)就会得到一堆量级离谱的数字。
| 类型名 | 位宽 | struct 格式符 | 说明 |
|---|---|---|---|
| BOOL | 8 | B | 0 为假,非 0 为真 |
| UINT8 | 8 | B | 无符号 |
| UINT32 | 32 | I | 无符号 |
| UINT64 | 64 | Q | 无符号 |
| INT32 | 32 | i | 二进制补码 |
| DOUBLE | 64 | d | IEEE 754 |
| VECTOR3D | 3×64 | 3d | 三个 double |
| VECTOR6D | 6×64 | 6d | 六个 double |
| VECTOR6INT32 | 6×32 | 6i | 六个 int32 |
| VECTOR6UINT32 | 6×32 | 6I | 六个 uint32 |
BOOL 我一般用B解再用!= 0转布尔,而不是直接上?。原因很实际:?在部分实现里对非 1 的非零值行为不完全一致,而 RTDE 明确说「非零即真」,自己转一次反而没有歧义,也方便在日志里打出原始字节。
2.3 手写一个最小客户端:请求协议版本、读控制器版本
不依赖任何第三方库,标准库的 socket 加 struct 就够把链路打通。下面这段是我常用来验证「网络通不通、端口开没开、协议版本对不对」的第一段代码,十几行,但能把 80% 的现场问题挡在门外:
import socket import struct RTDE_PORT = 30004 # 包类型码,与手册协议章节一致 REQ_PROTOCOL_VERSION = 86 # 'V' GET_CONTROL_VERSION = 118 # 'v' SETUP_OUTPUTS = 79 # 'O' SETUP_INPUTS = 73 # 'I' CONTROL_START = 83 # 'S' CONTROL_PAUSE = 80 # 'P' DATA_PACKAGE = 85 # 'U' def build_packet(pkt_type: int, payload: bytes = b"") -> bytes: """3 字节大端总长度(含 4 字节头) + 1 字节类型 + 载荷""" total = 4 + len(payload) return total.to_bytes(3, "big") + bytes([pkt_type]) + payload def recv_exact(sock: socket.socket, n: int) -> bytes: """必须读满 n 字节,TCP 不保证一次 recv 给你一整包""" buf = bytearray() while len(buf) < n: chunk = sock.recv(n - len(buf)) if not chunk: raise ConnectionError("对端关闭连接") buf.extend(chunk) return bytes(buf) def recv_packet(sock: socket.socket): head = recv_exact(sock, 4) size = int.from_bytes(head[:3], "big") ptype = head[3] payload = recv_exact(sock, size - 4) if size > 4 else b"" return ptype, payload if __name__ == "__main__": # 关掉 Nagle,125 Hz 的小包经不起 40 ms 的合并延迟 sock = socket.create_connection(("192.168.1.10", RTDE_PORT), timeout=3.0) sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) # 1) 请求协议版本,载荷是 2 字节大端版本号 sock.sendall(build_packet(REQ_PROTOCOL_VERSION, struct.pack(">H", 2))) ptype, payload = recv_packet(sock) assert ptype == REQ_PROTOCOL_VERSION, f"意外的返回包类型 {ptype}" print("协议版本协商结果:", payload[0]) # 1 = 接受 # 2) 读控制器版本,返回 4 个 UINT32 sock.sendall(build_packet(GET_CONTROL_VERSION)) ptype, payload = recv_packet(sock) assert ptype == GET_CONTROL_VERSION major, minor, bugfix, build = struct.unpack(">IIII", payload) print(f"控制器版本 {major}.{minor}.{bugfix}-{build}") sock.close()这段代码有两个地方值得单独说。第一处是recv_exact:很多人图省事直接sock.recv(4096),然后拿返回值当整包解析,在局域网里跑一天可能都没事,一旦现场交换机有负载或者走的是虚拟网卡,半包概率立刻上来,表现出来就是「每隔几分钟丢一帧、字段值错乱」。第二处是TCP_NODELAY:默认的 Nagle 算法会把小包攒到一定量再发,对 125 Hz 这种周期数据是灾难,单包延迟能到几十毫秒,而且抖动极大,画轨迹时会看到明显的锯齿。
build_packet里长度写成4 + len(payload),这个 4 就是包头自身。如果有人按len(payload)写,控制器会直接判包非法,返回 TEXT_MESSAGE 报错或者干脆断连,日志里只留下一句「连接被重置」,非常难定位。
版本号这一步不是必须的,但我建议每次都做。手册里明确说了协议可能被更新,客户端可以主动请求一个特定版本;如果控制器不接受,你能在启动阶段就知道,而不是等到解析数据时才发现字段对不上。至于GET_URCONTROL_VERSION,它真正的价值在于:当你怀疑某个字段「应该存在但读不到」时,先看版本号,再回去翻字段表里的「Introduced in version」那一列,能省掉大量瞎猜。
3. 配方怎么定:2048 字节预算内的字段取舍
3.1 先算预算,再挑字段
手册里有一句容易被忽略的话:客户端提交的输入/输出字段名列表,最大 2048 字节。这个限制针对的是字段名的字符串总长度,不是数据包大小。每个字段名以\0结尾,所以actual_TCP_pose实际占 15 字节,input_double_register_47占 25 字节。按平均 20 字节算,理论上能塞 100 个字段。
实际项目里没人这么干。字段越多,数据包越大,单包解析耗时越长,而 125 Hz 意味着你只有 8 ms 的预算。我一般的经验值是 10 到 20 个字段:
| 用途 | 字段组合 | 大致名字开销 |
|---|---|---|
| 轨迹绘制 | timestamp、actual_q、actual_TCP_pose、actual_TCP_speed、speed_scaling | 约 90 字节 |
| 状态监控 | robot_mode、safety_mode、safety_status、runtime_state、robot_status_bits、safety_status_bits | 约 130 字节 |
| I/O 联动 | actual_digital_input_bits、actual_digital_output_bits、standard_analog_input0/1 | 约 110 字节 |
| 力控补偿 | actual_TCP_force、ft_raw_wrench、actual_current、target_current | 约 95 字节 |
| 寄存器交互 | output_int_register_24/25/26、output_bit_register_64/65 | 约 100 字节 |
挑字段时有三个判断依据。第一,这个字段是给谁看的。如果只是给人看的监控面板,10 Hz 都嫌多,没必要挤进 125 Hz 的循环里;如果是给控制算法用的,那才值得占位。第二,这个字段有没有替代品。比如actual_TCP_pose和target_TCP_pose,前者是实际位姿、后者是规划位姿,做延迟测量时两个都要,做纯监控时只要前者。第三,版本兼容性。手册里标了3.10.0 / 5.4.0这类后缀的字段,在老控制器上根本不存在,如果你的客户端要在多个现场通用,就得先读版本号再决定字段列表,而不是写死一份。
注意:字段名大小写敏感,下划线位置固定,寄存器后缀是十进制数字不是十六进制。
input_double_register_24和input_double_register_024是两个完全不同的东西,后者不存在。
3.2 setup 阶段的三次往返与 recipe id 的分配
整个建立流程是固定顺序的:先协商协议版本,再分别建立输出配方和输入配方,最后 START。输出配方和输入配方各自独立,各自拿到一个 id,互不干扰。这意味着你可以只建输出配方(纯采集),也可以只建输入配方(纯下发),不必两边都配。
字段名列表的序列化格式很简单:每个名字后面跟一个\0,直接首尾相接,没有数量前缀、没有分隔符。SETUP_OUTPUTS 在名字列表之前多一个 8 字节的大端 double,表示期望的输出频率。这个频率字段很多人填错——它不是「我想要多快就多快」,控制器有自己的实时线程节奏,你填 500 也只是让它尽力而为,实际还是受限于控制周期。
返回值的处理是这一步的关键。SETUP_OUTPUTS 的返回是「1 字节 recipe id + N 个类型码」,N 等于你提交的字段个数,顺序与你提交的顺序严格一致。SETUP_INPUTS 的返回只有一个 recipe id,没有类型码,因为输入的序列化格式由你提交的顺序决定,控制器只是照单接收。如果某个字段名写错了,常见的表现是返回的类型码个数对不上,或者控制器返回一个不在你映射表里的类型值——这时候正确的做法是逐个字段二分排查,而不是去猜「可能是版本问题」。
3.3 建立输出配方的代码与类型校验
下面这段把 setup 的完整往返写出来了,重点看两个地方:字段名列表的拼装方式,以及返回类型码与本地映射表的比对逻辑。
# RTDE 类型码到 struct 格式符的映射 # 这几个值在协议里是固定的,但别把业务逻辑硬编码在数字上, # 建议启动时打印一次,和控制器实际返回比对 TYPE_FMT = { 1: "B", # BOOL,按字节解再转 bool 更稳 2: "B", # UINT8 3: "I", # UINT32 4: "Q", # UINT64 5: "i", # INT32 6: "d", # DOUBLE 7: "3d", # VECTOR3D 8: "6d", # VECTOR6D 9: "6i", # VECTOR6INT32 10: "6I", # VECTOR6UINT32 } def encode_names(names): """每个字段名以 \0 结尾首尾相接,没有数量前缀""" return b"".join(n.encode("ascii") + b"\x00" for n in names) def setup_outputs(sock, names, frequency=125.0): # 先算一下预算,超了控制器会直接拒绝,不如本地先拦 name_bytes = encode_names(names) if len(name_bytes) > 2048: raise ValueError(f"字段名列表 {len(name_bytes)} 字节,超过 2048 上限") payload = struct.pack(">d", frequency) + name_bytes sock.sendall(build_packet(SETUP_OUTPUTS, payload)) ptype, resp = recv_packet(sock) if ptype != SETUP_OUTPUTS: raise RuntimeError(f"setup 返回类型异常: {ptype}") if not resp: raise RuntimeError("setup 被拒绝,返回为空") recipe_id = resp[0] type_codes = list(resp[1:]) if len(type_codes) != len(names): raise RuntimeError( f"返回类型数量 {len(type_codes)} 与请求字段数 {len(names)} 不匹配," "大概率是某个字段名拼错了" ) unknown = [t for t in type_codes if t not in TYPE_FMT] if unknown: raise RuntimeError(f"出现未知类型码 {unknown},先别猜,打印协议原文比对") return recipe_id, type_codesencode_names里没有加总数前缀,这是最容易写错的地方之一。有些人会下意识在前面补一个len(names),结果控制器把第一个名字当成别的解析,整个 setup 直接失败,错误信息又非常模糊。
类型码校验那段是血泪经验。现场遇到过一次,字段名多打了一个下划线,控制器返回的类型码个数比请求少一个,客户端如果只按「有几个解几个」来写,后面所有字段全部错位一个位置,actual_q里读出的是actual_qd的值,看起来「能跑」,但数据全是错的。加一句数量比对,这个问题在启动阶段就会炸出来,而不是等到客户投诉轨迹画歪了才发现。
frequency参数填 125.0 是手册给出的默认节奏。如果你的应用不需要这么快(比如只做状态面板),填 10.0 或 50.0 都可以,控制器会按你请求的频率发,能省下不少带宽和解析开销。但别指望填 200 就能拿到 200 Hz,实时线程的优先级高于 RTDE 接口,最终节奏由控制器说了算。
4. 125 Hz 同步循环:解包、回写与节奏控制
4.1 按 recipe 顺序还原输出字段
进入循环后,控制器会持续推送 DATA_PACKAGE,包里第一个字节是这个输出配方的 recipe id,后面就是按你提交顺序排列的序列化数据。因为类型是已知的、顺序是已知的、长度也是已知的,所以只需要维护一个偏移量,从头往后切就行,不需要任何分隔符或者长度前缀。
def build_parser(names, type_codes): """预编译解析计划,避免每帧重复算格式串""" plan = [] fmt_all = ">" for name, tc in zip(names, type_codes): fmt = TYPE_FMT[tc] plan.append((name, fmt)) fmt_all += fmt expected = struct.calcsize(fmt_all) return fmt_all, expected, plan def parse_data_package(payload, fmt_all, expected, plan): """payload 已经去掉了 1 字节 recipe id""" if len(payload) != expected: # 长度不符一定是协议层出问题了,别硬解 raise ValueError(f"数据长度 {len(payload)} 与预期 {expected} 不符") values = struct.unpack(fmt_all, payload) out, idx = {}, 0 for name, fmt in plan: n = len(fmt) if fmt[0].isdigit() else 1 # 处理 6d / 3d / 6i 这类多值类型 if fmt[-1] in "diI" and fmt[:-1].isdigit(): n = int(fmt[:-1]) else: n = 1 out[name] = values[idx] if n == 1 else list(values[idx: idx + n]) idx += n return outbuild_parser在循环外调用一次,把所有字段拼成一个大的格式串,这个做法比「每个字段单独 unpack 一次」快一个数量级。125 Hz 意味着每 8 ms 就有一帧,一帧 15 个字段,如果字段逐个解析,光是struct.unpack的调用开销就能吃掉不少余量。
parse_data_package里那个长度校验看起来多余,其实很有用。它能在数据流错位的第一帧就报警,而不是让你在半小时后对着一堆 NaN 发呆。长度不符的原因通常只有两个:要么前面的包头长度算错了,要么上一帧没读完就接着读下一帧——两种情况都得停下来重连,硬解没有任何意义。
plan里对多值类型的处理稍微绕,是因为6d在struct.unpack返回的元组里展开成 6 个独立元素,而不是一个列表。如果把actual_q当成一个标量取,你只会拿到q[0],剩下的五个被后面字段顺走,整包错位。VECTOR6D、VECTOR3D、VECTOR6INT32都属于这一类,写解析器时必须显式处理。
实际循环大概长这样:
recipe_id, type_codes = setup_outputs(sock, OUTPUT_NAMES, 125.0) fmt_all, expected, plan = build_parser(OUTPUT_NAMES, type_codes) sock.sendall(build_packet(CONTROL_START)) ptype, resp = recv_packet(sock) assert ptype == CONTROL_START and resp[0] == 1, "START 被拒绝" try: while running: ptype, payload = recv_packet(sock) if ptype == DATA_PACKAGE: if payload[0] != recipe_id: continue # 不是本配方的包,直接跳过 frame = parse_data_package(payload[1:], fmt_all, expected, plan) handle(frame) # 业务处理,尽量轻,别在这里做耗时计算 elif ptype == 77: # TEXT_MESSAGE,控制器主动报的告警 log_text_message(payload) finally: sock.sendall(build_packet(CONTROL_PAUSE))handle里千万别做重活。我见过有人在里面直接跑 OpenCV 图像处理,单帧耗时 30 ms,结果接收缓冲区一路堆到几十帧,延迟越滚越大,画出来的轨迹比实际落后好几秒。正确的做法是把帧塞进一个有界队列,消费线程另起,队列满了就丢最旧的——反正 RTDE 跳过的包本来也不会补发,你丢一帧和控制器丢一帧没有本质区别。
4.2 输入包的掩码机制与「值不变就不发」
输入方向和输出方向不对称。输出是控制器按固定节奏推,输入是你想发就发,值不变就不需要发,控制器会保留上一次收到的值。手册里明确写了这一点,但很多人还是习惯按 125 Hz 无脑刷,结果控制器每个控制周期都要处理你的包,负载白白增加。
数字量输出必须配合掩码字段使用。以标准数字输出为例,standard_digital_output_mask是一个 UINT8 位掩码,哪一位为 1 表示这一路要用本次的standard_digital_output值覆盖,为 0 表示保持原状。这两个字段必须成对出现,只发值不发掩码,控制器不知道该不该采纳。
| 输出对象 | 掩码字段 | 值字段 | 位宽 |
|---|---|---|---|
| 标准数字输出 | standard_digital_output_mask | standard_digital_output | UINT8 |
| 可配置数字输出 | configurable_digital_output_mask | configurable_digital_output | UINT8 |
| 标准模拟输出 | standard_analog_output_mask(位 0-1) | standard_analog_output_0/1 | UINT8 + DOUBLE×2 |
| 模拟输出类型 | standard_analog_output_type | 与上同步 | UINT8 |
| 速度滑块 | speed_slider_mask | speed_slider_fraction | UINT32 + DOUBLE |
模拟输出还有一层:standard_analog_output_type决定这一路是电流(0)还是电压(1),同样是位 0 对应通道 0、位 1 对应通道 1。发值之前如果类型没设对,输出会是零或者满量程,现场表现是「夹爪不动」或者「夹爪一直夹死」,排查时很容易往机械方向想,其实是这个字节的事。
构造输入包的代码比解析简单,因为格式是你自己定的:
def build_input_payload(recipe_id, values, input_plan): body = b"" for name, fmt in input_plan: v = values[name] if fmt == "6d": body += struct.pack(">6d", *v) elif fmt == "3d": body += struct.pack(">3d", *v) else: body += struct.pack(">" + fmt, v) return bytes([recipe_id]) + body # 例:把通用整数寄存器 24 改成 1,其余不动 values["input_int_register_24"] = 1 sock.sendall(build_packet(DATA_PACKAGE, build_input_payload(in_recipe_id, values, input_plan)))这里有个硬约束必须记住:同一个变量在同一时刻只允许一个 RTDE 客户端控制。现场如果同时开了示教器上的某个 URCap 和你的外部程序,两边都在写同一个寄存器,表现是值来回跳,或者你写的值压根不生效。排查这类问题的第一件事不是看代码,而是确认还有没有别的客户端连着。
4.3 用 timestamp 算实际周期,不要用本地时钟
手册里timestamp的定义是「控制器启动以来经过的秒数」,DOUBLE 类型。这个字段的价值不在于显示时间,而在于它是控制器自己的时间基准。如果你用本地time.time()去算两帧间隔,得到的是「网络传输 + 调度抖动 + 控制器节奏」的混合结果;用timestamp相减,得到的才是控制器真实的发包间隔。
正常情况下这个间隔应该稳定在 8 ms 附近。如果统计出来是 8 ms、8 ms、24 ms、8 ms 这种模式,说明控制器跳过了两帧——手册里说得很清楚,实时循环优先级高于 RTDE 接口,资源紧张时会跳过若干输出包,而且跳过的包不会补发,控制器永远只发最新数据。这个特性对画轨迹没影响(点稀疏一点而已),但对做积分运算的算法就是致命的:你以为拿到了连续 100 帧,实际上是 100 个不等间隔的采样点。
last_ts = None for frame in frames: ts = frame["timestamp"] if last_ts is not None: dt = ts - last_ts if dt > 0.012: # 明显超过 8 ms,记一次跳包 skipped = round(dt / 0.008) - 1 stats["skipped"] += skipped stats["dt"].append(dt) last_ts = ts这段统计我建议常驻在客户端里,隔一段时间打一行日志。它是最便宜的健康度指标:跳包率长期低于 1% 说明链路和控制器负载都正常;一旦超过 5%,就该去看看是不是有别的进程在抢 CPU,或者网络路径上有没有人做深度包检测。
5. 避坑与排查:五条让同步循环悄悄跑偏的记录
5.1 现象、原因与处置
第一条:跑几分钟后字段值开始错乱,重启就好。现象是解析出来的actual_q时不时变成一个量级离谱的数,日志里没有任何异常。原因基本可以锁定在recv上——用了sock.recv(4096)直接当整包,局域网空闲时一次能收全,网络一忙就出现半包,后面的字节被当成新包头解析,整条流错位。处置办法是把接收封装成「先读 4 字节头、按长度读满载荷」的两段式,任何情况下都不假设一次 recv 等于一包。这个改动只有十几行,但能干掉一大类玄学问题。
第二条:写进寄存器的值不生效,或者过一会儿被改回去。现象是你明明发了input_int_register_24 = 1,读回来还是 0,或者短暂变成 1 又变回去。原因是同一个变量被两个客户端同时控制,手册里写死了「同一时刻只允许一个 RTDE 客户端控制某个变量」。现场常见的是示教器上跑着一个后台程序也在写同一个寄存器段。处置办法是先把寄存器分配表画出来,谁写哪一段写清楚;排查阶段可以先把其他客户端 PAUSE 掉,只留一个,确认是不是抢占。
第三条:客户端显示 125 Hz,但画出来的轨迹明显滞后。现象是帧率统计看着正常,实际轨迹比机器人慢半秒以上。原因不在网络,在消费端——handle里做了耗时操作,接收线程被阻塞,内核缓冲区越堆越满。TCP 的缓冲区是有限但很大的,堆到几百帧才反压,这时候延迟已经很难看。处置办法是接收与处理解耦,用一个固定长度的deque(maxlen=N)做缓冲,满了自动丢最旧的;同时给handle里的每一步加计时,超过 2 ms 就告警。
第四条:double 类型字段读出来是 1e-300 或者 0。现象是所有 DOUBLE 和 VECTOR6D 字段都不对,整数字段却正常。原因是字节序搞反了——struct.unpack的格式串没加>,走了本机小端。这个错误有个很典型的特征:整数看着「像那么回事」(因为小端大端对小的正整数只影响高位),浮点数则完全崩坏。处置办法是把格式串统一在一个地方生成,全局搜一遍有没有裸的"d"或"<",并且写一个单元测试,用已知字节序列验证解析结果。
第五条:机器人运动出现轻微抖动,或 RTDE 丢包率突然升高。现象是启用 RTDE 客户端后,原本平滑的运动开始有细微抖动。原因是客户端跑在控制柜那台机器上,和控制器的实时线程抢 CPU。手册里专门提醒过:跑在控制柜上的好处是没有网络延迟,坏处是争抢计算资源;客户端必须以普通操作系统优先级运行,图像处理这类重活应该放到外部 PC。处置办法是把客户端挪到外部 PC,或者至少在控制柜上把进程优先级调低、关掉不必要的后台任务。
5.2 上线前的一张自检表
每次新现场部署,我都会照着下面这张表过一遍,五分钟的事,能省掉后面几小时的扯皮:
| 检查项 | 通过标准 | 不通过的典型表现 |
|---|---|---|
| 端口连通 | 30004 可建立 TCP 连接 | 连接超时或被拒绝 |
| 协议版本 | REQUEST_PROTOCOL_VERSION 返回 1 | 返回 0,后续 setup 全失败 |
| 控制器版本 | 主次版本号与字段表后缀对得上 | 某些字段 setup 时类型码缺失 |
| 字段名列表长度 | 小于 2048 字节 | setup 直接被拒 |
| 首帧解析 | 长度校验通过,数值在合理范围 | 长度不符或出现天文数字 |
| 跳包率 | 连续 5 分钟低于 1% | 周期性出现 24 ms 间隔 |
| 输入生效 | 写入后读回一致 | 写 1 读 0,或被改回 |
| 客户端位置 | 重活不在控制柜上跑 | 机器人抖动、丢包率上升 |
这张表里最容易被跳过的是「首帧解析」和「跳包率」两项,因为它们需要写额外代码。但恰恰是这两项,能在系统还没上线的时候就把字节序错误、字段错位、资源争抢这三类问题暴露出来。
6. 进阶:用寄存器做软件握手,用时间戳量化端到端延迟
配方跑通之后,真正决定这套东西能不能上产线的,是两个工程问题:外部程序怎么知道控制器「已经准备好了」,以及你怎么证明延迟在可接受范围内。
第一个问题用通用寄存器就能解决,不需要任何额外协议。约定output_int_register_24作为状态字、input_int_register_24作为命令字,双方各写各的:控制器侧的程序把状态字置 1 表示「初始化完成」、置 2 表示「正在执行」、置 3 表示「完成」;外部客户端读状态字、写命令字。这套握手的优点是极其轻量,一个 UINT32 就够,而且天然具备「控制器重启后状态归零」的特性,客户端只要发现状态字从非零掉回 0,就知道对面重启了,该重新初始化。
# 客户端侧的握手状态机,跑在 RTDE 同步循环里 STATE_IDLE, STATE_READY, STATE_RUNNING, STATE_DONE = 0, 1, 2, 3 def handshake(frame, cmd): ctrl_state = frame["output_int_register_24"] if ctrl_state == STATE_IDLE and cmd == "start": return None # 控制器还没准备好,继续等 if ctrl_state == STATE_READY and cmd == "start": return {"input_int_register_24": 1} # 发启动命令 if ctrl_state == STATE_DONE: return {"input_int_register_24": 0} # 复位命令字 return None第二个问题稍微麻烦一点。RTDE 的timestamp是控制器时间,本地时钟是另一套基准,两者不能直接相减。可行的办法是做一次单点对齐:在某个时刻记录(控制器 timestamp, 本地 monotonic)作为基准对,之后每一帧都用「当前本地 monotonic − 基准本地」和「当前 timestamp − 基准控制器」做差,差值就是这段时间内累积的漂移。正常情况下这个漂移应该很小(几十毫秒以内),如果持续单调增长,说明链路上有积压,消费端跟不上。
我一般会在客户端里常驻一个滑动窗口,统计最近 500 帧的漂移均值和最大值,超过阈值就打日志。这个指标比单纯的帧率统计有用得多——帧率只告诉你「有没有包」,漂移告诉你「包是不是及时被处理了」。
最后一个习惯:每次换现场,先只订阅timestamp一个字段,空跑十分钟。字段最少,解析路径最短,任何异常都只可能来自网络或调度,排除了业务逻辑的干扰。这十分钟里看三件事:间隔分布是否集中在 8 ms、有没有周期性的长间隔、连接会不会自己断。三件事都干净了,再把业务字段一个个加回去。从那以后我每次接新控制器都强制走一遍这个空跑流程,宁可多花十分钟,也不想在产线调试的半夜对着错位的数据流抓头发。
希望帮到你。
本文还有配套的精品资源,点击获取