iomgaa
7e7f32a13d
- ceus：灌注指标改为对病灶去暗区（roi_perf）和紫圈（roi_pan）两组算，roi_all 指标删除； D1–D4 仍要 roi_all、roi_perf 两条都过才定位（负责人 2026-10-17：医生要正常区与病灶对比看） - build/library：tic.json 只出两组 11 个指标与 7 个槽位的视频时间；医生截图挪进 ultrasound/keyframes/key.png，UltrasoundRoute 去掉 key_image、keyframes 多一个 key - review：TIC参数.csv 取代 TIC结果.csv；问题清单按原因对照表写成大白话，对照表外的原因报错； 程序原文进 snapshot/问题清单_原文.csv；说明.pdf 整份重写为合作者口吻 - 文档：结构.md、医生核对包、装配两个库、CEUS定量 任务书 04。 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
397 lines
16 KiB
Python
397 lines
16 KiB
Python
"""build 装入 CEUS 结果：C15 跳过、tic.json 只有白名单字段、关键帧去附加数据、运行目录不合规时停下。
|
|
|
|
tic.json 的预期内容照任务书 04 第二部分手写：紫圈（`roi_pan`）和病灶去暗区（`roi_perf`）两组各
|
|
11 个灌注指标（拟合没过整组为 null，单项算不出为 null），7 个槽位定位到的视频时间（没取到为
|
|
null），此外什么都没有。超声一路的文件是 report.txt、tic.json 和 keyframes/ 下的 key.png 加定位到
|
|
的槽位。
|
|
"""
|
|
|
|
import csv
|
|
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, cast
|
|
|
|
import pytest
|
|
from PIL import Image
|
|
|
|
from cadx.answers import read_aliases
|
|
from cadx.build import BuildSummary, run_build
|
|
from cadx.library import list_aliases, read_ultrasound, scan_for_leaks
|
|
from fake_ceus_run import perfusion, summary, write_run
|
|
from fake_delivery import REPO_ROOT, make_delivery, write_config
|
|
|
|
NOW = datetime(2026, 10, 16, 11, 0, 0).astimezone()
|
|
RUN_ID = "20261016-100000-abcdef0"
|
|
LOCATED = ("D1", "D2", "F1")
|
|
ALL_SLOTS = ("B0", "D1", "D2", "D3", "D4", "F1", "F2")
|
|
|
|
|
|
def _build(tmp_path: Path, count: int, run_dir: Path) -> BuildSummary:
|
|
root = make_delivery(count).write(tmp_path / "delivery")
|
|
return run_build(write_config(tmp_path, root, 1, None, run_dir), REPO_ROOT, NOW)
|
|
|
|
|
|
def _skipped(answers_dir: Path) -> list[list[str]]:
|
|
with (answers_dir / "skipped.csv").open(encoding="utf-8-sig", newline="") as handle:
|
|
return list(csv.reader(handle))[1:]
|
|
|
|
|
|
def _png_chunks(data: bytes) -> list[str]:
|
|
chunks: list[str] = []
|
|
offset = 8
|
|
while offset < len(data):
|
|
length = int.from_bytes(data[offset : offset + 4], "big")
|
|
chunks.append(data[offset + 4 : offset + 8].decode("ascii"))
|
|
offset += 12 + length
|
|
return chunks
|
|
|
|
|
|
def test_c15_skips_failed_and_missing_cases(tmp_path: Path) -> None:
|
|
run = write_run(
|
|
tmp_path / "runs" / RUN_ID,
|
|
tmp_path / "delivery",
|
|
{"U001": LOCATED, "U004": LOCATED},
|
|
[("U002", 2, "两半都没有绿框")],
|
|
)
|
|
result = _build(tmp_path, 4, run)
|
|
assert [r.case_no for r in read_aliases(result.answers_dir)] == ["001", "004"]
|
|
assert len(list_aliases(result.library_dir)) == 2
|
|
skipped = _skipped(result.answers_dir)
|
|
assert [row[:4] for row in skipped] == [
|
|
["002", "C15", "CEUS 定量结果齐全", f"{RUN_ID}/failed.csv"],
|
|
["003", "C15", "CEUS 定量结果齐全", f"{RUN_ID}/U003"],
|
|
]
|
|
assert json.loads(skipped[0][5]) == "第 2 步失败：两半都没有绿框"
|
|
assert "没有这一例的结果目录" in json.loads(skipped[1][5])
|
|
assert result.skipped_by_check == {"C15": 2}
|
|
manifest = json.loads((result.library_dir / "manifest.json").read_text(encoding="utf-8"))
|
|
assert manifest["ceus_source"] == RUN_ID
|
|
|
|
|
|
# 四例各自定位到的槽位和 summary.json 的改法：
|
|
# - U001：紫圈、病灶去暗区两组都有（紫圈 = perfusion(10.0)），7 个槽位都取到；
|
|
# - U002：只有病灶去暗区一组（假运行的原样），其中 WoAUC、FT 两项算不出；只取到 D1、D2、F1；
|
|
# - U003：两组都 null（病灶两条曲线拟合都没过，紫圈不可用）；只取到 F1、F2；
|
|
# - U004：只有紫圈一组，其中 RT 一项算不出；F2 没取到。
|
|
_CASE_SLOTS: dict[str, tuple[str, ...]] = {
|
|
"U001": ALL_SLOTS,
|
|
"U002": LOCATED,
|
|
"U003": ("F1", "F2"),
|
|
"U004": ("B0", "D1", "D2", "D3", "D4", "F1"),
|
|
}
|
|
|
|
|
|
def _both_groups(content: dict[str, Any]) -> None:
|
|
content["reference_valid"] = True
|
|
content["fits"]["roi_pan"].update(success=True, failure_reason=None)
|
|
content["perfusion_metrics"]["roi_pan"] = perfusion(10.0)
|
|
|
|
|
|
def _perf_items_missing(content: dict[str, Any]) -> None:
|
|
values = content["perfusion_metrics"]["roi_perf"]["values"]
|
|
values["WoAUC_au_s"] = None
|
|
values["FT_s"] = None
|
|
|
|
|
|
def _no_groups(content: dict[str, Any]) -> None:
|
|
for curve in ("roi_all", "roi_perf"):
|
|
content["fits"][curve].update(success=False, failure_reason="credible_peak_not_found")
|
|
content["lesion_fit_ready"] = False
|
|
content["perfusion_metrics"]["roi_perf"] = None
|
|
|
|
|
|
def _pan_only(content: dict[str, Any]) -> None:
|
|
_both_groups(content)
|
|
content["fits"]["roi_perf"].update(success=False, failure_reason="parameter_on_boundary")
|
|
content["lesion_fit_ready"] = False
|
|
content["perfusion_metrics"]["roi_perf"] = None
|
|
content["perfusion_metrics"]["roi_pan"]["values"]["RT_s"] = None
|
|
|
|
|
|
_EDITS = {"U001": _both_groups, "U002": _perf_items_missing, "U003": _no_groups, "U004": _pan_only}
|
|
|
|
_PAN = {
|
|
"PE_au": 10.0,
|
|
"WiAUC_au_s": 20.0,
|
|
"RT_s": 30.0,
|
|
"mTTl_s": 40.0,
|
|
"TTP_s": 50.0,
|
|
"WiR_au_per_s": 60.0,
|
|
"WiPI_au": 70.0,
|
|
"WoAUC_au_s": 80.0,
|
|
"WiWoAUC_au_s": 90.0,
|
|
"FT_s": 100.0,
|
|
"WoR_au_per_s": 110.0,
|
|
}
|
|
_PERF = {
|
|
"PE_au": 1.0,
|
|
"WiAUC_au_s": 2.0,
|
|
"RT_s": 3.0,
|
|
"mTTl_s": 4.0,
|
|
"TTP_s": 5.0,
|
|
"WiR_au_per_s": 6.0,
|
|
"WiPI_au": 7.0,
|
|
"WoAUC_au_s": 8.0,
|
|
"WiWoAUC_au_s": 9.0,
|
|
"FT_s": 10.0,
|
|
"WoR_au_per_s": 11.0,
|
|
}
|
|
# 假运行里定位到的槽位，视频时间 = 槽位序号 × 2.5 秒。
|
|
_ALL_TIMES = {"B0": 0.0, "D1": 2.5, "D2": 5.0, "D3": 7.5, "D4": 10.0, "F1": 12.5, "F2": 15.0}
|
|
|
|
EXPECTED_TIC = {
|
|
"U001": {
|
|
"perfusion_metrics": {"roi_pan": _PAN, "roi_perf": _PERF},
|
|
"keyframe_times_s": _ALL_TIMES,
|
|
},
|
|
"U002": {
|
|
"perfusion_metrics": {
|
|
"roi_pan": None,
|
|
"roi_perf": {**_PERF, "WoAUC_au_s": None, "FT_s": None},
|
|
},
|
|
"keyframe_times_s": {
|
|
"B0": None,
|
|
"D1": 2.5,
|
|
"D2": 5.0,
|
|
"D3": None,
|
|
"D4": None,
|
|
"F1": 12.5,
|
|
"F2": None,
|
|
},
|
|
},
|
|
"U003": {
|
|
"perfusion_metrics": {"roi_pan": None, "roi_perf": None},
|
|
"keyframe_times_s": {
|
|
"B0": None,
|
|
"D1": None,
|
|
"D2": None,
|
|
"D3": None,
|
|
"D4": None,
|
|
"F1": 12.5,
|
|
"F2": 15.0,
|
|
},
|
|
},
|
|
"U004": {
|
|
"perfusion_metrics": {"roi_pan": {**_PAN, "RT_s": None}, "roi_perf": None},
|
|
"keyframe_times_s": {**_ALL_TIMES, "F2": None},
|
|
},
|
|
}
|
|
_FORBIDDEN_TIC_KEYS = {
|
|
"fits",
|
|
"r2",
|
|
"success",
|
|
"roi_all",
|
|
"background_circle",
|
|
"reason",
|
|
"pan_curve_usable",
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def four_cases(tmp_path: Path) -> BuildSummary:
|
|
run = write_run(tmp_path / "runs" / RUN_ID, tmp_path / "delivery", _CASE_SLOTS, [])
|
|
for video_id, located in _CASE_SLOTS.items():
|
|
content = summary(video_id, located)
|
|
_EDITS[video_id](content)
|
|
(run / video_id / "summary.json").write_text(json.dumps(content), encoding="utf-8")
|
|
return _build(tmp_path, 4, run)
|
|
|
|
|
|
def _all_keys(node: object) -> set[str]:
|
|
if isinstance(node, dict):
|
|
items = cast(dict[str, Any], node)
|
|
return set(items) | {k for value in items.values() for k in _all_keys(value)}
|
|
return set()
|
|
|
|
|
|
def test_tic_json_is_exactly_the_agent_parameters(four_cases: BuildSummary) -> None:
|
|
by_video = {r.video_id: r.alias for r in read_aliases(four_cases.answers_dir)}
|
|
assert sorted(by_video) == sorted(EXPECTED_TIC)
|
|
for video_id, expected in EXPECTED_TIC.items():
|
|
tic_path = four_cases.library_dir / "cases" / by_video[video_id] / "ultrasound" / "tic.json"
|
|
assert json.loads(tic_path.read_text(encoding="utf-8")) == expected, video_id
|
|
assert read_ultrasound(four_cases.library_dir, by_video[video_id]).tic == expected
|
|
|
|
|
|
def test_tic_json_has_no_internal_keys(four_cases: BuildSummary) -> None:
|
|
for alias in list_aliases(four_cases.library_dir):
|
|
tic_path = four_cases.library_dir / "cases" / alias / "ultrasound" / "tic.json"
|
|
tic = json.loads(tic_path.read_text(encoding="utf-8"))
|
|
assert _all_keys(tic) & _FORBIDDEN_TIC_KEYS == set()
|
|
|
|
|
|
def test_ultrasound_files_are_exactly_report_tic_and_keyframes(
|
|
tmp_path: Path, four_cases: BuildSummary
|
|
) -> None:
|
|
for row in read_aliases(four_cases.answers_dir):
|
|
folder = four_cases.library_dir / "cases" / row.alias / "ultrasound"
|
|
files = sorted(p.relative_to(folder).as_posix() for p in folder.rglob("*") if p.is_file())
|
|
slots = [f"keyframes/{slot}.png" for slot in _CASE_SLOTS[row.video_id]]
|
|
assert files == sorted(["report.txt", "tic.json", "keyframes/key.png", *slots])
|
|
source = tmp_path / "delivery" / "CEUS_agent" / row.case_no / f"U{row.case_no}_key.png"
|
|
with Image.open(source) as original, Image.open(folder / "keyframes" / "key.png") as copy:
|
|
assert copy.mode == original.mode and copy.size == original.size
|
|
assert copy.tobytes() == original.tobytes()
|
|
route = read_ultrasound(four_cases.library_dir, row.alias)
|
|
assert route.keyframes["key"] == folder / "keyframes" / "key.png"
|
|
assert sorted(route.keyframes) == sorted(["key", *_CASE_SLOTS[row.video_id]])
|
|
|
|
|
|
def test_keyframes_metadata_dropped_and_pixels_kept(tmp_path: Path) -> None:
|
|
run = write_run(tmp_path / "runs" / RUN_ID, tmp_path / "delivery", {"U001": LOCATED}, [])
|
|
result = _build(tmp_path, 1, run)
|
|
(alias,) = list_aliases(result.library_dir)
|
|
route = read_ultrasound(result.library_dir, alias)
|
|
assert sorted(route.keyframes) == sorted(["key", *LOCATED])
|
|
for slot in LOCATED:
|
|
copied = route.keyframes[slot]
|
|
assert (
|
|
copied == result.library_dir / "cases" / alias / "ultrasound/keyframes" / f"{slot}.png"
|
|
)
|
|
with Image.open(run / "U001" / "keyframes" / f"{slot}.png") as original:
|
|
assert "Source" in original.info and "exif" in original.info
|
|
pixels = original.tobytes()
|
|
with Image.open(copied) as image:
|
|
assert image.mode == "RGB" and image.tobytes() == pixels
|
|
assert "Source" not in image.info and "exif" not in image.info
|
|
assert _png_chunks(copied.read_bytes()) == ["IHDR", "IDAT", "IEND"]
|
|
|
|
|
|
def test_null_ceus_run_has_only_key_image_in_keyframes(tmp_path: Path) -> None:
|
|
root = make_delivery(2).write(tmp_path / "delivery")
|
|
result = run_build(write_config(tmp_path, root, 1, None, None), REPO_ROOT, NOW)
|
|
for alias in list_aliases(result.library_dir):
|
|
route = read_ultrasound(result.library_dir, alias)
|
|
folder = result.library_dir / "cases" / alias / "ultrasound"
|
|
assert route.tic is None
|
|
assert route.keyframes == {"key": folder / "keyframes" / "key.png"}
|
|
files = sorted(p.relative_to(folder).as_posix() for p in folder.rglob("*") if p.is_file())
|
|
assert files == ["keyframes/key.png", "report.txt"]
|
|
|
|
|
|
def test_scan_finds_nothing_then_hits_video_id_in_tic_json(tmp_path: Path) -> None:
|
|
run = write_run(
|
|
tmp_path / "runs" / RUN_ID, tmp_path / "delivery", {"U001": LOCATED, "U002": LOCATED}, []
|
|
)
|
|
result = _build(tmp_path, 2, run)
|
|
forbidden = ["P001", "P002", "U001", "U002", "U001_roi2.mp4", "U002_roi2.mp4", "U999"]
|
|
assert scan_for_leaks(result.library_dir, forbidden, ("name",)) == []
|
|
alias = list_aliases(result.library_dir)[0]
|
|
tic_path = result.library_dir / "cases" / alias / "ultrasound" / "tic.json"
|
|
tic = json.loads(tic_path.read_text(encoding="utf-8"))
|
|
tic["keyframe_times_s"]["D4"] = "见 U002 第 3 帧"
|
|
tic_path.write_text(json.dumps(tic, ensure_ascii=False), encoding="utf-8")
|
|
hits = scan_for_leaks(result.library_dir, forbidden, ("name",))
|
|
assert [(Path(h.file).name, h.kind, h.term) for h in hits] == [("tic.json", "字符串", "U002")]
|
|
|
|
|
|
def test_keyframe_that_cannot_be_reencoded_skips_case_as_c14(tmp_path: Path) -> None:
|
|
run = write_run(
|
|
tmp_path / "runs" / RUN_ID, tmp_path / "delivery", {"U001": LOCATED, "U002": LOCATED}, []
|
|
)
|
|
Image.new("I;16", (4, 4)).save(run / "U002" / "keyframes" / "D2.png", format="PNG")
|
|
result = _build(tmp_path, 2, run)
|
|
assert [r.case_no for r in read_aliases(result.answers_dir)] == ["001"]
|
|
skipped = _skipped(result.answers_dir)
|
|
assert [row[:4] for row in skipped] == [
|
|
["002", "C14", "关键图能原样重编码", f"{RUN_ID}/U002/keyframes/D2.png"]
|
|
]
|
|
|
|
|
|
def _assert_stops(tmp_path: Path, run: Path, error: type[Exception]) -> None:
|
|
with pytest.raises(error):
|
|
_build(tmp_path, 1, run)
|
|
assert not (tmp_path / "library").exists() and not (tmp_path / "answers").exists()
|
|
|
|
|
|
def test_video_changed_since_run_stops(tmp_path: Path) -> None:
|
|
"""交付包里的视频内容和 CEUS 运行时记的 SHA-256 不同：运行不是在这份交付包上跑的，整次停下。"""
|
|
run = write_run(tmp_path / "runs" / RUN_ID, tmp_path / "delivery", {"U001": LOCATED}, [])
|
|
root = make_delivery(1).write(tmp_path / "delivery")
|
|
(root / "CEUS_agent" / "001" / "U001_roi2.mp4").write_bytes(b"another video")
|
|
with pytest.raises(ValueError, match=r"U001_roi2\.mp4 的 SHA-256"):
|
|
run_build(write_config(tmp_path, root, 1, None, run), REPO_ROOT, NOW)
|
|
assert not (tmp_path / "library").exists() and not (tmp_path / "answers").exists()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("delivery_bolus", "summary_offset", "message"),
|
|
[
|
|
(7, 5.0, "推注时间 7 秒"), # 交付包的推注时间改了
|
|
(5, 5, "应为 float"), # summary 里的推注时间不是浮点数
|
|
],
|
|
)
|
|
def test_injection_offset_mismatch_stops(
|
|
tmp_path: Path, delivery_bolus: int, summary_offset: object, message: str
|
|
) -> None:
|
|
run = write_run(tmp_path / "runs" / RUN_ID, tmp_path / "delivery", {"U001": LOCATED}, [])
|
|
content = summary("U001", LOCATED)
|
|
content["injection_offset_s"] = summary_offset
|
|
(run / "U001" / "summary.json").write_text(json.dumps(content), encoding="utf-8")
|
|
delivery = make_delivery(1)
|
|
delivery.case(1).ultrasound["推注开始时间s"] = delivery_bolus
|
|
root = delivery.write(tmp_path / "delivery")
|
|
with pytest.raises(ValueError, match=message):
|
|
run_build(write_config(tmp_path, root, 1, None, run), REPO_ROOT, NOW)
|
|
assert not (tmp_path / "library").exists() and not (tmp_path / "answers").exists()
|
|
|
|
|
|
def test_run_on_other_delivery_dir_stops(tmp_path: Path) -> None:
|
|
"""快照里记的交付包路径与 build 的不同（规范化后比）：整次停下；写法不同但同一路径照常装。"""
|
|
root = make_delivery(1).write(tmp_path / "delivery")
|
|
same = write_run(
|
|
tmp_path / "runs" / RUN_ID, tmp_path / "x" / ".." / "delivery", {"U001": LOCATED}, []
|
|
)
|
|
result = run_build(write_config(tmp_path, root, 1, None, same), REPO_ROOT, NOW)
|
|
assert len(list_aliases(result.library_dir)) == 1
|
|
other = write_run(tmp_path / "runs" / "other", tmp_path / "delivery_old", {"U001": LOCATED}, [])
|
|
build_dir = tmp_path / "second"
|
|
build_dir.mkdir()
|
|
with pytest.raises(ValueError, match=r"paths\.delivery_dir"):
|
|
run_build(write_config(build_dir, root, 1, None, other), REPO_ROOT, NOW)
|
|
assert not (build_dir / "library").exists() and not (build_dir / "answers").exists()
|
|
|
|
|
|
def test_run_dir_missing_stops(tmp_path: Path) -> None:
|
|
_assert_stops(tmp_path, tmp_path / "runs" / RUN_ID, FileNotFoundError)
|
|
|
|
|
|
def test_partial_run_dir_stops(tmp_path: Path) -> None:
|
|
run = write_run(
|
|
tmp_path / "runs" / f"{RUN_ID}.partial", tmp_path / "delivery", {"U001": LOCATED}, []
|
|
)
|
|
_assert_stops(tmp_path, run, ValueError)
|
|
|
|
|
|
def test_run_dir_without_failed_csv_stops(tmp_path: Path) -> None:
|
|
run = write_run(tmp_path / "runs" / RUN_ID, tmp_path / "delivery", {"U001": LOCATED}, [])
|
|
(run / "failed.csv").unlink()
|
|
_assert_stops(tmp_path, run, FileNotFoundError)
|
|
|
|
|
|
def test_run_dir_without_snapshot_stops(tmp_path: Path) -> None:
|
|
run = write_run(tmp_path / "runs" / RUN_ID, tmp_path / "delivery", {"U001": LOCATED}, [])
|
|
(run / "snapshot" / "snapshot.json").unlink()
|
|
_assert_stops(tmp_path, run, FileNotFoundError)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("extra", "remove"),
|
|
[
|
|
("B0.tmp", None), # B0.png 与同主干的 B0.tmp 并存：多一个文件
|
|
("B0.jpg", "B0.png"), # 只有 B0.jpg：扩展名不是 .png
|
|
],
|
|
)
|
|
def test_keyframe_files_must_be_exactly_slot_pngs(
|
|
tmp_path: Path, extra: str, remove: str | None
|
|
) -> None:
|
|
run = write_run(tmp_path / "runs" / RUN_ID, tmp_path / "delivery", {"U001": ("B0", "D1")}, [])
|
|
folder = run / "U001" / "keyframes"
|
|
(folder / extra).write_bytes((folder / "B0.png").read_bytes())
|
|
if remove is not None:
|
|
(folder / remove).unlink()
|
|
with pytest.raises(ValueError, match="定位到的槽位不一致"):
|
|
_build(tmp_path, 1, run)
|
|
assert not (tmp_path / "library").exists()
|