From c582fe426f5c0c994cc8dbdfcda8b1ff8a448914 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Fri, 16 Oct 2026 05:55:57 -0400 Subject: [PATCH] =?UTF-8?q?feat(build):=20=E4=BB=BB=E5=8A=A1=E4=B9=A6=2002?= =?UTF-8?q?=20=E7=AC=AC=E5=9B=9B=E9=98=B6=E6=AE=B5=E4=BB=A3=E7=A0=81?= =?UTF-8?q?=E2=80=94=E2=80=94build=20=E8=A3=85=E5=85=A5=20CEUS=20=E7=BB=93?= =?UTF-8?q?=E6=9E=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - build.yaml 加必填字段 ceus_run（CEUS 运行目录或 null）。 - 不为 null 时按视频号找结果；在 failed.csv 里或缺结果目录的整例跳过，记 C15；关键帧图沿用 C14 检查。 - 资料库超声一路加 keyframes/<槽位>.png（去附加数据重编码）和 tic.json（白名单逐字段构造： 四条曲线的拟合成败与 R²、灌注指标、紫圈曲线是否可用、7 个槽位的定位与视频时间或原因）。 - manifest 的 CEUS 来源写运行号；泄露扫描已覆盖 tic.json，补命中测试。 - 真实构建等配准改完、全量 CEUS 运行之后再做。 Co-Authored-By: Claude Opus 5.5 --- config/build.yaml | 2 + research-wiki/explanation/结构.md | 23 +-- research-wiki/guides/装配两个库.md | 31 ++-- src/cadx/build/_assemble.py | 47 +++++- src/cadx/build/_ceus.py | 151 ++++++++++++++++++++ src/cadx/build/_config.py | 3 + src/cadx/library/__init__.py | 3 +- src/cadx/library/_read.py | 20 ++- src/cadx/library/_write.py | 32 ++++- tests/fake_ceus_run.py | 118 ++++++++++++++++ tests/fake_delivery.py | 11 +- tests/test_build_ceus.py | 220 +++++++++++++++++++++++++++++ tests/test_build_outputs.py | 5 +- tests/test_build_stops.py | 16 ++- tests/test_ceus_quant.py | 36 ++++- 15 files changed, 683 insertions(+), 35 deletions(-) create mode 100644 src/cadx/build/_ceus.py create mode 100644 tests/fake_ceus_run.py create mode 100644 tests/test_build_ceus.py diff --git a/config/build.yaml b/config/build.yaml index 4c42338..95b583a 100644 --- a/config/build.yaml +++ b/config/build.yaml @@ -8,3 +8,5 @@ alias_seed: 20261016 # 编码列是否按映射表解码。映射表还没向临床方拿到，先不解码，code_maps 必须是 null。 decode_codes: false code_maps: null +# CEUS 定量的运行目录（runs/ceus/<运行号>/），或 null（不装 CEUS 结果）。 +ceus_run: null diff --git a/research-wiki/explanation/结构.md b/research-wiki/explanation/结构.md index 7a560de..7649843 100644 --- a/research-wiki/explanation/结构.md +++ b/research-wiki/explanation/结构.md @@ -153,14 +153,20 @@ runs/ceus/<运行号>/ ────► ② 装配两个库 config/bu ``` library/<构建号>/ - manifest.json 构建清单：每个文件的 SHA-256、构建时的 git 提交、配置哈希、所用 CEUS 结果的来源 + manifest.json 构建清单：每个文件的 SHA-256、构建时的 git 提交、配置哈希、所用 CEUS 运行号（没装为 null） cases/<化名>/ clinical/clinical.json 临床那一路 - ultrasound/ 超声那一路：report.txt、key.png、keyframes/*.png、tic.json - （后两项等 CEUS 定量落地后由 build 加入，目前只有前两项） + ultrasound/ 超声那一路：report.txt、key.png；装了 CEUS 结果时另有 + keyframes/<槽位>.png 和 tic.json mr/report.txt MR 那一路 ``` +`build.yaml` 的 `ceus_run` 指向某次 CEUS 定量的运行目录时，build 才往超声一路装 CEUS 结果，为 null 时不装。装的时候： + +- `keyframes/<槽位>.png`：下文关键帧表里定位到的那几张（槽位即 B0、D1 等），和 `key.png` 一样去掉附加数据重新编码；每例最多 8 张图（7 张关键帧加 `key.png`）。 +- `tic.json`：从这一例的 `summary.json` 里逐字段挑出，此外什么都不写：四条曲线（`roi_all` 病灶整体、`roi_perf` 病灶去暗区、`roi_pan` 紫圈即医生圈的一块正常胰腺、`background_circle` 背景圆）各自的拟合是否成功（`success`）和拟合优度 R²（`r2`）；病灶两条曲线的灌注指标（`perfusion_metrics`，拟合没过质量门为 null）；紫圈曲线是否可用（`pan_curve_usable`）；7 个槽位各自是否定位到（`located`）、定位到的视频时间（`video_time_s`，秒）、没定位的原因（`reason`）。视频号、文件路径、配置、帧号、拟合参数都不写。挑哪些字段只在 `src/cadx/build/_ceus.py` 里列。 +- CEUS 在这一例上失败（在 `failed.csv` 里）或运行目录里没有这一例的结果，这一例整例跳过，记 C15。失败的病例要不要改成「超声不可评价、仍参与会诊」等 `../待定问题/03-病例排除标准.md` 定。 + 每一路能看见什么： | 内容 | 临床 | 超声 | MR | @@ -245,7 +251,7 @@ X 是超声报告工作簿的「推注开始时间s」，含义是视频第 0 | 文件 | 给谁用 | 内容 | | --- | --- | --- | -| `build.yaml` | `scripts/build.py` | `delivery_dir`（交付包位置）、`library_root`、`answers_root`（两个库的上级目录）、`alias_seed`（生成化名的随机种子）、`decode_codes`（编码解码开关）、`code_maps`（映射表位置或 null）；字段定义和校验在 `src/cadx/build/_config.py`，逐项说明见 `../guides/装配两个库.md` | +| `build.yaml` | `scripts/build.py` | `delivery_dir`（交付包位置）、`library_root`、`answers_root`（两个库的上级目录）、`alias_seed`（生成化名的随机种子）、`decode_codes`（编码解码开关）、`code_maps`（映射表位置或 null）、`ceus_run`（CEUS 运行目录或 null）；字段定义和校验在 `src/cadx/build/_config.py`，逐项说明见 `../guides/装配两个库.md` | | `code_maps.yaml` | build（经 `build.yaml` 引用） | 编码映射表：交付包里每个用数字编码的列，每个数字代表什么（如性别的 1、2）。还没向临床方拿到，尚未建 | | `ceus.yaml` | `scripts/ceus_quant.py` | CEUS 定量的全部参数，每个参数用注释写明出处 | | `consult.yaml` | `scripts/consult.py` | 会诊的实验配置与模型用途 | @@ -339,13 +345,14 @@ Python 环境用 uv；安装与检查命令见 `../guides/开发环境.md`。 ### library（`src/cadx/library/`） -- `write_library(build_dir, build_id, git_commit, git_clean, config_sha256, codes_decoded, uncompared, ceus_source, cases)`：把一次构建的各例三路内容写进一个已存在的空目录，关键图去掉附加数据重新编码，最后写 `manifest.json`。关键图不能保真重编码时报错停下（说明 build 漏了 C14）。 +- `write_library(build_dir, build_id, git_commit, git_clean, config_sha256, codes_decoded, uncompared, ceus_source, cases)`：把一次构建的各例三路内容写进一个已存在的空目录，关键图和关键帧去掉附加数据重新编码，有 CEUS 结果时写 `tic.json`，最后写 `manifest.json`（`ceus_source` 是 CEUS 运行号或 None，与每例的 `ceus` 是否为 None 必须配套）。图不能保真重编码时报错停下（说明 build 漏了 C14）。 - `check_key_image(path)`：关键图能逐像素保真重编码时返回 None，不能时（不是 PNG、位深或颜色类型不行、带 tRNS 透明色）返回原因。 -- `LibraryCase`：要写进资料库的一例：化名、临床一路的字段、超声描述、关键图源文件、MR 描述。 +- `LibraryCase`：要写进资料库的一例：化名、临床一路的字段、超声描述、关键图源文件、MR 描述、CEUS 部分（`LibraryCeus` 或 None）。 +- `LibraryCeus`：一例超声一路的 CEUS 部分：build 挑好的 tic.json 内容、槽位 → 关键帧源图。 - `list_aliases(build_dir)`：一次构建的全部化名（按化名排序）。 - `read_clinical(build_dir, alias)`、`read_ultrasound(build_dir, alias)`、`read_mr(build_dir, alias)`：按路读出一例，读时核对 SHA-256，拒绝 `.partial` 目录。 - `ClinicalRoute`：临床一路：是否已解码、22 列字段。 -- `UltrasoundRoute`：超声一路：报告「所见」原文、关键图路径。 +- `UltrasoundRoute`：超声一路：报告「所见」原文、关键图路径、tic.json 的内容（没装 CEUS 时为 None）、槽位 → 关键帧路径（没装时为空）。 - `MrRoute`：MR 一路：报告「所见」原文。 - `scan_for_leaks(build_dir, forbidden_strings, forbidden_keys)`：在构建目录的全部 `.txt`、`.json` 里找禁止的字符串（NFKC 归一、不分大小写、按子串）和禁止的 JSON 键名，返回命中列表。 - `LeakHit`：一处命中：文件（相对构建目录）、类别（「字符串」或「键名」）、命中的词。 @@ -390,7 +397,7 @@ Python 环境用 uv；安装与检查命令见 `../guides/开发环境.md`。 ### build（`src/cadx/build/`） -- `run_build(config_path, repo_root, now)`：完整做一次构建：读配置和映射表、读交付包、对每例做 C14（`check_key_image` 返回原因的整例跳过，记进跳过清单）、取化名、先写进 `.partial` 目录再改名，交出 `BuildSummary`。 +- `run_build(config_path, repo_root, now)`：完整做一次构建：读配置和映射表、读交付包、对每例做 C14（`check_key_image` 返回原因的整例跳过，记进跳过清单）；`ceus_run` 不为 null 时核对运行目录（不合规整次停下），对每例做 C15（CEUS 失败或没有结果的整例跳过）、关键帧也过 C14，按白名单构造 tic.json；取化名、先写进 `.partial` 目录再改名，交出 `BuildSummary`。 - `BuildSummary`：供终端打印的计数：构建号、两个库的路径、病例总数、进库数、跳过数、认不出编号的行数、按检查编号分的跳过数、复核清单条数、跳过清单说明句。 - `BuildConfig`：`build.yaml` 的字段定义（pydantic，字段必填、不许多写、读入后不可改）。 - `load_build_config(path, repo_root)`：读并校验 `build.yaml`，相对路径按仓库根目录解析。 diff --git a/research-wiki/guides/装配两个库.md b/research-wiki/guides/装配两个库.md index ed04add..116a3ed 100644 --- a/research-wiki/guides/装配两个库.md +++ b/research-wiki/guides/装配两个库.md @@ -6,7 +6,8 @@ - **四类胰腺囊性肿瘤**：IPMN（导管内乳头状黏液性肿瘤）、SPN（实性假乳头状肿瘤）、MCN（黏液性囊性肿瘤）、SCN（浆液性囊性肿瘤）。项目要专家判断病灶属于哪一类，所以资料库里出现这些名字就等于泄露答案。 - **金标准**：术后或穿刺的病理结论，交付表里叫「病理类型」或 `Diagnosis`，用 1–4 的数字记。 -- **CEUS、MR**：CEUS 是超声造影视频（往静脉打造影剂后录的超声）；MR 是磁共振检查，本项目只有它的文字报告。这一步不处理 CEUS 视频，只检查它在不在。 +- **CEUS、MR**：CEUS 是超声造影视频（往静脉打造影剂后录的超声）；MR 是磁共振检查，本项目只有它的文字报告。这一步不处理 CEUS 视频，只检查它在不在；CEUS 定量（`scripts/ceus_quant.py`，见 `../任务书/02-CEUS定量.md`）另外先跑，这一步按 `ceus_run` 读它的结果。 +- **TIC、关键帧、槽位**：TIC（时间-强度曲线）是病灶区域平均亮度随时间的变化；关键帧是按 TIC 挑出的 7 个时刻的画面，每个时刻叫一个槽位（B0、D1–D4、F1、F2，含义见 `../explanation/结构.md` 第四部分的关键帧表）。 - **交付包**：临床方交来的一批原始资料，目前是 `data/交付包_20260908_40例/`：三张汇总表（临床工作簿、超声报告、MR 报告）和每例一个病例目录（关键图、CEUS 视频、单例报告表）。 - **单例报告表**：每个病例目录里的一份 Excel，只有一行，列与超声汇总表相同（7 列）。它和汇总表重复，用来核对两份是否一致。 - **所见、结论段**：一份影像报告通常分「所见」（描述看到了什么）和「印象 / 结论」（医生的诊断倾向）。交付包的报告文字只有所见，这一步检查里面没有混进结论段（C11）。 @@ -32,6 +33,7 @@ | `alias_seed` | 整数 | 生成化名的随机种子。同一个种子、同一批通过的病例，得到同样的化名 | | `decode_codes` | true / false | 编码列是否按映射表解码 | | `code_maps` | 路径或 null | 映射表位置 | +| `ceus_run` | 路径或 null | 某次 CEUS 定量的运行目录（`runs/ceus/<运行号>/`）；null 表示不装 CEUS 结果，超声一路只有报告和关键图 | 相对路径**相对仓库根目录**解析，不相对配置文件所在目录，也不相对你敲命令时所在的目录。路径不能写空串（空串会变成当前目录）。 @@ -43,6 +45,10 @@ 为什么做成开关：映射表还没从临床方拿到，不能编造含义；先用 false 跑通，映射表到位后改成 true 重新构建一次（负责人 2026-10-16：解码做成可选开关，映射表是独立的配置文件）。诊断编码和其他编码列一样归这个开关管；诊断不进资料库，所以开关对诊断只影响答案库。 +### ceus_run + +指向一次**跑完的** CEUS 运行目录。目录不存在、名字以 `.partial` 结尾（那次运行没跑完）、缺 `failed.csv` 或 `snapshot/snapshot.json`，整次构建在建目录之前停下。目录合规时，按视频号找每例的结果，找不到或 CEUS 在这一例上失败的按 C15 跳过（第四节）。 + ### 映射表的格式 映射表目前**还没有**（`config/code_maps.yaml` 尚未建）。拿到后按下面的格式写：每个编码列一项，列名映射到「数字 → 含义」。 @@ -73,7 +79,7 @@ uv run python scripts/build.py --config config/build.yaml uv run python scripts/scan_library.py --library library/<构建号> --answers answers/<构建号> ``` -它从答案库读出这次构建的真实病例号和视频号，加上脚本里独立写的亚型名、诊断列名、交付文件名词表，扫资料库的全部 `.txt`、`.json`（不扫 PNG：PNG 的附加数据在写入时已经去掉，像素上烧录的信息栏按 `AGENTS.md` 硬性约束 3 不遮蔽，文字扫描也查不了像素），并检查 JSON 里有没有 `diagnosis`、`label`、`gold`、`pathology`、`name`、`row` 这类键名。只输出命中的条数和类别，不输出命中的内容；全部为 0 才说明资料库里没有这些东西。两个目录的构建号不同时直接报错。词表在脚本里另写一份、不取自 delivery 的常量，是为了让扫描不依赖被检查的代码本身：delivery 漏了一个词，扫描还能查出来。这个脚本同时读两个库，所以只能放在 `scripts/`，不能进 `src/cadx`（会诊代码不许碰答案库）。 +它从答案库读出这次构建的真实病例号和视频号，加上脚本里独立写的亚型名、诊断列名、交付文件名词表，扫资料库的全部 `.txt`、`.json`（含 `tic.json`；不扫 PNG：PNG 的附加数据在写入时已经去掉，像素上烧录的信息栏按 `AGENTS.md` 硬性约束 3 不遮蔽，文字扫描也查不了像素），并检查 JSON 里有没有 `diagnosis`、`label`、`gold`、`pathology`、`name`、`row` 这类键名。只输出命中的条数和类别，不输出命中的内容；全部为 0 才说明资料库里没有这些东西。两个目录的构建号不同时直接报错。词表在脚本里另写一份、不取自 delivery 的常量，是为了让扫描不依赖被检查的代码本身：delivery 漏了一个词，扫描还能查出来。这个脚本同时读两个库，所以只能放在 `scripts/`，不能进 `src/cadx`（会诊代码不许碰答案库）。 ## 三、产物在哪 @@ -89,18 +95,22 @@ cases/<化名>/ clinical/clinical.json ultrasound/report.txt ultrasound/key.png + ultrasound/keyframes/<槽位>.png （ceus_run 不为 null 时；只有定位到的槽位） + ultrasound/tic.json （ceus_run 不为 null 时） mr/report.txt ``` | 文件 | 内容 | | --- | --- | -| `manifest.json` | 构建号、git 提交号、工作区是否干净、配置文件（解码时含映射表）的 SHA-256、`codes_decoded`、`uncompared`（没比对的项）、`ceus_source`（所用 CEUS 结果的来源，这一步是 null）、化名列表（按化名排序）、每个文件的相对路径和 SHA-256。不含真实病例号、交付文件名和诊断 | +| `manifest.json` | 构建号、git 提交号、工作区是否干净、配置文件（解码时含映射表）的 SHA-256、`codes_decoded`、`uncompared`（没比对的项）、`ceus_source`（所用 CEUS 运行号，`ceus_run` 为 null 时是 null）、化名列表（按化名排序）、每个文件的相对路径和 SHA-256。不含真实病例号、交付文件名和诊断 | | `clinical.json` | `{"codes_decoded": ..., "fields": {列名: 值}}`。字段是临床表去掉编号和诊断后的 22 列：性别、年龄、5 个症状、5 项既往史、5 项化验的原值和档位。没测的化验写 null | | `ultrasound/report.txt` | 超声报告的「所见」原文，UTF-8，不加标题 | | `ultrasound/key.png` | 医生截的超声关键图，只保留像素和颜色模式重新编码，不带文本块、EXIF、ICC 等附加数据（附加数据里可能有文件名、时间等线索）。超声帧上烧录的信息栏不遮蔽（`AGENTS.md` 硬性约束 3） | +| `ultrasound/keyframes/<槽位>.png` | CEUS 运行里定位到的关键帧整幅画面，和 `key.png` 一样去掉附加数据重新编码 | +| `ultrasound/tic.json` | 四条 TIC 曲线的拟合是否成功和 R²、病灶两条曲线的灌注指标、紫圈曲线是否可用、7 个槽位是否定位到及视频时间或未定位原因。字段逐个从 CEUS 的 `summary.json` 挑出，不含视频号、路径、配置、帧号（`结构.md` 第四部分） | | `mr/report.txt` | MR 报告的「所见」原文 | -性别、年龄只在临床一路：两份报告表里医院也复制了这两列，但影像两路先看到会被人口学先验带偏（`结构.md` 第四部分）。以后 CEUS 定量落地，超声一路再加关键帧和灌注指标。 +性别、年龄只在临床一路：两份报告表里医院也复制了这两列，但影像两路先看到会被人口学先验带偏（`结构.md` 第四部分）。 读资料库要经 `cadx.library` 的 `read_clinical`、`read_ultrasound`、`read_mr`：读时逐个核对文件的 SHA-256 与清单一致，被改过就报错。 @@ -147,18 +157,18 @@ CSV 用带 BOM 的 UTF-8（文件开头有一个不可见的标记），医生 交付资料里的问题分两种（`AGENTS.md`「错误怎么处理」）： - **整张表的问题**（T1–T5）：一例也读不出来，整次构建停下，报错信息以检查编号开头，写明哪个文件哪一项不符。 -- **只落在某一个病例上的问题**（C01–C14）：跳过这一整例，接着处理下一例，在 `skipped.csv` 里记一行。这一例不进资料库，也不进答案库的 `aliases.csv`、`diagnoses.csv`。 +- **只落在某一个病例上的问题**（C01–C15）：跳过这一整例，接着处理下一例，在 `skipped.csv` 里记一行。这一例不进资料库，也不进答案库的 `aliases.csv`、`diagnoses.csv`。 -先对三张表的每一行和每个病例目录做 C01（编号认不认得出）；认得出的，再逐例按 C02 到 C13 的顺序检查，**遇到第一项没过就停下**，清单上每例只记这一项（C01 记的是认不出的那一行，不算某一例，见下文）。后面的检查没做过，不代表没问题：医生改好这一项后，重新构建时它可能卡在后面某一项。这句说明写在答案库 `manifest.json` 的 `skipped_note` 里，终端也会打印；CSV 文件开头不能加注释行，所以没写在 `skipped.csv` 里。为什么整例跳过而不是停下整次构建：负责人 2026-10-16「都是遇到整个病例跳过，让医生核对去吧」。评价时要报告跳过了几例、各因为什么。 +先对三张表的每一行和每个病例目录做 C01（编号认不认得出）；认得出的，再逐例按 C02 到 C13、C14（关键图）、C15（CEUS 结果）、C14（关键帧图）的顺序检查，**遇到第一项没过就停下**，清单上每例只记这一项（C01 记的是认不出的那一行，不算某一例，见下文）。后面的检查没做过，不代表没问题：医生改好这一项后，重新构建时它可能卡在后面某一项。这句说明写在答案库 `manifest.json` 的 `skipped_note` 里，终端也会打印；CSV 文件开头不能加注释行，所以没写在 `skipped.csv` 里。为什么整例跳过而不是停下整次构建：负责人 2026-10-16「都是遇到整个病例跳过，让医生核对去吧」。评价时要报告跳过了几例、各因为什么。 ### skipped.csv 的列 | 列 | 含义 | | --- | --- | | 病例 | 三位病例号。C01（编号认不出）的行不知道是哪个病人，写成「表的路径 第 N 行」（N 是 Excel 行号，第 1 行是表头）；病例目录名不在 1–999 之间时写成「病例目录 目录名」 | -| 检查编号 | C01–C14 | +| 检查编号 | C01–C15 | | 检查说明 | 这项检查的一句话说明 | -| 文件 | 涉及的文件，相对交付包根目录，多个用「; 」隔开 | +| 文件 | 涉及的文件，相对交付包根目录，多个用「; 」隔开；C15 和关键帧图的 C14 相对 CEUS 运行的上级目录（以运行号开头） | | 列 | 涉及的列，多个用「; 」隔开；不涉及具体列（如目录里的文件不对）时为空 | | 涉及值 | JSON 字符串：不合规的原值、对不上的各表的值、命中的词和所在句子等，供医生核对 | @@ -166,7 +176,7 @@ C01 的行不计入病例数，也不计入终端的「跳过病例」，单独 ### 检查编号速查 -规则细节以代码为准：T1–T4 在 `src/cadx/delivery/_tables.py`，T5 在 `src/cadx/build/_config.py`，C01–C10 在 `src/cadx/delivery/_cases.py`，C11–C13 的词表和正则在 `src/cadx/delivery/_text.py`，C14 由 build 调 library 的 `check_key_image`（`src/cadx/library/_write.py`）在写库前做，表名、工作表名、表头在 `src/cadx/delivery/_shape.py`。每条规则为什么这样定见任务书 `../任务书/01-装配两个库.md` 第二部分。 +规则细节以代码为准：T1–T4 在 `src/cadx/delivery/_tables.py`，T5 在 `src/cadx/build/_config.py`，C01–C10 在 `src/cadx/delivery/_cases.py`，C11–C13 的词表和正则在 `src/cadx/delivery/_text.py`，C14 由 build 调 library 的 `check_key_image`（`src/cadx/library/_write.py`）在写库前做，C15 在 `src/cadx/build/_ceus.py`，表名、工作表名、表头在 `src/cadx/delivery/_shape.py`。每条规则为什么这样定见任务书 `../任务书/01-装配两个库.md` 第二部分。 | 编号 | 查什么 | 不过时 | | --- | --- | --- | @@ -188,7 +198,8 @@ C01 的行不计入病例数，也不计入终端的「跳过病例」，单独 | C11 | 报告文字里没有结论段标题：段首（全文开头或换行之后）出现「印象」「结论」「提示」等关键词加冒号 | 跳过这一例 | | C12 | 报告文字里没有亚型名（缩写、中文名及其常见变体） | 跳过这一例 | | C13 | 报告文字里没有人口学信息：没有「数字岁」，没有「男」「女」两个字 | 跳过这一例 | -| C14 | 关键图能原样重编码：是 PNG，位深和颜色类型能逐像素保真（8 位灰度、RGB、灰度加透明通道、RGBA，或 1 位灰度），不带 tRNS 透明色；否则重编码会悄悄降位或丢透明。涉及值写不过的原因 | 跳过这一例 | +| C14 | 关键图能原样重编码：是 PNG，位深和颜色类型能逐像素保真（8 位灰度、RGB、灰度加透明通道、RGBA，或 1 位灰度），不带 tRNS 透明色；否则重编码会悄悄降位或丢透明。装 CEUS 结果时，每张关键帧图也查这一条（规则相同，「文件」一栏是哪张关键帧）。涉及值写不过的原因 | 跳过这一例 | +| C15 | CEUS 定量结果齐全（只在 `ceus_run` 不为 null 时查）：这一例的视频号不在那次运行的 `failed.csv` 里，且运行目录里有这一例的结果目录。涉及值写失败在第几步和原因，或「没有这一例的结果目录」。失败的病例以后要不要改成「超声不可评价、仍参与会诊」等 `../待定问题/03-病例排除标准.md` 定 | 跳过这一例 | 几点说明： diff --git a/src/cadx/build/_assemble.py b/src/cadx/build/_assemble.py index 6b39d86..22c3647 100644 --- a/src/cadx/build/_assemble.py +++ b/src/cadx/build/_assemble.py @@ -15,6 +15,7 @@ from pathlib import Path from typing import cast from cadx.answers import AliasRow, DiagnosisRow, ReviewRow, SkippedRow, write_answers +from cadx.build._ceus import CeusRun, case_ceus, open_ceus_run from cadx.build._config import load_build_config, load_code_maps from cadx.delivery import ( CODED_COLUMNS, @@ -25,7 +26,7 @@ from cadx.delivery import ( read_delivery, review_sentences, ) -from cadx.library import LibraryCase, check_key_image, write_library +from cadx.library import LibraryCase, LibraryCeus, check_key_image, write_library from cadx.runs import make_build_id, read_git_state, write_snapshot ALIAS_ALPHABET = string.ascii_lowercase + string.digits @@ -104,6 +105,35 @@ def _check_key_images( return kept, skipped +def _attach_ceus( + cases: Sequence[CaseRecord], run: CeusRun +) -> tuple[list[tuple[CaseRecord, LibraryCeus]], list[SkipRecord]]: + """C15：CEUS 失败或没有结果的病例整例跳过；关键帧图同 key.png 一样过 C14。""" + kept: list[tuple[CaseRecord, LibraryCeus]] = [] + skipped: list[SkipRecord] = [] + for case in cases: + ceus = case_ceus(run, case) + if isinstance(ceus, SkipRecord): + skipped.append(ceus) + continue + reasons = {slot: check_key_image(path) for slot, path in ceus.keyframe_sources.items()} + bad = next(((slot, r) for slot, r in reasons.items() if r is not None), None) + if bad is None: + kept.append((case, ceus)) + continue + skipped.append( + SkipRecord( + case=case.case_no, + check=C14, + description=C14_DESCRIPTION, + files=(f"{run.run_id}/{case.video_id}/keyframes/{bad[0]}.png",), + columns=(), + value_json=json.dumps(bad[1], ensure_ascii=False), + ) + ) + return kept, skipped + + def _sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() @@ -116,12 +146,22 @@ def _rename(partial: Path, final: Path) -> None: def run_build(config_path: Path, repo_root: Path, now: datetime) -> BuildSummary: - """读配置和交付包，装配一对同号的资料库与答案库。`now` 是本地时间，定构建号。""" + """读配置和交付包（`ceus_run` 不为 null 时还读那次 CEUS 运行），装配一对同号的资料库与答案库。 + + `now` 是本地时间，定构建号。 + """ config = load_build_config(config_path, repo_root) code_maps = load_code_maps(config.code_maps) if config.code_maps is not None else None delivery = read_delivery(config.delivery_dir, code_maps) passed, key_image_skips = _check_key_images(delivery.cases, config.delivery_dir) skipped = [*delivery.skipped, *key_image_skips] + ceus_run = None if config.ceus_run is None else open_ceus_run(config.ceus_run) + ceus_of: dict[str, LibraryCeus | None] = dict.fromkeys((c.case_no for c in passed), None) + if ceus_run is not None: + with_ceus, ceus_skips = _attach_ceus(passed, ceus_run) + passed = [case for case, _ in with_ceus] + ceus_of = {case.case_no: ceus for case, ceus in with_ceus} + skipped.extend(ceus_skips) git = read_git_state(repo_root) build_id = make_build_id(now, git) @@ -149,7 +189,7 @@ def run_build(config_path: Path, repo_root: Path, now: datetime) -> BuildSummary config_sha256={name: _sha256(path) for name, path in configs.items()}, codes_decoded=code_maps is not None, uncompared=delivery.uncompared, - ceus_source=None, + ceus_source=None if ceus_run is None else ceus_run.run_id, cases=[ LibraryCase( alias=aliases[case.case_no], @@ -157,6 +197,7 @@ def run_build(config_path: Path, repo_root: Path, now: datetime) -> BuildSummary ultrasound_report=case.ultrasound_report, key_image_source=case.key_image, mr_report=case.mr_report, + ceus=ceus_of[case.case_no], ) for case in by_alias ], diff --git a/src/cadx/build/_ceus.py b/src/cadx/build/_ceus.py new file mode 100644 index 0000000..e191d92 --- /dev/null +++ b/src/cadx/build/_ceus.py @@ -0,0 +1,151 @@ +"""读一次 CEUS 定量运行（`runs/ceus/<运行号>/`），按白名单挑出超声一路的 tic.json 和关键帧（C15）。 + +运行目录的布局由 `scripts/ceus_quant.py` 定：`failed.csv`、`snapshot/snapshot.json`、 +`<视频号>/summary.json`、`<视频号>/keyframes/<槽位>.png`。tic.json 的字段只在这里列， +视频号、路径、配置、帧号和拟合参数都不进去。 +""" + +import csv +import json +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from cadx.delivery import CaseRecord, SkipRecord +from cadx.library import LibraryCeus + +C15 = "C15" +C15_DESCRIPTION = "CEUS 定量结果齐全" + +_FAILED_HEADER = ["video_id", "step", "reason"] +# 四条曲线：病灶整体、病灶去暗区、紫圈（医生圈的正常胰腺对照）、背景圆。 +_CURVES = ("roi_all", "roi_perf", "roi_pan", "background_circle") +_LESION_CURVES = ("roi_all", "roi_perf") +_SLOTS = ("B0", "D1", "D2", "D3", "D4", "F1", "F2") +_PERFUSION_LABELS = ("signal_domain", "time_origin", "area_curve") +# 11 个灌注指标（同旧 replay summary 的 perfusion_metrics.values）。 +_PERFUSION_VALUES = ( + "PE_au", + "WiAUC_au_s", + "RT_s", + "mTTl_s", + "TTP_s", + "WiR_au_per_s", + "WiPI_au", + "WoAUC_au_s", + "WiWoAUC_au_s", + "FT_s", + "WoR_au_per_s", +) +_SUPPORTING_TIMES = ("TI", "mTT", "TO") +# 未定位原因由 ceus 按固定模板加数值生成，不含路径；出现这些字符说明模板变了，停下检查。 +_PATH_MARKS = ("/", "\\", ".mp4", ".png") + + +@dataclass(frozen=True) +class CeusRun: + """一次已完成的 CEUS 运行：运行号、目录、failed.csv 里的失败（视频号 → 步号、原因）。""" + + run_id: str + run_dir: Path + failed: dict[str, tuple[str, str]] + + +def open_ceus_run(run_dir: Path) -> CeusRun: + """核对运行目录（存在、不是 `.partial`、有 failed.csv 和快照），读 failed.csv。""" + if run_dir.name.endswith(".partial"): + raise ValueError(f"CEUS 运行目录是没跑完的 .partial：{run_dir}") + if not run_dir.is_dir(): + raise FileNotFoundError(f"CEUS 运行目录不存在：{run_dir}") + if not (run_dir / "snapshot" / "snapshot.json").is_file(): + raise FileNotFoundError(f"CEUS 运行目录缺快照 snapshot/snapshot.json：{run_dir}") + with (run_dir / "failed.csv").open(encoding="utf-8", newline="") as handle: + rows = list(csv.reader(handle)) + if not rows or rows[0] != _FAILED_HEADER: + raise ValueError(f"failed.csv 的表头应为 {_FAILED_HEADER}：{run_dir}") + return CeusRun(run_dir.name, run_dir, {row[0]: (row[1], row[2]) for row in rows[1:]}) + + +def _number(value: object) -> float | int | None: + if value is not None and (isinstance(value, bool) or not isinstance(value, int | float)): + raise ValueError(f"summary.json 里应为数值或 null 的字段是 {value!r}") + return value + + +def _typed[T](value: object, kind: type[T]) -> T: + if not isinstance(value, kind): + raise ValueError(f"summary.json 里应为 {kind.__name__} 的字段是 {value!r}") + return value + + +def _perfusion(section: dict[str, Any] | None) -> dict[str, Any] | None: + if section is None: + return None + return { + **{key: _typed(section[key], str) for key in _PERFUSION_LABELS}, + "values": {key: _number(section["values"][key]) for key in _PERFUSION_VALUES}, + "supporting_times_s": { + key: _number(section["supporting_times_s"][key]) for key in _SUPPORTING_TIMES + }, + } + + +def _keyframe(anchor: dict[str, Any], video_id: str) -> dict[str, Any]: + located = anchor["status"] == "ok" + time_s, reason = _number(anchor["frame_time_video_s"]), anchor["reason"] + if located != (time_s is not None) or located != (reason is None): + raise ValueError(f"summary.json 的关键帧状态、时间、原因互相矛盾：{anchor}") + if reason is not None and ( + not isinstance(reason, str) or video_id in reason or any(m in reason for m in _PATH_MARKS) + ): + raise ValueError(f"关键帧未定位原因含视频号或路径样的字符，不能给专家：{reason!r}") + return {"located": located, "video_time_s": time_s, "reason": reason} + + +def tic_view(summary: dict[str, Any], video_id: str) -> dict[str, Any]: + """从一例的 summary.json 按白名单逐字段构造 tic.json 的内容。""" + return { + "fits": { + name: { + "success": _typed(summary["fits"][name]["success"], bool), + "r2": _number(summary["fits"][name]["quality"]["r2"]), + } + for name in _CURVES + }, + "perfusion_metrics": { + name: _perfusion(summary["perfusion_metrics"][name]) for name in _LESION_CURVES + }, + "pan_curve_usable": _typed(summary["reference_valid"], bool), + "keyframes": {slot: _keyframe(summary["anchor_frames"][slot], video_id) for slot in _SLOTS}, + } + + +def case_ceus(run: CeusRun, case: CaseRecord) -> LibraryCeus | SkipRecord: + """一例的 CEUS 结果；在 failed.csv 里或没有结果目录的，交出 C15 跳过记录。""" + + def c15(files: str, value: str) -> SkipRecord: + return SkipRecord( + case=case.case_no, + check=C15, + description=C15_DESCRIPTION, + files=(f"{run.run_id}/{files}",), + columns=(), + value_json=json.dumps(value, ensure_ascii=False), + ) + + if case.video_id in run.failed: + step, reason = run.failed[case.video_id] + return c15("failed.csv", f"第 {step} 步失败：{reason}") + case_dir = run.run_dir / case.video_id + if not case_dir.is_dir(): + return c15(case.video_id, "CEUS 运行目录里没有这一例的结果目录") + summary = json.loads((case_dir / "summary.json").read_text(encoding="utf-8")) + if summary["video_id"] != case.video_id: + raise ValueError(f"summary.json 的视频号与所在目录不一致：{case_dir}") + tic = tic_view(summary, case.video_id) + located = [slot for slot in _SLOTS if tic["keyframes"][slot]["located"]] + images = sorted(p.stem for p in (case_dir / "keyframes").iterdir()) + if images != sorted(located): + raise ValueError(f"keyframes/ 里的图与 summary.json 定位到的槽位不一致：{case_dir}") + sources = {slot: case_dir / "keyframes" / f"{slot}.png" for slot in located} + return LibraryCeus(tic=tic, keyframe_sources=sources) diff --git a/src/cadx/build/_config.py b/src/cadx/build/_config.py index 8feab47..666aeee 100644 --- a/src/cadx/build/_config.py +++ b/src/cadx/build/_config.py @@ -54,6 +54,8 @@ class BuildConfig(BaseModel): alias_seed: int decode_codes: bool code_maps: Annotated[Path | None, Field(strict=False), BeforeValidator(_reject_empty_path)] + # 某次 CEUS 定量的运行目录（`runs/ceus/<运行号>/`）；null 表示这次构建不装 CEUS 结果。 + ceus_run: Annotated[Path | None, Field(strict=False), BeforeValidator(_reject_empty_path)] @model_validator(mode="after") def _code_maps_match_switch(self) -> Self: @@ -73,6 +75,7 @@ def load_build_config(path: Path, repo_root: Path) -> BuildConfig: "library_root": repo_root / config.library_root, "answers_root": repo_root / config.answers_root, "code_maps": None if config.code_maps is None else repo_root / config.code_maps, + "ceus_run": None if config.ceus_run is None else repo_root / config.ceus_run, } ) diff --git a/src/cadx/library/__init__.py b/src/cadx/library/__init__.py index c069344..9b85ea4 100644 --- a/src/cadx/library/__init__.py +++ b/src/cadx/library/__init__.py @@ -13,12 +13,13 @@ from cadx.library._read import ( read_ultrasound, ) from cadx.library._scan import LeakHit, scan_for_leaks -from cadx.library._write import LibraryCase, check_key_image, write_library +from cadx.library._write import LibraryCase, LibraryCeus, check_key_image, write_library __all__ = [ "ClinicalRoute", "LeakHit", "LibraryCase", + "LibraryCeus", "MrRoute", "UltrasoundRoute", "check_key_image", diff --git a/src/cadx/library/_read.py b/src/cadx/library/_read.py index f594a31..61cb490 100644 --- a/src/cadx/library/_read.py +++ b/src/cadx/library/_read.py @@ -8,8 +8,10 @@ from typing import Any, cast from cadx.library._write import ( CLINICAL_JSON, KEY_IMAGE, + KEYFRAMES_DIR, MANIFEST, MR_REPORT, + TIC_JSON, ULTRASOUND_REPORT, case_file, sha256_file, @@ -26,10 +28,13 @@ class ClinicalRoute: @dataclass(frozen=True) class UltrasoundRoute: - """超声一路：报告「所见」原文和医生截的关键图（给路径）。""" + """超声一路：报告「所见」原文、医生截的关键图（给路径）；装了 CEUS 结果时还有 tic.json 的内容和 + 定位到的关键帧（槽位 → 路径），没装时 `tic` 为 None、`keyframes` 为空。""" report: str key_image: Path + tic: dict[str, Any] | None + keyframes: dict[str, Path] @dataclass(frozen=True) @@ -75,7 +80,18 @@ def read_clinical(build_dir: Path, alias: str) -> ClinicalRoute: def read_ultrasound(build_dir: Path, alias: str) -> UltrasoundRoute: """读一例的超声一路。""" report = _text(_verified(build_dir, alias, ULTRASOUND_REPORT)) - return UltrasoundRoute(report=report, key_image=_verified(build_dir, alias, KEY_IMAGE)) + key_image = _verified(build_dir, alias, KEY_IMAGE) + manifest = _manifest(build_dir) + if manifest["ceus_source"] is None: + return UltrasoundRoute(report=report, key_image=key_image, tic=None, keyframes={}) + prefix = case_file(alias, f"{KEYFRAMES_DIR}/") + keyframes = { + Path(key).stem: _verified(build_dir, alias, key.removeprefix(f"cases/{alias}/")) + for key in manifest["files"] + if key.startswith(prefix) + } + tic = json.loads(_text(_verified(build_dir, alias, TIC_JSON))) + return UltrasoundRoute(report=report, key_image=key_image, tic=tic, keyframes=keyframes) def read_mr(build_dir: Path, alias: str) -> MrRoute: diff --git a/src/cadx/library/_write.py b/src/cadx/library/_write.py index f6cc093..7a3524f 100644 --- a/src/cadx/library/_write.py +++ b/src/cadx/library/_write.py @@ -6,6 +6,8 @@ cases/<化名>/clinical/clinical.json cases/<化名>/ultrasound/report.txt cases/<化名>/ultrasound/key.png + cases/<化名>/ultrasound/keyframes/<槽位>.png （装了 CEUS 结果时；只有定位到的槽位） + cases/<化名>/ultrasound/tic.json （装了 CEUS 结果时） cases/<化名>/mr/report.txt ``` """ @@ -22,6 +24,8 @@ MANIFEST = "manifest.json" CLINICAL_JSON = "clinical/clinical.json" ULTRASOUND_REPORT = "ultrasound/report.txt" KEY_IMAGE = "ultrasound/key.png" +KEYFRAMES_DIR = "ultrasound/keyframes" +TIC_JSON = "ultrasound/tic.json" MR_REPORT = "mr/report.txt" _PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n" @@ -30,15 +34,24 @@ _PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n" _FAITHFUL = frozenset({(8, 0), (8, 2), (8, 4), (8, 6), (1, 0)}) +@dataclass(frozen=True) +class LibraryCeus: + """超声一路的 CEUS 部分：`tic` 由 build 按白名单挑好 写成 tic.json；槽位 → 关键帧源图。""" + + tic: Mapping[str, object] + keyframe_sources: Mapping[str, Path] + + @dataclass(frozen=True) class LibraryCase: - """要写进资料库的一例：三路各自的内容。""" + """要写进资料库的一例：三路各自的内容。`ceus` 为 None 表示这次构建不装 CEUS 结果。""" alias: str clinical_fields: Mapping[str, str | int | float | None] ultrasound_report: str key_image_source: Path mr_report: str + ceus: LibraryCeus | None def case_file(alias: str, relative: str) -> str: @@ -98,9 +111,14 @@ def write_library( ceus_source: str | None, cases: Sequence[LibraryCase], ) -> None: - """把一次构建写进已经存在的空目录 `build_dir`，最后写清单。""" + """把一次构建写进已经存在的空目录 `build_dir`，最后写清单。 + + `ceus_source` 是所用 CEUS 运行号，为 None 时每例的 `ceus` 都必须是 None，反之都不能是 None。 + """ files: dict[str, str] = {} for case in cases: + if (case.ceus is None) != (ceus_source is None): + raise ValueError(f"化名 {case.alias} 的 CEUS 结果与 ceus_source={ceus_source!r} 不配套") case_dir = build_dir / "cases" / case.alias for relative in (CLINICAL_JSON, ULTRASOUND_REPORT, MR_REPORT): (case_dir / relative).parent.mkdir(parents=True, exist_ok=True) @@ -111,7 +129,15 @@ def write_library( (case_dir / ULTRASOUND_REPORT).write_text(case.ultrasound_report, encoding="utf-8") _write_clean_png(case.key_image_source, case_dir / KEY_IMAGE) (case_dir / MR_REPORT).write_text(case.mr_report, encoding="utf-8") - for relative in (CLINICAL_JSON, ULTRASOUND_REPORT, KEY_IMAGE, MR_REPORT): + written = [CLINICAL_JSON, ULTRASOUND_REPORT, KEY_IMAGE, MR_REPORT] + if case.ceus is not None: + (case_dir / KEYFRAMES_DIR).mkdir() + for slot, source in case.ceus.keyframe_sources.items(): + _write_clean_png(source, case_dir / KEYFRAMES_DIR / f"{slot}.png") + written.append(f"{KEYFRAMES_DIR}/{slot}.png") + _write_json(case_dir / TIC_JSON, dict(case.ceus.tic)) + written.append(TIC_JSON) + for relative in written: files[case_file(case.alias, relative)] = sha256_file(case_dir / relative) _write_json( build_dir / MANIFEST, diff --git a/tests/fake_ceus_run.py b/tests/fake_ceus_run.py new file mode 100644 index 0000000..c4d9941 --- /dev/null +++ b/tests/fake_ceus_run.py @@ -0,0 +1,118 @@ +"""测试用假 CEUS 运行目录：布局照 `scripts/ceus_quant.py` 写出的（failed.csv、snapshot/、 +`<视频号>/summary.json`、`<视频号>/keyframes/<槽位>.png`），数值全部编造，不调显卡。 + +summary.json 故意带上 tic.json 不该有的字段（视频号、分割缓存键、帧号、拟合参数、失败原因代码、 +推注时间、紫圈检出率），用来测 build 只按白名单挑字段。关键帧 PNG 带文本块和 EXIF。 +""" + +import csv +import json +from pathlib import Path +from typing import Any + +from PIL import Image, PngImagePlugin + +SLOTS = ("B0", "D1", "D2", "D3", "D4", "F1", "F2") +CURVES = ("roi_all", "roi_perf", "roi_pan", "background_circle") +UNLOCATED_REASON = "拟合曲线上没有 D4 时刻（编造的原因）" + + +def _perfusion(scale: float) -> dict[str, Any]: + names = ("PE_au", "WiAUC_au_s", "RT_s", "mTTl_s", "TTP_s", "WiR_au_per_s", "WiPI_au") + more = ("WoAUC_au_s", "WiWoAUC_au_s", "FT_s", "WoR_au_per_s") + return { + "signal_domain": "db_display", + "time_origin": "video_start", + "area_curve": "fitted_minus_baseline", + "values": {name: scale * (i + 1) for i, name in enumerate((*names, *more))}, + "supporting_times_s": {"TI": 3.0, "mTT": 9.5, "TO": None}, + } + + +def _fit(success: bool, r2: float | None) -> dict[str, Any]: + return { + "success": success, + "failure_reason": None if success else "credible_peak_not_found", + "curve_model": "lognormal", + "baseline_source": "observed", + "valid_samples": 120, + "duplicate_samples": 3, + "parameters": { + "area": 1.0, + "mu": 2.0, + "sigma": 0.5, + "arrival_offset_s": 4.0, + "baseline": 9, + }, + "quality": {"r2": r2, "nrmse": 0.1, "residual_noise_ratio": 1.2, "multi_peak_flag": False}, + "model_arrival_time_video_s": 4.0, + "model_landmark_times_video_s": {"D2": 6.0, "D3": 9.0, "D4": 15.0}, + "model_landmark_intensities": {"D2": 1.0, "D3": 2.0, "D4": 1.5}, + } + + +def summary(video_id: str, located: tuple[str, ...]) -> dict[str, Any]: + """一例的 summary.json：`located` 里的槽位定位到（视频时间=槽位序号×2.5 秒），余下未定位。""" + anchors = { + slot: { + "frame_index": index * 25 if slot in located else None, + "frame_time_video_s": index * 2.5 if slot in located else None, + "time_since_injection_s": index * 2.5 + 5.0 if slot in located else None, + "status": "ok" if slot in located else "anchor_undefined", + "reason": None if slot in located else UNLOCATED_REASON, + "anchor_time_video_s": index * 2.5, + "time_error_s": 0.0 if slot in located else None, + "source_curve": "roi_perf", + } + for index, slot in enumerate(SLOTS) + } + return { + "video_id": video_id, + "segment_cache_key": "f" * 64, + "injection_offset_s": 5.0, + "reference_valid": False, + "pan_detected_fraction": 0.4, + "early_truncation_flag": False, + "lesion_fit_ready": True, + "fits": { + "roi_all": _fit(True, 0.93), + "roi_perf": _fit(True, 0.91), + "roi_pan": _fit(False, None), + "background_circle": _fit(False, 0.2), + }, + "perfusion_metrics": {"roi_all": _perfusion(1.0), "roi_perf": None}, + "anchor_frames": anchors, + "available_dynamic_slots": [s for s in ("B0", "D1", "D2", "D3", "D4") if s in located], + } + + +def keyframe_png(path: Path, shade: int) -> None: + """一张 8 位 RGB 关键帧，带文本块和 EXIF（写进资料库前要去掉）。""" + pixels = bytes(v for i in range(12) for v in (shade, i * 20, 200 - i * 10)) + image = Image.frombytes("RGB", (4, 3), pixels) + info = PngImagePlugin.PngInfo() + info.add_text("Source", "U999_roi2.mp4") + exif = Image.Exif() + exif[0x010E] = "U999" # ImageDescription + image.save(path, format="PNG", pnginfo=info, exif=exif.tobytes()) + + +def write_run( + run_dir: Path, + results: dict[str, tuple[str, ...]], + failed: list[tuple[str, int, str]], +) -> Path: + """写一个已完成的运行目录：`results` 是视频号 → 定位到的槽位，`failed` 是 failed.csv 的行。""" + (run_dir / "snapshot").mkdir(parents=True) + (run_dir / "snapshot" / "snapshot.json").write_text("{}", encoding="utf-8") + with (run_dir / "failed.csv").open("w", encoding="utf-8", newline="") as handle: + writer = csv.writer(handle) + writer.writerow(("video_id", "step", "reason")) + writer.writerows(failed) + for video_id, located in results.items(): + (run_dir / video_id / "keyframes").mkdir(parents=True) + content = json.dumps(summary(video_id, located), ensure_ascii=False, indent=2) + (run_dir / video_id / "summary.json").write_text(content, encoding="utf-8") + for shade, slot in enumerate(located): + keyframe_png(run_dir / video_id / "keyframes" / f"{slot}.png", shade * 30) + return run_dir diff --git a/tests/fake_delivery.py b/tests/fake_delivery.py index 9a33d99..6afbee0 100644 --- a/tests/fake_delivery.py +++ b/tests/fake_delivery.py @@ -235,7 +235,11 @@ def _write_png(path: Path, number: int) -> None: def write_config( - tmp_path: Path, delivery_dir: Path, seed: int, code_maps: dict[str, dict[int, str]] | None + tmp_path: Path, + delivery_dir: Path, + seed: int, + code_maps: dict[str, dict[int, str]] | None, + ceus_run: Path | None, ) -> Path: """在 tmp_path 写一份 build.yaml（路径都写绝对路径）；给了映射表就同时写 code_maps.yaml。""" maps_path = None @@ -249,6 +253,7 @@ def write_config( "alias_seed": seed, "decode_codes": code_maps is not None, "code_maps": None if maps_path is None else str(maps_path), + "ceus_run": None if ceus_run is None else str(ceus_run), } path = tmp_path / "build.yaml" path.write_text(yaml.safe_dump(config, allow_unicode=True), encoding="utf-8") @@ -262,6 +267,6 @@ def build( code_maps: dict[str, dict[int, str]] | None, now: datetime, ) -> BuildSummary: - """写出假交付包和配置，跑一次构建。""" + """写出假交付包和配置，跑一次不装 CEUS 结果的构建。""" delivery_dir = delivery.write(tmp_path / "delivery") - return run_build(write_config(tmp_path, delivery_dir, seed, code_maps), REPO_ROOT, now) + return run_build(write_config(tmp_path, delivery_dir, seed, code_maps, None), REPO_ROOT, now) diff --git a/tests/test_build_ceus.py b/tests/test_build_ceus.py new file mode 100644 index 0000000..d016f99 --- /dev/null +++ b/tests/test_build_ceus.py @@ -0,0 +1,220 @@ +"""build 装入 CEUS 结果：C15 跳过、tic.json 只有白名单字段、关键帧去附加数据、运行目录不合规时停下。 + +预期的 tic.json 键集合照任务书 02 第四阶段的文字手写：四条曲线各自拟合是否成功和拟合优度（R²）； +病灶两条曲线的灌注指标（同旧 replay summary 的 perfusion_metrics 段）；紫圈曲线是否可用； +7 个槽位各自是否定位到、定位到的视频时间、未定位的原因。 +""" + +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 UNLOCATED_REASON, 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") + +_PERFUSION_KEYS = { + "signal_domain", + "time_origin", + "area_curve", + "values", + "supporting_times_s", +} +_PERFUSION_VALUES = { + "PE_au", + "WiAUC_au_s", + "RT_s", + "mTTl_s", + "TTP_s", + "WiR_au_per_s", + "WiPI_au", + "WoAUC_au_s", + "WiWoAUC_au_s", + "FT_s", + "WoR_au_per_s", +} + + +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 _key_tree(node: Any) -> Any: + """嵌套字典的键结构：字典换成「键 → 子结构」，其余值换成 None。""" + if isinstance(node, dict): + return {key: _key_tree(value) for key, value in cast(dict[str, Any], node).items()} + return None + + +def test_c15_skips_failed_and_missing_cases(tmp_path: Path) -> None: + run = write_run( + tmp_path / "runs" / RUN_ID, + {"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 + + +def test_tic_json_has_exactly_the_listed_fields(tmp_path: Path) -> None: + result = _build(tmp_path, 1, write_run(tmp_path / "runs" / RUN_ID, {"U001": LOCATED}, [])) + (alias,) = list_aliases(result.library_dir) + tic = read_ultrasound(result.library_dir, alias).tic + perfusion = { + **dict.fromkeys(_PERFUSION_KEYS - {"values", "supporting_times_s"}), + "values": dict.fromkeys(_PERFUSION_VALUES), + "supporting_times_s": dict.fromkeys(("TI", "mTT", "TO")), + } + slot = {"located": None, "video_time_s": None, "reason": None} + curves = ("roi_all", "roi_perf", "roi_pan", "background_circle") + assert _key_tree(tic) == { + "fits": {name: {"success": None, "r2": None} for name in curves}, + "perfusion_metrics": {"roi_all": perfusion, "roi_perf": None}, + "pan_curve_usable": None, + "keyframes": {s: slot for s in ("B0", "D1", "D2", "D3", "D4", "F1", "F2")}, + } + assert tic is not None + assert tic["fits"]["roi_all"] == {"success": True, "r2": 0.93} + assert tic["fits"]["roi_pan"] == {"success": False, "r2": None} + assert tic["pan_curve_usable"] is False + assert tic["perfusion_metrics"]["roi_all"]["values"]["PE_au"] == 1.0 + # 假数据里 D1 是第 2 个槽位：视频时间 1 × 2.5 秒。 + assert tic["keyframes"]["D1"] == {"located": True, "video_time_s": 2.5, "reason": None} + assert tic["keyframes"]["D4"] == { + "located": False, + "video_time_s": None, + "reason": UNLOCATED_REASON, + } + + +def test_keyframes_metadata_dropped_and_pixels_kept(tmp_path: Path) -> None: + run = write_run(tmp_path / "runs" / RUN_ID, {"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(LOCATED) + for slot, copied in route.keyframes.items(): + 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_no_ceus_files(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) + assert route.tic is None and route.keyframes == {} + assert sorted( + p.name for p in (result.library_dir / "cases" / alias / "ultrasound").iterdir() + ) == [ + "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, {"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["keyframes"]["D4"]["reason"] = "见 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_reason_with_video_id_stops_build(tmp_path: Path) -> None: + run = write_run(tmp_path / "runs" / RUN_ID, {"U001": LOCATED}, []) + content = summary("U001", LOCATED) + content["anchor_frames"]["D4"]["reason"] = "U001 的 D4 不在视频里" + (run / "U001" / "summary.json").write_text(json.dumps(content), encoding="utf-8") + with pytest.raises(ValueError, match="视频号或路径"): + _build(tmp_path, 1, run) + assert not (tmp_path / "library").exists() + + +def test_keyframe_that_cannot_be_reencoded_skips_case_as_c14(tmp_path: Path) -> None: + run = write_run(tmp_path / "runs" / RUN_ID, {"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_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", {"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, {"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, {"U001": LOCATED}, []) + (run / "snapshot" / "snapshot.json").unlink() + _assert_stops(tmp_path, run, FileNotFoundError) diff --git a/tests/test_build_outputs.py b/tests/test_build_outputs.py index 2eb50bb..52c18cc 100644 --- a/tests/test_build_outputs.py +++ b/tests/test_build_outputs.py @@ -217,7 +217,7 @@ def _png_chunks(data: bytes) -> list[str]: def _build_with_key_image(tmp_path: Path, image: Image.Image) -> None: root = make_delivery(2).write(tmp_path / "delivery") image.save(root / "CEUS_agent" / "002" / "U002_key.png", format="PNG") - run_build(write_config(tmp_path, root, 1, None), REPO_ROOT, NOW) + run_build(write_config(tmp_path, root, 1, None, None), REPO_ROOT, NOW) def _assert_key_image_skipped(tmp_path: Path, reason: str) -> None: @@ -254,7 +254,7 @@ def test_key_image_palette_source_skips_case(tmp_path: Path) -> None: def test_key_image_not_png_skips_case(tmp_path: Path) -> None: root = make_delivery(2).write(tmp_path / "delivery") Image.new("RGB", (4, 4)).save(root / "CEUS_agent" / "002" / "U002_key.png", format="JPEG") - run_build(write_config(tmp_path, root, 1, None), REPO_ROOT, NOW) + run_build(write_config(tmp_path, root, 1, None, None), REPO_ROOT, NOW) _assert_key_image_skipped(tmp_path, "不是 PNG") @@ -270,6 +270,7 @@ def test_write_library_rejects_unchecked_key_image(tmp_path: Path) -> None: ultrasound_report="", key_image_source=source, mr_report="", + ceus=None, ) with pytest.raises(ValueError, match="check_key_image"): write_library(target, "b", "c", True, {}, False, (), None, [case]) diff --git a/tests/test_build_stops.py b/tests/test_build_stops.py index 4eea030..db35523 100644 --- a/tests/test_build_stops.py +++ b/tests/test_build_stops.py @@ -28,7 +28,7 @@ def _assert_no_finished_build(tmp_path: Path) -> None: def _run(tmp_path: Path, root: Path, code_maps: dict[str, dict[int, str]] | None) -> None: - run_build(write_config(tmp_path, root, 7, code_maps), REPO_ROOT, NOW) + run_build(write_config(tmp_path, root, 7, code_maps, None), REPO_ROOT, NOW) def test_t1_missing_summary_table(tmp_path: Path) -> None: @@ -111,7 +111,7 @@ def _write_yaml(tmp_path: Path, body: str) -> Path: return path -_BASE = "delivery_dir: d\nlibrary_root: l\nanswers_root: a\nalias_seed: 1\n" +_BASE = "delivery_dir: d\nlibrary_root: l\nanswers_root: a\nalias_seed: 1\nceus_run: null\n" def test_config_decode_true_needs_code_maps(tmp_path: Path) -> None: @@ -152,12 +152,24 @@ def test_config_relative_paths_resolve_against_repo_root(tmp_path: Path) -> None assert config.library_root == Path("/repo/l") assert config.answers_root == Path("/repo/a") assert config.code_maps is None + assert config.ceus_run is None + body = _BASE.replace("ceus_run: null", "ceus_run: runs/ceus/r") + path = _write_yaml(tmp_path, body + "decode_codes: false\ncode_maps: null\n") + assert load_build_config(path, Path("/repo")).ceus_run == Path("/repo/runs/ceus/r") + + +def test_config_ceus_run_missing(tmp_path: Path) -> None: + body = _BASE.replace("ceus_run: null\n", "") + path = _write_yaml(tmp_path, body + "decode_codes: false\ncode_maps: null\n") + with pytest.raises(ValidationError, match="ceus_run"): + load_build_config(path, tmp_path) def test_committed_config_is_valid() -> None: config = load_build_config(REPO_ROOT / "config" / "build.yaml", REPO_ROOT) assert config.decode_codes is False assert config.code_maps is None + assert config.ceus_run is None def test_config_duplicate_key(tmp_path: Path) -> None: diff --git a/tests/test_ceus_quant.py b/tests/test_ceus_quant.py index bdbc169..b415976 100644 --- a/tests/test_ceus_quant.py +++ b/tests/test_ceus_quant.py @@ -13,6 +13,7 @@ import csv import importlib.util import json from collections.abc import Callable, Mapping, Sequence +from datetime import datetime from pathlib import Path from types import ModuleType from typing import Any @@ -23,9 +24,12 @@ import numpy.typing as npt import pytest import yaml +from cadx.answers import read_aliases +from cadx.build import run_build from cadx.ceus import ANCHOR_SLOTS, LognormalParameters, evaluate_lognormal +from cadx.library import list_aliases, read_ultrasound from conftest import CEUS_YAML -from fake_delivery import make_delivery +from fake_delivery import REPO_ROOT, make_delivery, write_config _SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "ceus_quant.py" _WIDTH, _HEIGHT, _SPLIT = 400, 200, 200 @@ -177,6 +181,36 @@ def test_run_writes_outputs_and_failed_case( assert "U001" not in out and "U002" not in out +def test_build_reads_real_run_layout( + tmp_path: Path, + setup: tuple[ModuleType, _CountingSegment, dict[str, Any], Callable[[str], None]], +) -> None: + """build 读这个脚本真实写出的运行目录：第 1 例带 tic.json 和关键帧进库，第 2 例记 C15。""" + module, _, document, next_run_id = setup + next_run_id("run-a") + module.main(["--config", str(_write_config(tmp_path, document, "ceus.yaml"))]) + run_dir = tmp_path / "runs" / "run-a" + delivery_dir = Path(document["paths"]["delivery_dir"]) + build_dir = tmp_path / "build" + build_dir.mkdir() + result = run_build( + write_config(build_dir, delivery_dir, 1, None, run_dir), + REPO_ROOT, + datetime(2026, 10, 16, 12, 0, 0).astimezone(), + ) + assert [r.case_no for r in read_aliases(result.answers_dir)] == ["001"] + assert result.skipped_by_check == {"C15": 1} + (alias,) = list_aliases(result.library_dir) + route = read_ultrasound(result.library_dir, alias) + summary = json.loads((run_dir / "U001" / "summary.json").read_text(encoding="utf-8")) + located = {s for s, a in summary["anchor_frames"].items() if a["status"] == "ok"} + assert set(route.keyframes) == located + assert route.tic is not None + assert set(route.tic) == {"fits", "perfusion_metrics", "pan_curve_usable", "keyframes"} + assert route.tic["keyframes"]["F1"]["video_time_s"] == pytest.approx(15.0, abs=0.1) + assert route.tic["keyframes"]["F2"]["located"] is False + + def test_unreadable_video_fails_at_step_1_and_next_case_runs( tmp_path: Path, monkeypatch: pytest.MonkeyPatch,