CADx-4/tests/test_ceus_align_review.py

"""`scripts/ceus_align_review.py`：配准目视核对的出图和计分（任务书 02「配准的目视核对」）。

出图：假交付包两例合成分屏视频（`fake_ceus_video`，两半扇形对齐，配准 (0, 0)），对照取 1 例。
计分：key.csv 和两份判读手写，预期按任务书的规则手算。表头按任务书和协调者给的列名独立写。
"""

import csv
import hashlib
import importlib.util
import math
import re
from pathlib import Path
from types import ModuleType
from typing import Any

import cv2
import pytest
import yaml

from conftest import CEUS_YAML
from fake_ceus_video import write_video
from fake_delivery import make_delivery

_SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "ceus_align_review.py"
_KEY_HEADER = ["id", "video_id", "is_control", "dx", "dy", "video_sha256"]
_VERDICT_HEADER = ["编号", "判定", "依据"]
_REVIEW_HEADER = ["video_sha256", "dx", "dy", "passed", "verdict_a", "verdict_b"]


def _load_script() -> ModuleType:
    spec = importlib.util.spec_from_file_location("ceus_align_review", _SCRIPT)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _read(path: Path) -> list[list[str]]:
    with path.open(encoding="utf-8", newline="") as handle:
        return list(csv.reader(handle))


def _write(path: Path, rows: list[list[str]]) -> None:
    with path.open("w", encoding="utf-8", newline="") as handle:
        csv.writer(handle).writerows(rows)


def test_render_writes_shuffled_images_controls_and_key(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
    delivery_dir = make_delivery(2).write(tmp_path / "delivery")
    videos = {
        "U001": delivery_dir / "CEUS_agent" / "001" / "U001_roi2.mp4",
        "U002": delivery_dir / "CEUS_agent" / "002" / "U002_roi2.mp4",
    }
    write_video(videos["U001"], green_box=True)
    write_video(videos["U002"], green_box=False)  # 第 2 步才会失败，第 1 步照常出图
    document: dict[str, Any] = yaml.safe_load(CEUS_YAML.read_text(encoding="utf-8"))
    document["paths"].update(
        delivery_dir=str(delivery_dir),
        align_runs_root=str(tmp_path / "align"),
        align_review=str(tmp_path / "unused_review.csv"),  # 出图用不到，文件不存在也行
    )
    document["align_check"]["control_count"] = 1
    config = tmp_path / "ceus.yaml"
    config.write_text(yaml.safe_dump(document, allow_unicode=True), encoding="utf-8")
    module = _load_script()

    def fake_run_id(now: object, git: object) -> str:
        return "run-a"

    monkeypatch.setattr(module, "make_build_id", fake_run_id)
    module.main(["render", "--config", str(config)])

    run_dir = tmp_path / "align" / "run-a"
    assert not (tmp_path / "align" / "run-a.partial").exists()
    key = _read(run_dir / "key.csv")
    assert key[0] == _KEY_HEADER
    rows = key[1:]
    # 2 例真实 + 1 例对照，编号是 001–003 的一个排列，每个编号两张图。
    assert sorted(row[0] for row in rows) == ["001", "002", "003"]
    names = sorted(p.name for p in (run_dir / "images").iterdir())
    assert names == sorted(f"{row[0]}_{kind}.png" for row in rows for kind in ("checker", "edges"))
    assert all(re.fullmatch(r"\d{3}_(checker|edges)\.png", name) for name in names)
    real = [row for row in rows if row[2] == "false"]
    controls = [row for row in rows if row[2] == "true"]
    assert sorted(row[1] for row in real) == ["U001", "U002"]
    assert [(row[3], row[4]) for row in real] == [("0", "0"), ("0", "0")]
    (control,) = controls
    # 对照的平移 = 真实配准 (0, 0) + 长度 15–30 像素的随机错开（取整后容差 1 像素）。
    assert 14 <= math.hypot(int(control[3]), int(control[4])) <= 31
    for row in rows:
        assert row[5] == hashlib.sha256(videos[row[1]].read_bytes()).hexdigest()
    checker = cv2.imread(str(run_dir / "images" / f"{control[0]}_checker.png"))
    edges = cv2.imread(str(run_dir / "images" / f"{control[0]}_edges.png"))
    assert checker is not None and edges is not None and checker.shape == edges.shape
    assert _read(run_dir / "failed.csv") == [["video_id", "step", "reason"]]
    assert (run_dir / "snapshot" / "ceus.yaml").read_bytes() == config.read_bytes()
    out = capsys.readouterr().out
    assert "U001" not in out and "U002" not in out


# 手写的 key.csv：两例真实（001、003）、一例对照（002）。
_KEY = [
    _KEY_HEADER,
    ["001", "V1", "false", "-24", "0", "a" * 64],
    ["002", "V1", "true", "-4", "17", "a" * 64],
    ["003", "V2", "false", "183", "-2", "b" * 64],
]


def _score(module: ModuleType, run_dir: Path, a: list[list[str]], b: list[list[str]]) -> None:
    _write(run_dir / "key.csv", _KEY)
    _write(run_dir / "verdicts_a.csv", [_VERDICT_HEADER, *a])
    _write(run_dir / "verdicts_b.csv", [_VERDICT_HEADER, *b])
    module.main(["score", "--run-dir", str(run_dir)])


def test_score_writes_review_when_controls_caught(
    tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
    """对照两人都判「没对齐」；001 两人都判对齐 → 通过，003 有一人看不清 → 不通过。"""
    module = _load_script()
    a = [["001", "对齐", "血管连续"], ["002", "没对齐", "边缘错开"], ["003", "对齐", "扇形重合"]]
    b = [["001", "对齐", "血管连续"], ["002", "没对齐", "边缘错开"], ["003", "看不清", "太暗"]]
    _score(module, tmp_path, a, b)

    assert _read(tmp_path / "review.csv") == [
        _REVIEW_HEADER,
        ["a" * 64, "-24", "0", "true", "对齐", "对齐"],
        ["b" * 64, "183", "-2", "false", "对齐", "看不清"],
    ]
    out = capsys.readouterr().out
    assert "通过 1，不通过 1" in out
    assert "「看不清」1 例" in out and "不一致 1 例" in out
    # ceus_quant 读得回这份表。
    review = _load_quant().read_review(tmp_path / "review.csv")
    assert review["a" * 64].offset == (-24, 0) and review["a" * 64].passed is True
    assert review["b" * 64].passed is False


def _load_quant() -> ModuleType:
    spec = importlib.util.spec_from_file_location("ceus_quant", _SCRIPT.parent / "ceus_quant.py")
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


@pytest.mark.parametrize("control_verdict", ["对齐", "看不清"])
def test_score_is_void_when_a_control_is_missed(
    tmp_path: Path, capsys: pytest.CaptureFixture[str], control_verdict: str
) -> None:
    module = _load_script()
    a = [["001", "对齐", "x"], ["002", "没对齐", "x"], ["003", "对齐", "x"]]
    b = [["001", "对齐", "x"], ["002", control_verdict, "x"], ["003", "对齐", "x"]]
    with pytest.raises(SystemExit) as raised:
        _score(module, tmp_path, a, b)
    assert raised.value.code == 1
    assert not (tmp_path / "review.csv").exists()
    assert "作废" in capsys.readouterr().out


@pytest.mark.parametrize(
    ("rows", "message"),
    [
        ([["001", "对齐", "x"], ["002", "没对齐", "x"]], "缺"),  # 缺 003
        (
            [
                ["001", "对齐", "x"],
                ["002", "没对齐", "x"],
                ["003", "对齐", "x"],
                ["004", "对齐", "x"],
            ],
            "多",
        ),
        ([["001", "对齐", "x"], ["002", "没对齐", "x"], ["003", "差不多", "x"]], "判定只能是"),
        (
            [
                ["001", "对齐", "x"],
                ["001", "对齐", "x"],
                ["002", "没对齐", "x"],
                ["003", "对齐", "x"],
            ],
            "重复",
        ),
    ],
)
def test_score_rejects_malformed_verdicts(
    tmp_path: Path, rows: list[list[str]], message: str
) -> None:
    module = _load_script()
    good = [["001", "对齐", "x"], ["002", "没对齐", "x"], ["003", "对齐", "x"]]
    with pytest.raises(ValueError, match=message):
        _score(module, tmp_path, rows, good)
    assert not (tmp_path / "review.csv").exists()