Spaces:

Stouching-Robotics
/

DataFlowing

Running

DataFlowing / processor /app /processing /modules /data_quality.py

stouchinglhh

Claude

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

bb59211 about 1 month ago

6.47 kB

"""Data Cleaning —— 数据清洗（质检）：一个节点统管确定性与 AI 两类检查。
显示名是 Data Cleaning；slug 仍是 ``data_quality``（持久化契约，改它会让已保存
工作流的 nodeType 对不上）。两者不一致是有意的 —— 见下方端口 key 的同类说明。
本节点由两个旧节点合并而来，两者功能高度重叠：
    ai_quality_review  视频检查(黑屏/冻结/丢帧/解码) + AI 标注覆盖率
    data_cleaning      视频检查 + UMI/手套/EGO 检查（第 2 期接入）
历史原因：视频检查最早"寄居"在 ``ai_quality_review`` 里，好让纯视频工作流
也有质量门禁（见 ``ai_annotation.py`` 的 ``video_quality_review_node_config``
docstring）。现在合并成一张卡片，语义与配置都归一处。
**两个旧 slug 都作为别名保留**（``workflow_types.LEGACY_TO_CANONICAL``），
已保存的工作流打开时自动迁移，连线不受影响 —— 端口的 key 刻意保持不变：
    输入  data      所有旧连线都用它
    输出  reviewed  旧连线用它接 lerobot_export
检查分两层，与清洗方案 PDF §4 的两栏一一对应：
    机器可直接判断（确定性、免费、可每集跑）
      ├─ 视频：解码 / 黑屏 / 冻结 / 丢帧
      ├─ UMI：帧对齐 / action / 夹爪量程 / SLAM 连续性
      ├─ 手套：全零 / 死点 / 漂移 / 饱和
      └─ EGO：曝光 / 关键点缺失 / 左右手
    需联合判断或复核（有成本、可选）
      └─ AI 标注覆盖率；将来的 VLM 语义复核
检查项的算法实现与工作流适配器分开，与 ``app/processing/black_glove`` 同构：
    app/processing/cleaning/                ← 契约 + 检查项 + 报告组装
    app/processing/modules/data_quality.py  ← 本文件：artifact 契约 + 端口 + 配置
新增检查项 = 在 ``app/processing/cleaning/checks/`` 下放一个 .py，本文件不用改。
**只出结论，不产出数据** —— 原始数据一字不改，也不生成 clean_v1 之类的副本。
区间信息完整保留在报告的 ``ranges`` 里，下游按 ``training_use`` 过滤即可。
数据不对时**卡在人工审核**：判 FAIL/ERROR 的批次会被推回 ``to_review``
（就是审核页的入口），并在 ``cleaning_summary.blocking`` 上留一个粘性标记，
让 AI 标注 / 视频门禁的自动批准路径让路。WARN 只记录、不拦。
"""

from __future__ import annotations

from app.processing import ProcessingModule, JobContext, ArtifactRef
from app.processing.registry import register
from app.processing.theme import REVIEW_COLOR


@register
class DataQualityModule(ProcessingModule):
    slug = "data_quality"
    version = "1.0"
    category = "review"
    label = "Data Cleaning"
    # 必须取自 web/static/iconify-preload.js 的白名单 —— 图标是离线预载的,
    # 不在表内的名字 iconify-icon 会去 api.iconify.design 拉,调色板一次渲染
    # 全部节点,一个漏网图标就能让整块面板卡住。check-circle 表示"校验",
    # 与同组 review 节点的 eye(human) 不冲突。
    # 改图标后请跑 scripts/check_icon_preload.sh。
    icon = "ant-design:check-circle-outlined"
    color = REVIEW_COLOR

    # 端口 key 是持久化契约 —— 取值必须与合并前的 ai_quality_review 逐字一致,
    # 否则已保存工作流的连线会因 sourceHandle/targetHandle 对不上而失效。
    # 详见 workflow_types.py 里前端对 ``out.key === inp.key`` 的校验。
    inputs = ({"key": "data", "label": "Data"},)
    outputs = ({"key": "reviewed", "label": "Reviewed Data"},)

    # 配置结构（前端设置弹窗按设备卡片写入）::
    #
    #     {"devices": {
    #        "gripper_device": {
    #          "checks": {
    #            "umi.slam_continuity": {"enabled": true,
    #                                    "params": {"max_linear_mps": 3.0}},
    #            "video.freeze":       {"enabled": false}
    #          }}}}
    #
    # 引擎侧由 ``cleaning.engine.resolve_config()`` 解析。
    default_config = {"devices": {}}

    # ★ 刻意留空 —— 配置全部走专用弹窗（DeviceQualityModal），不走通用
    #   config_schema 渲染。
    #
    #   通用那套是**平铺字段列表**，表达不了"按设备卡片分组 + 每项带阈值"的
    #   嵌套结构；硬塞会变成一长串 enable_xxx 开关。曾短暂用过那套，结果是
    #   三个阈值在节点里和检查项里各存一份（frame_tolerance 与
    #   video.frame_drop.tolerance_ratio 同义），迟早不一致。
    #
    #   现在阈值只有一处定义：检查项的 ``default_params``；节点只存"覆盖值"。
    config_schema = ()

    # 真实检查在 run 完成回调里跑（那时才能读到合并后的 episode 数据），
    # 与 human_review / 旧 ai_quality_review 的分工一致。
    # 注：该字段当前无消费方，仅作语义标注（全仓 grep 只见 schema 与文档）。
    execution_target = "server"
    capabilities = ("data_quality", "quality_gate", "quality_review")

    def run(self, ctx: JobContext) -> dict[str, ArtifactRef] | None:
        """透传上游 artifact。上游没连 → skipped。
        **真正的检查不在 ``run()`` 里跑** —— 在 worker 的 run 完成回调里
        （``api/worker.py`` 调 ``cleaning.trigger.spawn_cleaning_for_run``）。
        为什么不在 ``run()``：DAG 执行期 worker 是在 staging 目录里干活、
        产物要等 ``_publish_episode_result`` 才合并回批次目录。在这一步读
        批次目录，拿到的是**上一轮**的数据 —— 尤其 ``action`` 列，它由
        ``umi_slam_action`` 派生产出，读早了会全是采集端的零占位。
        完成回调时数据已经合并完毕，与 ``human_review`` / 旧
        ``ai_quality_review`` 的分工一致。
        检查所需的数据**不靠端口传** —— 完成回调手上就有
        ``localstore.get_episode()`` 的记录（``path`` + ``episode_index``），
        检查项直接读批次目录。
        ★ 进来的是 ``episode_index`` 不是可选的：``path`` 是**项目目录**，
          几十集共享，漏传会拿第 0 集的数据去比全部集的视频。
        """
        if not ctx.incoming:
            ctx.skip("No upstream data — skipped")
        return dict(ctx.incoming)

	"""Data Cleaning —— 数据清洗（质检）：一个节点统管确定性与 AI 两类检查。

	显示名是 Data Cleaning；slug 仍是 ``data_quality``（持久化契约，改它会让已保存
	工作流的 nodeType 对不上）。两者不一致是有意的 —— 见下方端口 key 的同类说明。

	本节点由两个旧节点合并而来，两者功能高度重叠：

	ai_quality_review 视频检查(黑屏/冻结/丢帧/解码) + AI 标注覆盖率
	data_cleaning 视频检查 + UMI/手套/EGO 检查（第 2 期接入）

	历史原因：视频检查最早"寄居"在 ``ai_quality_review`` 里，好让纯视频工作流
	也有质量门禁（见 ``ai_annotation.py`` 的 ``video_quality_review_node_config``
	docstring）。现在合并成一张卡片，语义与配置都归一处。

	两个旧 slug 都作为别名保留（``workflow_types.LEGACY_TO_CANONICAL``），
	已保存的工作流打开时自动迁移，连线不受影响 —— 端口的 key 刻意保持不变：

	输入 data 所有旧连线都用它
	输出 reviewed 旧连线用它接 lerobot_export

	检查分两层，与清洗方案 PDF §4 的两栏一一对应：

	机器可直接判断（确定性、免费、可每集跑）
	├─ 视频：解码 / 黑屏 / 冻结 / 丢帧
	├─ UMI：帧对齐 / action / 夹爪量程 / SLAM 连续性
	├─ 手套：全零 / 死点 / 漂移 / 饱和
	└─ EGO：曝光 / 关键点缺失 / 左右手

	需联合判断或复核（有成本、可选）
	└─ AI 标注覆盖率；将来的 VLM 语义复核

	检查项的算法实现与工作流适配器分开，与 ``app/processing/black_glove`` 同构：

	app/processing/cleaning/ ← 契约 + 检查项 + 报告组装
	app/processing/modules/data_quality.py ← 本文件：artifact 契约 + 端口 + 配置

	新增检查项 = 在 ``app/processing/cleaning/checks/`` 下放一个 .py，本文件不用改。

	只出结论，不产出数据 —— 原始数据一字不改，也不生成 clean_v1 之类的副本。
	区间信息完整保留在报告的 ``ranges`` 里，下游按 ``training_use`` 过滤即可。

	数据不对时卡在人工审核：判 FAIL/ERROR 的批次会被推回 ``to_review``
	（就是审核页的入口），并在 ``cleaning_summary.blocking`` 上留一个粘性标记，
	让 AI 标注 / 视频门禁的自动批准路径让路。WARN 只记录、不拦。
	"""

	from __future__ import annotations

	from app.processing import ProcessingModule, JobContext, ArtifactRef
	from app.processing.registry import register
	from app.processing.theme import REVIEW_COLOR


	@register
	class DataQualityModule(ProcessingModule):
	slug = "data_quality"
	version = "1.0"
	category = "review"
	label = "Data Cleaning"
	# 必须取自 web/static/iconify-preload.js 的白名单 —— 图标是离线预载的,
	# 不在表内的名字 iconify-icon 会去 api.iconify.design 拉,调色板一次渲染
	# 全部节点,一个漏网图标就能让整块面板卡住。check-circle 表示"校验",
	# 与同组 review 节点的 eye(human) 不冲突。
	# 改图标后请跑 scripts/check_icon_preload.sh。
	icon = "ant-design:check-circle-outlined"
	color = REVIEW_COLOR

	# 端口 key 是持久化契约 —— 取值必须与合并前的 ai_quality_review 逐字一致,
	# 否则已保存工作流的连线会因 sourceHandle/targetHandle 对不上而失效。
	# 详见 workflow_types.py 里前端对 ``out.key === inp.key`` 的校验。
	inputs = ({"key": "data", "label": "Data"},)
	outputs = ({"key": "reviewed", "label": "Reviewed Data"},)

	# 配置结构（前端设置弹窗按设备卡片写入）::
	#
	# {"devices": {
	# "gripper_device": {
	# "checks": {
	# "umi.slam_continuity": {"enabled": true,
	# "params": {"max_linear_mps": 3.0}},
	# "video.freeze": {"enabled": false}
	# }}}}
	#
	# 引擎侧由 ``cleaning.engine.resolve_config()`` 解析。
	default_config = {"devices": {}}

	# ★ 刻意留空 —— 配置全部走专用弹窗（DeviceQualityModal），不走通用
	# config_schema 渲染。
	#
	# 通用那套是平铺字段列表，表达不了"按设备卡片分组 + 每项带阈值"的
	# 嵌套结构；硬塞会变成一长串 enable_xxx 开关。曾短暂用过那套，结果是
	# 三个阈值在节点里和检查项里各存一份（frame_tolerance 与
	# video.frame_drop.tolerance_ratio 同义），迟早不一致。
	#
	# 现在阈值只有一处定义：检查项的 ``default_params``；节点只存"覆盖值"。
	config_schema = ()

	# 真实检查在 run 完成回调里跑（那时才能读到合并后的 episode 数据），
	# 与 human_review / 旧 ai_quality_review 的分工一致。
	# 注：该字段当前无消费方，仅作语义标注（全仓 grep 只见 schema 与文档）。
	execution_target = "server"
	capabilities = ("data_quality", "quality_gate", "quality_review")

	def run(self, ctx: JobContext) -> dict[str, ArtifactRef] \| None:
	"""透传上游 artifact。上游没连 → skipped。

	真正的检查不在 ``run()`` 里跑 —— 在 worker 的 run 完成回调里
	（``api/worker.py`` 调 ``cleaning.trigger.spawn_cleaning_for_run``）。

	为什么不在 ``run()``：DAG 执行期 worker 是在 staging 目录里干活、
	产物要等 ``_publish_episode_result`` 才合并回批次目录。在这一步读
	批次目录，拿到的是上一轮的数据 —— 尤其 ``action`` 列，它由
	``umi_slam_action`` 派生产出，读早了会全是采集端的零占位。
	完成回调时数据已经合并完毕，与 ``human_review`` / 旧
	``ai_quality_review`` 的分工一致。

	检查所需的数据不靠端口传 —— 完成回调手上就有
	``localstore.get_episode()`` 的记录（``path`` + ``episode_index``），
	检查项直接读批次目录。

	★ 进来的是 ``episode_index`` 不是可选的：``path`` 是项目目录，
	几十集共享，漏传会拿第 0 集的数据去比全部集的视频。
	"""
	if not ctx.incoming:
	ctx.skip("No upstream data — skipped")
	return dict(ctx.incoming)