Spaces:
Running
Running
stouchinglhh
collector: 更新采集工程至 v1.3.0（UMI 夹爪全链接入：Fays S80M 双目 SLAM 位姿/轨迹 + Sightac 左右触觉与 250×250 力矩阵 + DECXIN RGB；双臂双夹爪同录，rig1 旧契约不变 rig2 加前缀；两 rig 独立 CPU 亲和分区；SLAM 位姿坐标约定修正 X=右/Y=正对/Z=上；左目显示稳 15fps 且 raw 流断线自动重连；轨迹并入 episode parquet 列）
af37f5b | """ | |
| 统一设备 worker 注册表 + 面板开关分派 + 录制元数据口径（纯 Python， | |
| 零 Qt 依赖）。 | |
| 具体开启/关闭动作（建槽、弹窗、widget 接线）留在 UI 侧回调，本模块 | |
| 只做核心口径： | |
| - DeviceManager 注册表容器：条目骨架构造 + 查询 + 元数据/抽帧状态 | |
| - dispatch_toggle 面板开关分派（录制锁双保险 + kind 路由） | |
| - build_device_meta / reset_s80m_record_state / teardown_all | |
| 注册表级纯口径（离线测试直接注入假条目即可） | |
| 注册表条目形状（与旧主窗口 _workers 口径一致）： | |
| kind ∈ {"uvc", "d435", "s80m", "data_ble", "ble", "gripper"} | |
| d435/s80m 条目由各自 manager 的 new_entry 构造 | |
| （core.d435_manager / core.s80m_manager），uvc/ble/glove/gripper | |
| 骨架由本类构造。 | |
| """ | |
| from __future__ import annotations | |
| from config import settings | |
| class DeviceManager: | |
| """统一设备 worker 注册表（多路并发核心）。 | |
| entries: dev_key → entry dict（主窗口 _workers 直接引用同一 dict， | |
| 离线测试注入假条目沿用同一形状）。UI 侧负责把条目塞进注册表， | |
| 本类只提供骨架构造、查询与录制口径。 | |
| """ | |
| def __init__(self): | |
| self.entries: dict = {} | |
| # ── 条目骨架 ── | |
| def uvc_entry(slots, label) -> dict: | |
| """UVC 条目（无专属字段，槽位即状态）。""" | |
| return {"kind": "uvc", "slots": list(slots), "label": label} | |
| def ble_entry(slot: str, label: str) -> dict: | |
| """无数据蓝牙（耳机类）占位条目。""" | |
| return {"kind": "ble", "slots": [slot], "label": label} | |
| def glove_entry(slot: str, role: str, glove_widget, label: str) -> dict: | |
| """数据手套条目：传感器列名 + 画面 widget（close 时停 BLE/撤画面）。""" | |
| return {"kind": "data_ble", "slots": [slot], | |
| "sensor_column": role, "glove": glove_widget, | |
| "label": label} | |
| def gripper_entry(slots, label: str, serial: str = "") -> dict: | |
| """UMI 夹爪条目：6 个网格槽 + ESP 序列号（fays_serial 开链后回填）。""" | |
| return {"kind": "gripper", "slots": list(slots), | |
| "label": label, "serial": serial, | |
| "fays_serial": "", "sensor_columns": []} | |
| # ── 查询 ── | |
| def get(self, dev_key: str) -> dict | None: | |
| return self.entries.get(dev_key) | |
| def has_kind(self, kind: str) -> bool: | |
| return any(e["kind"] == kind for e in self.entries.values()) | |
| def has_serial(self, kind: str, serial: str) -> bool: | |
| """按 serial 查重（d435 同机重复开关检测）。""" | |
| return any(e["kind"] == kind and e.get("serial") == serial | |
| for e in self.entries.values()) | |
| # ── 录制口径 ── | |
| def build_device_meta(self) -> list: | |
| """按注册表构建录制设备信息（写入 meta 的 devices 段）。""" | |
| return build_device_meta(self.entries) | |
| def reset_s80m_record_state(self): | |
| """录制起止时重置 50→30 抽帧状态（桶号 + 待写 IMU 缓冲）。""" | |
| reset_s80m_record_state(self.entries) | |
| def teardown_all(self, close_fns: dict) -> None: | |
| """关闭全部已开启设备（遍历注册表分类清理，未知 kind 兜底弹出）。""" | |
| teardown_all(self.entries, close_fns) | |
| def build_device_meta(entries: dict) -> list: | |
| """按注册表构建录制设备信息（写入 meta 的 devices 段）。 | |
| 手套：无画面槽位，附 parquet 传感器列名； | |
| 无数据蓝牙（ble）不进 devices 段。 | |
| """ | |
| meta = [] | |
| for key, e in entries.items(): | |
| if e["kind"] == "data_ble": | |
| meta.append({ | |
| "key": key, "kind": e["kind"], | |
| "name": settings.device_name(key) or e.get("label", ""), | |
| "slots": [], | |
| "sensor_column": e.get("sensor_column", "")}) | |
| continue | |
| if e["kind"] not in ("uvc", "d435", "s80m", "gripper"): | |
| continue | |
| d = {"key": key, "kind": e["kind"], | |
| "name": settings.device_name(key) or e.get("label", ""), | |
| "slots": list(e.get("slots", []))} | |
| if e.get("serial"): | |
| d["serial"] = str(e["serial"]) | |
| if e["kind"] == "gripper": | |
| if e.get("fays_serial"): | |
| d["fays_serial"] = str(e["fays_serial"]) | |
| if e.get("sensor_columns"): | |
| d["sensor_columns"] = list(e["sensor_columns"]) | |
| meta.append(d) | |
| return meta | |
| def reset_s80m_record_state(entries: dict) -> None: | |
| """录制起止时重置 50→30 抽帧状态（桶号 + 待写 IMU 缓冲 + 空桶统计）。""" | |
| for e in entries.values(): | |
| if e["kind"] == "s80m": | |
| e["last_bucket"] = {} | |
| e["pending_imu"] = [] | |
| e["drop_watch"] = {"last_mono_ns": None, "dropped": 0, | |
| "elapsed": 0, "alerted": False} | |
| def teardown_all(entries: dict, close_fns: dict) -> None: | |
| """关闭全部已开启设备（遍历注册表分类清理，未知 kind 兜底弹出）。 | |
| s80m/d435 的关闭回调会自行 unregister（pop），故遍历键副本。 | |
| """ | |
| for key in list(entries): | |
| entry = entries.get(key) | |
| if not entry: | |
| continue | |
| fn = (close_fns or {}).get(entry["kind"]) | |
| if fn: | |
| fn(key) | |
| else: | |
| entries.pop(key, None) | |
| entries.clear() | |
| def dispatch_toggle(dev, on: bool, is_recording: bool, | |
| open_fns: dict, close_fns: dict) -> bool: | |
| """面板开关分派口径（录制锁双保险 + kind 路由）。 | |
| open_fns / close_fns: kind → 可调用（具体动作在 UI 侧， | |
| open 接收 dev 并返回是否成功，close 接收 dev_key）。 | |
| 返回 opened：打开成功 True；录制中/打开失败/缺回调 False | |
| （UI 侧按 on 区分「回退勾选」与「关闭路径」）。 | |
| """ | |
| if on: | |
| if is_recording: | |
| return False # 面板已锁死，双保险：UI 侧回退勾选 | |
| fn = (open_fns or {}).get(dev.kind) | |
| return bool(fn(dev)) if fn else False | |
| fn = (close_fns or {}).get(dev.kind) | |
| if fn: | |
| fn(dev.key) | |
| return False | |