"""清洗报告组装 —— PDF §5 的「四层结果」。 第一层 数据流 哪条流好、哪条流坏 第二层 时间区间 问题发生在哪几秒、哪些帧 第三层 episode 这一集整体什么结论 第四层 训练用途 能用于什么训练、不能用于什么 全部是纯函数 —— 输入是检查项产出的 ``StreamResult`` / ``Finding``, 输出是可 JSON 序列化的 dict。不碰磁盘、不碰 cv2,因此完全可单测。 """ from __future__ import annotations from datetime import datetime, timezone from typing import Any, Iterable from app.processing.cleaning.contract import ( Finding, Range, StreamResult, PASS, WARN, FAIL, PENDING, ERROR, SEVERITY_ORDER, SEVERITY_LABELS, worst, is_blocking, ) SCHEMA_VERSION = 2 # 「训练用途」判定表 —— 声明式,不是 if-else 堆出来的。 # requires 里的信道必须全部具备,该用途才成立。 # PDF §7 要求"说明 episode 适合纯视觉、触觉或联合训练",这张表就是那个说明。 TRAINING_USE_RULES: tuple[dict[str, Any], ...] = ( {"use": "pure_vision", "label": "Pure vision", "requires": ("rgb", "time")}, {"use": "vision_gripper_joint", "label": "Vision + gripper", "requires": ("rgb", "gripper_state", "time")}, {"use": "slam_trajectory", "label": "SLAM trajectory", "requires": ("slam", "time")}, {"use": "vision_tactile_joint", "label": "Vision + tactile", "requires": ("rgb", "tactile", "time")}, {"use": "bimanual_tactile", "label": "Bimanual tactile", "requires": ("tactile", "time")}, ) def _utcnow() -> str: return datetime.now(timezone.utc).isoformat() def build_timeline(streams: Iterable[StreamResult | dict]) -> list[dict]: """把各条流里的区间拍平成一条时间轴,按时间排序去重。 详情页要一次拉完整条时间轴,而 streams 是嵌套的 —— 所以顶层冗余一份。 单一来源、单一方向(从 streams 派生),有测试保证一致。 """ items: list[dict] = [] for stream in streams: if isinstance(stream, StreamResult): for item in stream.ranges: items.append(_range_row(item.to_dict(), stream.stream)) elif isinstance(stream, dict): for item in stream.get("ranges") or []: if isinstance(item, dict): items.append(_range_row(item, str(stream.get("stream") or ""))) items.sort(key=lambda row: (row["start_sec"], row["end_sec"], row["code"])) deduped: list[dict] = [] for row in items: if deduped and all( row[key] == deduped[-1][key] for key in ("start_sec", "end_sec", "code", "stream") ): continue deduped.append(row) return deduped def _range_row(item: dict, stream: str) -> dict: return { "stream": item.get("stream") or stream, "channel": item.get("channel") or "", "start_sec": float(item.get("start_sec") or 0.0), "end_sec": float(item.get("end_sec") or 0.0), "start_frame": item.get("start_frame"), "end_frame": item.get("end_frame"), "code": str(item.get("code") or ""), "severity": str(item.get("severity") or WARN), "rule": str(item.get("rule") or ""), "message": str(item.get("message") or ""), } def rollup_episode(streams: Iterable[StreamResult | dict], findings: Iterable[Finding | dict] = ()) -> dict: """第三层:整集结论。纯函数。 取所有流 + 所有 finding 里最严重的那个状态。 """ statuses: list[str] = [] failed: list[str] = [] warned: list[str] = [] pending: list[str] = [] for stream in streams: if isinstance(stream, StreamResult): name, status = stream.stream, stream.status else: name, status = str(stream.get("stream") or ""), str(stream.get("status") or PASS) statuses.append(status) if is_blocking(status): failed.append(name) elif status == PENDING: pending.append(name) elif status == WARN: warned.append(name) for finding in findings: status = finding.status if isinstance(finding, Finding) else str(finding.get("status") or PASS) statuses.append(status) overall = worst(statuses) if statuses else PASS pieces: list[str] = [] if failed: pieces.append(f"{len(failed)} stream(s) failed") if warned: pieces.append(f"{len(warned)} stream(s) with warnings") if pending: pieces.append(f"{len(pending)} stream(s) pending review") return { "status": overall, "status_label": SEVERITY_LABELS.get(overall, overall), "failed_streams": failed, "warn_streams": warned, "pending_streams": pending, "summary": "; ".join(pieces) if pieces else "All checks passed", } def rollup_training_use(channels: Iterable[str], episode_status: str = PASS) -> list[dict]: """第四层:这个 episode 能用于哪些训练。纯函数。 信道不具备 → 该用途判 fail("未采集"),这是**用途维度**的失败, 不代表数据本身坏 —— 所以 reason 写清楚是"未采集"而不是"质量差"。 """ have = {str(item) for item in channels} rows: list[dict] = [] for rule in TRAINING_USE_RULES: required = tuple(rule["requires"]) missing = [channel for channel in required if channel not in have] if missing: rows.append({ "use": rule["use"], "label": rule["label"], "status": FAIL, "reason": f"not collected: {', '.join(missing)}", "requires": list(required), "missing": missing, }) else: rows.append({ "use": rule["use"], "label": rule["label"], "status": episode_status, "reason": "" if episode_status == PASS else "data quality problem", "requires": list(required), "missing": [], }) return rows def passed_for(status: str) -> bool: """该状态是否算"通过"(只有 PASS 算通过,WARN 不算)。""" return status == PASS def build_report(streams: Iterable[StreamResult], findings: Iterable[Finding] = (), *, episode_id: str = "", project: str = "", modalities: dict[str, Any] | None = None, channels: Iterable[str] = (), ruleset_revision: str = "", template_source: str = "", elapsed_ms: int = 0) -> dict: """组装完整报告。纯函数 —— 唯一的副作用是 ``_utcnow()``。""" stream_list = list(streams) finding_list = list(findings) episode = rollup_episode(stream_list, finding_list) use_channels = {str(item) for item in channels} for stream in stream_list: use_channels.update(stream.channels) return { "schema_version": SCHEMA_VERSION, "episode_id": episode_id, "project": project, "computed_at": _utcnow(), "elapsed_ms": int(elapsed_ms), "ruleset_revision": ruleset_revision, "template_source": template_source, # 模态清单 —— PDF §1 的「数据清单」 "modalities": dict(modalities or {}), # 判定总入口:只有 PASS 算通过 "passed": passed_for(episode["status"]), # 四层结果 "streams": [ item.to_dict() if isinstance(item, StreamResult) else dict(item) for item in stream_list ], "ranges": build_timeline(stream_list), "episode": episode, "training_use": rollup_training_use(use_channels, episode["status"]), "findings": [ item.to_dict() if isinstance(item, Finding) else dict(item) for item in finding_list ], } def empty_report(*, episode_id: str = "", project: str = "", reason: str = "no_data") -> dict: """没有任何数据可检查时的占位报告。 ``passed`` 恒为 False —— 没有数据不等于数据合格(fail closed)。 """ return { "schema_version": SCHEMA_VERSION, "episode_id": episode_id, "project": project, "computed_at": _utcnow(), "elapsed_ms": 0, "ruleset_revision": "", "template_source": "", "modalities": {}, "passed": False, "streams": [], "ranges": [], "episode": { "status": ERROR, "status_label": SEVERITY_LABELS[ERROR], "failed_streams": [], "warn_streams": [], "pending_streams": [], "summary": reason, }, "training_use": [], "findings": [], "reason": reason, }