diff --git a/scripts/lib/manual_actions.py b/scripts/lib/manual_actions.py new file mode 100644 index 0000000..578ae10 --- /dev/null +++ b/scripts/lib/manual_actions.py @@ -0,0 +1,38 @@ +#!/usr/bin/env python3 +"""人工动作的**唯一措辞**来源 —— 让同一件事在体检报告里只出现一次。 + +为什么单独一个文件 +------------------ +体检报告最后会汇总"要人工做的事",而去重是**按整串文字**做的 +(见 ``real_preflight.Report.manual_actions``)。同一件事如果两处各写一句 +意思相同、字面不同的话,去重就形同虚设 —— 现场看到的是: + + 1. 机器人不在网里:确认网线接控制柜 User Port、PC 与臂同网段, + 先跑 bash scripts/check_robot_network.sh(当前目标 192.168.2.101) + ... + 3. 确认网线接的是控制柜用户口、PC 与机器人同网段 + (先跑 bash scripts/check_robot_network.sh)。 + +两条说的是同一件事,人却要读两遍、还会怀疑"是不是两个不同的问题"。 +真正的根因(`check_network` 判据 + `rdk_status` 的状态分类)分在两个文件里、 +还跑在两个不同的 Python 上(3.11 / 3.10,见 rdk_status 的模块说明), +互相 import 不了,所以把措辞抽到这里:只依赖标准库,谁都能 import。 + +改这里的文字等于改现场看到的操作步骤,动之前先想清楚。 +""" + +from __future__ import annotations + +# 机器人不在网里(ping 不通)。措辞里**不带具体 IP** —— IP 已经在上面的 +# 判据明细里打过了;带上 IP 就会让 check_network 和 rdk_status 的句子不一样, +# 去重立刻失效。 +NET_UNREACHABLE_ACTION = ( + "确认网线接的是控制柜用户口、PC 与机器人同网段" + "(先跑 bash scripts/check_robot_network.sh)。" +) + +# 通了网也连不上 RDK 时的三种常见原因(与厂商报错 a) b) d) 对齐)。 +RDK_REMOTE_MODE_ACTION = "确认示教器上已开启 Remote/Ethernet(RDK)。" +RDK_STACK_HOLDING_ACTION = ( + "如果 ros2_control 已经在跑,RDK 可能被它占着 —— 先停掉栈再试。" +) diff --git a/scripts/lib/rdk_status.py b/scripts/lib/rdk_status.py index 9eb7dbf..b0194cf 100644 --- a/scripts/lib/rdk_status.py +++ b/scripts/lib/rdk_status.py @@ -32,6 +32,16 @@ import sys from pathlib import Path from typing import Dict, List, Optional, Sequence, Tuple +# rdk_status.py 必须在 **3.10** 下跑(flexivrdk 与 ROS 的 3.11 ABI 不兼容), +# 所以这里只 import 标准库 + 同目录下同样只用标准库的 manual_actions。 +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from manual_actions import ( # noqa: E402 + NET_UNREACHABLE_ACTION, + RDK_REMOTE_MODE_ACTION, + RDK_STACK_HOLDING_ACTION, +) + # 结论级别:ok / warn / fail(与 scripts/lib/usb_health.py 同一套词) LEVEL_OK = "ok" LEVEL_WARN = "warn" @@ -65,10 +75,11 @@ def classify_state( LEVEL_FAIL, f"连不上 RDK:{error or '未知原因'}", [ - "确认网线接的是控制柜用户口、PC 与机器人同网段(先跑 " - "bash scripts/check_robot_network.sh)。", - "确认示教器上已开启 Remote/Ethernet(RDK)。", - "如果 ros2_control 已经在跑,RDK 可能被它占着 —— 先停掉栈再试。", + # 措辞来自 manual_actions:体检报告里 check_network 也会给出 + # 同一条,两处字面一致才能去重成一条(否则人读两遍)。 + NET_UNREACHABLE_ACTION, + RDK_REMOTE_MODE_ACTION, + RDK_STACK_HOLDING_ACTION, ], ) if fault: diff --git a/scripts/lib/real_preflight.py b/scripts/lib/real_preflight.py new file mode 100644 index 0000000..d65b01e --- /dev/null +++ b/scripts/lib/real_preflight.py @@ -0,0 +1,722 @@ +#!/usr/bin/env python3 +"""真机开工前的**一次性体检**:把「看起来都起来了,其实动不了」的原因一次说清。 + +为什么必须有这个 +---------------- +2026-09-28 无人值守实测,一次开工踩了三个**互不相干、现象却很像**的坑: + + 1. 示教器还在 Manual 模式 → ``ros2_control`` 永远卡在 wait_for_operational, + 栈起来了、话题在、``/joint_states`` 是空的(全是 0)。现场看起来像"驱动没配好"。 + 2. 总线供电的小 Hub 在内核里卡死 → 挂它下面的四台设备一起消失 + (腕部 D405 + 顶置 D435 + 左右视触觉)。现场看起来像"相机坏了"。 + 3. 上一轮的栈没退干净 → spawner 进程永远等一个不存在的 controller_manager, + 这一轮再起栈,报错指向别处。 + +单独看每一个都不难,但它们**同时存在**时,日志里全是别人的症状,人会先怀疑软件。 +所以这里把它们收成一条命令,按「根因在前、症状在后」输出,并给出**人工要做什么**。 + +设计约束(都是踩过的) +---------------------- +* **绝不挂住**。任何子进程都有超时,sysfs 读取有超时(见 ``usb_health.SysfsReader``)。 + 一个卡死的 USB 设备曾让体检自己挂到超时被杀 —— 那等于什么都没查。 +* **单项失败不连坐**。某个检查抛异常时把它转成一条结论,其余照跑。 +* **人工动作去重后按优先级排**,因为现场是按顺序做的。 + +用法:: + + bash scripts/run_real_preflight.sh --arm 89 + python3 scripts/lib/real_preflight.py --sn Rizon4-063989 # 已在真机环境里时 +""" + +from __future__ import annotations + +import argparse +import json +import os +import shutil +import subprocess +import sys +from dataclasses import dataclass, field, asdict +from pathlib import Path +from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple + +ROOT = Path(__file__).resolve().parents[2] +SCRIPTS_LIB = ROOT / "scripts" / "lib" +sys.path.insert(0, str(SCRIPTS_LIB)) + +# 人工动作的措辞只有这一个来源:同一个文件也被 rdk_status.py import, +# 两处字面一致才能被 Report.manual_actions 去重成一条。 +from manual_actions import NET_UNREACHABLE_ACTION # noqa: E402 + +LEVEL_OK = "ok" +LEVEL_WARN = "warn" +LEVEL_FAIL = "fail" + +#: 级别排序:fail 最先报 +_LEVEL_RANK = {LEVEL_FAIL: 0, LEVEL_WARN: 1, LEVEL_OK: 2} + +DEFAULT_RDK_PYTHON = os.environ.get("RDK_PYTHON") or "/home/cc/rdk_py310/bin/python" + +#: 上一轮没退干净的进程特征。留着它们会让这一轮的错误指向别处: +#: · ros2_control_node 占着 RDK(同一个 SN 是独占访问) +#: · spawner 永远在等一个不存在的 controller_manager,日志里全是 WARN +#: · 相机节点占着 USB(USB 是独占的,新进程只会报"打不开设备") +LEFTOVER_PATTERNS: Tuple[Tuple[str, str], ...] = ( + ("ros2_control_node", "ros2_control 还在跑(它占着 RDK,RDK 对同一台臂是独占的)"), + ("stack.launch", "上一次的 arm_bringup 栈还在"), + ("ros2 run arm_application realsense_color_pub", "相机节点还在跑(USB 独占)"), + ("arm_application/tactile", "触觉服务还在跑(V4L 独占)"), + ("dh_pgea", "夹爪节点还在跑(串口独占)"), + ("controller_manager/spawner", "孤儿 spawner 还在等 controller_manager"), +) + + +@dataclass +class Check: + """一项检查的结论。``level`` 只允许 ok/warn/fail。""" + + key: str + title: str + level: str + lines: List[str] = field(default_factory=list) + #: 需要**人工**做的事(软件做不了的)。会被汇总、去重后按检查顺序输出。 + actions: List[str] = field(default_factory=list) + + def as_dict(self) -> Dict[str, Any]: + return asdict(self) + + +@dataclass +class Report: + checks: List[Check] = field(default_factory=list) + + @property + def level(self) -> str: + if any(c.level == LEVEL_FAIL for c in self.checks): + return LEVEL_FAIL + if any(c.level == LEVEL_WARN for c in self.checks): + return LEVEL_WARN + return LEVEL_OK + + def manual_actions(self) -> List[str]: + """汇总人工动作,去重且保持检查顺序(= 现场该做的顺序)。""" + out: List[str] = [] + for c in self.checks: + for a in c.actions: + if a not in out: + out.append(a) + return out + + def as_dict(self) -> Dict[str, Any]: + return { + "level": self.level, + "checks": [c.as_dict() for c in self.checks], + "manual_actions": self.manual_actions(), + } + + +# --------------------------------------------------------------------------- +# 单项检查 +# --------------------------------------------------------------------------- + + +def _json_tail(text: str) -> Dict[str, Any]: + """从一段可能混着日志的文本里取**最后一个完整的 JSON 对象**。 + + 为什么不 ``json.loads(text[text.find('{'):])``:RDK 的 C++ 日志里也有花括号, + JSON 后面还可能拖日志行,那样解析必然失败 —— 失败之后只能把原始文本打出来, + 结论就丢了(实测就是这个现象)。``raw_decode`` 允许后面有剩余文本, + 所以逐个候选起点试一遍、取最后成功的那个即可。 + """ + dec = json.JSONDecoder() + best: Dict[str, Any] = {} + for i, ch in enumerate(text): + if ch != "{": + continue + try: + obj, _end = dec.raw_decode(text[i:]) + except json.JSONDecodeError: + continue + if isinstance(obj, dict): + best = obj + return best + + +def _run(cmd: Sequence[str], timeout_s: float) -> Tuple[int, str, str]: + """跑一条命令并**一定有超时**。返回 (退出码, stdout, stderr)。 + + 超时返回码 124(与 GNU timeout 一致)。体检里任何一处不能无限等 —— + 卡死的 USB / 半连的机器人都会让普通的 subprocess.run 永远不返回。 + + stdout / stderr **分开返回**(不合并):RDK 自己的日志走 stderr, + 而 ``rdk_status.py --json`` 的 JSON 在 stdout。合并之后 JSON 后面会拖一串 + 日志,``json.loads`` 必然失败 —— 实测踩过,表现为"体检说读不到 RDK 状态", + 而实际那句话就在输出里。 + """ + try: + p = subprocess.run( + list(cmd), capture_output=True, text=True, timeout=timeout_s + ) + except subprocess.TimeoutExpired as exc: + out = exc.stdout or "" + err = exc.stderr or "" + if isinstance(out, bytes): # pragma: no cover - text=True 时不会走到 + out = out.decode(errors="replace") + if isinstance(err, bytes): # pragma: no cover + err = err.decode(errors="replace") + return 124, out, (err + f"\n超时({timeout_s:.0f}s):{' '.join(cmd)}") + except FileNotFoundError as exc: + return 127, "", str(exc) + return p.returncode, p.stdout or "", p.stderr or "" + + +def check_leftovers( + ps_output: str, running_cmds: Sequence[str] = () +) -> Check: + """上一轮的残留进程。 + + ``running_cmds`` 是**当前进程自己**的命令行(用于排除自己)—— + 以前用 ``pkill -f`` 清理时把自己的 shell 也匹配掉了,教训是: + 匹配规则必须能区分"别人的进程"和"我这条命令"。这里改成显式排除。 + """ + hits: List[str] = [] + for line in ps_output.splitlines(): + if not line.strip(): + continue + if any(frag in line for frag in running_cmds): + continue + for pat, why in LEFTOVER_PATTERNS: + if pat in line and why not in hits: + hits.append(why) + if not hits: + return Check("leftovers", "遗留进程", LEVEL_OK, ["没有上一轮残留的进程"]) + return Check( + "leftovers", + "遗留进程", + LEVEL_WARN, + ["上一轮有进程没退干净,这一轮的现象会指向别处:"] + [f" · {h}" for h in hits], + actions=[ + "先清干净再起栈:pkill -f '[r]os2_control_node';" + "pkill -f '[a]rm_bringup.*stack.launch';pkill -f '[s]pawner'" + ], + ) + + +def check_network(ping_code: int, ping_out: str, arp_lines: Sequence[str], ip: str) -> Check: + """机器人是否在网里。 + + 额外查一条**现场特有的坑**:两台臂的 ARP 指向同一个 MAC。2026-09-28 实测 + 到一次:``.100``(91)与 ``.101``(89)在 ARP 表里是**同一个 MAC** + ``88:88:88:88:87:88``,而 ping ``.100`` 稳定、ping ``.101`` 丢 50~90%, + 且**先 ping 一次 .100 之后 .101 丢包立刻加重**。这是同一台设备(或两台同 + MAC 的设备)在同一 L2 段上争抢转发 —— 它不是"网络不通",而是"通得不稳", + 足以让 RDK 的握手时好时坏。 + + 这条只涉及**目标 IP 自己**的撞车才报;与目标无关的撞车只在 PING 不通时 + 作为解释提一句,避免正常的双机配置被误报。 + """ + lines: List[str] = [] + actions: List[str] = [] + level = LEVEL_OK + + last = [ln for ln in ping_out.splitlines() if ln.strip()] + if ping_code == 0: + lines.append(f"ping {ip}: 通") + else: + level = LEVEL_FAIL + lines.append(f"ping {ip}: 不通") + if last: + lines.append(f" {last[-1].strip()[:160]}") + actions.append(NET_UNREACHABLE_ACTION) + + # 同一 MAC 对应多个 IP:按目标 IP 是否牵涉其中决定措辞与级别。 + macs: Dict[str, List[str]] = {} + for line in arp_lines: + parts = line.split() + if len(parts) >= 5 and parts[0].startswith("192.168."): + macs.setdefault(parts[4].lower(), []).append(parts[0]) + for mac, ips in macs.items(): + if len(ips) < 2: + continue + involves_target = ip in ips + lines.append( + f" ARP:{' / '.join(sorted(ips))} 指向同一个 MAC {mac}" + ) + if involves_target: + # 目标币自己牵涉其中:这是"通得不稳"的根因候选,比 ping 失败更值钱。 + if level == LEVEL_OK: + level = LEVEL_WARN + actions.append( + f"两个 IP({' / '.join(sorted(ips))})在 ARP 里是同一个 MAC —— 要么是" + "同一台设备配了两个地址,要么是两台臂的 MAC 出厂就相同(同一 L2 段" + "会互相抢转发,表现为 ping 通但丢包、RDK 时连时断)。" + "先把另一台臂的网线拔掉/换到独立网段,只留当前这台,再重跑体检。" + ) + if not involves_target and level == LEVEL_FAIL: + lines.append(" (与目标 IP 无关,仅供参考)") + break + + return Check("network", "机器人在网", level, lines, actions=actions) + + +def check_arm_rdk(payload: Dict[str, Any], returncode: int) -> Check: + """机械臂能不能被使能(这是"能不能动"的唯一权威判据)。 + + 状态词与处置来自厂商 ``flexiv/rdk/data.hpp`` 的原话: + ``IN_MANUAL_MODE`` = "In Manual mode, need to switch to Auto (Remote) mode." + """ + if returncode == 124: + return Check( + "arm", + "机械臂 RDK", + LEVEL_FAIL, + ["读 RDK 状态超时 —— 通常是别的进程占着 RDK(见上一条遗留进程)"], + actions=["先清掉占用 RDK 的进程(ros2_control / 上一轮的栈),再重试"], + ) + level = str(payload.get("level") or LEVEL_FAIL) + lines = [str(payload.get("summary") or payload.get("error") or "")] + if payload.get("connected"): + lines.append( + f" mode={payload.get('mode', '?')} " + f"operational={payload.get('operational')} " + f"op_status={payload.get('op_status', '?')} " + f"fault={payload.get('fault')} estop_released={payload.get('estop_released')}" + ) + return Check( + "arm", + "机械臂 RDK", + level if level in _LEVEL_RANK else LEVEL_FAIL, + lines, + actions=[str(a) for a in (payload.get("actions") or [])], + ) + + +def check_usb(reader: Any = None) -> Check: + """USB 链路:RealSense 协商速率,以及 Hub 是否在内核里卡死。""" + import usb_health + + rd = reader or usb_health.SysfsReader() + devices = usb_health.read_devices(reader=rd) + findings = usb_health.findings_for(devices, rd.blocked) + level = LEVEL_FAIL if usb_health.has_errors(findings) else ( + LEVEL_WARN if any(f.level == LEVEL_WARN for f in findings) else LEVEL_OK + ) + actions = [] + if rd.blocked: + actions.append( + "某个 USB Hub 在内核里卡死(见上面 FAIL):给这个 Hub 断电重启," + "或把设备改插到主板后面板直连的口;软件复位需要 root" + ) + if any(f.level == "error" and "都没枚举到" in f.text for f in findings): + actions.append("一台 RealSense 都没枚举到:拔插一次,并确认没有被别的进程占着") + return Check( + "usb", + "USB 链路(相机)", + level, + [f"[{f.level}] {f.text}" for f in findings], + actions=actions, + ) + + +def check_v4l(pairs: Sequence[Tuple[str, str]], v4l_root: Optional[Path] = None) -> Check: + """视触觉的 V4L2 设备在不在,以及是"没插"还是"卡死"。""" + import v4l_health + + if not pairs: + return Check("v4l", "触觉相机(V4L2)", LEVEL_OK, ["本臂没有配触觉,跳过"]) + checks = ( + v4l_health.check_devices(pairs) + if v4l_root is None + else v4l_health.check_devices(pairs, v4l_root=v4l_root) + ) + level = LEVEL_FAIL if any(c.level == "error" for c in checks) else LEVEL_OK + actions = [] + if any(c.state == "wedged" for c in checks): + actions.append( + "触觉相机在内核里卡死(sysfs 还在、/dev/video* 没了):`sudo -n " + "/usr/local/bin/orisys-usb-reset ` 就是为这种情况准备的" + "(免密白名单,无人值守可直接跑);它也不行才拔插该 USB 口" + ) + elif level == LEVEL_FAIL: + actions.append("触觉相机没枚举到:确认插好(by-path 是按口认的,换口就会失配)") + return Check( + "v4l", "触觉相机(V4L2)", level, [c.text() for c in checks], actions=actions + ) + + +def check_cameras( + site_path: Path, sn: str, devices: Optional[Sequence[Any]] = None +) -> Check: + """本臂配置的**腕部相机**(eye-in-hand 那只)必须在 USB 上。 + + 为什么单独一条判据 + ------------------ + ``check_usb`` 只看 RealSense 的**协商速率**,对"这只相机**根本不在了**" + 没有话可说 —— 它遍历的是**发现到的**设备。2026-09-28 现场实测到这只: + + lsusb 里只剩顶置 D435,d405(89 的腕部相机、装配全程唯一的位置来源) + 整个消失,而体检报告**全绿**。 + + 也就是说,把相机碰掉了、或者它所在的 Hub 掉电了,报告照样说"可以开工"。 + 缺 D405 之后,踩的是完全另一类错误(点云为空 / 找不到工件),排查方向 + 会被带偏很远。所以这里按 site.yaml 的**配置意图**去核在不在: + ``robots[].camera.name`` → ``cameras..model``,比的是型号。 + + 为什么比型号而不是序列号:实测这台 D435 的 sysfs ``serial`` 是**空串** + (UVC 不一定把序列号透出来),拿序列号判会永远判"不在"。型号 + (``d405`` / ``d435``)来自 USB product 串,稳定得多。序列号只用来 + 在**型号对得上**时做一次"是不是本臂那只"的提醒(WARN,不阻断)。 + """ + import export_site_env as ex + + try: + data = ex.load_site(site_path) + _merged, robot = ex.apply_active_arm(data, sn) + except Exception as exc: # noqa: BLE001 + return Check( + "cameras", + "腕部相机(USB)", + LEVEL_WARN, + [f"读不了 {site_path} 或找不到 SN={sn}:{type(exc).__name__}: {exc}"], + ) + + cam_name = str(((robot.get("camera") or {}) or {}).get("name") or "") + if not cam_name: + return Check( + "cameras", "腕部相机(USB)", LEVEL_OK, ["本臂没配腕部相机(camera.name),跳过"] + ) + cam = (data.get("cameras") or {}).get(cam_name) or {} + if "sim" in cam_name or "sim" in str(cam.get("role") or ""): + return Check( + "cameras", + "腕部相机(USB)", + LEVEL_OK, + [f"本臂腕部相机 {cam_name} 是仿真源(真机不读),跳过"], + ) + + want_model = str(cam.get("model") or "").lower() + want_serial = str(cam.get("serial") or "") + if not want_model: + return Check( + "cameras", + "腕部相机(USB)", + LEVEL_WARN, + [f"cameras.{cam_name} 没写 model,无法核对是否插着"], + actions=[f"在 site.yaml 的 cameras.{cam_name} 里补上 model(如 d405)"], + ) + + if devices is None: + import usb_health + + devices = usb_health.read_devices(reader=usb_health.SysfsReader()) + + found = [d for d in devices if getattr(d, "model", "") == want_model] + realsense = [d for d in devices if getattr(d, "is_realsense", False)] + lines = [ + f"本臂腕部相机:{cam_name}(期望 {want_model}" + + (f",SN {want_serial}" if want_serial else "") + + ")", + "USB 上现在有的 RealSense:" + + ( + "、".join(f"{d.model or d.product}({d.path})" for d in realsense) + or "一台都没有" + ), + ] + if not found: + return Check( + "cameras", + "腕部相机(USB)", + LEVEL_FAIL, + lines, + actions=[ + f"本臂腕部相机 {want_model}({cam_name})没在 USB 上 —— 装配全程的位置" + f"都来自它,没有它后面每一步都会以别的样子报错。查:相机线/腕部延长线" + f"插紧(它常挂在腕部/第三轴的活动线上,反复弯折最先断)," + f"USB Hub 是否还在(lsusb),换一个口再 lsusb 确认。" + ], + ) + + level = LEVEL_OK + # 型号对上了,再看序列号对不对:两台臂各有一只同型号,插反了要能看出来。 + serials = [d.serial for d in found if d.serial] + if want_serial: + if serials and want_serial not in serials: + level = LEVEL_WARN + lines.append( + f" [warn] 型号对得上,但序列号是 {serials},配置里写的是 {want_serial}" + " —— 可能把另一台臂的腕部相机插到了本臂上" + ) + elif not serials: + lines.append(" (这只相机 sysfs 里没透出序列号,无法核对是不是本臂那只)") + return Check("cameras", "腕部相机(USB)", level, lines) + + +def check_geometry(site_path: Path, sn: str) -> Check: + """手眼矩阵自洽 + 悬停高度会不会把工具捅进工件。""" + import real_geometry_check + + try: + rep = real_geometry_check.geometry_report(site_path, sn) + except Exception as exc: # noqa: BLE001 —— 单项失败不能连坐 + return Check( + "geometry", + "几何(手眼 / 悬停)", + LEVEL_WARN, + [f"跑不了几何检查:{type(exc).__name__}: {exc}"], + actions=["确认工作区已构建(install/ 里有 arm_application),再重跑体检"], + ) + lines: List[str] = [] + problems: List[str] = [] + he = rep["handeye"] + lines.append(f"手眼:{he.get('file')}") + if he.get("translation_m"): + t = he["translation_m"] + lines.append(f" 平移 [{t[0]:+.4f}, {t[1]:+.4f}, {t[2]:+.4f}] m") + for p in he.get("problems") or []: + problems.append(p) + lines.append(f" [fail] {p}") + if not (he.get("problems") or []): + lines.append(" [ok ] 矩阵自洽(正交、det=+1、量级合理)") + + hv = rep["hover"] + lines.append( + f"悬停:法兰 {hv['hole_hover_z_m']:.4f} − 工具长 " + f"{hv['tool_tip_below_flange_m'] * 1000:.1f}mm,工件面 " + f"{hv['workpiece_surface_z_m']:.4f}" + + ("(未标定)" if hv.get("uncalibrated_surface") else "") + ) + for p in hv.get("problems") or []: + problems.append(p) + lines.append(f" [fail] {p}") + actions = [] + if hv.get("uncalibrated_surface"): + actions.append( + "实测工件上表面 z 并写回 site.yaml 的 robots[].gauge_assembly." + "workpiece_surface_z_m(跑 scripts/run_stage2_locate.sh 会把它打出来)" + ) + return Check( + "geometry", + "几何(手眼 / 悬停)", + LEVEL_FAIL if problems else LEVEL_OK, + lines, + actions=actions, + ) + + +def check_environment(site_path: Path, sn: str) -> Check: + """把这一轮到底在读哪份配置、认哪台设备打出来(换臂/换口最容易错)。""" + import export_site_env as ex + + try: + data = ex.load_site(site_path) + _merged, robot = ex.apply_active_arm(data, sn) + except Exception as exc: # noqa: BLE001 + return Check( + "env", + "现场配置", + LEVEL_FAIL, + [f"读不了 {site_path} 或找不到 SN={sn}:{exc}"], + actions=[f"确认 site.yaml 与 --arm/--sn 一致(当前 {sn})"], + ) + lines = [ + f"site={site_path}", + f"SN={ex.resolve_robot_sn(data, sn)[0]} " + f"arm={robot.get('arm')} {robot.get('label') or ''}".rstrip(), + f"IP={robot.get('ip')} iface={robot.get('iface') or (data.get('net') or {}).get('iface')}", + ] + cams = data.get("cameras") or {} + for key, cam in cams.items(): + if not isinstance(cam, dict): + continue + res = f"{cam.get('width')}x{cam.get('height')}" + if cam.get("fps"): + res += f"@{cam['fps']}" + role = str(cam.get("role") or "") + tag = "(仿真用,真机不读)" if "sim" in role or str(key).startswith("sim_") else "" + lines.append(f"cameras.{key}: serial={cam.get('serial')} {res} {role}{tag}".rstrip()) + grip = robot.get("gripper") or {} + if grip: + lines.append( + f"gripper: {grip.get('model')} port={grip.get('port')} " + f"auto_init={grip.get('auto_init')}" + ) + if grip.get("auto_init"): + lines.append( + " 注意 auto_init=true:上电会先把手指张到满行程," + "爪里已经夹着工件时会把工件放掉" + ) + return Check("env", "现场配置", LEVEL_OK, lines) + + +# --------------------------------------------------------------------------- +# 编排 +# --------------------------------------------------------------------------- + + +def _ps_output() -> str: + exe = shutil.which("ps") or "/bin/ps" + code, out, _err = _run([exe, "-eo", "pid,cmd"], 5.0) + return out if code == 0 else "" + + +def run_all( + site_path: Path, + sn: str, + rdk_python: str = DEFAULT_RDK_PYTHON, + iface: str = "", +) -> Report: + """跑完整套体检。顺序 = 输出的顺序 = 现场该处理的顺序。""" + rep = Report() + + # 0) 现场配置:先把"在读哪份配置"钉住,后面所有结论都依赖它。 + env_check = check_environment(site_path, sn) + rep.checks.append(env_check) + + # 1) 遗留进程:它会让后面每一项的现象都指向别处,所以先查。 + rep.checks.append( + check_leftovers(_ps_output(), running_cmds=(sys.argv[0] if sys.argv else "",)) + ) + + # 2) 网络:RDK 连不上的前提条件。 + ip = "" + try: + import export_site_env as ex + + data = ex.load_site(site_path) + _merged, robot = ex.apply_active_arm(data, sn) + ip = str(robot.get("ip") or "") + iface = iface or str(robot.get("iface") or (data.get("net") or {}).get("iface") or "") + except Exception: # noqa: BLE001 + pass + if ip: + code, out, _err = _run(["ping", "-c", "2", "-W", "1", ip], 8.0) + _arp_code, arp, _arp_err = _run(["ip", "neigh", "show"], 5.0) + rep.checks.append(check_network(code, out, arp.splitlines() if arp else [], ip)) + + # 3) 机械臂:能不能使能。 + status_py = SCRIPTS_LIB / "rdk_status.py" + if status_py.is_file(): + cmd = [rdk_python, str(status_py), "--sn", sn, "--json"] + if iface: + cmd += ["--iface", iface] + # 磁盘上没有的 SN 会走完整搜索超时;给足但绝不无限等。 + code, out, err = _run(cmd, 90.0) + payload = _json_tail(out) or _json_tail(err) or {} + if not payload: + payload = { + "level": LEVEL_FAIL, + "error": (out + err).strip()[-400:] or "读不到 RDK 状态", + } + rep.checks.append(check_arm_rdk(payload, code)) + + # 4) USB / 触觉 / 几何:都不需要使能机械臂,随时可查。 + # 腕部相机排在 USB 之前:USB 那条只说"速率对不对",这条说"在不在", + # 而"不在"是更硬的阻断,必须先报。 + for fn in ( + lambda: check_cameras(site_path, sn), + lambda: check_usb(), + lambda: check_v4l(_tactile_pairs(site_path, sn)), + lambda: check_geometry(site_path, sn), + ): + try: + rep.checks.append(fn()) + except Exception as exc: # noqa: BLE001 —— 单项失败不连坐 + rep.checks.append( + Check( + "unknown", + "检查项异常", + LEVEL_WARN, + [f"{type(exc).__name__}: {exc}"], + ) + ) + + rep.checks.sort(key=lambda c: _LEVEL_RANK.get(c.level, 3)) + return rep + + +def _tactile_pairs(site_path: Path, sn: str) -> List[Tuple[str, str]]: + """从 site.yaml 取本臂触觉的 by-path(按口认设备,换口就会失配)。 + + 注意 ``robots[].tactile`` 的写法有两种:布尔开关(本臂装没装)和 + 参数覆盖字典。以前只按字典处理,遇到 ``tactile: true`` 就 + ``dict.update(True)`` 抛 TypeError —— 现象是"整个触觉检查项没跑", + 而不是报错,很容易被当成"这台没触觉"。 + """ + import export_site_env as ex + + try: + data = ex.load_site(site_path) + _merged, robot = ex.apply_active_arm(data, sn) + except Exception: # noqa: BLE001 + return [] + tac = dict(data.get("tactile") or {}) + override = robot.get("tactile") + if isinstance(override, dict): + tac.update(override) + elif override is not None and not override: + return [] # 显式关掉 + if not tac: + return [] + pairs: List[Tuple[str, str]] = [] + for side, key in (("左指", "left_video"), ("右指", "right_video")): + path = str(tac.get(key) or "") + if path: + pairs.append((side, path)) + return pairs + + +def render(rep: Report) -> str: + marks = {LEVEL_OK: "OK ", LEVEL_WARN: "WARN", LEVEL_FAIL: "FAIL"} + out: List[str] = [] + for c in rep.checks: + out.append(f"[{marks.get(c.level, '? ')}] {c.title}") + for line in c.lines: + if line: + out.append(f" {line}") + out.append("") + out.append("=" * 68) + if rep.level == LEVEL_OK: + out.append("结论:GO —— 所有检查通过,可以起栈跑流程。") + else: + out.append( + f"结论:NO-GO({'有硬阻断' if rep.level == LEVEL_FAIL else '有告警'})" + ) + acts = rep.manual_actions() + if acts: + out.append("") + out.append("要人工做的事(按顺序,软件做不了的部分):") + for i, a in enumerate(acts, 1): + out.append(f" {i}. {a}") + else: + out.append(" (没有需要人工处理的项,见上面 WARN 的具体内容)") + return "\n".join(out) + + +def main(argv: Optional[Sequence[str]] = None) -> int: + ap = argparse.ArgumentParser(description="真机开工前的一次性体检") + ap.add_argument("--sn", default="", help="机械臂 SN 或别名(89/91)") + ap.add_argument("--site", default=str(ROOT / "config" / "site.yaml")) + ap.add_argument("--iface", default="") + ap.add_argument("--rdk-python", default=DEFAULT_RDK_PYTHON) + ap.add_argument("--json", action="store_true") + ap.add_argument("--out", default="", help="把结论写到这个文件(JSON)") + args = ap.parse_args(argv) + + if not args.sn: + ap.error("--sn 必填(89 / 91 / Rizon4-XXXXXX)") + + rep = run_all(Path(args.site), args.sn, args.rdk_python, args.iface) + if args.json: + print(json.dumps(rep.as_dict(), ensure_ascii=False, indent=2)) + else: + print(render(rep)) + if args.out: + Path(args.out).write_text( + json.dumps(rep.as_dict(), ensure_ascii=False, indent=2) + "\n", + encoding="utf-8", + ) + return 0 if rep.level == LEVEL_OK else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/run_real_preflight.sh b/scripts/run_real_preflight.sh new file mode 100755 index 0000000..dd2bc00 --- /dev/null +++ b/scripts/run_real_preflight.sh @@ -0,0 +1,80 @@ +#!/usr/bin/env bash +# 真机开工前的**一次性体检**(一条命令说清"能不能开工、不能的话人要去做什么")。 +# +# bash ~/ros2/scripts/run_real_preflight.sh --arm 89 +# bash ~/ros2/scripts/run_real_preflight.sh --arm 89 --json # 给脚本消费 +# bash ~/ros2/scripts/run_real_preflight.sh --arm 89 --watch-usb 10 +# +# 查什么(顺序就是现场该处理的顺序:根因在前,症状在后): +# 1. 现场配置:这一轮在读哪份 site.yaml、认哪台臂/哪台相机/哪个串口 +# 2. 遗留进程:上一轮没退干净的栈会让这一轮的现象指向别处 +# 3. 网络:机器人是否在网(含"两台臂 ARP 撞同一个 MAC"这个坑) +# 4. 机械臂 RDK:能不能使能(Manual 模式是**硬阻断**,只能人工在示教器上切) +# 5. USB:RealSense 协商速率 + Hub 是否在内核里卡死 +# 6. 触觉 V4L2:设备在不在,是"没插"还是"卡死" +# 7. 几何:手眼矩阵是否自洽 + 悬停高度会不会把工具捅进工件 +# +# 退出码:0 = 全绿(GO);1 = 有需要处理的项(NO-GO,看输出里"要人工做的事")。 +# +# 为什么要有这个而不是让人挨个跑:这些故障**同时出现时现象会互相串** +#(Hub 卡死会让相机消失,相机消失又会让 S00 自检报"缺相机", +# 而真正拦住机械臂的其实是示教器模式)。分开查会让人先怀疑软件。 +# +# 注意:这里**不加** `set -u` —— conda/RoboStack 的钩子会引用未绑定的变量。 +set -eo pipefail + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +WATCH_USB=0 +JSON=false + +# --sn/--arm 必须先解析:load_site_config.sh 靠 ROBOT_SN 决定选哪台臂 +# shellcheck disable=SC1091 +source "${ROOT}/scripts/lib/parse_arm_args.sh" +parse_arm_args "$@" +set -- "${PARSE_ARM_REST[@]}" + +while [[ $# -gt 0 ]]; do + case "$1" in + --watch-usb) WATCH_USB="${2:?--watch-usb 需要秒数}"; shift 2 ;; + --json) JSON=true; shift ;; + -h|--help) sed -n '2,20p' "$0"; exit 0 ;; + *) echo "Unknown: $1" >&2; exit 1 ;; + esac +done + +# shellcheck disable=SC1091 +source "${ROOT}/scripts/env.sh" >/dev/null 2>&1 || true +# shellcheck disable=SC1091 +source "${ROOT}/scripts/lib/load_site_config.sh" + +# site.yaml 不存在时不要直接失败:网络与 RDK 自检在配置前就该能用, +# 而"配置没写"本身也是体检要报出来的一项。 +if [[ "${SITE_LOAD_STATUS}" == "2" ]]; then + echo "ERROR: site.yaml 有问题,先按上面的提示修好再体检。" >&2 + exit 2 +fi + +SN="${SITE_ROBOT_SN:-${ROBOT_SN:-}}" +if [[ -z "${SN}" ]]; then + echo "ERROR: 没选到机械臂。给命令加 --arm <臂号>,或在 site.yaml 里填 robot.sn。" >&2 + exit 2 +fi + +echo "=== 真机开工体检 arm=${SITE_ROBOT_ARM:-?} ${SITE_ROBOT_LABEL:-} (${SN}) ===" +echo + +ARGS=(--sn "${SN}" --site "${SITE_CONFIG:-${ROOT}/config/site.yaml}") +[[ -n "${SITE_NET_IFACE:-}" ]] && ARGS+=(--iface "${SITE_NET_IFACE}") +[[ "${JSON}" == true ]] && ARGS+=(--json) + +RC=0 +if [[ "${WATCH_USB}" != "0" ]]; then + # 先单独盯一会儿 USB:Hub 半死时会"设备在 / 不在"反复跳, + # 单次采样只会看到一个瞬间,看不出它在抖。 + echo "--- USB 链路盯 ${WATCH_USB}s(看掉线/重连)---" + python3 "${ROOT}/scripts/lib/usb_health.py" --watch "${WATCH_USB}" || RC=$? + echo +fi + +python3 "${ROOT}/scripts/lib/real_preflight.py" "${ARGS[@]}" || RC=$? +exit "${RC}" diff --git a/src/arm_application/test/test_real_preflight.py b/src/arm_application/test/test_real_preflight.py new file mode 100644 index 0000000..49e5616 --- /dev/null +++ b/src/arm_application/test/test_real_preflight.py @@ -0,0 +1,456 @@ +"""开工体检的编排判据:**根因必须排在症状前面**,且任何一项都不能把体检拖垮。 + +这个文件钉三件事,都是这次无人值守实测踩出来的: + + 1. **顺序**。现场同时存在多个故障时,日志里全是"别人的症状": + 示教器还在 Manual(拦住机械臂)、总线 Hub 在内核里卡死(相机全没了)、 + 上一轮栈没退干净(spawner 永远在等)。人看到的是"相机不见了", + 于是去查相机 —— 而真正的硬阻断是示教器模式。所以 fail 必须排在最前, + 并且"要人工做的事"要按这个顺序给。 + 2. **不挂**。一个卡死的 USB 设备曾让体检自己挂到超时被杀(sysfs 读取永久阻塞)。 + 所以任何外部调用都必须有超时,且超时要变成一条结论。 + 3. **不连坐**。某一项自己抛异常时,其余检查必须照跑 —— 否则一个坏掉的 + 可选检查会让"机械臂能不能动"这个最关键的问题没有答案。 +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[3] / "scripts" / "lib")) + +import real_preflight as pf # noqa: E402 +from real_preflight import Check, Report, check_leftovers, check_network # noqa: E402 + + +# --------------------------------------------------------------------------- +# 汇总与顺序 +# --------------------------------------------------------------------------- + + +def test_level_is_fail_if_anything_fails() -> None: + rep = Report([Check("a", "A", pf.LEVEL_OK), Check("b", "B", pf.LEVEL_FAIL)]) + assert rep.level == pf.LEVEL_FAIL + rep2 = Report([Check("a", "A", pf.LEVEL_OK), Check("b", "B", pf.LEVEL_WARN)]) + assert rep2.level == pf.LEVEL_WARN + assert Report([Check("a", "A", pf.LEVEL_OK)]).level == pf.LEVEL_OK + + +def test_manual_actions_are_deduped_and_ordered() -> None: + """同一句建议被两项都提到时只出现一次(现场按清单做,重复会让人以为漏了)。""" + rep = Report( + [ + Check("a", "A", pf.LEVEL_FAIL, actions=["插网线", "开关"]), + Check("b", "B", pf.LEVEL_FAIL, actions=["插网线"]), + ] + ) + assert rep.manual_actions() == ["插网线", "开关"] + + +def test_root_cause_is_reported_before_the_symptom() -> None: + """一条 fail 的"根因"与一条 fail 的"症状"同时存在时,两者都要在, + 但建议清单里根因的处置必须排在前面。""" + root = Check( + "arm", + "机械臂 RDK", + pf.LEVEL_FAIL, + ["示教器在 Manual 模式"], + actions=["示教器切到 Auto (Remote)"], + ) + symptom = Check( + "usb", "USB", pf.LEVEL_FAIL, ["一台相机都没枚举到"], actions=["拔插相机"] + ) + rep = Report([symptom, root]) + rep.checks.sort(key=lambda c: pf._LEVEL_RANK.get(c.level, 3)) # run_all 的做法 + assert [c.key for c in rep.checks] == ["usb", "arm"] # 同级保持稳定顺序 + # 顺序由 run_all 决定(根因检查项先加入),这里只钉住"根因那条在前"的编排意图: + assert pf.Report([root, symptom]).manual_actions()[0] == "示教器切到 Auto (Remote)" + + +def test_run_all_puts_arm_before_symptom_checks() -> None: + """真编排里的顺序:机械臂(硬阻断)在 USB/触觉/几何之前。""" + src = (Path(pf.__file__)).read_text(encoding="utf-8") + body = src[src.index("def run_all") : src.index("def _tactile_pairs")] + assert body.index("check_arm_rdk") < body.index("check_usb"), ( + "机械臂(能不能动)必须排在 USB 之前 —— 它是唯一的人工硬阻断" + ) + assert body.index("check_leftovers") < body.index("check_network"), ( + "遗留进程要先查:它会让后面每一项的现象都指向别处" + ) + + +def test_render_says_go_only_when_everything_is_ok() -> None: + go = pf.render(Report([Check("a", "A", pf.LEVEL_OK, ["一切正常"])])) + assert "GO" in go and "NO-GO" not in go + nogo = pf.render( + Report([Check("a", "A", pf.LEVEL_FAIL, ["坏了"], actions=["去修"])]) + ) + assert "NO-GO" in nogo and "要人工做的事" in nogo and "1. 去修" in nogo + + +# --------------------------------------------------------------------------- +# 单项判据 +# --------------------------------------------------------------------------- + + +def test_leftovers_reports_orphan_spawner_and_ros2_control() -> None: + ps = ( + " 100 /x/lib/controller_manager/spawner joint_state_broadcaster\n" + " 200 /y/lib/controller_manager/ros2_control_node\n" + " 300 /usr/bin/htop\n" + ) + c = check_leftovers(ps) + assert c.level == pf.LEVEL_WARN + assert any("ros2_control" in ln for ln in c.lines) + assert c.actions, "要说清怎么清,不能只说'有残留'" + + +def test_leftovers_does_not_match_its_own_command_line() -> None: + """曾经的坑:pkill -f 把自己的 shell 也匹配掉了(命令里含同样的字符串)。""" + ps = " 42 /bin/bash -c pkill -f '[a]rm_bringup.*stack.launch'\n" + c = check_leftovers(ps, running_cmds=("/bin/bash -c pkill",)) + assert c.level == pf.LEVEL_OK + + +def test_network_ok_when_ping_works() -> None: + c = check_network(0, "3 received", [], "192.168.2.101") + assert c.level == pf.LEVEL_OK and not c.actions + + +def test_network_flags_same_mac_for_two_robot_ips() -> None: + """2026-09-28 实测:.100/.101 是同一个 MAC,ping .101 丢 50~90%。""" + arp = [ + "192.168.2.100 dev enp4s0 lladdr 88:88:88:88:87:88 REACHABLE", + "192.168.2.101 dev enp4s0 lladdr 88:88:88:88:87:88 REACHABLE", + ] + c = check_network(0, "ok", arp, "192.168.2.101") + assert c.level == pf.LEVEL_WARN, "ping 通但同 MAC 抢转发,必须提一句" + assert any("88:88:88:88:87:88" in ln for ln in c.lines) + assert c.actions and "拔掉" in c.actions[0] + # 与目标无关的同 MAC 不打扰(正常双机配置不该被误报) + c2 = check_network(0, "ok", arp, "192.168.2.55") + assert c2.level == pf.LEVEL_OK + + +def test_network_ping_fail_is_fail_and_shows_only_the_summary_line() -> None: + out = "PING 192.168.2.101 (192.168.2.101) 56(84) bytes of data.\n\n--- statistics ---\n2 packets transmitted, 0 received, 100% packet loss\n" + c = check_network(1, out, [], "192.168.2.101") + assert c.level == pf.LEVEL_FAIL + assert not any("bytes of data" in ln for ln in c.lines), "不要把 ping 原文刷进来" + + +# --------------------------------------------------------------------------- +# RDK 结论的翻译 +# --------------------------------------------------------------------------- + + +def test_arm_manual_mode_is_a_hard_block_with_a_manual_action() -> None: + payload = { + "connected": True, + "level": "fail", + "summary": "示教器在 Manual 模式(operational_status=OperationalStatus.IN_MANUAL_MODE)", + "actions": ["示教器切到 Auto (Remote)"], + "mode": "IDLE", + "operational": False, + "op_status": "OperationalStatus.IN_MANUAL_MODE", + "fault": False, + "estop_released": True, + } + c = pf.check_arm_rdk(payload, 0) + assert c.level == pf.LEVEL_FAIL + assert c.actions == ["示教器切到 Auto (Remote)"] + assert any("IN_MANUAL_MODE" in ln for ln in c.lines) + + +def test_arm_timeout_points_at_the_running_stack() -> None: + c = pf.check_arm_rdk({}, 124) + assert c.level == pf.LEVEL_FAIL + assert any("超时" in ln for ln in c.lines) + assert c.actions, "超时也要给下一步(RDK 被占用是最常见原因)" + + +def test_unknown_level_from_payload_is_treated_as_fail() -> None: + """级别词表外的值不能当成功 —— 脚本按 ok/warn/fail 判定。""" + c = pf.check_arm_rdk({"level": "weird"}, 0) + assert c.level == pf.LEVEL_FAIL + + +# --------------------------------------------------------------------------- +# JSON 提取(RDK 的日志会混进 stdout/stderr) +# --------------------------------------------------------------------------- + + +def test_json_tail_survives_surrounding_logs() -> None: + text = ( + '[info] Flexiv RDK v1.9.1 | {not json}\n' + '{"connected": true, "level": "ok"}\n' + "[info] >>>>> Flexiv RDK exited <<<<<\n" + ) + assert pf._json_tail(text) == {"connected": True, "level": "ok"} + + +def test_json_tail_picks_the_last_object() -> None: + text = '{"a": 1}\nnoise\n{"b": 2}' + assert pf._json_tail(text) == {"b": 2} + + +def test_json_tail_returns_empty_when_there_is_nothing() -> None: + assert pf._json_tail("no json here") == {} + assert pf._json_tail("") == {} + + +def test_json_tail_is_not_confused_by_braces_in_logs() -> None: + text = "warning: {unknown token} in config\n" '{"level": "fail", "error": "x"}' + assert pf._json_tail(text)["level"] == "fail" + + +# --------------------------------------------------------------------------- +# 触觉配置的两种写法 +# --------------------------------------------------------------------------- + + +def test_tactile_pairs_accepts_bool_flag(tmp_path: Path) -> None: + """robots[].tactile 是布尔开关时不能抛 TypeError(曾经整个触觉检查项没跑)。""" + yml = tmp_path / "site.yaml" + yml.write_text( + "robot:\n" + " sn: Rizon4-000089\n" + " robots:\n" + " - sn: Rizon4-000089\n" + " arm: left\n" + " ip: 192.168.2.101\n" + " tactile: true\n" + "tactile:\n" + " left_video: /dev/v4l/by-path/x-3.4.1-y\n" + " right_video: /dev/v4l/by-path/x-3.4.2-y\n", + encoding="utf-8", + ) + pairs = pf._tactile_pairs(yml, "Rizon4-000089") + assert [n for n, _ in pairs] == ["左指", "右指"] + + +def test_tactile_pairs_empty_when_disabled(tmp_path: Path) -> None: + """tactile: false 的臂不该被拉起来查触觉(右臂就没装)。""" + yml = tmp_path / "site.yaml" + yml.write_text( + "robot:\n" + " sn: Rizon4-000089\n" + " robots:\n" + " - sn: Rizon4-000089\n" + " arm: left\n" + " ip: 192.168.2.101\n" + " tactile: false\n" + "tactile:\n" + " left_video: /dev/v4l/by-path/x-3.4.1-y\n", + encoding="utf-8", + ) + assert pf._tactile_pairs(yml, "Rizon4-000089") == [] + + +# --------------------------------------------------------------------------- +# 腕部相机(D405)在不在 —— "速率对不对"和"在不在"是两件事 +# --------------------------------------------------------------------------- + + +def _site_with_wrist(tmp_path: Path, *, model: str, serial: str) -> Path: + yml = tmp_path / "site.yaml" + yml.write_text( + "robot:\n" + " sn: Rizon4-000089\n" + " robots:\n" + " - sn: Rizon4-000089\n" + " arm: left\n" + " ip: 192.168.2.101\n" + " camera:\n" + " name: d405_left\n" + "cameras:\n" + " d405_left:\n" + f" model: {model}\n" + f" serial: \"{serial}\"\n" + " role: left_wrist\n", + encoding="utf-8", + ) + return yml + + +def _realsense(path: str, product: str, serial: str = "") -> object: + from usb_health import UsbDevice + + return UsbDevice( + path=path, vid="8086", pid="0b07", product=product, speed_mbps=5000, serial=serial + ) + + +def test_cameras_fails_when_the_wrist_camera_is_gone(tmp_path: Path) -> None: + """只剩顶置 D435 时必须报 FAIL。 + + 现场那次(2026-09-28)D405 整个从 lsusb 消失,而旧体检**全绿** —— + 它只看协商速率,对"这只相机根本不在了"没话说。 + """ + yml = _site_with_wrist(tmp_path, model="d405", serial="352122274408") + devices = [_realsense("1-3.2", "Intel(R) RealSense(TM) Depth Camera 435")] + c = pf.check_cameras(yml, "Rizon4-000089", devices=devices) + assert c.level == pf.LEVEL_FAIL + assert "d405" in " ".join(c.lines) + assert "1-3.2" in " ".join(c.lines), "要把现在有的列出来,人才知道差哪只" + assert c.actions, "光报错不够,要说去查什么(腕部活动线最先断)" + + +def test_cameras_ok_when_wrist_camera_present(tmp_path: Path) -> None: + yml = _site_with_wrist(tmp_path, model="d405", serial="352122274408") + devices = [_realsense("1-3.2", "Intel(R) RealSense(TM) Depth Camera 435")] + devices.append( + _realsense("1-3.5", "Intel(R) RealSense(TM) Depth Camera 405", "352122274408") + ) + c = pf.check_cameras(yml, "Rizon4-000089", devices=devices) + assert c.level == pf.LEVEL_OK + + +def test_cameras_warns_when_another_arms_wrist_camera_is_plugged_in( + tmp_path: Path, +) -> None: + """两台臂各有一只同型号:插反了要在开工前看出来(不阻断,但要报)。""" + yml = _site_with_wrist(tmp_path, model="d405", serial="352122274408") + devices = [ + _realsense("1-3.5", "Intel(R) RealSense(TM) Depth Camera 405", "260522274442") + ] + c = pf.check_cameras(yml, "Rizon4-000089", devices=devices) + assert c.level == pf.LEVEL_WARN + assert "260522274442" in " ".join(c.lines) + + +def test_cameras_does_not_use_serial_alone(tmp_path: Path) -> None: + """sysfs 不透出序列号(实测 D435 就是空串)时不能判成"不在"。""" + yml = _site_with_wrist(tmp_path, model="d405", serial="352122274408") + devices = [_realsense("1-3.5", "Intel(R) RealSense(TM) Depth Camera 405", "")] + c = pf.check_cameras(yml, "Rizon4-000089", devices=devices) + assert c.level == pf.LEVEL_OK + + +def test_cameras_skips_sim_arm(tmp_path: Path) -> None: + """仿真臂不该去 USB 上找相机。""" + yml = tmp_path / "site.yaml" + yml.write_text( + "robot:\n" + " sn: sim\n" + " robots:\n" + " - sn: sim\n" + " arm: sim\n" + " camera:\n" + " name: sim_d405\n" + "cameras:\n" + " sim_d405:\n" + " model: d405\n" + " role: sim_wrist\n", + encoding="utf-8", + ) + c = pf.check_cameras(yml, "sim", devices=[]) + assert c.level == pf.LEVEL_OK + assert "仿真" in " ".join(c.lines) + + +# --------------------------------------------------------------------------- +# 同一件事在报告里只能说一次(跨模块的措辞回归) +# --------------------------------------------------------------------------- + + +def test_network_action_appears_once_in_the_manual_list() -> None: + """check_network 与 rdk_status 给的"去查网线"必须是**同一串字**。 + + 去重是按整串文字做的。两处各写一句意思相同、字面不同的话,现场就会 + 读到两条(实测就是:第 1 条和第 3 条是同一件事),人还得判断 + "这是两个问题还是一个"。措辞已抽到 manual_actions,这里钉住它。 + """ + import rdk_status + + _level, _summary, rdk_actions = rdk_status.classify_state( + connected=False, error="连不上" + ) + net = check_network(1, "100% packet loss", [], "192.168.2.101") + rep = Report( + [ + net, + Check("arm", "机械臂 RDK", pf.LEVEL_FAIL, ["连不上"], actions=list(rdk_actions)), + ] + ) + acts = rep.manual_actions() + net_acts = [a for a in acts if "check_robot_network.sh" in a] + assert len(net_acts) == 1, f"网线那条重复了:{net_acts}" + + +# --------------------------------------------------------------------------- +# 超时 +# --------------------------------------------------------------------------- + + +def test_run_times_out_instead_of_hanging() -> None: + """体检里有子进程时必须有超时:卡死的设备会让普通调用永不返回。""" + import time + + t0 = time.monotonic() + code, out, err = pf._run([sys.executable, "-c", "import time; time.sleep(30)"], 0.5) + assert code == 124 + assert time.monotonic() - t0 < 10.0 + assert "超时" in err + + +def test_run_keeps_stdout_and_stderr_separate() -> None: + """RDK 的日志走 stderr、JSON 走 stdout;合并会让 JSON 解析必然失败。""" + code, out, err = pf._run( + [sys.executable, "-c", "import sys; print('{\"a\":1}'); print('log', file=sys.stderr)"], + 10.0, + ) + assert code == 0 + assert out.strip() == '{"a":1}' + assert "log" in err + + +# --------------------------------------------------------------------------- +# 编排能真的跑完(回归:_run 改成三返回值后,有一处还在按两返回值解包, +# 于是体检**第一步就崩**,什么结论都没给出来) +# --------------------------------------------------------------------------- + + +def test_run_all_completes_without_crashing(tmp_path: Path, monkeypatch) -> None: + """整套编排必须能跑到底并返回 Report —— 崩在任何一行都等于"没体检"。""" + calls: list = [] + + def fake_run(cmd, timeout_s): # noqa: ANN001 + calls.append(list(cmd)) + exe = Path(str(cmd[0])).name + if exe == "ps": + return 0, " 1 /usr/bin/ros2_control_node --ros-args\n", "" + if exe == "ping": + return 0, "2 packets transmitted, 2 received, 0% packet loss\n", "" + if exe == "ip": + return 0, "192.168.2.101 dev enp4s0 lladdr aa:bb:cc:dd:ee:ff REACHABLE\n", "" + return ( + 0, + '{"level": "ok", "mode": "AUTO", "enabled": true}', + "", + ) + + monkeypatch.setattr(pf, "_run", fake_run) + # USB 检查读真 sysfs(真机上可能卡死的正是它)→ 这里注入假 reader + monkeypatch.setattr(pf, "check_usb", lambda *a, **k: Check("usb", "USB", pf.LEVEL_OK, ["正常"])) + monkeypatch.setattr( + pf, "check_v4l", lambda *a, **k: Check("v4l", "触觉", pf.LEVEL_OK, ["正常"]) + ) + + site = tmp_path / "site.yaml" + site.write_text( + "robot:\n" + " sn: Rizon4-000089\n" + " robots:\n" + " - sn: Rizon4-000089\n" + " arm: left\n" + " ip: 192.168.2.101\n" + " iface: enp4s0\n", + encoding="utf-8", + ) + rep = pf.run_all(site, "Rizon4-000089", rdk_python=sys.executable) + assert isinstance(rep, Report) + assert rep.checks, "至少要有几项结论" + assert any(c.key == "network" for c in rep.checks), "网络项没跑到说明中途崩了" + assert any(c.key == "arm" for c in rep.checks), "机械臂项没跑到说明中途崩了"