Spaces:

Stouching-Robotics
/

DataFlowing

Running

DataFlowing / processor /app /processing /cleaning /contract.py

stouchinglhh

Claude

sync processor from GitLab c697b6e: 版本 1.19.4 / 前端 1.12.4 —— 服务端 i18n(首屏不再闪英文)、推理模块并入模型页、前端抽出 shared/、缓存键改用前端号

bb59211 about 1 month ago

10.7 kB

"""清洗契约 —— 模态、信道、严重度的公共定义。
这个文件把"平台里同一个东西有好几个名字"这件事收敛到一处:
    device_naming.DEVICE_LABELS        mono_rgb / stereo_rgb / ...
    workflow_dispatch.semantic_type    同上（同一套值，只是没人敢这么说）
    api/projects._device_input_sources mono_camera / stereo_camera  ← 名字不一样
    处理模块 slug                       mono_camera / stereo_camera
**本模块不新建第 7 套枚举** —— DEVICE_MODALITIES 就是 device_naming 那 6 个值的
权威声明，其余几套通过 ``normalize_device_modality()`` 单向映射进来。新增设备类型时
只需改这一处，测试会锁住映射完整性。
信道(channel) 是比设备模态更细的粒度: PDF 里"IMU 可选""SLAM 必须"这类判断
是**信道级**的，硬塞进设备模态会毁掉它。所以设备模态 → 信道的推导放在这里。
"""

from __future__ import annotations

from dataclasses import dataclass, field as dc_field
from typing import Any, Iterable

# ── 设备模态 ──────────────────────────────────────────────────
# 与 app/device_naming.py:15 DEVICE_LABELS 的键**逐字一致**。
# 改这里必须同步改那里，tests/test_cleaning_contract.py 会断言两边相等。
DEVICE_MODALITIES: tuple[str, ...] = (
    "mono_rgb",
    "stereo_rgb",
    "rgbd_camera",
    "stereo_rgbd_camera",
    "glove_sensor",
    "gripper_device",
)

# 「摄像机类」模态 —— 判定 collection_type 时要用
CAMERA_MODALITIES: frozenset[str] = frozenset({
    "mono_rgb", "stereo_rgb", "rgbd_camera", "stereo_rgbd_camera",
})

# 会产出 RGB 视频的**全部**设备卡片 —— ``video.*`` 类检查适用于它们每一个。
# 比 CAMERA_MODALITIES 多一个 gripper_device：UMI 夹爪自带鱼眼相机，
# 视频检查对同样它成立。
VIDEO_CAPABLE_MODALITIES: tuple[str, ...] = (
    "mono_rgb", "stereo_rgb", "rgbd_camera", "stereo_rgbd_camera",
    "gripper_device",
)

# ── 信道 ──────────────────────────────────────────────────────
# 检查项按信道声明需求，而不是按设备模态 —— 这样一条"SLAM 连续性"检查
# 既能被 UMI 夹爪触发，将来也能被别的带 SLAM 的设备复用。
CHANNELS: tuple[str, ...] = (
    "time",           # 时间戳 / 帧号
    "rgb",            # RGB 视频
    "depth",          # 深度
    "slam",           # SLAM 位姿/轨迹
    "gripper_state",  # 夹爪开合
    "force",          # 力 / 触觉（夹爪力）
    "tactile",        # 触觉阵列（手套）
    "imu",            # 惯性单元
    "keypoints",      # 手部关键点
    "action",         # 动作（派生列）
    # 设备的连接状态（``status.left_glove`` 这类**字符串**列）。
    # 单独算一个信道而不是塞进 tactile 的 series：它是"设备这一帧在不在"，
    # 与阵列读数正交 —— 手套掉线时阵列会静默保持最后一个值，看起来完全正常。
    "device_status",
)

# 设备模态 → 该设备**可能**提供的信道。
# 注意这是"可能"不是"必然" —— 实际有哪些信道由 collect.py 从真实列名推导，
# 这里只用来回答"这个模态需不需要某类检查"。
MODALITY_CHANNELS: dict[str, tuple[str, ...]] = {
    "mono_rgb":           ("rgb", "time"),
    "stereo_rgb":         ("rgb", "time"),
    "rgbd_camera":        ("rgb", "depth", "time"),
    "stereo_rgbd_camera": ("rgb", "depth", "time"),
    "glove_sensor":       ("tactile", "device_status", "imu", "time"),
    "gripper_device":     ("rgb", "slam", "gripper_state", "force", "action", "time"),
}

# 非设备来源的信道 —— 由处理节点产出，不属于任何输入设备
DERIVED_CHANNELS: frozenset[str] = frozenset({"keypoints", "action"})


def channels_for(modalities: Iterable[str]) -> tuple[str, ...]:
    """一组设备模态覆盖的全部信道（去重、保序）。纯函数。"""
    seen: dict[str, None] = {}
    for modality in modalities:
        for channel in MODALITY_CHANNELS.get(str(modality), ()):
            seen.setdefault(channel, None)
    return tuple(seen)


def pose_is_valid(item) -> bool:
    """位姿能不能用 —— 空元组和**全零占位**都不算。
    ★ 全零占位长得像合法位姿：长度 7、能索引、能相减。但它的四元数模是 0，
      任何涉及它的差分都会算出"瞬移"：
          两个零四元数的点积 = 0 → acos(0) = π/2 → 夹角 180°
      30fps 下就是 180°×30 = **94.25 rad/s** —— 一个物理上荒谬的角速度。
      实测 Test94 的 ``observation.slam_pose`` 有 437/599 行是全零，SLAM 连续性
      因此报出 465 处"跳变"（599 帧里几乎每一帧），把整集假判成 FAIL 并拦住批次。
    判据只看四元数模：**位置可以合法地是 (0,0,0)**（原点就是合法位姿），
    四元数模为 0 则一定不是合法朝向。
    放在 contract 而不是 collect：检查项要用它，而 collect 依赖 checks
    （检查项自动发现），collect ← checks 的导入会成环。
    """
    if not item or len(item) < 7:
        return False
    try:
        return sum(float(value) ** 2 for value in item[3:7]) > 1e-12
    except (TypeError, ValueError):
        return False


def normalize_device_modality(value: object) -> str | None:
    """把别处的设备/输入类型名映射到 canonical 设备模态。
    覆盖两套现存命名:
      * ``stereo_camera`` / ``mono_camera``（api/projects、processing 模块 slug）
      * ``fisheye_camera`` / ``rgb_camera``（历史工作流节点类型）
    认不出来返回 ``None``（调用方应跳过该值，而不是终止）。
    """
    text = str(value or "").strip()
    if not text:
        return None
    if text in DEVICE_MODALITIES:
        return text
    return _ALIASES.get(text)


# 别名表 —— 只列真实存在于代码里的历史/并行命名。
# 新增别名时同步加进 tests/test_cleaning_contract.py 的用例。
_ALIASES: dict[str, str] = {
    # api/projects._device_input_sources 与处理模块 slug 用的是 camera 后缀
    "mono_camera": "mono_rgb",
    "stereo_camera": "stereo_rgb",
    "fisheye_camera": "mono_rgb",     # 历史节点，镜头类型对清洗不可见
    "rgb_camera": "mono_rgb",         # 历史节点
    "rgbd": "rgbd_camera",
    "stereo_rgbd": "stereo_rgbd_camera",
    # 中文/展示名兜底
    "umi": "gripper_device",
    "umi_gripper": "gripper_device",
    "glove": "glove_sensor",
}


# ── 严重度 ────────────────────────────────────────────────────
PASS = "pass"
WARN = "warn"
FAIL = "fail"
PENDING = "pending_review"
ERROR = "error"

# 越靠前越严重 —— worst() 与排序都用它
SEVERITY_ORDER: tuple[str, ...] = (ERROR, FAIL, PENDING, WARN, PASS)

# 面向用户的展示文案（报告里的 ``status_label``）。界面是英文的，这里保持一致。
SEVERITY_LABELS: dict[str, str] = {
    PASS: "Pass",
    WARN: "Warning",
    FAIL: "Fail",
    PENDING: "Pending review",
    ERROR: "Error",
}


def worst(statuses: Iterable[str]) -> str:
    """取一组状态里最严重的；空集合视为 PASS。纯函数。"""
    rank = {status: index for index, status in enumerate(SEVERITY_ORDER)}
    result = PASS
    best = len(SEVERITY_ORDER)
    for status in statuses:
        index = rank.get(str(status), len(SEVERITY_ORDER))
        if index < best:
            best = index
            result = str(status)
    return result


def is_blocking(status: str) -> bool:
    """该状态是否应当拦截（退回人工审核）。"""
    return str(status) in (FAIL, ERROR)


# ── 结构 ──────────────────────────────────────────────────────
@dataclass(frozen=True)
class Range:
    """一段时间区间 —— PDF §5「第二层结果」。"""

    start_sec: float
    end_sec: float
    code: str
    severity: str
    stream: str = ""
    channel: str = ""
    start_frame: int | None = None
    end_frame: int | None = None
    rule: str = ""
    message: str = ""
    evidence: dict[str, Any] = dc_field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        return {
            "start_sec": round(float(self.start_sec), 3),
            "end_sec": round(float(self.end_sec), 3),
            "code": self.code,
            "severity": self.severity,
            "stream": self.stream,
            "channel": self.channel,
            "start_frame": self.start_frame,
            "end_frame": self.end_frame,
            "rule": self.rule,
            "message": self.message,
            "evidence": dict(self.evidence or {}),
        }


@dataclass(frozen=True)
class StreamResult:
    """一条数据流的检查结果 —— PDF §5「第一层结果」。"""

    stream: str
    modality: str = ""
    label: str = ""
    channels: tuple[str, ...] = ()
    status: str = PASS
    reasons: tuple[str, ...] = ()
    metrics: dict[str, Any] = dc_field(default_factory=dict)
    ranges: tuple[Range, ...] = ()

    def to_dict(self) -> dict[str, Any]:
        return {
            "stream": self.stream,
            "modality": self.modality,
            "label": self.label,
            "channels": list(self.channels),
            "status": self.status,
            "reasons": list(self.reasons),
            "metrics": dict(self.metrics or {}),
            "ranges": [item.to_dict() for item in self.ranges],
        }


@dataclass(frozen=True)
class Finding:
    """一次检查产出的结论 —— 检查项 run() 的返回值单元。"""

    rule: str
    version: str
    status: str
    scope: str                        # stream | episode | cross_modal
    stream: str = ""
    message: str = ""
    metrics: dict[str, Any] = dc_field(default_factory=dict)
    ranges: tuple[Range, ...] = ()
    elapsed_ms: int = 0

    def to_dict(self) -> dict[str, Any]:
        return {
            "rule": self.rule,
            "version": self.version,
            "status": self.status,
            "scope": self.scope,
            "stream": self.stream,
            "message": self.message,
            "metrics": dict(self.metrics or {}),
            "ranges": [item.to_dict() for item in self.ranges],
            "elapsed_ms": self.elapsed_ms,
        }

	"""清洗契约 —— 模态、信道、严重度的公共定义。

	这个文件把"平台里同一个东西有好几个名字"这件事收敛到一处:

	device_naming.DEVICE_LABELS mono_rgb / stereo_rgb / ...
	workflow_dispatch.semantic_type 同上（同一套值，只是没人敢这么说）
	api/projects._device_input_sources mono_camera / stereo_camera ← 名字不一样
	处理模块 slug mono_camera / stereo_camera

	本模块不新建第 7 套枚举 —— DEVICE_MODALITIES 就是 device_naming 那 6 个值的
	权威声明，其余几套通过 ``normalize_device_modality()`` 单向映射进来。新增设备类型时
	只需改这一处，测试会锁住映射完整性。

	信道(channel) 是比设备模态更细的粒度: PDF 里"IMU 可选""SLAM 必须"这类判断
	是信道级的，硬塞进设备模态会毁掉它。所以设备模态 → 信道的推导放在这里。
	"""

	from __future__ import annotations

	from dataclasses import dataclass, field as dc_field
	from typing import Any, Iterable

	# ── 设备模态 ──────────────────────────────────────────────────
	# 与 app/device_naming.py:15 DEVICE_LABELS 的键逐字一致。
	# 改这里必须同步改那里，tests/test_cleaning_contract.py 会断言两边相等。
	DEVICE_MODALITIES: tuple[str, ...] = (
	"mono_rgb",
	"stereo_rgb",
	"rgbd_camera",
	"stereo_rgbd_camera",
	"glove_sensor",
	"gripper_device",
	)

	# 「摄像机类」模态 —— 判定 collection_type 时要用
	CAMERA_MODALITIES: frozenset[str] = frozenset({
	"mono_rgb", "stereo_rgb", "rgbd_camera", "stereo_rgbd_camera",
	})

	# 会产出 RGB 视频的全部设备卡片 —— ``video.*`` 类检查适用于它们每一个。
	# 比 CAMERA_MODALITIES 多一个 gripper_device：UMI 夹爪自带鱼眼相机，
	# 视频检查对同样它成立。
	VIDEO_CAPABLE_MODALITIES: tuple[str, ...] = (
	"mono_rgb", "stereo_rgb", "rgbd_camera", "stereo_rgbd_camera",
	"gripper_device",
	)

	# ── 信道 ──────────────────────────────────────────────────────
	# 检查项按信道声明需求，而不是按设备模态 —— 这样一条"SLAM 连续性"检查
	# 既能被 UMI 夹爪触发，将来也能被别的带 SLAM 的设备复用。
	CHANNELS: tuple[str, ...] = (
	"time", # 时间戳 / 帧号
	"rgb", # RGB 视频
	"depth", # 深度
	"slam", # SLAM 位姿/轨迹
	"gripper_state", # 夹爪开合
	"force", # 力 / 触觉（夹爪力）
	"tactile", # 触觉阵列（手套）
	"imu", # 惯性单元
	"keypoints", # 手部关键点
	"action", # 动作（派生列）
	# 设备的连接状态（``status.left_glove`` 这类字符串列）。
	# 单独算一个信道而不是塞进 tactile 的 series：它是"设备这一帧在不在"，
	# 与阵列读数正交 —— 手套掉线时阵列会静默保持最后一个值，看起来完全正常。
	"device_status",
	)

	# 设备模态 → 该设备可能提供的信道。
	# 注意这是"可能"不是"必然" —— 实际有哪些信道由 collect.py 从真实列名推导，
	# 这里只用来回答"这个模态需不需要某类检查"。
	MODALITY_CHANNELS: dict[str, tuple[str, ...]] = {
	"mono_rgb": ("rgb", "time"),
	"stereo_rgb": ("rgb", "time"),
	"rgbd_camera": ("rgb", "depth", "time"),
	"stereo_rgbd_camera": ("rgb", "depth", "time"),
	"glove_sensor": ("tactile", "device_status", "imu", "time"),
	"gripper_device": ("rgb", "slam", "gripper_state", "force", "action", "time"),
	}

	# 非设备来源的信道 —— 由处理节点产出，不属于任何输入设备
	DERIVED_CHANNELS: frozenset[str] = frozenset({"keypoints", "action"})


	def channels_for(modalities: Iterable[str]) -> tuple[str, ...]:
	"""一组设备模态覆盖的全部信道（去重、保序）。纯函数。"""
	seen: dict[str, None] = {}
	for modality in modalities:
	for channel in MODALITY_CHANNELS.get(str(modality), ()):
	seen.setdefault(channel, None)
	return tuple(seen)


	def pose_is_valid(item) -> bool:
	"""位姿能不能用 —— 空元组和全零占位都不算。

	★ 全零占位长得像合法位姿：长度 7、能索引、能相减。但它的四元数模是 0，
	任何涉及它的差分都会算出"瞬移"：

	两个零四元数的点积 = 0 → acos(0) = π/2 → 夹角 180°

	30fps 下就是 180°×30 = 94.25 rad/s —— 一个物理上荒谬的角速度。
	实测 Test94 的 ``observation.slam_pose`` 有 437/599 行是全零，SLAM 连续性
	因此报出 465 处"跳变"（599 帧里几乎每一帧），把整集假判成 FAIL 并拦住批次。

	判据只看四元数模：位置可以合法地是 (0,0,0)（原点就是合法位姿），
	四元数模为 0 则一定不是合法朝向。

	放在 contract 而不是 collect：检查项要用它，而 collect 依赖 checks
	（检查项自动发现），collect ← checks 的导入会成环。
	"""
	if not item or len(item) < 7:
	return False
	try:
	return sum(float(value) ** 2 for value in item[3:7]) > 1e-12
	except (TypeError, ValueError):
	return False


	def normalize_device_modality(value: object) -> str \| None:
	"""把别处的设备/输入类型名映射到 canonical 设备模态。

	覆盖两套现存命名:
	* ``stereo_camera`` / ``mono_camera``（api/projects、processing 模块 slug）
	* ``fisheye_camera`` / ``rgb_camera``（历史工作流节点类型）
	认不出来返回 ``None``（调用方应跳过该值，而不是终止）。
	"""
	text = str(value or "").strip()
	if not text:
	return None
	if text in DEVICE_MODALITIES:
	return text
	return _ALIASES.get(text)


	# 别名表 —— 只列真实存在于代码里的历史/并行命名。
	# 新增别名时同步加进 tests/test_cleaning_contract.py 的用例。
	_ALIASES: dict[str, str] = {
	# api/projects._device_input_sources 与处理模块 slug 用的是 camera 后缀
	"mono_camera": "mono_rgb",
	"stereo_camera": "stereo_rgb",
	"fisheye_camera": "mono_rgb", # 历史节点，镜头类型对清洗不可见
	"rgb_camera": "mono_rgb", # 历史节点
	"rgbd": "rgbd_camera",
	"stereo_rgbd": "stereo_rgbd_camera",
	# 中文/展示名兜底
	"umi": "gripper_device",
	"umi_gripper": "gripper_device",
	"glove": "glove_sensor",
	}


	# ── 严重度 ────────────────────────────────────────────────────
	PASS = "pass"
	WARN = "warn"
	FAIL = "fail"
	PENDING = "pending_review"
	ERROR = "error"

	# 越靠前越严重 —— worst() 与排序都用它
	SEVERITY_ORDER: tuple[str, ...] = (ERROR, FAIL, PENDING, WARN, PASS)

	# 面向用户的展示文案（报告里的 ``status_label``）。界面是英文的，这里保持一致。
	SEVERITY_LABELS: dict[str, str] = {
	PASS: "Pass",
	WARN: "Warning",
	FAIL: "Fail",
	PENDING: "Pending review",
	ERROR: "Error",
	}


	def worst(statuses: Iterable[str]) -> str:
	"""取一组状态里最严重的；空集合视为 PASS。纯函数。"""
	rank = {status: index for index, status in enumerate(SEVERITY_ORDER)}
	result = PASS
	best = len(SEVERITY_ORDER)
	for status in statuses:
	index = rank.get(str(status), len(SEVERITY_ORDER))
	if index < best:
	best = index
	result = str(status)
	return result


	def is_blocking(status: str) -> bool:
	"""该状态是否应当拦截（退回人工审核）。"""
	return str(status) in (FAIL, ERROR)


	# ── 结构 ──────────────────────────────────────────────────────
	@dataclass(frozen=True)
	class Range:
	"""一段时间区间 —— PDF §5「第二层结果」。"""

	start_sec: float
	end_sec: float
	code: str
	severity: str
	stream: str = ""
	channel: str = ""
	start_frame: int \| None = None
	end_frame: int \| None = None
	rule: str = ""
	message: str = ""
	evidence: dict[str, Any] = dc_field(default_factory=dict)

	def to_dict(self) -> dict[str, Any]:
	return {
	"start_sec": round(float(self.start_sec), 3),
	"end_sec": round(float(self.end_sec), 3),
	"code": self.code,
	"severity": self.severity,
	"stream": self.stream,
	"channel": self.channel,
	"start_frame": self.start_frame,
	"end_frame": self.end_frame,
	"rule": self.rule,
	"message": self.message,
	"evidence": dict(self.evidence or {}),
	}


	@dataclass(frozen=True)
	class StreamResult:
	"""一条数据流的检查结果 —— PDF §5「第一层结果」。"""

	stream: str
	modality: str = ""
	label: str = ""
	channels: tuple[str, ...] = ()
	status: str = PASS
	reasons: tuple[str, ...] = ()
	metrics: dict[str, Any] = dc_field(default_factory=dict)
	ranges: tuple[Range, ...] = ()

	def to_dict(self) -> dict[str, Any]:
	return {
	"stream": self.stream,
	"modality": self.modality,
	"label": self.label,
	"channels": list(self.channels),
	"status": self.status,
	"reasons": list(self.reasons),
	"metrics": dict(self.metrics or {}),
	"ranges": [item.to_dict() for item in self.ranges],
	}


	@dataclass(frozen=True)
	class Finding:
	"""一次检查产出的结论 —— 检查项 run() 的返回值单元。"""

	rule: str
	version: str
	status: str
	scope: str # stream \| episode \| cross_modal
	stream: str = ""
	message: str = ""
	metrics: dict[str, Any] = dc_field(default_factory=dict)
	ranges: tuple[Range, ...] = ()
	elapsed_ms: int = 0

	def to_dict(self) -> dict[str, Any]:
	return {
	"rule": self.rule,
	"version": self.version,
	"status": self.status,
	"scope": self.scope,
	"stream": self.stream,
	"message": self.message,
	"metrics": dict(self.metrics or {}),
	"ranges": [item.to_dict() for item in self.ranges],
	"elapsed_ms": self.elapsed_ms,
	}