机器人视频质检（一）：帧级检测器的实现与标定

{
    "codebase_version": "v3.0",
    "robot_type": "bi_yam_follower",
    "total_episodes": 32246,
    "total_frames": 76046658,
    "fps": 30,
    "features": {
        "observation.images.left": {
            "dtype": "video",
            "shape": [360, 640, 3],
            "info": { "video.codec": "av1", "video.fps": 30 }
        },
        "observation.state": {
            "dtype": "float32",
            "shape": [14],
            "names": [
                "left_joint_0.pos", "left_joint_1.pos", "left_joint_2.pos",
                "left_joint_3.pos", "left_joint_4.pos", "left_joint_5.pos",
                "left_gripper.pos",
                "right_joint_0.pos", "right_joint_1.pos", "right_joint_2.pos",
                "right_joint_3.pos", "right_joint_4.pos", "right_joint_5.pos",
                "right_gripper.pos"
            ]
        }
    }
}

import pyarrow.parquet as pq

cols = [
    "episode_index", "length", "tasks",
    "videos/observation.images.left/from_timestamp",
    "videos/observation.images.left/to_timestamp",
    "videos/observation.images.left/file_index",
]
meta = pq.read_table(
    "data/allenai__MolmoAct2-BimanualYAM-Dataset/meta/episodes/chunk-000/file-000.parquet",
    columns=cols,
).to_pandas()

print(meta.head())

# stage 1: 指标提取 —— 唯一昂贵的一段，10^5 帧/分钟量级
rows = extract_metrics(video_path)

# stage 2: 质量评估 —— 微秒级，纯阈值比较
results = detect_all(rows)

# stage 3: 可用性判定 —— 微秒级
summary = build_summary(video_path, results, probe_info)

import cv2
import numpy as np


def _frame_moments(bgr, black_pixel, bright_pixel, changed_pixel_threshold):
    gray = cv2.cvtColor(bgr, cv2.COLOR_BGR2GRAY)
    h, w = gray.shape
    n = h * w

    mean_gray = float(gray.mean())

    # 逐像素取颜色通道的最大值来判定黑/白，而不是用灰度均值。
    # 纯红像素 (0,0,220) 的灰度值很低，但它显然不是黑的。
    b, g, r = cv2.split(bgr)
    ch_max = cv2.max(cv2.max(b, g), r)
    black_ratio = float((ch_max < black_pixel).sum()) / n
    bright_ratio = float((ch_max > bright_pixel).sum()) / n

    lap = cv2.Laplacian(gray, cv2.CV_16S)
    lap_var = float(lap.var())

    gx = cv2.Sobel(gray, cv2.CV_32F, 1, 0, ksize=3)
    gy = cv2.Sobel(gray, cv2.CV_32F, 0, 1, ksize=3)
    tenengrad = float(np.mean(gx.astype(np.float64) ** 2 + gy.astype(np.float64) ** 2))

    return mean_gray, black_ratio, bright_ratio, lap_var, tenengrad

@dataclass(frozen=True)
class BlackThresholds:
    black_pixel: int = 20
    min_black_ratio: float = 0.99
    min_duration_s: float = 0.5


class BlackDetector(Detector):
    name = "black"
    severity = Status.FAIL
    dominates = ("blur", "freeze")

    def flag_frames(self, rows):
        return np.array([r.black_ratio >= self.t.min_black_ratio for r in rows], dtype=bool)

@dataclass(frozen=True)
class FreezeThresholds:
    max_mad: float = 0.01
    changed_pixel_threshold: int = 3
    max_changed_ratio: float = 1e-06
    min_duration_s: float = 1.0


class FreezeDetector(Detector):
    def flag_frames(self, rows):
        out = []
        for r in rows:
            if r.frame_index == 0:
                out.append(False)   # 没有前一帧可比
                continue
            out.append(r.mad <= self.t.max_mad and r.changed_ratio <= self.t.max_changed_ratio)
        return np.array(out, dtype=bool)

@dataclass(frozen=True)
class BlurThresholds:
    max_lap_var: float = 12.0
    min_duration_s: float = 0.4

@dataclass(frozen=True)
class ExposureThresholds:
    dark_pixel: int = 20
    bright_pixel: int = 245
    min_dark_ratio: float = 0.75
    min_bright_ratio: float = 0.75
    max_brightness_jump: float = 20.0
    min_duration_s: float = 0.5

def detect_all(rows, detectors=None):
    if detectors is None:
        detectors = [registry()[n]() for n in ("black", "exposure", "blur", "freeze")]

    results = []
    owned = None
    for det in detectors:
        res = det.run(rows, suppressed=owned)
        results.append(res)
        mask = det.flag_frames(rows)
        # 只有通过时长门槛的帧才算"被拥有"——
        # 两帧的模糊抖动不该盖住它底下一个真实的冻结。
        if res.intervals:
            own = np.zeros(len(rows), dtype=bool)
            for iv in res.intervals:
                for i, r in enumerate(rows):
                    if iv.start_ns <= r.timestamp_ns < iv.end_ns:
                        own[i] = True
            mask = own
        owned = mask if owned is None else (owned | mask)
    return results

def find_intervals(flags, timestamps_ns, min_duration_ns=0):
    if len(flags) != len(timestamps_ns):
        raise ValueError(f"flags ({len(flags)}) and timestamps_ns ({len(timestamps_ns)}) length mismatch")
    if not flags:
        return []

    n = len(flags)
    intervals = []
    i = 0
    while i < n:
        if not flags[i]:
            i += 1
            continue
        j = i
        while j + 1 < n and flags[j + 1]:
            j += 1
        start = timestamps_ns[i]
        # 帧 [i..j] 的结束时刻是第 j+1 帧的开始时刻。
        if j + 1 < n:
            end = timestamps_ns[j + 1]
        else:
            step = timestamps_ns[-1] - timestamps_ns[-2] if n >= 2 else 0
            end = timestamps_ns[-1] + max(step, 0)
        intervals.append(Interval(int(start), int(end)))
        i = j + 1

    if min_duration_ns > 0:
        intervals = [iv for iv in intervals if iv.duration_ns >= min_duration_ns]
    return intervals

def merge_intervals(intervals, max_gap_ns=0):
    """max_gap_ns 桥接短于阈值的间隙。

    从操作员的视角，隔 0.2 秒好帧的两段坏画面是"一次事故"；
    从检测器的视角是两段。两个参数都在，才能既表达"一次事故"，
    又不谎报坏帧到底在哪。
    """

def inject_video(src, dst, injections, start_frame=0, num_frames=600, fps=30.0,
                 blur_ksize=21, blur_sigma=6.0, exposure_gain=3.0,
                 seed=20261103, fourcc_name="FFV1"):
    """从 src 构造 dst，把 injections 注入进去。

    返回实际注入的列表（含解析后的帧号），调用方据此给检测器打分。
    """

PARQUET_SCHEMA_VERSION = "1.0"

FRAME_METRICS_SCHEMA = pa.schema([
    pa.field("episode_index", pa.int64()),
    pa.field("camera", pa.string()),
    pa.field("video_path", pa.string()),
    pa.field("frame_index", pa.int64()),
    pa.field("timestamp_ns", pa.int64()),
    pa.field("timestamp_s", pa.float64()),
    pa.field("mean_gray", pa.float64()),
    pa.field("black_ratio", pa.float64()),
    pa.field("bright_ratio", pa.float64()),
    pa.field("lap_var", pa.float64()),
    pa.field("tenengrad", pa.float64()),
    pa.field("mad", pa.float64()),
    pa.field("changed_ratio", pa.float64()),
    pa.field("schema_version", pa.string()),
])

{
  "schema_version": "1.1",
  "status": "WARN",
  "total_frames": 900,
  "duration_s": 30.0,
  "fps": 30.0,
  "detectors": {
    "blur": {
      "name": "blur",
      "status": "WARN",
      "severity": "WARN",
      "bad_frames": 120,
      "total_frames": 900,
      "bad_ratio": 0.133333,
      "intervals": [
        { "start_ns": 9999999900, "end_ns": 13999999860, "duration_ns": 3999999960 }
      ],
      "thresholds": { "max_lap_var": 12.0, "min_duration_s": 0.4 },
      "detail": { "min_lap_var": 1.0495, "p01_lap_var": 1.0694, "median_lap_var": 35.0216 }
    }
  },
  "file_layer": {
    "size_bytes": 57908532,
    "width": 640, "height": 360,
    "codec_name": "ffv1", "pix_fmt": "bgra",
    "container_nb_frames": 900
  }
}

python run_qc.py path/to/video.mp4 --camera left --episode 7 --out out/

def run_metric(key, attr, cands, higher_is_bad, skip_first=False, defect_case=None):
    bad_case = defect_case or key
    clean = series("clean", attr)
    defect = series(bad_case, attr)
    ...

def run(self, rows, suppressed=None):
    mask = self.flag_frames(rows)
    if suppressed is not None and len(suppressed) == len(mask):
        mask = mask & ~suppressed

    ivs = self.intervals(rows, mask)

    # 状态从"通过了时长门槛的帧"推导，而不是从原始标记数推导。
    # 两帧的模糊抖动短于 min_duration，不该是对这一段 episode 的警告。
    bad = _frames_in_intervals(rows, ivs) if ivs else 0
    ...

robot_video_qc/
├── __init__.py
├── metrics.py      # 指标提取（stage 1）
├── detectors.py    # 四个检测器 + 所有权仲裁
├── intervals.py    # 半开纳秒区间
├── checks.py       # L1 文件检查、L2 时间检查
├── summary.py      # L4 判定
├── contract.py     # Parquet / JSON 契约
├── calibrate.py    # 阈值扫描
├── inject.py       # 缺陷注入（无损）
├── pairing.py      # 跨模态配对（第二篇）
└── pipeline.py     # 三段式流水线
run_qc.py           # 命令行入口