主界面一屏要同时看臂、夹爪、相机和触觉,按钮再被拉高就会叠到画面上。相机按原分辨率走回调取最新帧,场景一动不再等红外齐帧把彩色丢掉。模型和相机都留在工位窗口里,启动默认不再弹 RViz。

验收:test_listed_arms 与 test_camera_preview 共 20 项通过。D435 1280×720 发布约 29 帧/秒。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
arm_dev
2026-10-15 16:20:35 +08:00
parent 76c9b9cc49
commit 7942014763
44 changed files with 1664 additions and 599 deletions
+57 -14
View File
@@ -117,10 +117,48 @@ def classify_state(
)
def _probe(sn: str, iface: str, enable: bool) -> Dict[str, object]:
"""连一次机器人并读状态。返回给 classify_state 的关键字参数。"""
import flexivrdk # 延迟导入:让本模块在没有 flexivrdk 的环境里也能被单测导入
def pack_status(state: Dict[str, object]) -> Dict[str, object]:
"""把探针结果收成给界面的 JSON。
``q`` 不能传进 ``classify_state``:那个函数只认使能字段。
曾经用 ``classify_state(**state)``,关节角一读到就抛 TypeError,
JSON 根本没打出来,界面就永远停在“还没连上”。
"""
fields = dict(state)
q = fields.pop("q", None)
level, summary, actions = classify_state(**fields) # type: ignore[arg-type]
payload: Dict[str, object] = {
**fields,
"level": level,
"summary": summary,
"actions": actions,
}
if isinstance(q, list) and len(q) == 7:
payload["q"] = q
return payload
def _open_robot(sn: str, iface: str, pc_ip: str):
"""按 IPv4 白名单连机器人。网卡名不能直接交给 RDK。"""
import flexivrdk
from rdk_helpers import rdk_whitelist_lists
# 空名单才退回不指定网卡。指定了 eth0 这种名字时,FastDDS 会把真地址滤掉,
# 表现是 ping 得通、这里一直搜到超时。
candidates = rdk_whitelist_lists(iface, pc_ip) or [[]]
last = ""
for listen in candidates:
try:
if listen:
return flexivrdk.Robot(sn, list(listen), True, False), ""
return flexivrdk.Robot(sn), ""
except Exception as exc: # noqa: BLE001
last = str(exc)
return None, last or "连不上 RDK"
def _probe(sn: str, iface: str, enable: bool, pc_ip: str = "") -> Dict[str, object]:
"""连一次机器人并读状态。返回给 classify_state 的关键字参数,另加关节角 ``q``。"""
out: Dict[str, object] = {
"connected": False,
"operational": False,
@@ -130,13 +168,9 @@ def _probe(sn: str, iface: str, enable: bool) -> Dict[str, object]:
"estop_released": True,
"error": "",
}
try:
if iface:
robot = flexivrdk.Robot(sn, [iface], True, False)
else:
robot = flexivrdk.Robot(sn)
except Exception as exc: # noqa: BLE001
out["error"] = str(exc)
robot, error = _open_robot(sn, iface, pc_ip)
if robot is None:
out["error"] = error
return out
out["connected"] = True
@@ -157,6 +191,13 @@ def _probe(sn: str, iface: str, enable: bool) -> Dict[str, object]:
out["op_status"] = str(_safe(robot.operational_status, ""))
out["fault"] = bool(_safe(robot.fault, False))
out["estop_released"] = bool(_safe(robot.estop_released, True))
# 只读关节角,给界面画姿态。读不到就空着,绝不在这里 Enable。
try:
q = [float(v) for v in list(robot.states().q)[:7]]
except Exception: # noqa: BLE001
q = []
if len(q) == 7 and all(v == v for v in q):
out["q"] = q
return out
@@ -164,18 +205,20 @@ def main(argv: Optional[Sequence[str]] = None) -> int:
ap = argparse.ArgumentParser(description="读一次机械臂状态并给结论")
ap.add_argument("--sn", required=True)
ap.add_argument("--iface", default="")
ap.add_argument("--pc-ip", default="", help="本机机器人网卡的 IPv4,用来做 RDK 白名单")
ap.add_argument("--enable", action="store_true", help="顺手尝试 Enable(会真的使能)")
ap.add_argument("--out", default="", help="把 KEY=VALUE 写到这个文件")
ap.add_argument("--json", action="store_true", help="把结果打到 stdout(JSON)")
args = ap.parse_args(argv)
state = _probe(args.sn, args.iface, args.enable)
level, summary, actions = classify_state(**state) # type: ignore[arg-type]
payload = {**state, "level": level, "summary": summary, "actions": actions}
state = _probe(args.sn, args.iface, args.enable, args.pc_ip)
payload = pack_status(state)
level = str(payload["level"])
actions = list(payload["actions"])
if args.out:
lines = [f"{k}={v}" for k, v in payload.items() if k != "actions"]
lines.append("actions=" + " | ".join(actions))
lines.append("actions=" + " | ".join(str(item) for item in actions))
Path(args.out).write_text("\n".join(lines) + "\n", encoding="utf-8")
if args.json:
print(json.dumps(payload, ensure_ascii=False, indent=2))