iomgaa
7e7f32a13d
- ceus：灌注指标改为对病灶去暗区（roi_perf）和紫圈（roi_pan）两组算，roi_all 指标删除； D1–D4 仍要 roi_all、roi_perf 两条都过才定位（负责人 2026-10-17：医生要正常区与病灶对比看） - build/library：tic.json 只出两组 11 个指标与 7 个槽位的视频时间；医生截图挪进 ultrasound/keyframes/key.png，UltrasoundRoute 去掉 key_image、keyframes 多一个 key - review：TIC参数.csv 取代 TIC结果.csv；问题清单按原因对照表写成大白话，对照表外的原因报错； 程序原文进 snapshot/问题清单_原文.csv；说明.pdf 整份重写为合作者口吻 - 文档：结构.md、医生核对包、装配两个库、CEUS定量 任务书 04。 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
263 lines
12 KiB
Python
263 lines
12 KiB
Python
"""ceus 对外两段接口的端到端测试：合成一段分屏造影视频，第二段喂合成掩膜（不调分割模型）。
|
||
|
||
视频 400×200、10 fps、40 秒（400 帧），分屏线在 x=200（这一列全黑）。造影画面在右半：病灶是整帧
|
||
[250, 369] × [60, 159] 的方块，医生的绿框是它的空心描边；病灶内部亮度按 Lognormal 团注曲线从
|
||
基线 40 升到约 170 再回落，内部另有一块一直发黑的囊（暗区）；每帧加一点随机噪声，免得相邻帧
|
||
被当成重复帧。两半各有一块位置相同、带纹理、随时间轻微起伏的扇形区（`fake_split_scene`，
|
||
配准要靠它），左半是灰阶画面，右半（造影）调暗。
|
||
|
||
手算的预期：
|
||
- 绿框在右半 → 造影画面是右半，分割那一半是左半 [0, 200)；
|
||
- 绿框挪到左半局部坐标：x = 250 − 200 = 50 到 369 − 200 = 169；
|
||
- 推注时间 X = 5 秒 → F1 在视频第 20 − 5 = 15 秒（第 150 帧），F2 在第 60 − 5 = 55 秒，
|
||
超过 40 秒的视频末尾，未定位。
|
||
"""
|
||
|
||
import dataclasses
|
||
import json
|
||
import math
|
||
|
||
import numpy as np
|
||
import pytest
|
||
|
||
from cadx.ceus import (
|
||
ANCHOR_SLOTS,
|
||
CeusConfig,
|
||
LognormalFitResult,
|
||
LognormalParameters,
|
||
NoValidContourError,
|
||
SegmentationInput,
|
||
TicResult,
|
||
VideoFile,
|
||
compute_lognormal_perfusion_metrics,
|
||
compute_tic_for_video,
|
||
crop_frame,
|
||
evaluate_lognormal,
|
||
fit_lognormal_curve,
|
||
keyframe_images,
|
||
locate_contrast_side,
|
||
pack_tic_curves,
|
||
pack_trajectory,
|
||
prepare_segmentation,
|
||
quant_summary,
|
||
quantify_case,
|
||
register_halves,
|
||
render_case_overlay,
|
||
render_tic_png,
|
||
split_views,
|
||
tic_plot_markers,
|
||
unpack_tic_curves,
|
||
unpack_trajectory,
|
||
)
|
||
from fake_ceus_video import BOLUS, FPS, FRAMES, HEIGHT, LESION, SPLIT, WIDTH, write_video
|
||
|
||
_INJECTION_OFFSET_S = 5.0
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def config() -> CeusConfig:
|
||
from conftest import load_ceus_section
|
||
|
||
return CeusConfig.model_validate(load_ceus_section())
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def prepared(tmp_path_factory: pytest.TempPathFactory, config: CeusConfig) -> SegmentationInput:
|
||
path = tmp_path_factory.mktemp("ceus") / "synthetic.mp4"
|
||
write_video(path, green_box=True)
|
||
return prepare_segmentation(path, config)
|
||
|
||
|
||
def _lesion_masks(prepared: SegmentationInput) -> np.ndarray:
|
||
"""合成「分割结果」：每帧都是病灶方块挪到左半局部坐标后的位置（略往里收 3 像素）。"""
|
||
view = prepared.contrast.target_view
|
||
masks = np.zeros((prepared.frame_count, view.height, view.width), dtype=bool)
|
||
x1, y1, x2, y2 = LESION
|
||
masks[:, y1 + 3 : y2 - 2, x1 - SPLIT + 3 : x2 - SPLIT - 2] = True
|
||
return masks
|
||
|
||
|
||
def test_prepare_segmentation_finds_contrast_half_and_maps_prompts(
|
||
prepared: SegmentationInput,
|
||
) -> None:
|
||
assert prepared.frame_count == FRAMES
|
||
assert prepared.frame_size == (WIDTH, HEIGHT)
|
||
assert prepared.contrast.side == "right"
|
||
# 两半扇形对齐：两种配准都估出 (0, 0)；造影在右半，平移取反后仍是 (0, 0)。
|
||
assert prepared.registration.correlation == (0, 0)
|
||
assert prepared.registration.disagreement_px <= 1
|
||
assert prepared.offset == (0, 0)
|
||
target = prepared.contrast.target_view
|
||
assert (target.x1, target.x2) == (0, prepared.split_x)
|
||
assert abs(prepared.split_x - SPLIT) <= 1
|
||
target_frames = list(prepared.target_frames())
|
||
assert len(target_frames) == FRAMES
|
||
assert all(frame.shape == (target.height, target.width, 3) for frame in target_frames)
|
||
assert sorted(prepared.prompt_masks) == [p.frame_index for p in prepared.contrast.prompt_frames]
|
||
assert 0 in prepared.prompt_masks
|
||
xs = np.where(prepared.prompt_masks[0])[1]
|
||
# 描边 3 像素宽、找绿框时再膨胀 2 像素、mp4 压缩会糊开边缘，容差放到 5 像素。
|
||
assert abs(int(xs.min()) - 50) <= 5
|
||
assert abs(int(xs.max()) - 169) <= 5
|
||
|
||
|
||
def test_streaming_prepare_matches_whole_video_in_memory(
|
||
prepared: SegmentationInput, config: CeusConfig
|
||
) -> None:
|
||
"""逐帧解码的第 1–2 步与「整段帧读进内存再交给各步」逐像素相同（改成流式之前就是后一种）。"""
|
||
frames = [frame for _, frame in VideoFile(prepared.video_path).iter_frames()]
|
||
split_x, left, right = split_views(frames, config.views)
|
||
registration = register_halves(
|
||
frames, left, right, config.registration, config.prompt_source, config.pan_source
|
||
)
|
||
contrast = locate_contrast_side(frames, left, right, config.prompt_source, config.keyframes)
|
||
assert prepared.split_x == split_x
|
||
assert prepared.registration == registration
|
||
assert prepared.contrast.side == contrast.side
|
||
assert (prepared.contrast.contrast_view, prepared.contrast.target_view) == (
|
||
contrast.contrast_view,
|
||
contrast.target_view,
|
||
)
|
||
assert [p.frame_index for p in prepared.contrast.prompt_frames] == [
|
||
p.frame_index for p in contrast.prompt_frames
|
||
]
|
||
for ours, whole in zip(prepared.contrast.prompt_frames, contrast.prompt_frames, strict=True):
|
||
assert np.array_equal(ours.mask, whole.mask)
|
||
target_frames = list(prepared.target_frames())
|
||
assert len(target_frames) == len(frames)
|
||
for ours, frame in zip(target_frames, frames, strict=True):
|
||
assert np.array_equal(ours, crop_frame(frame, contrast.target_view))
|
||
assert np.array_equal(
|
||
prepared.first_contrast_frame, crop_frame(frames[0], contrast.contrast_view)
|
||
)
|
||
|
||
|
||
def test_quantify_case_end_to_end(prepared: SegmentationInput, config: CeusConfig) -> None:
|
||
result = quantify_case(prepared, _lesion_masks(prepared), _INJECTION_OFFSET_S, config)
|
||
|
||
assert result.trajectory.source_polygons.shape == (FRAMES, 64, 2)
|
||
assert result.lesion_fit_ready is True
|
||
assert list(result.anchors) == list(ANCHOR_SLOTS)
|
||
for slot in ("D1", "D2", "D3"):
|
||
assert result.anchors[slot].status == "ok", result.anchors[slot].reason
|
||
assert result.anchors["F1"].frame_index == 150
|
||
assert result.anchors["F1"].frame_time_video_s == pytest.approx(15.0, abs=0.06)
|
||
assert result.anchors["F2"].status == "not_covered"
|
||
assert result.anchors["F2"].reason is not None
|
||
assert result.perfusion["roi_perf"] is not None
|
||
peak_time = BOLUS.arrival_offset_s + math.exp(BOLUS.mu - BOLUS.sigma**2)
|
||
assert result.perfusion["roi_perf"].ttp_s == pytest.approx(peak_time, abs=0.5)
|
||
# 画面上没有紫圈。
|
||
assert result.pan_detected_fraction == 0.0
|
||
assert result.tic.reference_valid is False
|
||
# 暗区（囊）被找出来，去暗区曲线比病灶整体亮。
|
||
peak_row = result.tic.rows[round(peak_time * FPS)]
|
||
assert peak_row["dark_area_px"] > 0
|
||
assert peak_row["roi_perf_raw"] > peak_row["roi_all_raw"]
|
||
|
||
restored = unpack_trajectory(pack_trajectory(result.trajectory))
|
||
assert restored.calibration["contrast_side"] == "right"
|
||
curves = unpack_tic_curves(pack_tic_curves(result.curves))
|
||
assert curves.injection_offset_s == pytest.approx(_INJECTION_OFFSET_S)
|
||
|
||
summary = quant_summary(result)
|
||
json.dumps(summary, allow_nan=False)
|
||
assert summary["anchor_frames"]["F1"]["frame_index"] == 150
|
||
assert summary["perfusion_metrics"]["roi_perf"] is not None
|
||
# 没有紫圈：紫圈那一组整组为 null；病灶整体不算指标。
|
||
assert set(summary["perfusion_metrics"]) == {"roi_perf", "roi_pan"}
|
||
assert summary["perfusion_metrics"]["roi_pan"] is None
|
||
|
||
markers = tic_plot_markers(result.anchors)
|
||
assert [marker.label.split()[0] for marker in markers][:3] == ["D1", "D2", "D3"]
|
||
assert render_tic_png(result.tic, markers=markers).startswith(b"\x89PNG")
|
||
assert len(render_case_overlay(prepared, result)) > 0
|
||
images = keyframe_images(prepared, result.anchors)
|
||
assert "F1" in images and "F2" not in images
|
||
assert all(image.shape == (HEIGHT, WIDTH, 3) for image in images.values())
|
||
|
||
|
||
def test_empty_segmentation_fails_case_at_step_4(
|
||
prepared: SegmentationInput, config: CeusConfig
|
||
) -> None:
|
||
empty = np.zeros_like(_lesion_masks(prepared))
|
||
with pytest.raises(NoValidContourError) as raised:
|
||
quantify_case(prepared, empty, _INJECTION_OFFSET_S, config)
|
||
assert raised.value.step == 4
|
||
|
||
|
||
def test_mask_shape_mismatch_is_rejected(prepared: SegmentationInput, config: CeusConfig) -> None:
|
||
masks = _lesion_masks(prepared)[:-1]
|
||
with pytest.raises(ValueError, match="分割掩膜"):
|
||
quantify_case(prepared, masks, _INJECTION_OFFSET_S, config)
|
||
|
||
|
||
def _clean_fit(values: np.ndarray, times: np.ndarray, config: CeusConfig) -> LognormalFitResult:
|
||
"""直接调拟合函数拟一条曲线（基线 40、噪声 0.2，前 20 帧当基线段）。"""
|
||
return fit_lognormal_curve(
|
||
times,
|
||
values,
|
||
baseline=40.0,
|
||
baseline_noise=0.2,
|
||
baseline_end_index=20,
|
||
duplicate_mask=np.zeros(times.size, dtype=bool),
|
||
sample_mask=None,
|
||
cfg=config.tic_analysis.fitting,
|
||
anchor_cfg=config.tic_analysis.anchor_frames,
|
||
)
|
||
|
||
|
||
def test_pan_metrics_computed_and_lesion_gate_unchanged(
|
||
prepared: SegmentationInput, config: CeusConfig, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""紫圈拟合过、病灶整体拟合没过：紫圈与病灶去暗区各算一组指标，病灶整体不算；
|
||
D1–D4 照「病灶两条曲线都过才定位」不定位。
|
||
|
||
合成视频里没有紫圈，所以把第 6–7 步交出的拟合换掉：紫圈换成一条干净 Lognormal 曲线的拟合
|
||
（峰与病灶不同），病灶整体换成一条平线的拟合（没有可信的峰，拟合不过）。
|
||
"""
|
||
times = np.arange(FRAMES) / FPS
|
||
pan_truth = LognormalParameters(
|
||
area=600.0, mu=2.0, sigma=0.5, arrival_offset_s=3.0, baseline=40.0
|
||
)
|
||
pan_fit = _clean_fit(evaluate_lognormal(times, pan_truth), times, config)
|
||
flat_fit = _clean_fit(np.full(times.size, 40.0), times, config)
|
||
assert pan_fit.success is True and flat_fit.success is False
|
||
|
||
def patched(*args: object, **kwargs: object) -> TicResult:
|
||
tic = compute_tic_for_video(*args, **kwargs) # pyright: ignore[reportArgumentType]
|
||
return dataclasses.replace(tic, fits={**tic.fits, "roi_pan": pan_fit, "roi_all": flat_fit})
|
||
|
||
monkeypatch.setattr("cadx.ceus._pipeline.compute_tic_for_video", patched)
|
||
result = quantify_case(prepared, _lesion_masks(prepared), _INJECTION_OFFSET_S, config)
|
||
|
||
assert result.tic.fits["roi_perf"].success is True
|
||
assert result.lesion_fit_ready is False
|
||
for slot in ("B0", "D1", "D2", "D3", "D4"):
|
||
assert result.anchors[slot].status != "ok"
|
||
assert result.anchors[slot].frame_index is None
|
||
assert set(result.perfusion) == {"roi_perf", "roi_pan"}
|
||
assert result.perfusion["roi_pan"] == compute_lognormal_perfusion_metrics(pan_fit)
|
||
|
||
summary = quant_summary(result)
|
||
assert set(summary["perfusion_metrics"]) == {"roi_perf", "roi_pan"}
|
||
expected = compute_lognormal_perfusion_metrics(pan_fit)
|
||
assert expected is not None
|
||
# 11 项指标的键名与 LognormalPerfusionMetrics 字段的对应，照任务书 04 列的指标逐个手写。
|
||
assert summary["perfusion_metrics"]["roi_pan"]["values"] == {
|
||
"PE_au": expected.pe_au,
|
||
"WiAUC_au_s": expected.wi_auc_au_s,
|
||
"RT_s": expected.rt_s,
|
||
"mTTl_s": expected.mttl_s,
|
||
"TTP_s": expected.ttp_s,
|
||
"WiR_au_per_s": expected.wir_au_per_s,
|
||
"WiPI_au": expected.wipi_au,
|
||
"WoAUC_au_s": expected.wo_auc_au_s,
|
||
"WiWoAUC_au_s": expected.wiwo_auc_au_s,
|
||
"FT_s": expected.ft_s,
|
||
"WoR_au_per_s": expected.wor_au_per_s,
|
||
}
|
||
# 紫圈的峰与病灶不同：两组指标不是同一组数。
|
||
assert summary["perfusion_metrics"]["roi_pan"] != summary["perfusion_metrics"]["roi_perf"]
|
||
assert summary["lesion_fit_ready"] is False
|