iomgaa
1de88a5d0a
- 内存：prepare_segmentation 改为流式解码，只留分屏和配准的抽样帧，绿框逐帧在两半上找； 分割那一半由 SegmentationInput.target_frames() 到用时再解码，segment_video 边读边写 jpg； MedSAM2 init_state 的读帧换成直接在存放设备上建张量（不改 MedSAM2 源码，结果逐位相同）； 传播结果写进预分配数组。帧数最多的一段视频常驻内存峰值 11.2 GiB → 2.3 GiB，分辨率最大的 一段原代码被 OOM 杀掉、现为 4.2 GiB；两段的掩码和全部产物与改前逐字节相同，33 段视频的 第 1–2 步结果与改前一致，29 例配准与 review.csv 一致。 - 快照：ceus_quant 开跑时写 snapshot/（git、ceus.yaml、review.csv、uv.lock、segment.json）， 配置和核对表从快照副本读；收尾由新函数 runs.finish_snapshot 只新建 finished.json。 build 与 ceus_align_review 同样改用 finish_snapshot 记结束时间。 - build：summary.json 加 video_sha256；装 CEUS 结果时核对快照里的 delivery_dir、逐例核对 视频 SHA-256 和推注时间，不同就整次停下；新字段做类型检查，不进 tic.json。 来源：主会话任务（问题 1–3 及补充的 delivery_dir 比对）。 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
244 lines
12 KiB
Python
244 lines
12 KiB
Python
"""配准的目视核对（任务书 02 第三阶段）：`render --config <ceus.yaml>` 出图，`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:])
|