Spaces:
Running
Running
stouchinglhh
Claude
sync processor from GitLab c697b6e: 版本 1.19.4 / 前端 1.12.4 —— 服务端 i18n(首屏不再闪英文)、推理模块并入模型页、前端抽出 shared/、缓存键改用前端号
bb59211 | """Data cleaning / validation — runs on uploaded episodes before review. | |
| Checks: | |
| 1. Frame alignment — parquet frame count vs video frame count | |
| 2. Sensor completeness — observation.state not all-zero | |
| 3. Video decode — MP4 files readable by OpenCV | |
| """ | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| from typing import Optional | |
| from sqlalchemy.ext.asyncio import AsyncSession | |
| from app.config import settings | |
| from app.models import Episode | |
| def utcnow(): | |
| return datetime.now(timezone.utc) | |
| async def validate_episode(ep: Episode, session_dir: Optional[Path], db: AsyncSession) -> dict: | |
| """Run all validation checks on a single episode. | |
| Returns the cleaning_report dict: {passed, checks: [{name, passed, detail}], summary} | |
| """ | |
| checks = [] | |
| # ── 1. Frame alignment check ── | |
| checks.append(await _check_frame_alignment(ep, session_dir)) | |
| # ── 2. Sensor completeness ── | |
| checks.append(await _check_sensor_completeness(ep, session_dir)) | |
| # ── 3. Video decode check ── | |
| checks.append(await _check_video_decode(ep)) | |
| # ── 4. Device connectivity check ── | |
| checks.append(await _check_device_connectivity(ep, session_dir)) | |
| passed = all(c["passed"] for c in checks) | |
| failed_names = [c["name"] for c in checks if not c["passed"]] | |
| report = { | |
| "passed": passed, | |
| "checks": checks, | |
| "summary": "All checks passed" if passed else f"Failed: {', '.join(failed_names)}", | |
| "validated_at": utcnow().isoformat(), | |
| } | |
| # Update episode | |
| ep.cleaning_report = report | |
| ep.processing_started_at = utcnow() | |
| ep.review_ready_at = utcnow() | |
| if passed: | |
| # Map old legacy statuses appropriately | |
| if ep.status not in ("reviewed", "approved", "rejected"): | |
| ep.status = "to_review" | |
| else: | |
| if ep.status not in ("reviewed", "approved"): | |
| ep.status = "failed" | |
| await db.flush() | |
| return report | |
| # ── Individual checks ────────────────────────────────────────── | |
| async def _check_frame_alignment(ep: Episode, session_dir: Optional[Path]) -> dict: | |
| """Compare parquet frame count with video frame count.""" | |
| if not session_dir or not session_dir.exists(): | |
| return {"name": "frame_alignment", "passed": True, | |
| "detail": "No session dir, skipped"} | |
| try: | |
| import pandas as pd | |
| parquet_frames = 0 | |
| seen_indices = set() | |
| for data_dir in session_dir.rglob("data"): | |
| if not data_dir.is_dir(): | |
| continue | |
| for pq_file in data_dir.rglob("*.parquet"): | |
| try: | |
| df = pd.read_parquet(pq_file) | |
| if "frame_index" in df.columns: | |
| for fi in df["frame_index"]: | |
| seen_indices.add(int(fi)) | |
| except Exception: | |
| continue | |
| parquet_frames = len(seen_indices) | |
| video_frames = ep.frame_count or 1 | |
| if parquet_frames == 0: | |
| return {"name": "frame_alignment", "passed": True, | |
| "detail": "No parquet data to compare"} | |
| ratio = abs(parquet_frames - video_frames) / max(video_frames, 1) | |
| passed = ratio < 0.10 # < 10% deviation | |
| return { | |
| "name": "frame_alignment", | |
| "passed": passed, | |
| "detail": f"parquet={parquet_frames}, video={video_frames}, dev={ratio:.1%}", | |
| } | |
| except Exception as e: | |
| return {"name": "frame_alignment", "passed": True, | |
| "detail": f"Check error (skipped): {e}"} | |
| async def _check_sensor_completeness(ep: Episode, session_dir: Optional[Path]) -> dict: | |
| """Check that observation.state data is present (not all-zero for all frames).""" | |
| if not session_dir or not session_dir.exists(): | |
| return {"name": "sensor_completeness", "passed": True, | |
| "detail": "No session dir, skipped"} | |
| try: | |
| import pandas as pd | |
| import numpy as np | |
| for data_dir in session_dir.rglob("data"): | |
| if not data_dir.is_dir(): | |
| continue | |
| for pq_file in data_dir.rglob("*.parquet"): | |
| try: | |
| df = pd.read_parquet(pq_file) | |
| # Look for observation columns | |
| obs_cols = [c for c in df.columns if c.startswith("observation.")] | |
| if not obs_cols: | |
| continue | |
| # Sample first 10 frames, check not all zero | |
| sample = df[obs_cols[0]].head(min(10, len(df))) | |
| all_zero = True | |
| for val in sample: | |
| if isinstance(val, (list, np.ndarray)): | |
| arr = np.array(val, dtype=np.float32) | |
| if arr.max() > 0: | |
| all_zero = False | |
| break | |
| if all_zero: | |
| return { | |
| "name": "sensor_completeness", | |
| "passed": False, | |
| "detail": "All sampled observation values are zero", | |
| } | |
| return { | |
| "name": "sensor_completeness", | |
| "passed": True, | |
| "detail": f"Checked {len(sample)} frames, data present", | |
| } | |
| except Exception: | |
| continue | |
| # No observation columns found — video-only episode, OK | |
| return {"name": "sensor_completeness", "passed": True, | |
| "detail": "No observation columns (video-only episode)"} | |
| except Exception as e: | |
| return {"name": "sensor_completeness", "passed": True, | |
| "detail": f"Check error (skipped): {e}"} | |
| async def _check_video_decode(ep: Episode) -> dict: | |
| """Verify the associated MP4 video is readable by OpenCV.""" | |
| try: | |
| import cv2 | |
| except ImportError: | |
| return {"name": "video_decode", "passed": True, | |
| "detail": "OpenCV not available, skipped"} | |
| video_path = None | |
| if ep.meta and ep.meta.get("video_path"): | |
| p = settings.storage_root / ep.meta["video_path"] | |
| if p.exists(): | |
| video_path = p | |
| if not video_path: | |
| # Try scanning sessions dir | |
| sessions_root = settings.storage_root / "sessions" | |
| if sessions_root.exists(): | |
| for mp4 in sessions_root.rglob("*.mp4"): | |
| video_path = mp4 | |
| break | |
| if not video_path: | |
| return {"name": "video_decode", "passed": True, | |
| "detail": "No video file found to check"} | |
| try: | |
| cap = cv2.VideoCapture(str(video_path)) | |
| if not cap.isOpened(): | |
| return {"name": "video_decode", "passed": False, | |
| "detail": f"Cannot open: {video_path.name}"} | |
| ret, _ = cap.read() | |
| cap.release() | |
| if not ret: | |
| return {"name": "video_decode", "passed": False, | |
| "detail": f"Cannot read first frame: {video_path.name}"} | |
| return {"name": "video_decode", "passed": True, | |
| "detail": f"Video readable: {video_path.name}"} | |
| except Exception as e: | |
| return {"name": "video_decode", "passed": False, | |
| "detail": f"Decode error: {e}"} | |
| async def _check_device_connectivity(ep: Episode, session_dir: Optional[Path]) -> dict: | |
| """Scan parquet for status.* columns — flag disconnected devices as failed.""" | |
| if not session_dir or not session_dir.exists(): | |
| return {"name": "device_connectivity", "passed": True, | |
| "detail": "No session dir, skipped"} | |
| try: | |
| import pandas as pd | |
| disconnected_devices: set[str] = set() | |
| for data_dir in session_dir.rglob("data"): | |
| if not data_dir.is_dir(): | |
| continue | |
| for pq_file in data_dir.rglob("*.parquet"): | |
| try: | |
| df = pd.read_parquet(pq_file) | |
| status_cols = [c for c in df.columns if c.startswith("status.")] | |
| for col in status_cols: | |
| if "disconnected" in df[col].values: | |
| device_name = col.split(".", 1)[1] | |
| disconnected_devices.add(device_name) | |
| except Exception: | |
| continue | |
| if disconnected_devices: | |
| return { | |
| "name": "device_connectivity", | |
| "passed": False, | |
| "detail": f"Disconnected: {', '.join(sorted(disconnected_devices))}", | |
| } | |
| return { | |
| "name": "device_connectivity", | |
| "passed": True, | |
| "detail": "All devices connected", | |
| } | |
| except Exception as e: | |
| return {"name": "device_connectivity", "passed": True, | |
| "detail": f"Check error (skipped): {e}"} | |
| # ── 文件级校验(无数据库版本)────────────────────────── | |
| def validate_batch(batch_dir: Path) -> dict: | |
| """校验一个批次目录(无数据库):帧对齐 / 传感器 / 视频解码。""" | |
| checks = [ | |
| _check_frame_alignment_fs(batch_dir), | |
| _check_sensor_completeness_fs(batch_dir), | |
| _check_video_decode_fs(batch_dir), | |
| ] | |
| passed = all(c["passed"] for c in checks) | |
| failed_names = [c["name"] for c in checks if not c["passed"]] | |
| return { | |
| "passed": passed, | |
| "checks": checks, | |
| "summary": "All checks passed" if passed else f"Failed: {', '.join(failed_names)}", | |
| "validated_at": utcnow().isoformat(), | |
| } | |
| def _check_frame_alignment_fs(batch_dir: Path) -> dict: | |
| """parquet 帧数 vs 视频帧数(±10%)。""" | |
| try: | |
| import pandas as pd | |
| seen = set() | |
| for data_dir in batch_dir.rglob("data"): | |
| if not data_dir.is_dir(): | |
| continue | |
| for pq_file in data_dir.rglob("*.parquet"): | |
| try: | |
| df = pd.read_parquet(pq_file) | |
| if "frame_index" in df.columns: | |
| seen.update(int(fi) for fi in df["frame_index"]) | |
| except Exception: | |
| continue | |
| parquet_frames = len(seen) | |
| if parquet_frames == 0: | |
| return {"name": "frame_alignment", "passed": True, "detail": "No parquet data"} | |
| # 2026-09-23 修:此处此前直接用 cv2 却没 import,异常被下方 except 吞掉, | |
| # 整个帧对齐检查在文件级路径下**永远静默通过**(detail 里写的是 | |
| # "Check error (skipped)" 但 passed=True,不会引起任何人注意)。 | |
| try: | |
| import cv2 | |
| except ImportError: | |
| return {"name": "frame_alignment", "passed": True, | |
| "detail": "OpenCV not available — video frame count skipped"} | |
| mp4s = sorted(batch_dir.rglob("*.mp4")) | |
| if not mp4s: | |
| # 与 _check_video_decode_fs 同款处理:没有视频就没有可比对的对象, | |
| # 判"未适用"而不是"不匹配"——只有传感器的批次(如纯手套)是合法的。 | |
| # 修复 cv2 漏 import 后,这里若不给守卫会退化成 dev=10000% 的误报。 | |
| return {"name": "frame_alignment", "passed": True, | |
| "detail": "No video to compare against"} | |
| video_frames = 0 | |
| for mp4 in mp4s: | |
| cap = cv2.VideoCapture(str(mp4)) | |
| try: | |
| video_frames = max(video_frames, int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) or 0) | |
| finally: | |
| cap.release() | |
| ratio = abs(parquet_frames - video_frames) / max(video_frames, 1) | |
| return {"name": "frame_alignment", "passed": ratio < 0.10, | |
| "detail": f"parquet={parquet_frames}, video={video_frames}, dev={ratio:.1%}"} | |
| except Exception as e: | |
| # 保持 passed=True(不改变既有行为),但 detail 必须说清是"降级通过", | |
| # 免得再出现"看起来全过了其实根本没检查"的情况。 | |
| return {"name": "frame_alignment", "passed": True, | |
| "detail": f"Degraded pass — check errored: {e}"} | |
| def _check_sensor_completeness_fs(batch_dir: Path) -> dict: | |
| """采样前 10 帧,检查 observation 数据非全零。""" | |
| try: | |
| import pandas as pd | |
| import numpy as np | |
| for data_dir in batch_dir.rglob("data"): | |
| if not data_dir.is_dir(): | |
| continue | |
| for pq_file in data_dir.rglob("*.parquet"): | |
| try: | |
| df = pd.read_parquet(pq_file) | |
| obs_cols = [c for c in df.columns if c.startswith("observation.")] | |
| if not obs_cols: | |
| continue | |
| sample = df[obs_cols[0]].head(min(10, len(df))) | |
| for val in sample: | |
| if isinstance(val, (list, np.ndarray)): | |
| if np.array(val, dtype=np.float32).max() > 0: | |
| return {"name": "sensor_completeness", "passed": True, | |
| "detail": f"Checked {len(sample)} frames, data present"} | |
| return {"name": "sensor_completeness", "passed": False, | |
| "detail": "All sampled observation values are zero"} | |
| except Exception: | |
| continue | |
| return {"name": "sensor_completeness", "passed": True, | |
| "detail": "No observation columns (video-only episode)"} | |
| except Exception as e: | |
| return {"name": "sensor_completeness", "passed": True, | |
| "detail": f"Check error (skipped): {e}"} | |
| def _check_video_decode_fs(batch_dir: Path) -> dict: | |
| """验证批次内第一个 MP4 可被 OpenCV 读取。""" | |
| try: | |
| import cv2 | |
| except ImportError: | |
| return {"name": "video_decode", "passed": True, "detail": "OpenCV not available"} | |
| mp4s = sorted(batch_dir.rglob("*.mp4")) | |
| if not mp4s: | |
| return {"name": "video_decode", "passed": True, "detail": "No video file"} | |
| try: | |
| cap = cv2.VideoCapture(str(mp4s[0])) | |
| if not cap.isOpened(): | |
| return {"name": "video_decode", "passed": False, | |
| "detail": f"Cannot open: {mp4s[0].name}"} | |
| ret, _ = cap.read() | |
| cap.release() | |
| return {"name": "video_decode", "passed": ret, | |
| "detail": f"Video readable: {mp4s[0].name}"} | |
| except Exception as e: | |
| return {"name": "video_decode", "passed": False, "detail": f"Decode error: {e}"} | |