"""原因对照表:每种程序原因 → 给医生的那句话,逐条手写预期(任务书 04「原因的写法」)。 程序原因按源码里的写法造(`_anchor_frames.py`、`_tic_fitting.py`、`ceus/` 各步的异常、 `ceus_quant.py` 的目视核对、delivery 的 C01–C13、build 的 C14、C15),数值是编的。 """ import ast import json import re from pathlib import Path from typing import Any import pytest from cadx.answers import SkippedRow from cadx.review._problems import ( UnknownReasonError, case_label, ceus_failure, check_plain, curve_problems, group_reason, skip_problem, ) from fake_ceus_run import summary FORBIDDEN = re.compile(r"C[0-9]|roi_|_|\{|\[|band|[0-9]{8}-[0-9]{6}") BROKEN = "视频文件本身有问题(帧率、帧数或时间信息不对),程序读不了" NO_SPLIT = "视频里找不到左右两屏的分界线" MISALIGNED = "视频左右两屏的位置对不齐" @pytest.mark.parametrize( ("step", "reason", "expected"), [ (1, "配准目视核对未通过", f"{MISALIGNED}(我们人工逐例看过,这一例没对上)"), (1, "cv2 打不开视频:/x/U001_roi2.mp4", "视频文件打不开"), (1, "探测到非法 fps=0.0:/x/U001.mp4", BROKEN), (1, "探测到非法元数据 frame_count=0 width=1 height=1:/x/U001.mp4", BROKEN), (1, "视频包含非法显示时间戳 -1.0:/x/U001.mp4", BROKEN), (1, "视频显示时间戳不严格递增:第 3 帧 0.1,前一帧 0.2:/x/U001.mp4", BROKEN), (1, "时间戳帧数与容器记录的帧数不一致:3 / 4:/x/U001.mp4", BROKEN), (1, "解码帧数 3 与容器记录的帧数 4 不一致:/x/U001.mp4", BROKEN), (1, "帧尺寸不一致:首帧 (4, 6, 3),第 2 帧 (4, 8, 3)。", "视频前后画面大小不一致"), (1, "帧序列为空,无法求跨帧平均灰度。", "视频里一帧画面都没有"), (1, "帧序列为空,无法找分屏线。", "视频里一帧画面都没有"), ( 1, "中部 band 内未找到稳定暗缝(谷值 40.2 未显著低于 中位 51.0),疑为无缝或全黑画面。", NO_SPLIT, ), (1, "非法 search_band:要求 0<=lo None: assert ceus_failure(step, reason) == f"{expected},程序没法处理这段视频" REPO = Path(__file__).resolve().parents[1] # ceus 各步一例失败时抛的 CeusCaseError 子类 → 第几步(各类的 *_STEP 常量)。 _CASE_ERRORS = { "VideoDecodeError": 1, "SplitDetectionError": 1, "RegistrationError": 1, "ContrastSideError": 2, "NoValidContourError": 4, } def _raised_messages() -> list[tuple[str, int, str]]: """源码里每处 raise 一例失败的地方:(位置,第几步,消息,格式化部分代成 1)。""" found: list[tuple[str, int, str]] = [] files = [ *sorted((REPO / "src" / "cadx" / "ceus").glob("*.py")), REPO / "scripts" / "ceus_quant.py", ] for path in files: for node in ast.walk(ast.parse(path.read_text(encoding="utf-8"))): if not (isinstance(node, ast.Raise) and isinstance(node.exc, ast.Call)): continue func = node.exc.func name = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "") args = node.exc.args if name in _CASE_ERRORS: step, message = _CASE_ERRORS[name], args[0] elif name == "CeusCaseError": first = args[0] assert isinstance(first, ast.Constant) and isinstance(first.value, int) step, message = first.value, args[1] else: continue if isinstance(message, ast.Constant): text = str(message.value) else: assert isinstance(message, ast.JoinedStr), f"{path}:{node.lineno}" text = "".join( str(v.value) if isinstance(v, ast.Constant) else "1" for v in message.values ) found.append((f"{path.name}:{node.lineno}", step, text)) return found def test_every_raised_ceus_failure_is_in_the_table() -> None: """ceus 和 ceus_quant 里每一处 raise 一例失败的消息,对照表都认得(新加一处就会在这里报出)。""" messages = _raised_messages() assert len(messages) >= 20 for where, step, text in messages: assert not FORBIDDEN.search(ceus_failure(step, text)), where @pytest.mark.parametrize( ("step", "reason"), [(3, "分割模型报错"), (2, "配准目视核对未通过"), (1, "编造的原因")], ) def test_unknown_ceus_failure_stops(step: int, reason: str) -> None: with pytest.raises(UnknownReasonError): ceus_failure(step, reason) FIT_CASES = { "insufficient_valid_samples": "能用的画面太少(很多帧认不出这块区域,或是重复帧),连不成曲线", "candidate_grid_unavailable": "能用的画面时间点太少,连不成曲线", "credible_peak_not_found": "曲线太平,没有明显的峰", "candidate_arrival_not_found": "峰之前找不到亮度明确、持续升起来的时刻", "insufficient_bolus_samples": "造影剂第一次流过的那一段能用的画面太少", "long_gap_in_wash_in": "变亮的那一段中间缺了一大段画面", "arrival_bounds_invalid": "峰离视频开头太近,或造影剂第一次流过的那一段太短", "optimizer_failed": "曲线形状不规则,平滑曲线怎么也贴不上", "parameter_on_boundary": ( "试着贴出来的曲线在峰高、峰出现的时间、曲线宽窄、开始变亮的时刻或增强前水平上," "有一项卡在了程序允许范围的边上" ), "model_peak_on_boundary": "试着贴出来的曲线,峰落在可用那一段的最开头或最末尾,峰的位置不可信", "fit_quality_failed": "曲线起伏杂乱,平滑曲线和它差得太多", } @pytest.mark.parametrize(("code", "phrase"), FIT_CASES.items()) def test_fit_reason_codes(code: str, phrase: str) -> None: content = summary("U001", ()) content["reference_valid"] = True content["fits"]["roi_pan"].update(success=False, failure_reason=code) assert group_reason(content, "roi_pan") == f"{phrase},没能拟合" content["fits"]["roi_perf"].update(success=False, failure_reason=code) assert group_reason(content, "roi_perf") == f"{phrase},没能拟合" def test_every_fit_reject_code_is_in_the_table() -> None: source = (REPO / "src" / "cadx" / "ceus" / "_tic_fitting.py").read_text(encoding="utf-8") codes = set(re.findall(r'reject\("([a-z_]+)"\)', source)) assert codes == set(FIT_CASES) def test_group_reason_pan_unusable_and_unknown_code() -> None: content = summary("U001", ()) assert group_reason(content, "roi_pan") == "视频里大部分时间认不出紫圈" content["fits"]["roi_perf"].update(success=False, failure_reason="new_code") with pytest.raises(UnknownReasonError): group_reason(content, "roi_perf") BOUNDARY = FIT_CASES["parameter_on_boundary"] LESION = ";B0、D1–D4 要病灶整体和病灶去暗区两条曲线都拟合上才取,所以这 5 个关键帧没有取" FIT_FAILED = "病灶两条曲线(roi_all、roi_perf)的 Lognormal 拟合没有都过质量门" def _curves(located: tuple[str, ...], edit: dict[str, Any]) -> list[tuple[str, str]]: content = summary("U001", located) content["reference_valid"] = True content["fits"]["roi_pan"].update(success=True, failure_reason=None) for slot, reason in edit.items(): content["anchor_frames"][slot]["reason"] = reason return curve_problems(content) ALL = ("B0", "D1", "D2", "D3", "D4", "F1", "F2") @pytest.mark.parametrize( ("slot", "reason", "expected"), [ ( "B0", "B0 锚点 3.70 秒的容差窗口 [3.20, 4.20] 秒 内," "观测到的基线段里、到达时刻之前没有可用帧", "B0 没有取:开始增强之前的那一小段里没有能用的画面", ), ( "B0", "拟合曲线升不过到达门槛,没有到达时刻", "B0 没有取:曲线上升得太少,定不出造影剂到达的时刻", ), ( "D1", "拟合曲线上没有 D1 时刻(升不过到达门槛)", "D1 没有取:曲线上升得太少,定不出造影剂到达的时刻", ), ( "B0", "视频开头没有观测到稳定的增强前基线(基线是估计的),B0 不定义", "B0 没有取:视频一开始病灶就已经在变亮,没录到增强之前的那一段", ), ( "D4", "峰后观测不够长,或洗出拐点落在观测范围之外,拟合曲线上没有 D4", "D4 没有取:峰值之后录得不够长,找不到变暗最快的时刻", ), ( "D3", "D3 时刻 -0.40 秒落在视频有效帧范围 [0.04, 40.20] 秒 之外", "D3 没有取:算出的时刻不在视频能用的画面范围里" "(D3 在视频第 -0.4 秒,能用的画面是第 0.0–40.2 秒)", ), ( "D2", "D2 容差窗口 [4.70, 5.30] 秒 内的候选帧都被时间更早的槽位占用,无法保持先后顺序", "D2 没有取:这个时刻附近能用的画面都不晚于前一个关键帧取的那一帧" "(D1–D4 要按时间先后各取一帧,后一个必须比前一个晚)", ), ( "D2", "D2 容差窗口 [4.70, 5.30] 秒 内没有可用帧(拟合区间内、不重复)", "D2 没有取:这个时刻附近没有能用的画面" "(附近都是重复帧,或已过了造影剂第一次流过的那一段)", ), ( "F1", "推注后第 20 秒在视频第 -5.00 秒(推注时间 X=25 秒)," "落在视频有效帧范围 [0.00, 40.20] 秒之外", "F1 没有取:换算到视频里的时刻不在录到的范围内" "(F1 是推注后第 20 秒,应在视频第 -5.0 秒,那时还没开始录)", ), ( "F2", "推注后第 60 秒在视频第 55.00 秒(推注时间 X=5 秒)," "落在视频有效帧范围 [0.00, 40.20] 秒之外", "F2 没有取:换算到视频里的时刻不在录到的范围内" "(F2 是推注后第 60 秒,应在视频第 55.0 秒,视频只录到第 40.2 秒)", ), ( "F2", "推注后第 60 秒在视频第 55.00 秒(推注时间 X=5 秒),落在没有不重复的帧之外", "F2 没有取:换算到视频里的时刻不在录到的范围内" "(F2 是推注后第 60 秒,应在视频第 55.0 秒,视频里没有能用的画面)", ), ( "F2", "视频第 55.00 秒 ±2 秒内没有不重复的帧", "F2 没有取:这个时刻前后没有能用的画面(都是重复帧)", ), ( "F1", "没有推注时间,推注后时刻换算不到视频时间", "F1 没有取:没有推注时间,换算不出视频里的时刻", ), ], ) def test_keyframe_reasons(slot: str, reason: str, expected: str) -> None: located = tuple(s for s in ALL if s != slot) assert _curves(located, {slot: reason}) == [("关键帧", expected)] def test_same_reason_merges_into_one_row() -> None: reason = "视频第 {} 秒 ±2 秒内没有不重复的帧" rows = _curves(("B0", "D1", "D2", "D3", "D4"), {"F1": reason.format(1), "F2": reason.format(2)}) assert rows == [("关键帧", "F1、F2 没有取:这个时刻前后没有能用的画面(都是重复帧)")] @pytest.mark.parametrize( ("failed", "expected"), [ ( {"roi_all": "parameter_on_boundary"}, f"病灶整体那条曲线没能拟合({BOUNDARY}){LESION}", ), ( {"roi_all": "parameter_on_boundary", "roi_perf": "parameter_on_boundary"}, f"病灶整体、病灶去暗区两条曲线都没能拟合(都是{BOUNDARY}){LESION}" ",TIC 参数里病灶一组也空着", ), ( {"roi_all": "credible_peak_not_found", "roi_perf": "fit_quality_failed"}, "病灶整体、病灶去暗区两条曲线都没能拟合(病灶整体:曲线太平,没有明显的峰;" f"病灶去暗区:曲线起伏杂乱,平滑曲线和它差得太多){LESION}" ",TIC 参数里病灶一组也空着", ), ], ) def test_lesion_fit_failure_is_one_row(failed: dict[str, str], expected: str) -> None: content = summary("U001", ("F1", "F2")) content["reference_valid"] = True content["fits"]["roi_pan"].update(success=True, failure_reason=None) for curve, code in failed.items(): content["fits"][curve].update(success=False, failure_reason=code) for slot in ("B0", "D1", "D2", "D3", "D4"): content["anchor_frames"][slot]["reason"] = FIT_FAILED assert curve_problems(content) == [("TIC 曲线", expected)] def test_roi_perf_fit_reason_counts_as_fit_failure() -> None: content = summary("U001", ("F1", "F2")) content["reference_valid"] = True content["fits"]["roi_pan"].update(success=True, failure_reason=None) content["fits"]["roi_perf"].update(success=False, failure_reason="optimizer_failed") for slot in ("B0", "D1", "D2", "D3", "D4"): content["anchor_frames"][slot]["reason"] = ( "roi_perf 曲线的 Lognormal 拟合没过质量门(optimizer_failed)" ) assert [stage for stage, _ in curve_problems(content)] == ["TIC 曲线"] def test_fit_failure_without_its_keyframes_stops() -> None: content = summary("U001", ALL) content["fits"]["roi_all"].update(success=False, failure_reason="optimizer_failed") with pytest.raises(ValueError, match="对不上"): curve_problems(content) def test_pan_rows() -> None: content = summary("U001", ALL) assert curve_problems(content) == [ ("紫圈", "视频里大部分时间认不出紫圈,紫圈那条曲线不能用,TIC 参数里胰腺实质一组空着") ] content["reference_valid"] = True content["fits"]["roi_pan"]["failure_reason"] = "long_gap_in_wash_in" assert curve_problems(content) == [ ( "TIC 曲线", "紫圈那条曲线没能拟合(变亮的那一段中间缺了一大段画面),TIC 参数里胰腺实质一组空着", ) ] def test_unknown_keyframe_reason_stops() -> None: with pytest.raises(UnknownReasonError): _curves(("B0",), {s: "编造的原因" for s in ALL if s != "B0"}) CLINICAL = "Clinical_agent/临床工作簿_40例.xlsx" ULTRASOUND = "CEUS_agent/超声报告文本_40例.xlsx" MR = "MR_agent/MR报告文本_40例.xlsx" DIAGNOSIS = "病理类型(1 IPMN,2 SPN,3 MCN,4 SCN)" def _row(case: str, check: str, files: tuple[str, ...], value: object) -> SkippedRow: return SkippedRow( case=case, check=check, description="(检查说明不用)", files=files, columns=(), value_json=json.dumps(value, ensure_ascii=False), ) NOT_IN = ",这一例没进资料库" @pytest.mark.parametrize( ("row", "expected"), [ ( _row(f"{CLINICAL} 第 5 行", "C01", (CLINICAL,), {"原始写法": "P1", "行号": 5}), "临床表第 5 行的病例编号认不出(「P1」)", ), ( _row("病例目录 1000", "C01", ("CEUS_agent/1000",), {"目录名": "1000"}), "病例文件夹名「1000」认不出是哪个病例", ), ( _row("003", "C02", (MR,), {"行号": [4, 9]}), "MR 表里这个病例号出现了 2 次(第 4、9 行)", ), ( _row( "003", "C03", (ULTRASOUND, "CEUS_agent/003/"), {"缺在": [ULTRASOUND, "CEUS_agent/003/"]}, ), "在超声表、病例文件夹里找不到这一例", ), ( _row("003", "C04", ("CEUS_agent/003",), {"文件": ["U003_key.png", "U003.mp4"]}), "病例文件夹里应当恰好有截图、造影视频、单例超声报告表三个文件,实际有 2 个", ), ( _row("003", "C05", (ULTRASOUND,), {"CEUS视频ID": "U030", "病例号": "003"}), "超声表里这一例的 CEUS视频ID 写的是「U030」,和病例号对不上", ), ( _row("003", "C06", ("CEUS_agent/003/U003_report.xlsx",), {"工作表": ["A", "B"]}), "单例超声报告表的工作表不对(应只有一张)", ), ( _row( "003", "C06", ("CEUS_agent/003/U003_report.xlsx",), {"表头": "与超声汇总表不一致"} ), "单例超声报告表的表头和超声汇总表不一样", ), ( _row("003", "C06", ("CEUS_agent/003/U003_report.xlsx",), {"数据行数": 2}), "单例超声报告表应只有一行数据,实际有 2 行", ), ( _row( "003", "C06", ("CEUS_agent/003/U003_report.xlsx", ULTRASOUND), { "年龄": {"单例报告表": "40岁", "超声汇总表": "41岁"}, "超声描述": {"单例报告表": "甲" * 31, "超声汇总表": "乙"}, }, ), "单例超声报告表和超声汇总表对不上:" "年龄(单例表 「40岁」,汇总表 「41岁」);超声描述(内容不一样)", ), ( _row("003", "C07", (CLINICAL,), {"Sex": 3, "CA199_Level": None}), "临床表这几项的取值不合规:性别(3)、CA19-9 是否异常(空着)", ), ( _row("003", "C08", (ULTRASOUND,), {"年龄": "四十", "推注开始时间s": -1}), "超声表这几项的取值不合规:年龄(「四十」)、推注开始时间(-1)", ), ( _row("003", "C09", (MR,), {DIAGNOSIS: 5, "MR描述": None}), "MR 表这几项的取值不合规:病理诊断(5)、MR描述(空着)", ), ( _row( "003", "C10", (CLINICAL, ULTRASOUND, MR), {"年龄": {"临床表": 28, "超声表": "63岁", "MR 表": "63岁"}}, ), "三张表的年龄对不上:临床表 28 岁,超声表 63 岁,MR 表 63 岁", ), ( _row( "003", "C10", (CLINICAL, ULTRASOUND, MR), {"性别": {"超声表": "男", "MR 表": "男", "临床表(解码后)": "女"}}, ), "三张表的性别对不上:超声表 男,MR 表 男,临床表 女", ), ( _row( "003", "C10", (CLINICAL, ULTRASOUND, MR), {"性别": {"超声表": "女", "MR 表": "男"}}, ), "超声表和 MR 表的性别对不上:超声表 女,MR 表 男" "(临床表的性别是数字编码,还没有编码对照表,这次没有拿来比)", ), ( _row("003", "C08", (ULTRASOUND,), {"年龄": "datetime.datetime(2026, 1, 1, 0, 0)"}), "超声表这几项的取值不合规:年龄(这一格是日期或时间格式)", ), ( _row("003", "C07", (CLINICAL,), {"Symptom_Fever": True, "Hx_Alcohol": False}), "临床表这几项的取值不合规:发热(这一格填的是 TRUE)、饮酒史(这一格填的是 FALSE)", ), ( _row( "003", "C06", ("CEUS_agent/003/U003_report.xlsx", ULTRASOUND), {"推注开始时间s": {"单例报告表": "datetime.time(0, 0, 5)", "超声汇总表": 5}}, ), "单例超声报告表和超声汇总表对不上:" "推注开始时间(单例表 这一格是日期或时间格式,汇总表 5)", ), ( _row( "003", "C11", (ULTRASOUND,), {"超声描述": [{"关键词": "超声提示", "句子": "超声提示:囊性灶"}]}, ), "超声报告的描述里有像结论段标题的写法「超声提示:」(所在句子:超声提示:囊性灶)", ), ( _row("003", "C12", (MR,), {"MR描述": ["SCN", "浆液性囊"]}), "MR报告的描述里出现了亚型名「SCN」、「浆液性囊」", ), ( _row("003", "C13", (ULTRASOUND, MR), {"超声描述": ["女"], "MR描述": ["45岁"]}), "超声报告的描述里出现了性别或年龄「女」;MR报告的描述里出现了性别或年龄「45岁」", ), ( _row("003", "C14", ("CEUS_agent/003/U003_key.png",), "不是 PNG"), "医生截图不是 PNG 格式", ), ( _row("003", "C14", ("CEUS_agent/003/U003_key.png",), "带 tRNS 透明色,重编码会丢透明"), "医生截图带透明色,另存会丢掉透明信息", ), ( _row( "003", "C14", ("20261017-080000-abcdef0/U003/keyframes/D2.png",), "位深 16、颜色类型 2 不能逐像素保真重编码", ), "程序从视频里取的 D2 关键帧的颜色格式特殊,程序没法一点不差地另存", ), ], ) def test_skip_reasons(row: SkippedRow, expected: str) -> None: assert skip_problem(row, {}) == ("资料核对", expected, expected + NOT_IN) def test_c15_merges_with_failed_csv() -> None: failed = _row( "002", "C15", ("RUN/failed.csv",), "第 2 步失败:左右两半都没有找到绿框,选不出提示帧" ) reason = "视频里找不到画好的绿框,程序没法处理这段视频" steps = {"002": (2, "左右两半都没有找到绿框,选不出提示帧")} assert skip_problem(failed, steps) == ("造影视频处理", reason, reason + NOT_IN) with pytest.raises(ValueError, match="对不上"): skip_problem(failed, {"002": (1, "两半配准不一致:")}) missing = _row("003", "C15", ("RUN/U003",), "CEUS 运行目录里没有这一例的结果目录") assert skip_problem(missing, {}) == ( "造影视频处理", "造影视频没有处理结果", "造影视频没有处理结果" + NOT_IN, ) @pytest.mark.parametrize( "row", [ _row("003", "C99", (), {}), _row("003", "C07", (CLINICAL,), {"NewColumn": 1}), _row("003", "C14", ("CEUS_agent/003/U003_key.png",), "编造的原因"), _row("003", "C15", ("RUN/U003",), "编造的原因"), _row("003", "C10", (CLINICAL,), {"身高": {"临床表": 1}}), _row("003", "C03", ("x/y.xlsx",), {"缺在": ["x/y.xlsx"]}), ], ) def test_unknown_skip_reason_stops(row: SkippedRow) -> None: with pytest.raises(UnknownReasonError): skip_problem(row, {}) def test_case_labels() -> None: assert case_label("007") == "P007" assert case_label(f"{CLINICAL} 第 5 行") == "临床表第 5 行" assert case_label("病例目录 1000") == "病例文件夹「1000」" def test_program_words_stop() -> None: for text in ("C15 失败", "roi_pan", "a_b", "{", "[x]", "band", "20261017-080000"): assert FORBIDDEN.search(text) is not None with pytest.raises(ValueError, match="程序用语"): check_plain(text) assert check_plain("视频里找不到画好的绿框") == "视频里找不到画好的绿框"