iomgaa/CADx-4
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2753ea7721a2d8835f42e8ca2615f2a42b25cd9d
CADx-4/tests/test_delivery_cases.py
T
iomgaa 3524b31e3e feat(ceus): 任务书 02 第三阶段——scripts/ceus_quant.py 
- delivery 的 CaseRecord 加视频路径和推注时间（C04、C08 已检查过）。
- scripts/ceus_quant.py：读 ceus.yaml，跑检查全部通过的病例；runs/ceus/<运行号>/ 先写 .partial
  再改名；第 1–6 步的单例失败记 failed.csv 后接着跑；终端只打印计数。
- 分割缓存 work/segment_cache/<键>/：键含视频 SHA-256、第 1–3 步参数、权重 SHA-256、MedSAM2 提交号。
- 视频解码错误改为第 1 步的单例失败（VideoDecodeError），不再停掉整次运行（审阅意见）。
- 暗区 kmeans 用 cv2 进程级随机数，结果受病例顺序影响；ceus 配置加必填 random_seed，每例第 4 步前设种子。
- segment 导出 cuda_peak_memory_bytes，供运行快照记录显存峰值。

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-16 05:30:29 -04:00

437 lines
17 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""逐例检查 C01–C13：只让一例违反一项，这一例跳过、记对检查编号，其余病例照常通过。
预期值照任务书 01 第二部分「单个病例的问题」的文字写。
"""
import shutil
from pathlib import Path
from cadx.delivery import DeliveryResult, read_delivery
from cadx.delivery._text import conclusion_headings
from fake_delivery import (
CLINICAL_PATH,
DIAGNOSIS_COLUMN,
MR_PATH,
MR_TEXT,
SAMPLE_CODE_MAPS,
ULTRASOUND_HEADER,
ULTRASOUND_PATH,
ULTRASOUND_TEXT,
FakeDelivery,
make_delivery,
)
def _read(
tmp_path: Path, delivery: FakeDelivery, code_maps: dict[str, dict[int, str]] | None
) -> DeliveryResult:
return read_delivery(delivery.write(tmp_path / "delivery"), code_maps)
def _assert_only_skipped(result: DeliveryResult, case: str, check: str, total: int) -> None:
assert [(s.case, s.check) for s in result.skipped] == [(case, check)]
passed = [f"{n:03d}" for n in range(1, total + 1) if f"{n:03d}" != case]
assert [c.case_no for c in result.cases] == passed
assert result.total_cases == total
def test_all_pass(tmp_path: Path) -> None:
result = _read(tmp_path, make_delivery(4), None)
assert [c.case_no for c in result.cases] == ["001", "002", "003", "004"]
assert result.skipped == ()
assert result.unrecognized_rows == 0
def test_c01_unrecognized_clinical_id_is_its_own_row_and_case_missing(tmp_path: Path) -> None:
delivery = make_delivery(4)
delivery.case(2).clinical["Case_ID"] = "P2" # 临床表要求 P 加三位数字
result = _read(tmp_path, delivery, None)
# 第 1 行是表头，病例 2 在第 3 行；同一个病人因此在 C03（缺在临床表）再记一条。
assert [(s.case, s.check) for s in result.skipped] == [
(f"{CLINICAL_PATH} 第 3 行", "C01"),
("002", "C03"),
]
assert result.unrecognized_rows == 1
assert result.total_cases == 4
assert [c.case_no for c in result.cases] == ["001", "003", "004"]
def test_c01_float_report_id_is_unrecognized(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(3).mr["患者ID"] = 3.5 # 浮点不是认得出的写法
result = _read(tmp_path, delivery, None)
assert [(s.case, s.check) for s in result.skipped] == [
(f"{MR_PATH} 第 4 行", "C01"),
("003", "C03"),
]
def test_c01_report_ids_accept_int_and_zero_padded_strings(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(1).ultrasound["患者ID"] = 1
delivery.case(2).mr["患者ID"] = "002"
delivery.case(2).ultrasound["患者ID"] = "02"
result = _read(tmp_path, delivery, None)
assert result.skipped == ()
assert len(result.cases) == 3
def test_c02_duplicate_id_in_one_table(tmp_path: Path) -> None:
delivery = make_delivery(4)
duplicate = delivery.case(2).mr
delivery.extra_rows[MR_PATH].append([duplicate[h] for h in delivery.headers[MR_PATH]])
_assert_only_skipped(_read(tmp_path, delivery, None), "002", "C02", 4)
def test_c03_case_directory_missing(tmp_path: Path) -> None:
delivery = make_delivery(4)
root = delivery.write(tmp_path / "delivery")
shutil.rmtree(root / "CEUS_agent" / "003")
_assert_only_skipped(read_delivery(root, None), "003", "C03", 4)
def test_c04_missing_key_image(tmp_path: Path) -> None:
delivery = make_delivery(4)
delivery.case(4).dir_files.remove("U004_key.png")
_assert_only_skipped(_read(tmp_path, delivery, None), "004", "C04", 4)
def test_c04_extra_file(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(1).dir_files.append("U001.mp4") # 两个视频共四个文件
_assert_only_skipped(_read(tmp_path, delivery, None), "001", "C04", 3)
def test_c04_roi2_video_with_roi2_report_passes(tmp_path: Path) -> None:
delivery = make_delivery(3)
assert delivery.case(2).dir_files == ["U002_key.png", "U002_roi2.mp4", "U002_roi2_report.xlsx"]
assert _read(tmp_path, delivery, None).skipped == ()
def test_c04_plain_video_with_plain_report_passes(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(3).dir_files = ["U003_key.png", "U003.mp4", "U003_report.xlsx"]
result = _read(tmp_path, delivery, None)
assert result.skipped == ()
assert len(result.cases) == 3
def test_c04_roi2_video_with_plain_report_is_skipped(tmp_path: Path) -> None:
delivery = make_delivery(3)
# 文件名被 gitleaks 的通用密钥规则误报，不是密钥。
delivery.case(1).dir_files = [
"U001_key.png",
"U001_roi2.mp4", # gitleaks:allow
"U001_report.xlsx",
]
_assert_only_skipped(_read(tmp_path, delivery, None), "001", "C04", 3)
def test_c04_plain_video_with_roi2_report_is_skipped(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(2).dir_files = ["U002_key.png", "U002.mp4", "U002_roi2_report.xlsx"]
_assert_only_skipped(_read(tmp_path, delivery, None), "002", "C04", 3)
def test_c06_reads_plain_report_when_paired_with_plain_video(tmp_path: Path) -> None:
delivery = make_delivery(3)
case = delivery.case(3)
case.dir_files = ["U003_key.png", "U003.mp4", "U003_report.xlsx"]
row = [case.ultrasound[h] for h in ULTRASOUND_HEADER]
row[-1] = ULTRASOUND_TEXT + "补充一句。"
case.case_report_rows = [list(ULTRASOUND_HEADER), row]
_assert_only_skipped(_read(tmp_path, delivery, None), "003", "C06", 3)
def test_c05_video_id_mismatch(tmp_path: Path) -> None:
delivery = make_delivery(4)
delivery.case(2).ultrasound["CEUS视频ID"] = "U003"
_assert_only_skipped(_read(tmp_path, delivery, None), "002", "C05", 4)
def test_c06_case_report_cell_differs(tmp_path: Path) -> None:
delivery = make_delivery(4)
case = delivery.case(3)
row = [case.ultrasound[h] for h in ULTRASOUND_HEADER]
row[-1] = ULTRASOUND_TEXT + "补充一句。"
case.case_report_rows = [list(ULTRASOUND_HEADER), row]
result = _read(tmp_path, delivery, None)
_assert_only_skipped(result, "003", "C06", 4)
assert result.skipped[0].columns == ("超声描述",)
def test_c06_case_report_cell_type_differs(tmp_path: Path) -> None:
delivery = make_delivery(3)
case = delivery.case(1)
row = [case.ultrasound[h] for h in ULTRASOUND_HEADER]
row[0] = 1 # 汇总表里是字符串 "1"
case.case_report_rows = [list(ULTRASOUND_HEADER), row]
_assert_only_skipped(_read(tmp_path, delivery, None), "001", "C06", 3)
def test_c06_case_report_wrong_sheet(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(2).case_report_sheet = "Sheet2"
_assert_only_skipped(_read(tmp_path, delivery, None), "002", "C06", 3)
def test_c06_case_report_two_data_rows(tmp_path: Path) -> None:
delivery = make_delivery(3)
case = delivery.case(2)
row = [case.ultrasound[h] for h in ULTRASOUND_HEADER]
case.case_report_rows = [list(ULTRASOUND_HEADER), row, row]
_assert_only_skipped(_read(tmp_path, delivery, None), "002", "C06", 3)
def test_c07_bool_sex_is_not_an_int(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(1).clinical["Sex"] = True # bool 是 int 的子类，不能当作 1
_assert_only_skipped(_read(tmp_path, delivery, None), "001", "C07", 3)
def test_c01_bool_report_id_is_unrecognized(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(1).ultrasound["患者ID"] = True
result = _read(tmp_path, delivery, None)
assert [(s.case, s.check) for s in result.skipped] == [
(f"{ULTRASOUND_PATH} 第 2 行", "C01"),
("001", "C03"),
]
def test_c07_sex_out_of_range(tmp_path: Path) -> None:
delivery = make_delivery(4)
delivery.case(1).clinical["Sex"] = 3
_assert_only_skipped(_read(tmp_path, delivery, None), "001", "C07", 4)
def test_c07_non_integer_age(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(2).clinical["Age"] = 32.5
_assert_only_skipped(_read(tmp_path, delivery, None), "002", "C07", 3)
def test_c07_symptom_empty(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(2).clinical["Symptom_Fever"] = None
_assert_only_skipped(_read(tmp_path, delivery, None), "002", "C07", 3)
def test_c07_lab_only_one_column_empty(tmp_path: Path) -> None:
delivery = make_delivery(4)
delivery.case(2).clinical["CA199"] = None # 档位列仍是 0
result = _read(tmp_path, delivery, None)
_assert_only_skipped(result, "002", "C07", 4)
assert set(result.skipped[0].columns) == {"CA199", "CA199_Level"}
def test_c07_lab_both_empty_is_fine(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(1).clinical["TBil"] = None
delivery.case(1).clinical["TBil_Level"] = None
assert _read(tmp_path, delivery, None).skipped == ()
def test_c07_value_missing_from_code_map(tmp_path: Path) -> None:
delivery = make_delivery(3)
maps = {**SAMPLE_CODE_MAPS, "Hx_Alcohol": {0: "否"}} # 映射表里没有 1
delivery.case(2).clinical["Hx_Alcohol"] = 1
_assert_only_skipped(_read(tmp_path, delivery, maps), "002", "C07", 3)
def test_c08_ultrasound_sex_invalid(tmp_path: Path) -> None:
delivery = make_delivery(4)
delivery.case(4).ultrasound["性别"] = "M"
_assert_only_skipped(_read(tmp_path, delivery, None), "004", "C08", 4)
def test_c08_negative_bolus_time(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(3).ultrasound["推注开始时间s"] = -1
_assert_only_skipped(_read(tmp_path, delivery, None), "003", "C08", 3)
def test_c08_empty_description(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(1).ultrasound["超声描述"] = None
_assert_only_skipped(_read(tmp_path, delivery, None), "001", "C08", 3)
def test_c09_mr_age_not_number_plus_sui(tmp_path: Path) -> None:
delivery = make_delivery(4)
delivery.case(3).mr["年龄"] = "33"
_assert_only_skipped(_read(tmp_path, delivery, None), "003", "C09", 4)
def test_c09_mr_diagnosis_out_of_range(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(2).mr[DIAGNOSIS_COLUMN] = 5
_assert_only_skipped(_read(tmp_path, delivery, None), "002", "C09", 3)
def test_c10_diagnosis_mismatch(tmp_path: Path) -> None:
delivery = make_delivery(4)
delivery.case(1).clinical["Diagnosis"] = 2 # 两张报告表仍是 1
result = _read(tmp_path, delivery, None)
_assert_only_skipped(result, "001", "C10", 4)
assert result.skipped[0].columns == ("诊断",)
def test_c10_age_mismatch(tmp_path: Path) -> None:
delivery = make_delivery(4)
delivery.case(2).mr["年龄"] = "99岁"
result = _read(tmp_path, delivery, None)
_assert_only_skipped(result, "002", "C10", 4)
assert result.skipped[0].columns == ("年龄",)
def test_c10_sex_mismatch_between_reports(tmp_path: Path) -> None:
delivery = make_delivery(4)
delivery.case(3).mr["性别"] = "男" if delivery.case(3).ultrasound["性别"] == "女" else "女"
result = _read(tmp_path, delivery, None)
_assert_only_skipped(result, "003", "C10", 4)
assert result.skipped[0].columns == ("性别",)
def test_c10_clinical_sex_compared_only_when_decoding(tmp_path: Path) -> None:
delivery = make_delivery(3)
case = delivery.case(2)
case.clinical["Sex"] = 1 if case.clinical["Sex"] == 2 else 2 # 报告表的性别不变
undecoded = _read(tmp_path / "a", delivery, None)
assert undecoded.skipped == ()
assert undecoded.uncompared == ("临床性别未比对",)
decoded = _read(tmp_path / "b", delivery, SAMPLE_CODE_MAPS)
_assert_only_skipped(decoded, "002", "C10", 3)
assert decoded.uncompared == ()
def test_c11_heading_at_start_with_fullwidth_colon(tmp_path: Path) -> None:
delivery = make_delivery(4)
delivery.case(2).ultrasound["超声描述"] = "结论：胰体部囊性病灶。" + ULTRASOUND_TEXT
_assert_only_skipped(_read(tmp_path, delivery, None), "002", "C11", 4)
def test_c11_heading_after_newline(tmp_path: Path) -> None:
delivery = make_delivery(4)
delivery.case(3).mr["MR描述"] = MR_TEXT + "\n印象 : 胰体部囊性病灶。"
result = _read(tmp_path, delivery, None)
_assert_only_skipped(result, "003", "C11", 4)
assert result.skipped[0].columns == ("MR描述",)
def test_c11_heading_mid_text_is_not_a_failure(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(1).ultrasound["超声描述"] = "胰体部见一无回声区。造影提示：囊壁呈等增强。"
result = _read(tmp_path, delivery, None)
assert result.skipped == ()
assert len(result.cases) == 3
def test_c12_subtype_name_fullwidth_lowercase(tmp_path: Path) -> None:
delivery = make_delivery(4)
delivery.case(4).mr["MR描述"] = MR_TEXT + "形态类似ｉｐｍｎ。"
_assert_only_skipped(_read(tmp_path, delivery, None), "004", "C12", 4)
def test_c12_chinese_subtype_name(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(1).ultrasound["超声描述"] = ULTRASOUND_TEXT + "符合浆液性囊腺瘤表现。"
_assert_only_skipped(_read(tmp_path, delivery, None), "001", "C12", 3)
def test_c11_heading_after_vertical_tab_or_form_feed() -> None:
# xlsx（XML）里写不进 \x0b、\x0c，openpyxl 拒绝生成，所以直接测规则函数。
for separator in ("\x0b", "\x0c"):
text = MR_TEXT + separator + "诊断：胰体部囊性病灶。"
assert [k for k, _ in conclusion_headings(text)] == ["诊断"]
def test_c11_heading_after_unicode_line_separator(tmp_path: Path) -> None:
for i, separator in enumerate(("\x85", "\u2028", "\u2029")):
delivery = make_delivery(3)
delivery.case(2).mr["MR描述"] = MR_TEXT + separator + "诊断：胰体部囊性病灶。"
_assert_only_skipped(_read(tmp_path / str(i), delivery, None), "002", "C11", 3)
def test_c11_ratio_sign_as_colon(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(1).ultrasound["超声描述"] = "超声提示∶胰体部囊性病灶。" + ULTRASOUND_TEXT
result = _read(tmp_path, delivery, None)
_assert_only_skipped(result, "001", "C11", 3)
assert '"关键词": "超声提示"' in result.skipped[0].value_json
def test_c12_new_subtype_terms(tmp_path: Path) -> None:
terms = (
"SPT",
"ｓｐｔｐ",
"IPMT",
"导管内乳头",
"实性假乳头",
"黏液囊性",
"粘液囊性",
"囊腺瘤",
"浆液性微囊",
)
for i, term in enumerate(terms):
delivery = make_delivery(3)
delivery.case(3).ultrasound["超声描述"] = ULTRASOUND_TEXT + f"形态与{term}相似。"
_assert_only_skipped(_read(tmp_path / str(i), delivery, None), "003", "C12", 3)
def test_c12_serous_alone_is_fine(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(1).ultrasound["超声描述"] = ULTRASOUND_TEXT + "囊液呈浆液性。"
assert _read(tmp_path, delivery, None).skipped == ()
def test_c13_age_in_report(tmp_path: Path) -> None:
delivery = make_delivery(4)
delivery.case(1).ultrasound["超声描述"] = "患者４５岁。" + ULTRASOUND_TEXT
_assert_only_skipped(_read(tmp_path, delivery, None), "001", "C13", 4)
def test_c13_age_variants(tmp_path: Path) -> None:
for i, age in enumerate(("四十五岁", "45 岁", "6周岁", "三十歲", "１２周岁")):
delivery = make_delivery(3)
delivery.case(2).mr["MR描述"] = f"患者{age}。" + MR_TEXT
_assert_only_skipped(_read(tmp_path / str(i), delivery, None), "002", "C13", 3)
def test_c13_sex_character_in_report(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(3).mr["MR描述"] = "女性患者。" + MR_TEXT
_assert_only_skipped(_read(tmp_path, delivery, None), "003", "C13", 3)
def test_only_first_failing_check_is_recorded(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(2).ultrasound["性别"] = "M" # C08
delivery.case(2).mr["年龄"] = "99岁" # C10
_assert_only_skipped(_read(tmp_path, delivery, None), "002", "C08", 3)
def test_skip_record_names_file_and_column(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(2).ultrasound["推注开始时间s"] = 2.5
record = _read(tmp_path, delivery, None).skipped[0]
assert record.files == (ULTRASOUND_PATH,)
assert record.columns == ("推注开始时间s",)
assert record.value_json == '{"推注开始时间s": 2.5}'
def test_case_record_carries_video_path_and_bolus_time(tmp_path: Path) -> None:
delivery = make_delivery(3)
delivery.case(2).dir_files = ["U002_key.png", "U002.mp4", "U002_report.xlsx"]
delivery.case(3).ultrasound["推注开始时间s"] = 0 # 0 是合法取值：视频第 0 秒正好推注
root = delivery.write(tmp_path / "delivery")
result = read_delivery(root, None)
folder = root / "CEUS_agent"
assert [c.video for c in result.cases] == [
folder / "001" / "U001_roi2.mp4",
folder / "002" / "U002.mp4",
folder / "003" / "U003_roi2.mp4",
]
# 假交付包的推注时间是 5，第 3 例改成 0。
assert [c.bolus_start_s for c in result.cases] == [5, 5, 0]
Reference in New Issue View Git Blame Copy Permalink