diff --git a/config/ceus.yaml b/config/ceus.yaml index c995b47..791840e 100644 --- a/config/ceus.yaml +++ b/config/ceus.yaml @@ -28,6 +28,10 @@ segment: # 分割设置，字段定义在 src/cadx/segment/_config.py。出自 ceus: config_version: "4" # 参数口径版本号，由人改。出自旧项目：Lognormal-v2 参数口径（design/0044） + # cv2 进程级随机数的种子，每例第 4–9 步开头设一次（第 5 步暗区的 kmeans 候选用 k-means++ 随机选初始 + # 中心）。不设时同一例的结果取决于同一进程里前面跑过几次 kmeans。本仓库为可复现新加（AGENTS.md + # 「可复现」），旧项目未设种子 + random_seed: 0 views: # 出自旧项目：findings §3.1 视图切分 layout: horizontal_split # 屏幕左右各一半。出自旧项目 findings §3.1 diff --git a/research-wiki/explanation/结构.md b/research-wiki/explanation/结构.md index 4bb098c..7a560de 100644 --- a/research-wiki/explanation/结构.md +++ b/research-wiki/explanation/结构.md @@ -293,8 +293,8 @@ Python 环境用 uv；安装与检查命令见 `../guides/开发环境.md`。 | 模块 | 预期行数 | 实际行数 | 旧项目对应部分的行数 | | --- | --- | --- | --- | | ceus | 6000–7500（基本照搬旧算法，去掉只为产品层服务的部分） | — | `ceus_roi/` 除 `service.py` 7330，`video_decoder.py` 135 | -| segment | 150–300 | 220 | `medsam2_local_track.py` 240 | -| delivery | 400–700 | 733 | `clinical.py` 266、`modalities.py` 522、`workbook.py` 99、`delivery.py` 134，共 1021，其中入库与数据库部分不迁 | +| segment | 150–300 | 233 | `medsam2_local_track.py` 240 | +| delivery | 400–700 | 739 | `clinical.py` 266、`modalities.py` 522、`workbook.py` 99、`delivery.py` 134，共 1021，其中入库与数据库部分不迁 | | library | 150–300 | 300 | `input_packages.py` 104 | | answers | 100–200 | 201 | — | | verdicts | 100–200 | — | — | @@ -304,7 +304,7 @@ Python 环境用 uv；安装与检查命令见 `../guides/开发环境.md`。 | runs | 100–250 | 86（只含快照部分） | — | | consult | 500–1000（会诊口径定后重估） | — | — | | evaluate | 300–600（评价口径定后重估） | — | 已删的 `core/eval` 622 | -| scripts/ 每个 | ≤ 300 | `build.py` 37、`scan_library.py` 88 | `t1_local_debug.py` 762、`tic_lognormal_replay.py` 627 | +| scripts/ 每个 | ≤ 300 | `build.py` 37、`scan_library.py` 88、`ceus_quant.py` 293 | `t1_local_debug.py` 762、`tic_lognormal_replay.py` 627 | --- @@ -328,7 +328,7 @@ Python 环境用 uv；安装与检查命令见 `../guides/开发环境.md`。 - `read_delivery(delivery_dir, code_maps)`：读交付包，做整张表的检查 T1–T4 和逐例检查 C01–C13，交出 `DeliveryResult`；`code_maps` 为 None 表示不解码。 - `DeliveryResult`：一次读取的结果：通过的病例（按病例号排序）、跳过记录、病例总数、认不出编号的行数、因没有映射表而没比对的项。 -- `CaseRecord`：一个检查全部通过的病例：三位病例号、视频号、临床一路的 22 列、超声描述、MR 描述、关键图路径、三张表的诊断编码。 +- `CaseRecord`：一个检查全部通过的病例：三位病例号、视频号、临床一路的 22 列、超声描述、MR 描述、关键图路径、CEUS 视频路径（`video`，C04 已检查恰好一段）、推注时间 X（`bolus_start_s`，超声表「推注开始时间s」，C08 已检查是非负整数）、三张表的诊断编码。 - `DiagnosisCodes`：一例在临床表、超声表、MR 表里各自的诊断编码（C10 保证三者相等）。 - `SkipRecord`：一条跳过记录：病例号（认不出编号时是「文件 第 N 行」或「病例目录 名字」）、检查编号、检查说明、涉及的文件和列、涉及值的 JSON 字符串。 - `DeliveryError`：整张表读不了（T1–T4）时抛出，消息以检查编号开头，整次构建停下。 @@ -370,14 +370,14 @@ Python 环境用 uv；安装与检查命令见 `../guides/开发环境.md`。 对外两段接口，中间由 `scripts/ceus_quant.py` 调 segment 做第 3 步： -- `prepare_segmentation(video_path, cfg)`：第 1–2 步，交出 `SegmentationInput`；找不到可信分屏线抛 `SplitDetectionError`（第 1 步），两半都有或都没有绿框抛 `ContrastSideError`（第 2 步）。 +- `prepare_segmentation(video_path, cfg)`：第 1–2 步，交出 `SegmentationInput`；视频打不开或读出的不合规（fps、尺寸、时间戳、帧数）抛 `VideoDecodeError`、找不到可信分屏线抛 `SplitDetectionError`（都是第 1 步），两半都有或都没有绿框抛 `ContrastSideError`（第 2 步）。 - `SegmentationInput`：视频的 fps、帧数、宽高、逐帧时间戳，分屏线，`contrast`（`ContrastSide`：哪一半是造影画面、造影视图和分割那一半视图 `target_view`、提示帧），`target_frames`（分割那一半的逐帧 BGR 图像），`prompt_masks`（提示帧帧号 → 绿框挪到 `target_view` 局部坐标的掩膜），造影那一半的第一帧。 - `quantify_case(prepared, target_masks, injection_offset_s, cfg)`：第 4–9 步。`target_masks` 是分割在 `target_view` 局部坐标上的 `(帧数, 高, 宽)` 布尔数组，`injection_offset_s` 是推注时间 X；分割一帧有效轮廓都没有抛 `NoValidContourError`（第 4 步）。拟合没过质量门不抛错。 - `QuantResult`：`trajectory`（`Trajectory`）、`tic`（`TicResult`）、`curves`（`TicCurves`）、推注时间、病灶两条曲线拟合是否都过质量门、`anchors`（7 个槽位的 `AnchorFrameLocation`）、`perfusion`（`roi_all`、`roi_perf` 各自的 `LognormalPerfusionMetrics`，拟合没过为 None）、紫圈检出率。 - `AnchorFrameLocation`：一个关键帧槽位的帧号、视频时间、推注后时刻、状态（`ok` 或未定位的类别）、未定位的原因、锚点时刻。 - 写产物用：`pack_trajectory(trajectory)`（trajectory.npz 字节）、`pack_tic_curves(curves)`（tic_curves.npz 字节）、`render_tic_png(tic, markers=tic_plot_markers(anchors))`（tic.png，标出 D1–D4）、`render_case_overlay(prepared, result)`（overlay.mp4 字节）、`keyframe_images(prepared, anchors)`（定位到的关键帧整幅原图，槽位 → 图像）、`quant_summary(result)`（拟合、灌注指标、紫圈检出率、7 个关键帧定位，可直接写成 JSON）；读回用 `unpack_trajectory`、`unpack_tic_curves`。 - `CeusConfig` 及各段的配置类：`config/ceus.yaml` 里 `ceus` 段的字段定义（pydantic，字段必填、不许多写、读入后不可改）；`paths` 段由 `scripts/ceus_quant.py` 定义。 -- `CeusCaseError`：只落在这一例上的失败，带失败的步号和原因；`SplitDetectionError`、`ContrastSideError`、`NoValidContourError` 是它的子类。脚本据此把这一例记进 `failed.csv`、接着跑下一例。 +- `CeusCaseError`：只落在这一例上的失败，带失败的步号和原因；`VideoDecodeError`、`SplitDetectionError`、`ContrastSideError`、`NoValidContourError` 是它的子类。脚本据此把这一例记进 `failed.csv`、接着跑下一例。 - `VideoFile(path)`：按路径用 cv2 读视频：`probe()` 给 fps、帧数、宽高；`iter_frames()` 每次从第 0 帧逐帧读；`frame_times_s()` 给逐帧显示时间戳。 - 其余导出（几何函数、各步的算法函数与结果类型，如 `split_views`、`locate_contrast_side`、`masks_to_smoothed_polygons`、`extract_dark_region_polygons`、`select_reference_region`、`compute_tic_for_video`、`fit_lognormal_curve`、`locate_anchor_frames`、`compute_lognormal_perfusion_metrics`、`render_overlay`）是两段接口的组成部分，供测试用。 @@ -386,6 +386,7 @@ Python 环境用 uv；安装与检查命令见 `../guides/开发环境.md`。 - `segment_video(frames, prompt_masks, cfg)`：把提示帧的病灶掩码用 MedSAM2 传播到每一帧；`frames` 是分割那一半的逐帧 BGR 图像，`prompt_masks` 是提示帧帧号 → 同尺寸布尔掩码，交出 `(帧数, 高, 宽)` 布尔数组（即 `quantify_case` 的 `target_masks`）。 - `SegmentConfig`：`config/ceus.yaml` 里 `segment` 段的字段定义（MedSAM2 代码目录、权重、模型配置名、设备、两个 offload 开关；pydantic，字段必填、不许多写、读入后不可改）。 - `medsam2_commit(medsam2_dir)`：读 MedSAM2 代码目录 `.git` 的 HEAD 提交号，给快照和分割缓存键用。 +- `cuda_peak_memory_bytes(device)`：本进程在该设备上分配过的显存峰值（字节），给运行快照用；非 CUDA 设备为 None，本进程没初始化过 CUDA（分割全读了缓存）为 0。放在 segment 里，是因为 `cadx` 里只有它引用 `torch`。 ### build（`src/cadx/build/`） diff --git a/scripts/ceus_quant.py b/scripts/ceus_quant.py new file mode 100644 index 0000000..59da9a5 --- /dev/null +++ b/scripts/ceus_quant.py @@ -0,0 +1,293 @@ +"""CEUS 定量：对交付包里检查全部通过的病例逐例跑第 1–9 步，写进 `runs/ceus/<运行号>/`。 + +用法：uv run python scripts/ceus_quant.py --config config/ceus.yaml +第 3 步（MedSAM2 分割）的结果按内容键缓存在 `/segment_cache/<键>/masks.npz`，键相同就 +直接读缓存、不调显卡。某一例在第 1–6 步失败（`CeusCaseError`）记进 `failed.csv`，接着跑下一例； +其余异常整次运行停下，留下 `.partial` 目录。终端只打印计数，不打印视频号。 +""" + +import argparse +import csv +import hashlib +import json +import sys +import time +from collections import Counter +from collections.abc import Callable, Hashable, Sequence +from datetime import datetime +from pathlib import Path +from typing import Annotated, Any + +import cv2 +import numpy as np +import numpy.typing as npt +import yaml +from pydantic import BaseModel, BeforeValidator, ConfigDict, Field + +from cadx.ceus import ( + ANCHOR_SLOTS, + CeusCaseError, + CeusConfig, + SegmentationInput, + keyframe_images, + pack_tic_curves, + pack_trajectory, + prepare_segmentation, + quant_summary, + quantify_case, + render_case_overlay, + render_tic_png, + tic_plot_markers, +) +from cadx.delivery import CaseRecord, read_delivery +from cadx.runs import make_build_id, read_git_state, write_snapshot +from cadx.segment import SegmentConfig, cuda_peak_memory_bytes, medsam2_commit, segment_video + +REPO_ROOT = Path(__file__).resolve().parents[1] + +# 第 1–2 步（`prepare_segmentation`）读的 ceus 段子段：views（分屏）、prompt_source（绿框颜色）、 +# keyframes（提示帧选择）。它们决定交给分割的帧和提示掩码，进分割缓存键。 +_STEP_1_2_CEUS_FIELDS = frozenset({"views", "prompt_source", "keyframes"}) +# segment 段里不改变分割结果的开关，不进缓存键；其余字段都进。 +_RESULT_NEUTRAL_SEGMENT_FIELDS = frozenset( + {"device", "offload_video_to_cpu", "offload_state_to_cpu"} +) +_FAILED_HEADER = ("video_id", "step", "reason") + + +def _reject_empty_path(value: object) -> object: + # Path("") 会变成 "."（当前目录），空串在转换前就拦下。 + if value == "": + raise ValueError("路径不能是空串") + return value + + +# YAML 里的路径是字符串；路径字段是唯一允许从字符串转换类型的地方。 +type _PathField = Annotated[Path, Field(strict=False), BeforeValidator(_reject_empty_path)] + + +class PathsConfig(BaseModel): + """`paths` 段：交付包、运行目录的上级、中间文件目录。相对路径相对仓库根目录。""" + + model_config = ConfigDict(strict=True, extra="forbid", frozen=True) + + delivery_dir: _PathField + runs_root: _PathField + work_root: _PathField + + +class QuantConfig(BaseModel): + """`ceus.yaml` 整份：`segment` 段交 segment 校验，`ceus` 段交 ceus 校验。""" + + model_config = ConfigDict(strict=True, extra="forbid", frozen=True) + + paths: PathsConfig + segment: SegmentConfig + ceus: CeusConfig + + +class _UniqueKeyLoader(yaml.SafeLoader): + """同一层出现重复键就报错：PyYAML 默认让后一个悄悄覆盖前一个。""" + + # 签名（含 deep 的默认值）照抄 PyYAML 的基类方法，覆盖时必须一致。 + def construct_mapping(self, node: yaml.MappingNode, deep: bool = False) -> dict[Hashable, Any]: + seen: set[Hashable] = set() + for key_node, _ in node.value: + key = self.construct_object(key_node, deep=deep) # pyright: ignore[reportUnknownMemberType] + if key in seen: + raise ValueError(f"YAML 第 {key_node.start_mark.line + 1} 行重复的键：{key!r}") + seen.add(key) + return super().construct_mapping(node, deep=deep) + + +def load_config(path: Path) -> QuantConfig: + """读并校验 `ceus.yaml`，路径保持文件里的写法（缓存键用这一份，不随仓库位置变）。""" + text = path.read_text(encoding="utf-8") + return QuantConfig.model_validate(yaml.load(text, Loader=_UniqueKeyLoader)) # noqa: S506 继承 SafeLoader + + +def _resolved(config: QuantConfig) -> QuantConfig: + """相对路径换成仓库根目录下的绝对路径（绝对路径原样保留）。""" + paths, segment = config.paths, config.segment + return config.model_copy( + update={ + "paths": paths.model_copy( + update={ + "delivery_dir": REPO_ROOT / paths.delivery_dir, + "runs_root": REPO_ROOT / paths.runs_root, + "work_root": REPO_ROOT / paths.work_root, + } + ), + "segment": segment.model_copy( + update={ + "medsam2_dir": REPO_ROOT / segment.medsam2_dir, + "checkpoint": REPO_ROOT / segment.checkpoint, + } + ), + } + ) + + +def file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + while chunk := handle.read(1 << 20): + digest.update(chunk) + return digest.hexdigest() + + +def segment_cache_key( + video_sha256: str, config: QuantConfig, checkpoint_sha256: str, commit: str +) -> str: + """分割缓存键：视频、第 1–3 步参数、权重、MedSAM2 提交号拼成规范 JSON 后的 SHA-256。""" + payload = { + "video_sha256": video_sha256, + "ceus": config.ceus.model_dump(mode="json", include=set(_STEP_1_2_CEUS_FIELDS)), + "segment": config.segment.model_dump( + mode="json", exclude=set(_RESULT_NEUTRAL_SEGMENT_FIELDS) + ), + "checkpoint_sha256": checkpoint_sha256, + "medsam2_commit": commit, + } + canonical = json.dumps(payload, sort_keys=True, ensure_ascii=False, separators=(",", ":")) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +def _masks( + prepared: SegmentationInput, key: str, cache_root: Path, segment_cfg: SegmentConfig +) -> tuple[npt.NDArray[np.bool_], bool]: + """读分割缓存；没有就调 MedSAM2，先写 `<键>.partial` 再改名。返回（掩码，是否命中缓存）。""" + final = cache_root / key + if final.exists(): + with np.load(final / "masks.npz", allow_pickle=False) as data: + return data["masks"], True + masks = segment_video(prepared.target_frames, prepared.prompt_masks, segment_cfg) + partial = cache_root / f"{key}.partial" + # 上次中途停下留下的 .partial 会让这里报错：work/ 可以删掉重算，删掉它再跑。 + partial.mkdir(parents=True, exist_ok=False) + np.savez_compressed(partial / "masks.npz", masks=masks) + partial.rename(final) + return masks, False + + +def _write_png(path: Path, image: npt.NDArray[np.uint8]) -> None: + ok, encoded = cv2.imencode(".png", image) + if not ok: + raise RuntimeError(f"关键帧编码 PNG 失败：{path}") + path.write_bytes(encoded.tobytes()) + + +def _run_case( + case: CaseRecord, cfg: QuantConfig, key_of: Callable[[str], str], run_dir: Path +) -> tuple[tuple[str, ...], bool]: + """跑一例并写出产物，返回（定位到的关键帧槽位，分割是否命中缓存）。 + + `key_of` 把视频的 SHA-256 换成分割缓存键（其余几样跨病例不变）。 + """ + prepared = prepare_segmentation(case.video, cfg.ceus) + key = key_of(file_sha256(case.video)) + masks, hit = _masks(prepared, key, cfg.paths.work_root / "segment_cache", cfg.segment) + result = quantify_case(prepared, masks, float(case.bolus_start_s), cfg.ceus) + + case_dir = run_dir / case.video_id + (case_dir / "keyframes").mkdir(parents=True, exist_ok=False) + (case_dir / "trajectory.npz").write_bytes(pack_trajectory(result.trajectory)) + (case_dir / "tic_curves.npz").write_bytes(pack_tic_curves(result.curves)) + summary = {"video_id": case.video_id, "segment_cache_key": key, **quant_summary(result)} + (case_dir / "summary.json").write_text( + json.dumps(summary, ensure_ascii=False, indent=2, allow_nan=False) + "\n", encoding="utf-8" + ) + markers = tic_plot_markers(result.anchors) + (case_dir / "tic.png").write_bytes(render_tic_png(result.tic, markers=markers)) + (case_dir / "overlay.mp4").write_bytes(render_case_overlay(prepared, result)) + images = keyframe_images(prepared, result.anchors) + for slot, image in images.items(): + _write_png(case_dir / "keyframes" / f"{slot}.png", image) + return tuple(images), hit + + +def main(argv: Sequence[str]) -> None: + parser = argparse.ArgumentParser(description="CEUS 定量：逐例跑第 1–9 步") + parser.add_argument("--config", type=Path, required=True, help="ceus.yaml 的路径") + config_path: Path = parser.parse_args(argv).config + started_at = datetime.now().astimezone() + clock = time.monotonic() + + written = load_config(config_path) + cfg = _resolved(written) + git = read_git_state(REPO_ROOT) + run_id = make_build_id(started_at, git) + run_dir = cfg.paths.runs_root / run_id + partial = cfg.paths.runs_root / f"{run_id}.partial" + for path in (run_dir, partial): + if path.exists(): + raise FileExistsError(f"运行目录已存在，不覆盖：{path}") + partial.mkdir(parents=True, exist_ok=False) + + delivery = read_delivery(cfg.paths.delivery_dir, None) + checkpoint_sha256 = file_sha256(cfg.segment.checkpoint) + commit = medsam2_commit(cfg.segment.medsam2_dir) + + # 缓存键用配置文件里的写法（路径不随仓库位置变）。 + def key_of(video_sha256: str) -> str: + return segment_cache_key(video_sha256, written, checkpoint_sha256, commit) + + failures: list[tuple[str, int, str]] = [] + located: Counter[str] = Counter() + cache_hits = 0 + total = len(delivery.cases) + for number, case in enumerate(delivery.cases, start=1): + try: + slots, hit = _run_case(case, cfg, key_of, partial) + except CeusCaseError as error: + failures.append((case.video_id, error.step, error.reason)) + print(f"[{number}/{total}] 失败在第 {error.step} 步", flush=True) + continue + located.update(slots) + cache_hits += hit + print(f"[{number}/{total}] 完成", flush=True) + + with (partial / "failed.csv").open("w", encoding="utf-8", newline="") as handle: + writer = csv.writer(handle) + writer.writerow(_FAILED_HEADER) + writer.writerows(failures) + peak = cuda_peak_memory_bytes(cfg.segment.device) + write_snapshot( + snapshot_dir=partial / "snapshot", + build_id=run_id, + git=git, + started_at=started_at, + finished_at=datetime.now().astimezone(), + uv_lock=REPO_ROOT / "uv.lock", + configs={"ceus.yaml": config_path}, + ) + segment_record = { + "checkpoint_sha256": checkpoint_sha256, + "medsam2_commit": commit, + "device": cfg.segment.device, + "cuda_peak_memory_bytes": peak, + "segment_cache_hits": cache_hits, + } + (partial / "snapshot" / "segment.json").write_text( + json.dumps(segment_record, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" + ) + # POSIX 的 rename 会悄悄覆盖空目录，所以改名前再查一次。 + if run_dir.exists(): + raise FileExistsError(f"运行目录已存在，不覆盖：{run_dir}") + partial.rename(run_dir) + + by_step = Counter(step for _, step, _ in failures) + print(f"运行目录：{run_dir}") + print(f"病例总数（检查全部通过）：{total}") + print(f"成功：{total - len(failures)}") + failed_text = "，".join(f"第 {step} 步 {n}" for step, n in sorted(by_step.items())) + print(f"按步分的失败数：{failed_text or '无'}") + print("关键帧按槽位分的定位数：" + "，".join(f"{s} {located[s]}" for s in ANCHOR_SLOTS)) + print(f"分割缓存命中：{cache_hits}") + print(f"耗时：{time.monotonic() - clock:.1f} 秒") + peak_text = "不适用（非 CUDA 设备）" if peak is None else f"{peak / 2**20:.0f} MiB" + print(f"显存峰值（torch 分配）：{peak_text}") + + +if __name__ == "__main__": + main(sys.argv[1:]) diff --git a/src/cadx/ceus/__init__.py b/src/cadx/ceus/__init__.py index 485a2be..2291d5f 100644 --- a/src/cadx/ceus/__init__.py +++ b/src/cadx/ceus/__init__.py @@ -135,7 +135,7 @@ from cadx.ceus._trajectory import ( pack_trajectory, unpack_trajectory, ) -from cadx.ceus._video import VideoFile +from cadx.ceus._video import VideoDecodeError, VideoFile __all__ = [ "ANCHOR_SLOTS", @@ -186,6 +186,7 @@ __all__ = [ "TicPlotMarker", "TicResult", "Trajectory", + "VideoDecodeError", "VideoFile", "ViewRect", "ViewsConfig", diff --git a/src/cadx/ceus/_config.py b/src/cadx/ceus/_config.py index 5364a6c..c112f60 100644 --- a/src/cadx/ceus/_config.py +++ b/src/cadx/ceus/_config.py @@ -539,12 +539,14 @@ class CeusConfig(BaseModel): """`config/ceus.yaml` 的 `ceus` 段：CEUS 定量的全部算法参数。 `config_version` 标记这一批产物用的是哪一套参数口径，由人改，不从内容哈希派生； - 不许是空串。 + 不许是空串。`random_seed` 是 cv2 进程级随机数的种子，`quantify_case` 每例开头设一次（第 5 步暗区 + 的 kmeans 候选用到随机数），让一例的结果与同一进程里先跑过哪些病例无关。 """ model_config = _STRICT config_version: str = Field(min_length=1) + random_seed: int views: ViewsConfig prompt_source: PromptSourceConfig pan_source: PanSourceConfig diff --git a/src/cadx/ceus/_pipeline.py b/src/cadx/ceus/_pipeline.py index d14a1fc..1ed0aff 100644 --- a/src/cadx/ceus/_pipeline.py +++ b/src/cadx/ceus/_pipeline.py @@ -20,6 +20,7 @@ from dataclasses import dataclass from pathlib import Path from typing import Any +import cv2 import numpy as np from cadx.ceus._anchor_frames import ( @@ -50,7 +51,7 @@ from cadx.ceus._tic_curves import TicCurves, from_tic_result from cadx.ceus._tic_fitting import LognormalFitResult, evaluate_lognormal from cadx.ceus._tic_plot import TicPlotMarker from cadx.ceus._trajectory import Trajectory -from cadx.ceus._video import VideoFile +from cadx.ceus._video import VideoDecodeError, VideoFile # 灌注指标只对病灶两条曲线算（同旧 replay 的 `_LESION_CURVES`）。 LESION_CURVES = ("roi_all", "roi_perf") @@ -125,7 +126,8 @@ class QuantResult: def prepare_segmentation(video_path: Path, cfg: CeusConfig) -> SegmentationInput: """第 1–2 步：解码、分屏、找绿框定造影那一半、选提示帧，交出分割的输入。 - 找不到可信分屏线抛 `SplitDetectionError`（第 1 步），两半都有或都没有绿框抛 `ContrastSideError` + 视频读不了或读出的不合规抛 `VideoDecodeError`、找不到可信分屏线抛 `SplitDetectionError`（都是 + 第 1 步），两半都有或都没有绿框抛 `ContrastSideError` （第 2 步），两者都是 `CeusCaseError`。 """ video = VideoFile(video_path) @@ -133,7 +135,7 @@ def prepare_segmentation(video_path: Path, cfg: CeusConfig) -> SegmentationInput frame_times_s = video.frame_times_s() full_frames = [frame for _, frame in video.iter_frames()] if len(full_frames) != frame_count: - raise RuntimeError( + raise VideoDecodeError( f"解码帧数 {len(full_frames)} 与容器记录的帧数 {frame_count} 不一致：{video_path}" ) @@ -194,6 +196,9 @@ def quantify_case( 灌注指标为 None。 """ _validate_masks(prepared, target_masks) + # 第 5 步暗区的 kmeans 候选（k-means++ 选初始中心）用 cv2 的进程级随机数；每例开头重设， + # 这一例的结果就与同一进程里先跑过什么无关。 + cv2.setRNGSeed(cfg.random_seed) contrast = prepared.contrast frame_count = prepared.frame_count prompt_frame_dicts: list[dict[str, Any]] = [ diff --git a/src/cadx/ceus/_video.py b/src/cadx/ceus/_video.py index 39a4f2a..f5660b2 100644 --- a/src/cadx/ceus/_video.py +++ b/src/cadx/ceus/_video.py @@ -2,6 +2,7 @@ 反复遍历靠重新解码，不缓存帧：一段 1260 帧的素材解一遍不到两秒，缓存下来要好几 GB（旧项目实测）。 每次遍历各开一个 `cv2.VideoCapture`，互不干扰。探测不到有效 fps、帧数或尺寸直接报错，不回退默认值。 +打不开、元数据或时间戳不合规是这一段视频的资料问题，抛 `VideoDecodeError`（第 1 步的单例失败）。 """ import math @@ -13,8 +14,21 @@ import cv2 import numpy as np import numpy.typing as npt +from cadx.ceus._errors import CeusCaseError from cadx.ceus._geometry import Frame +DECODE_STEP = 1 + + +class VideoDecodeError(CeusCaseError, RuntimeError): + """这段视频读不了或读出的不合规（打不开、fps 或尺寸非法、时间戳不严格递增、帧数对不上）。 + + 同时是 `RuntimeError`：旧测试按 `RuntimeError` 断言，语义不变。 + """ + + def __init__(self, reason: str) -> None: + super().__init__(DECODE_STEP, reason) + @dataclass(frozen=True) class VideoFile: @@ -26,7 +40,7 @@ class VideoFile: cap = cv2.VideoCapture(str(self.path)) if not cap.isOpened(): cap.release() - raise RuntimeError(f"cv2 打不开视频：{self.path}") + raise VideoDecodeError(f"cv2 打不开视频：{self.path}") return cap def probe(self) -> tuple[float, int, int, int]: @@ -40,9 +54,9 @@ class VideoFile: finally: cap.release() if not math.isfinite(fps) or fps <= 0: - raise RuntimeError(f"探测到非法 fps={fps}：{self.path}") + raise VideoDecodeError(f"探测到非法 fps={fps}：{self.path}") if frame_count <= 0 or width <= 0 or height <= 0: - raise RuntimeError( + raise VideoDecodeError( f"探测到非法元数据 frame_count={frame_count} " f"width={width} height={height}：{self.path}" ) @@ -74,9 +88,9 @@ class VideoFile: break timestamp_s = float(cap.get(cv2.CAP_PROP_POS_MSEC)) / 1000.0 if not math.isfinite(timestamp_s) or timestamp_s < 0.0: - raise RuntimeError(f"视频包含非法显示时间戳 {timestamp_s}：{self.path}") + raise VideoDecodeError(f"视频包含非法显示时间戳 {timestamp_s}：{self.path}") if times and timestamp_s <= times[-1]: - raise RuntimeError( + raise VideoDecodeError( "视频显示时间戳不严格递增：" f"第 {len(times)} 帧 {timestamp_s}，前一帧 {times[-1]}：{self.path}" ) @@ -84,7 +98,7 @@ class VideoFile: finally: cap.release() if len(times) != expected_count: - raise RuntimeError( + raise VideoDecodeError( f"时间戳帧数与容器记录的帧数不一致：{len(times)} / {expected_count}：{self.path}" ) return np.asarray(times, dtype=np.float64) diff --git a/src/cadx/delivery/_cases.py b/src/cadx/delivery/_cases.py index 065a89e..19551b0 100644 --- a/src/cadx/delivery/_cases.py +++ b/src/cadx/delivery/_cases.py @@ -77,6 +77,8 @@ class CaseRecord: ultrasound_report: str mr_report: str key_image: Path + video: Path # CEUS 视频（C04 已保证恰好一段，文件名带不带 roi2 标记都可能） + bolus_start_s: int # 超声表「推注开始时间s」：视频第 0 秒时已推注的秒数（C08 已保证非负整数） diagnosis: DiagnosisCodes @@ -181,13 +183,13 @@ def _index_rows( _VIDEO_REPORT_PAIRS = (("_roi2.mp4", "_roi2_report.xlsx"), (".mp4", "_report.xlsx")) -def _check_case_dir(case_no: str, folder: Path) -> str | SkipRecord: - """C04：通过时返回这一例单例报告表的文件名。""" +def _check_case_dir(case_no: str, folder: Path) -> tuple[str, str] | SkipRecord: + """C04：通过时返回这一例（视频文件名，单例报告表文件名）。""" u = video_id(case_no) names = sorted(p.name for p in folder.iterdir()) for video, report in _VIDEO_REPORT_PAIRS: if names == sorted((f"{u}_key.png", f"{u}{video}", f"{u}{report}")): - return f"{u}{report}" + return f"{u}{video}", f"{u}{report}" return _skip(case_no, "C04", (case_dir_path(case_no),), (), {"文件": names}) @@ -336,9 +338,10 @@ def _check_case( folder = delivery_dir / case_dir_path(case_no) u = video_id(case_no) - report_name = _check_case_dir(case_no, folder) - if isinstance(report_name, SkipRecord): - return report_name + dir_files = _check_case_dir(case_no, folder) + if isinstance(dir_files, SkipRecord): + return dir_files + video_name, report_name = dir_files if ultrasound["CEUS视频ID"] != u: value = {"CEUS视频ID": ultrasound["CEUS视频ID"], "病例号": case_no} @@ -370,7 +373,8 @@ def _check_case( if (skip := _text_hits(case_no, check, reports, find)) is not None: return skip - # C07 已保证临床各列是整数、浮点数或空；C07–C10 已保证三个诊断是相等的整数。 + # C07 已保证临床各列是整数、浮点数或空；C07–C10 已保证三个诊断是相等的整数； + # C08 已保证推注时间是非负整数。 return CaseRecord( case_no=case_no, video_id=u, @@ -378,6 +382,8 @@ def _check_case( ultrasound_report=reports[ULTRASOUND], mr_report=reports[MR], key_image=folder / f"{u}_key.png", + video=folder / video_name, + bolus_start_s=cast(int, bolus), diagnosis=DiagnosisCodes( clinical=cast(int, clinical["Diagnosis"]), ultrasound=cast(int, ultrasound[DIAGNOSIS_HEADER]), diff --git a/src/cadx/segment/__init__.py b/src/cadx/segment/__init__.py index ab62d35..d0accbd 100644 --- a/src/cadx/segment/__init__.py +++ b/src/cadx/segment/__init__.py @@ -5,6 +5,6 @@ """ from cadx.segment._config import SegmentConfig -from cadx.segment._medsam2 import medsam2_commit, segment_video +from cadx.segment._medsam2 import cuda_peak_memory_bytes, medsam2_commit, segment_video -__all__ = ["SegmentConfig", "medsam2_commit", "segment_video"] +__all__ = ["SegmentConfig", "cuda_peak_memory_bytes", "medsam2_commit", "segment_video"] diff --git a/src/cadx/segment/_medsam2.py b/src/cadx/segment/_medsam2.py index 8214ce1..1fd68a1 100644 --- a/src/cadx/segment/_medsam2.py +++ b/src/cadx/segment/_medsam2.py @@ -45,6 +45,19 @@ def medsam2_commit(medsam2_dir: Path) -> str: return completed.stdout.decode().strip() +def cuda_peak_memory_bytes(device: str) -> int | None: + """本进程从启动起在 `device` 上分配过的显存峰值（字节，`torch.cuda.max_memory_allocated`）。 + + `device` 不是 CUDA 设备时返回 None；本进程还没初始化过 CUDA（比如分割全部读了缓存）时返回 0， + 不为了查询去初始化显卡。 + """ + if not device.startswith("cuda"): + return None + if not torch.cuda.is_initialized(): + return 0 + return int(torch.cuda.max_memory_allocated(device)) + + def _check_inputs( frames: Sequence[Frame], prompt_masks: Mapping[int, Mask], cfg: SegmentConfig ) -> tuple[int, int]: diff --git a/tests/test_ceus_quant.py b/tests/test_ceus_quant.py new file mode 100644 index 0000000..bdbc169 --- /dev/null +++ b/tests/test_ceus_quant.py @@ -0,0 +1,286 @@ +"""`scripts/ceus_quant.py`：逐例循环、failed.csv、输出布局、分割缓存。 + +假交付包两例：第 1 例是能过质量门的合成分屏造影视频（造影在右半，绿框是病灶的空心描边），第 2 例 +同样的画面但没有绿框。分割换成计数的假函数（monkeypatch 脚本里引用的 `segment_video`），交出病灶 +方块挪到左半局部坐标的掩码，不调显卡。 + +手算的预期（视频 400×200、10 fps、40 秒，推注时间 X = 5 秒，同 test_ceus_pipeline）： +- 第 2 例两半都没有绿框 → 第 2 步失败，进 failed.csv，没有输出目录，也不调分割； +- 第 1 例 F1 在视频第 20 − 5 = 15 秒，有关键帧图；F2 在第 55 秒，超出视频，未定位，没有图。 +""" + +import csv +import importlib.util +import json +from collections.abc import Callable, Mapping, Sequence +from pathlib import Path +from types import ModuleType +from typing import Any + +import cv2 +import numpy as np +import numpy.typing as npt +import pytest +import yaml + +from cadx.ceus import ANCHOR_SLOTS, LognormalParameters, evaluate_lognormal +from conftest import CEUS_YAML +from fake_delivery import make_delivery + +_SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "ceus_quant.py" +_WIDTH, _HEIGHT, _SPLIT = 400, 200, 200 +_FPS, _FRAMES = 10.0, 400 +_LESION = (250, 60, 369, 159) # 整帧坐标 x1, y1, x2, y2（含端点） +_CYST = (290, 95, 320, 125) +_BOLUS = LognormalParameters(area=1300.0, mu=2.4, sigma=0.45, arrival_offset_s=4.0, baseline=40.0) + + +def _load_script() -> ModuleType: + spec = importlib.util.spec_from_file_location("ceus_quant", _SCRIPT) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _write_video(path: Path, green_box: bool) -> None: + rng = np.random.default_rng(3) + intensity = evaluate_lognormal(np.arange(_FRAMES) / _FPS, _BOLUS) + writer = cv2.VideoWriter(str(path), cv2.VideoWriter.fourcc(*"mp4v"), _FPS, (_WIDTH, _HEIGHT)) + x1, y1, x2, y2 = _LESION + for index in range(_FRAMES): + frame = np.zeros((_HEIGHT, _WIDTH, 3), dtype=np.uint8) + frame[:, :_SPLIT] = 120 + frame[:, _SPLIT + 1 :] = 30 + lesion = np.clip( + intensity[index] + rng.normal(0.0, 6.0, (y2 - y1 + 1, x2 - x1 + 1)), 0, 255 + ) + frame[y1 : y2 + 1, x1 : x2 + 1] = lesion.astype(np.uint8)[:, :, None] + cx1, cy1, cx2, cy2 = _CYST + frame[cy1 : cy2 + 1, cx1 : cx2 + 1] = 10 + if green_box: + cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 255, 0), thickness=3) + writer.write(frame) + writer.release() + + +def _config_dict(tmp_path: Path, delivery_dir: Path) -> dict[str, Any]: + document: dict[str, Any] = yaml.safe_load(CEUS_YAML.read_text(encoding="utf-8")) + document["paths"] = { + "delivery_dir": str(delivery_dir), + "runs_root": str(tmp_path / "runs"), + "work_root": str(tmp_path / "work"), + } + checkpoint = tmp_path / "fake_checkpoint.pt" + checkpoint.write_bytes(b"fake weights") + document["segment"].update( + {"medsam2_dir": str(tmp_path / "MedSAM2"), "checkpoint": str(checkpoint), "device": "cpu"} + ) + return document + + +def _write_config(tmp_path: Path, document: dict[str, Any], name: str) -> Path: + path = tmp_path / name + path.write_text(yaml.safe_dump(document, allow_unicode=True), encoding="utf-8") + return path + + +class _CountingSegment: + """假分割：每帧都是病灶方块挪到左半局部坐标后的位置（略往里收 3 像素）。""" + + def __init__(self) -> None: + self.calls = 0 + + def __call__( + self, + frames: Sequence[npt.NDArray[np.uint8]], + prompt_masks: Mapping[int, npt.NDArray[np.bool_]], + cfg: object, + ) -> npt.NDArray[np.bool_]: + self.calls += 1 + height, width = frames[0].shape[:2] + masks = np.zeros((len(frames), height, width), dtype=bool) + x1, y1, x2, y2 = _LESION + masks[:, y1 + 3 : y2 - 2, x1 - _SPLIT + 3 : x2 - _SPLIT - 2] = True + return masks + + +@pytest.fixture +def setup( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> tuple[ModuleType, _CountingSegment, dict[str, Any], Callable[[str], None]]: + """写假交付包（两例真视频），载入脚本并换掉分割、MedSAM2 提交号和运行号。""" + delivery_dir = make_delivery(2).write(tmp_path / "delivery") + _write_video(delivery_dir / "CEUS_agent" / "001" / "U001_roi2.mp4", green_box=True) + _write_video(delivery_dir / "CEUS_agent" / "002" / "U002_roi2.mp4", green_box=False) + module = _load_script() + segment = _CountingSegment() + monkeypatch.setattr(module, "segment_video", segment) + run_ids: list[str] = [] + + def fake_commit(medsam2_dir: Path) -> str: + return "c" * 40 + + def fake_run_id(now: object, git: object) -> str: + return run_ids[-1] + + monkeypatch.setattr(module, "medsam2_commit", fake_commit) + monkeypatch.setattr(module, "make_build_id", fake_run_id) + return module, segment, _config_dict(tmp_path, delivery_dir), run_ids.append + + +def test_run_writes_outputs_and_failed_case( + tmp_path: Path, + setup: tuple[ModuleType, _CountingSegment, dict[str, Any], Callable[[str], None]], + capsys: pytest.CaptureFixture[str], +) -> None: + module, segment, 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" + assert sorted(p.name for p in (tmp_path / "runs").iterdir()) == ["run-a"] + with (run_dir / "failed.csv").open(encoding="utf-8", newline="") as handle: + rows = list(csv.reader(handle)) + assert rows[0] == ["video_id", "step", "reason"] + assert [row[:2] for row in rows[1:]] == [["U002", "2"]] + assert "都没有" in rows[1][2] + assert not (run_dir / "U002").exists() + assert segment.calls == 1 + + case_dir = run_dir / "U001" + for name in ("trajectory.npz", "tic_curves.npz", "summary.json", "tic.png", "overlay.mp4"): + assert (case_dir / name).stat().st_size > 0, name + summary = json.loads((case_dir / "summary.json").read_text(encoding="utf-8")) + assert summary["video_id"] == "U001" + assert summary["injection_offset_s"] == 5.0 + assert list(summary["anchor_frames"]) == list(ANCHOR_SLOTS) + assert summary["perfusion_metrics"]["roi_perf"] is not None + assert "pan_detected_fraction" in summary + assert summary["anchor_frames"]["F2"]["status"] == "not_covered" + located = {s for s, a in summary["anchor_frames"].items() if a["frame_index"] is not None} + assert "F1" in located and "F2" not in located + keyframes = {p.stem for p in (case_dir / "keyframes").iterdir()} + assert keyframes == located + image = cv2.imread(str(case_dir / "keyframes" / "F1.png")) + assert image is not None and image.shape == (_HEIGHT, _WIDTH, 3) # 整幅原始画面 + + snapshot = run_dir / "snapshot" + assert (snapshot / "ceus.yaml").read_bytes() == (tmp_path / "ceus.yaml").read_bytes() + segment_record = json.loads((snapshot / "segment.json").read_text(encoding="utf-8")) + assert segment_record["medsam2_commit"] == "c" * 40 + assert segment_record["cuda_peak_memory_bytes"] is None # device 是 cpu + assert (snapshot / "snapshot.json").is_file() + + out = capsys.readouterr().out + assert "成功：1" in out and "第 2 步 1" in out + assert "U001" not in out and "U002" not in out + + +def test_unreadable_video_fails_at_step_1_and_next_case_runs( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + setup: tuple[ModuleType, _CountingSegment, dict[str, Any], Callable[[str], None]], +) -> None: + module, segment, document, next_run_id = setup + # 第 1 例换成损坏的视频（cv2 打不开），第 2 例换成正常视频。 + videos = Path(document["paths"]["delivery_dir"]) / "CEUS_agent" + (videos / "001" / "U001_roi2.mp4").write_bytes(b"not a real video") + _write_video(videos / "002" / "U002_roi2.mp4", green_box=True) + next_run_id("run-a") + module.main(["--config", str(_write_config(tmp_path, document, "ceus.yaml"))]) + + run_dir = tmp_path / "runs" / "run-a" + with (run_dir / "failed.csv").open(encoding="utf-8", newline="") as handle: + rows = list(csv.reader(handle)) + assert [row[:2] for row in rows[1:]] == [["U001", "1"]] + assert "打不开" in rows[1][2] + assert not (run_dir / "U001").exists() + assert (run_dir / "U002" / "summary.json").is_file() + assert segment.calls == 1 + + +def test_segment_cache_reused_and_keyed_on_step_1_3_params( + tmp_path: Path, + setup: tuple[ModuleType, _CountingSegment, dict[str, Any], Callable[[str], None]], +) -> None: + module, segment, document, next_run_id = setup + config = _write_config(tmp_path, document, "ceus.yaml") + next_run_id("run-a") + module.main(["--config", str(config)]) + assert segment.calls == 1 + cache = tmp_path / "work" / "segment_cache" + assert [p.name.endswith(".partial") for p in cache.iterdir()] == [False] + + next_run_id("run-b") + module.main(["--config", str(config)]) + assert segment.calls == 1 # 同样的输入，读缓存 + # 同一进程里第二次跑（cv2 的随机数已被第一次推进过）结果也逐位相同：quantify_case 每例重设种子。 + first = json.loads((tmp_path / "runs/run-a/U001/summary.json").read_text(encoding="utf-8")) + second = json.loads((tmp_path / "runs/run-b/U001/summary.json").read_text(encoding="utf-8")) + assert first == second + + # 改设备不改结果，不重算；改第 1 步的参数，键变了，重新分割。 + document["segment"]["device"] = "cuda:0" + document["segment"]["offload_state_to_cpu"] = True + next_run_id("run-c") + module.main(["--config", str(_write_config(tmp_path, document, "device.yaml"))]) + assert segment.calls == 1 + document["ceus"]["views"]["split_sample_frames"] = 20 + next_run_id("run-d") + module.main(["--config", str(_write_config(tmp_path, document, "views.yaml"))]) + assert segment.calls == 2 + assert len(list(cache.iterdir())) == 2 + + +def test_existing_run_directory_is_not_overwritten( + tmp_path: Path, + setup: tuple[ModuleType, _CountingSegment, dict[str, Any], Callable[[str], None]], +) -> None: + module, _, document, next_run_id = setup + config = _write_config(tmp_path, document, "ceus.yaml") + next_run_id("run-a") + module.main(["--config", str(config)]) + before = sorted(str(p) for p in (tmp_path / "runs").rglob("*")) + with pytest.raises(FileExistsError, match="运行目录已存在"): + module.main(["--config", str(config)]) + assert sorted(str(p) for p in (tmp_path / "runs").rglob("*")) == before + + +def _key(module: ModuleType, document: dict[str, Any]) -> str: + config = module.QuantConfig.model_validate(document) + return module.segment_cache_key("v" * 64, config, "w" * 64, "c" * 40) + + +@pytest.mark.parametrize( + ("section", "path", "value", "changes"), + [ + ("ceus", ("views", "split_search_band"), [0.3, 0.7], True), + ("ceus", ("prompt_source", "min_area"), 9000, True), + ("ceus", ("keyframes", "iou_threshold"), 0.9, True), + ("segment", ("sam2_config",), "configs/other.yaml", True), + ("segment", ("checkpoint",), "models/other.pt", True), + ("segment", ("device",), "cuda:1", False), + ("segment", ("offload_video_to_cpu",), True, False), + ("ceus", ("smoothing", "temporal_window"), 9, False), # 第 4 步 + ("ceus", ("tic_analysis", "fitting", "minimum_r2"), 0.5, False), # 第 7 步 + ], +) +def test_cache_key_covers_exactly_step_1_3_params( + tmp_path: Path, + section: str, + path: tuple[str, ...], + value: object, + changes: bool, +) -> None: + module = _load_script() + document = _config_dict(tmp_path, tmp_path) + base = _key(module, document) + node = document[section] + for name in path[:-1]: + node = node[name] + node[path[-1]] = value + assert (_key(module, document) != base) is changes + assert _key(module, document) == _key(module, document) # 同输入同键 + other = module.QuantConfig.model_validate(document) + assert module.segment_cache_key("x" * 64, other, "w" * 64, "c" * 40) != _key(module, document) diff --git a/tests/test_delivery_cases.py b/tests/test_delivery_cases.py index 8652e2b..0695286 100644 --- a/tests/test_delivery_cases.py +++ b/tests/test_delivery_cases.py @@ -418,3 +418,19 @@ def test_skip_record_names_file_and_column(tmp_path: Path) -> None: 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]