"""配准的目视核对（任务书 02 第三阶段）：`render --config ` 出图，`score --run-dir` 计分。 render 只做第 1 步、不用显卡：第 1 步成功的每例出棋盘拼图和边缘叠加图（只截扇形区），另随机挑 `align_check` 段定的例数把配准故意错开当阴性对照；图按打乱的编号放进 `images/`，对应关系写 `key.csv`（不给判读者）。score 对照 `verdicts_a.csv`、`verdicts_b.csv`：任一张对照被任一人判成 「对齐」或「看不清」就作废（不写 review.csv，退出码 1）；否则写 `review.csv`，两人都判「对齐」才 通过。终端只打印计数。 """ import argparse import csv import sys from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Any import cv2 import numpy as np import numpy.typing as npt from cadx import ceus from cadx.ceus import CeusConfig, HalfRegistration, ViewRect, crop_frame from cadx.delivery import read_delivery from cadx.runs import finish_snapshot, make_build_id, read_git_state, write_snapshot from ceus_quant import REPO_ROOT, REVIEW_HEADER, file_sha256, load_config, resolve_paths KEY_HEADER = ("id", "video_id", "is_control", "dx", "dy", "video_sha256") VERDICT_HEADER = ("编号", "判定", "依据") ALIGNED, MISALIGNED, UNCLEAR = "对齐", "没对齐", "看不清" _EDGE_FRACTION = 0.1 # 每半取扇形内梯度幅值最高的 10% 像素画成边缘 type _U8 = npt.NDArray[np.uint8] type _Gray = npt.NDArray[np.float32] type _Mask = npt.NDArray[np.bool_] @dataclass(frozen=True) class Material: """一例出图要的东西：两个视图；整帧的时间平均灰度、抽样中间那帧的灰度、绿框紫圈像素；两半扇形。""" left: ViewRect right: ViewRect mean: _Gray middle: _Gray marks: _Mask fan_left: _Mask fan_right: _Mask def step1(video_path: Path, cfg: CeusConfig) -> tuple[Material, HalfRegistration]: """第 1 步：逐帧解码只留分屏和配准要的抽样帧，找分屏线，配准两半。""" video = ceus.VideoFile(video_path) _, frame_count, _, _ = video.probe() split_at = set(ceus.sample_indices(frame_count, cfg.views.split_sample_frames)) register_at = set(ceus.sample_indices(frame_count, cfg.registration.sample_frames)) split_frames, frames = list[_U8](), list[_U8]() decoded = 0 for index, frame in video.iter_frames(): decoded += 1 if index in split_at: split_frames.append(frame) if index in register_at: frames.append(frame) if decoded != frame_count: raise ceus.VideoDecodeError(f"解码帧数 {decoded} 与容器记录的帧数 {frame_count} 不一致") _, left, right = ceus.split_views(split_frames, cfg.views) reg = ceus.register_halves( frames, left, right, cfg.registration, cfg.prompt_source, cfg.pan_source ) gray = np.stack([cv2.cvtColor(f, cv2.COLOR_BGR2GRAY) for f in frames]).astype(np.float32) std = gray.std(axis=0) fans = [ceus.fan_mask(crop_frame(std, view), cfg.registration) for view in (left, right)] marks = ceus.color_marks( frames, cfg.prompt_source, cfg.pan_source, cfg.registration.mark_dilate ) material = Material(left, right, gray.mean(0), gray[len(frames) // 2], marks, *fans) return material, reg def _shift(image: npt.NDArray[Any], offset: tuple[int, int], size: tuple[int, int]) -> Any: """右半按平移挪进左半坐标：结果 (x, y) = 右半 (x + dx, y + dy)，出界补 0。""" matrix = np.array([[1, 0, -offset[0]], [0, 1, -offset[1]]], dtype=np.float32) return cv2.warpAffine(image, matrix, size, flags=cv2.INTER_NEAREST) def _edges(gray: _Gray, fan: _Mask, marks: _Mask, sigma: float) -> Any: """扇形内、绿框紫圈之外梯度幅值最高的 `_EDGE_FRACTION` 像素，膨胀 1 像素便于看出重合。""" smooth = cv2.GaussianBlur(gray, (0, 0), sigma) magnitude = np.hypot(cv2.Sobel(smooth, cv2.CV_32F, 1, 0), cv2.Sobel(smooth, cv2.CV_32F, 0, 1)) keep = fan & ~marks # 绿框紫圈只画在一半上，它们的边缘不参与比对 edge = keep & (magnitude >= np.quantile(magnitude[keep], 1 - _EDGE_FRACTION)) return cv2.dilate(edge.astype(np.uint8), np.ones((3, 3), np.uint8)) > 0 def render_pair(m: Material, offset: tuple[int, int], sigma: float) -> tuple[_U8, _U8]: """按左半 → 右半的平移出（棋盘拼图，边缘叠加图），截到两半扇形的外接框，扇形外涂黑。""" left, right, size = m.left, m.right, (m.left.width, m.left.height) fan_r = _shift(m.fan_right.astype(np.uint8), offset, size) > 0 support = m.fan_left | fan_r ys, xs = np.nonzero(support) x1, x2, y1, y2 = int(xs.min()), int(xs.max()) + 1, int(ys.min()), int(ys.max()) + 1 block = max(8, round((x2 - x1) / 12)) yy, xx = np.mgrid[0 : left.height, 0 : left.width] first = ((yy - y1) // block + (xx - x1) // block) % 2 == 0 checker = np.where( first, crop_frame(m.middle, left), _shift(crop_frame(m.middle, right), offset, size) ) checker[~support] = 0 overlay = np.zeros((left.height, left.width, 3), dtype=np.uint8) marks_l = crop_frame(m.marks, left) marks_r = _shift(crop_frame(m.marks, right).astype(np.uint8), offset, size) > 0 edges_l = _edges(crop_frame(m.mean, left), m.fan_left, marks_l, sigma) edges_r = _edges(_shift(crop_frame(m.mean, right), offset, size), fan_r, marks_r, sigma) overlay[..., 2][edges_l] = 255 # 红：左半 overlay[..., 1][edges_r] = 255 # 绿：右半；重合处红 + 绿 = 黄 return np.clip(checker[y1:y2, x1:x2], 0, 255).astype(np.uint8), overlay[y1:y2, x1:x2] def render(config_path: Path) -> None: started_at = datetime.now().astimezone() cfg = resolve_paths(load_config(config_path)) git = read_git_state(REPO_ROOT) run_id = make_build_id(started_at, git) run_dir = cfg.paths.align_runs_root / run_id partial = cfg.paths.align_runs_root / f"{run_id}.partial" for path in (run_dir, partial): if path.exists(): raise FileExistsError(f"运行目录已存在，不覆盖：{path}") (partial / "images").mkdir(parents=True) rng = np.random.default_rng(cfg.align_check.seed) cases: list[tuple[str, str, Material, tuple[int, int]]] = [] failures: list[tuple[str, int, str]] = [] for case in read_delivery(cfg.paths.delivery_dir, None).cases: try: material, registration = step1(case.video, cfg.ceus) except ceus.CeusCaseError as error: failures.append((case.video_id, error.step, error.reason)) continue cases.append((case.video_id, file_sha256(case.video), material, registration.correlation)) # 对照例数多于成功例数时 rng.choice 报错（不放回抽样）。 controls = rng.choice(len(cases), size=cfg.align_check.control_count, replace=False) items: list[tuple[int, bool, tuple[int, int]]] = [(i, False, c[3]) for i, c in enumerate(cases)] for index in sorted(int(i) for i in controls): length = rng.uniform(*cfg.align_check.control_shift_px) angle = rng.uniform(0.0, 2 * np.pi) dx, dy = cases[index][3] shift = (round(length * np.cos(angle)), round(length * np.sin(angle))) items.append((index, True, (dx + shift[0], dy + shift[1]))) numbers = rng.permutation(len(items)) + 1 rows: list[tuple[str, str, str, int, int, str]] = [] for number, (index, is_control, offset) in zip(numbers.tolist(), items, strict=True): video_id, sha, material, _ = cases[index] name = f"{number:03d}" checker, edges = render_pair(material, offset, cfg.ceus.registration.blur_sigma) for kind, image in (("checker", checker), ("edges", edges)): if not cv2.imwrite(str(partial / "images" / f"{name}_{kind}.png"), image): raise RuntimeError(f"写 PNG 失败：{name}_{kind}.png") rows.append((name, video_id, str(is_control).lower(), offset[0], offset[1], sha)) with (partial / "key.csv").open("w", encoding="utf-8", newline="") as handle: csv.writer(handle).writerows([KEY_HEADER, *sorted(rows)]) with (partial / "failed.csv").open("w", encoding="utf-8", newline="") as handle: csv.writer(handle).writerows([("video_id", "step", "reason"), *failures]) configs = {"ceus.yaml": config_path} # align_check 段的随机数种子随 ceus.yaml 进快照 write_snapshot(partial / "snapshot", run_id, git, started_at, REPO_ROOT / "uv.lock", configs) finish_snapshot(partial / "snapshot", datetime.now().astimezone(), {}) if run_dir.exists(): raise FileExistsError(f"运行目录已存在，不覆盖：{run_dir}") partial.rename(run_dir) print(f"运行目录：{run_dir}") print(f"第 1 步成功 {len(cases)} 例、失败 {len(failures)} 例；对照 {len(controls)} 组") def _read_csv(path: Path, header: tuple[str, ...]) -> list[list[str]]: with path.open(encoding="utf-8", newline="") as handle: rows = list(csv.reader(handle)) if not rows or tuple(rows[0]) != header or any(len(r) != len(header) for r in rows): raise ValueError(f"{path} 的表头应为 {header}，且每行 {len(header)} 列") return rows[1:] def read_verdicts(path: Path, ids: set[str]) -> dict[str, str]: """一位判读者的判定，编号 → 判定。缺编号、多编号、重复编号、判定取值不合法都报错。""" rows = _read_csv(path, VERDICT_HEADER) verdicts = {row[0]: row[1] for row in rows} if len(verdicts) != len(rows): raise ValueError(f"{path} 有重复的编号") if set(verdicts) != ids: missing, extra = sorted(ids - set(verdicts)), sorted(set(verdicts) - ids) raise ValueError(f"{path} 的编号与出图不符：缺 {missing}，多 {extra}") bad = sorted({v for v in verdicts.values() if v not in (ALIGNED, MISALIGNED, UNCLEAR)}) if bad: raise ValueError(f"{path} 的判定只能是对齐、没对齐、看不清，收到 {bad}") return verdicts def score(run_dir: Path) -> int: """计分，返回退出码：0 写出 review.csv，1 本次核对作废。""" review_path = run_dir / "review.csv" if review_path.exists(): raise FileExistsError(f"review.csv 已存在，不覆盖：{review_path}") key = _read_csv(run_dir / "key.csv", KEY_HEADER) ids = {row[0] for row in key} a = read_verdicts(run_dir / "verdicts_a.csv", ids) b = read_verdicts(run_dir / "verdicts_b.csv", ids) controls = [row[0] for row in key if row[2] == "true"] real = [row for row in key if row[2] == "false"] caught_a, caught_b = (sum(v[i] == MISALIGNED for i in controls) for v in (a, b)) print(f"阴性对照 {len(controls)} 张：判读者 A 判「没对齐」{caught_a} 张，B {caught_b} 张") if caught_a < len(controls) or caught_b < len(controls): print("本次核对作废：有阴性对照被判成「对齐」或「看不清」，不写 review.csv") return 1 if len({row[5] for row in real}) != len(real): raise ValueError("key.csv 里同一段视频作为真实病例出现了两次") passed = [a[row[0]] == ALIGNED and b[row[0]] == ALIGNED for row in real] lines = [ (r[5], r[3], r[4], str(ok).lower(), a[r[0]], b[r[0]]) for r, ok in zip(real, passed, strict=True) ] with review_path.open("x", encoding="utf-8", newline="") as handle: csv.writer(handle).writerows([REVIEW_HEADER, *lines]) unclear = sum(UNCLEAR in (a[row[0]], b[row[0]]) for row in real) differ = sum(a[row[0]] != b[row[0]] for row in real) print(f"真实病例 {len(real)} 例：通过 {sum(passed)}，不通过 {len(real) - sum(passed)}") print(f"其中有人判「看不清」{unclear} 例，两人判定不一致 {differ} 例") return 0 def main(argv: list[str]) -> None: parser = argparse.ArgumentParser(description="配准的目视核对：出图、计分") commands = parser.add_subparsers(dest="command", required=True) commands.add_parser("render", help="出图").add_argument("--config", type=Path, required=True) commands.add_parser("score", help="计分").add_argument("--run-dir", type=Path, required=True) args = parser.parse_args(argv) if args.command == "render": render(args.config) elif score(args.run_dir) != 0: raise SystemExit(1) if __name__ == "__main__": main(sys.argv[1:])