iomgaa
150eb5f6ae
- ceus 迁入 polygons、dark_region、reference_region、tic_fitting、tic、anchor_frames、 perfusion_metrics、trajectory、tic_curves、tic_plot、overlay，旧测试照搬（改 import 与配置读取）。 - 对外两段接口：prepare_segmentation（第 1–2 步）与 quantify_case（第 4–9 步）。 - 第 3 条：删掉 D1 取医生标注时刻的模式，D1 只从拟合曲线算。 - 第 4 条：F1/F2 取视频第 20−X、60−X 秒（X 为视频第 0 秒时已推注的秒数），落在视频外记未定位。 - 绿框融合权重、紫圈颜色阈值照搬；关键帧容差窗口的出处写进 ceus.yaml 注释。 - segment：同一进程调用 MedSAM2，设置写进 ceus.yaml 的 segment 段，全部必填；gpu 测试真跑通过。 - 依赖加 scipy、matplotlib、medsam2（可编辑安装，不编译 CUDA 扩展）；uv 只解析 linux x86_64。 - 导入契约：只有 segment 引用 torch、sam2；segment 不引用其他 cadx 模块。 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
421 lines
16 KiB
Python
421 lines
16 KiB
Python
"""TIC 曲线 npz 打包 / 解包（tic_curves）单元测试（照搬旧项目 test_ceus_tic_curves.py）。
|
||
|
||
钉死曲线工件的单一 schema：完整往返还原全部键、存档内各键 dtype / shape 精确、
|
||
NaN 原样往返（**不许被填成 0**——那会把「这一帧没采到」变成「这一帧亮度是 0」）、
|
||
基线无效时三个基线键的编码（NaN / NaN / (-1,-1) / 0），以及防御：坏字节解包 raise、
|
||
长度不一致 pack raise、禁 pickle（含 object pickle 的恶意 npz 解包被拒）。
|
||
另有一条从 ``TicResult`` 构造的往返，确认 ``from_tic_result`` 认得行键。
|
||
结构照 ``tests/unit/test_ceus_trajectory.py``。
|
||
"""
|
||
|
||
import io
|
||
import pickle
|
||
from typing import Any
|
||
|
||
import numpy as np
|
||
import pytest
|
||
|
||
from cadx.ceus import (
|
||
CURVE_NAMES,
|
||
CURVES_SCHEMA_VERSION,
|
||
CeusConfig,
|
||
TicCurves,
|
||
TicCurveSeries,
|
||
compute_tic_for_video,
|
||
from_tic_result,
|
||
pack_tic_curves,
|
||
unpack_tic_curves,
|
||
)
|
||
from conftest import load_ceus_section
|
||
|
||
# 配置不设默认值：旧测试靠默认值构造的配置，改为读 config/ceus.yaml 的取值。
|
||
_CONFIG = CeusConfig.model_validate(load_ceus_section())
|
||
|
||
|
||
_T = 5
|
||
|
||
|
||
# Phase 1: 构造一个真实形态的 TicCurves（合成数值）
|
||
|
||
|
||
def _series(offset: float, *, valid: bool) -> TicCurveSeries:
|
||
"""一条曲线：中间插一个 NaN（该帧没采到），基线按 valid 给或不给。"""
|
||
raw = np.asarray([10.0, 20.0, np.nan, 40.0, 50.0], dtype=np.float32) + offset
|
||
processed = np.asarray([11.0, 21.0, 31.0, 41.0, 51.0], dtype=np.float32) + offset
|
||
common: dict[str, Any] = {
|
||
"candidate": processed.copy(),
|
||
"fitted": processed.copy(),
|
||
"processed": processed.copy() if valid else np.full(_T, np.nan, np.float32),
|
||
"valid_sample_mask": np.isfinite(raw),
|
||
"duplicate_sample_mask": np.asarray([False, False, True, False, False]),
|
||
"fit_success": valid,
|
||
"fit_failure_reason": "" if valid else "baseline_invalid",
|
||
"curve_model": "lognormal",
|
||
"smoothing_method": "",
|
||
"baseline_source": "observed" if valid else "unavailable",
|
||
"fit_area": 123.0 if valid else float("nan"),
|
||
"fit_mu": 2.0 if valid else float("nan"),
|
||
"fit_sigma": 0.5 if valid else float("nan"),
|
||
"fit_arrival_offset_s": 1.0 if valid else float("nan"),
|
||
"fit_baseline": 14.5 + offset if valid else float("nan"),
|
||
"fit_r2": 0.9 if valid else float("nan"),
|
||
"fit_nrmse": 0.1 if valid else float("nan"),
|
||
"residual_noise_ratio": 1.2 if valid else float("nan"),
|
||
"fit_interval_s": (0.0, 4.0) if valid else (float("nan"), float("nan")),
|
||
"multi_peak_flag": False,
|
||
"anchor_times_s": (1.0, 2.0, 3.0, 4.0) if valid else (float("nan"),) * 4,
|
||
}
|
||
if valid:
|
||
return TicCurveSeries(
|
||
raw=raw,
|
||
baseline_mean=15.0 + offset,
|
||
baseline_sd=2.5,
|
||
baseline_frames=(0, 2),
|
||
baseline_valid=True,
|
||
**common,
|
||
)
|
||
return TicCurveSeries(
|
||
raw=raw,
|
||
baseline_mean=float("nan"),
|
||
baseline_sd=float("nan"),
|
||
baseline_frames=(-1, -1),
|
||
baseline_valid=False,
|
||
**common,
|
||
)
|
||
|
||
|
||
def _make_curves(*, injection_offset_s: float = 3.5) -> TicCurves:
|
||
counts = {
|
||
key: np.arange(_T, dtype=np.int32) + index
|
||
for index, key in enumerate(
|
||
(
|
||
"roi_area_px",
|
||
"dark_area_px",
|
||
"pan_area_px",
|
||
"background_circle_area_px",
|
||
"valid_roi_all_px",
|
||
"valid_roi_perf_px",
|
||
"valid_pan_px",
|
||
"valid_background_circle_px",
|
||
"dark_component_count",
|
||
)
|
||
)
|
||
}
|
||
frame_indices = np.arange(_T, dtype=np.int32)
|
||
return TicCurves(
|
||
frame_indices=frame_indices,
|
||
time_video_s=(frame_indices / np.float32(10.0)).astype(np.float32),
|
||
injection_offset_s=injection_offset_s,
|
||
time_origin="injection",
|
||
fps=10.0,
|
||
config_version="test-v1",
|
||
algorithm_version="lognormal-v2",
|
||
curves_schema_version=CURVES_SCHEMA_VERSION,
|
||
curves={
|
||
# roi_pan 那条给成基线无效，覆盖「紫色没检出」那一路的编码。
|
||
name: _series(float(index) * 100.0, valid=(name != "roi_pan"))
|
||
for index, name in enumerate(CURVE_NAMES)
|
||
},
|
||
perf_mask_stable=np.ones(_T, dtype=bool),
|
||
reference_valid=False,
|
||
early_truncation_flag=True,
|
||
**counts,
|
||
)
|
||
|
||
|
||
# Phase 2: 完整往返 —— 全部键还原
|
||
|
||
|
||
def test_roundtrip_preserves_all_keys() -> None:
|
||
curves = _make_curves()
|
||
restored = unpack_tic_curves(pack_tic_curves(curves))
|
||
|
||
np.testing.assert_array_equal(restored.frame_indices, curves.frame_indices)
|
||
np.testing.assert_allclose(restored.time_video_s, curves.time_video_s)
|
||
assert restored.injection_offset_s == pytest.approx(curves.injection_offset_s)
|
||
assert restored.time_origin == curves.time_origin
|
||
assert restored.fps == pytest.approx(curves.fps)
|
||
assert restored.curves_schema_version == curves.curves_schema_version
|
||
assert restored.reference_valid is curves.reference_valid
|
||
assert restored.early_truncation_flag is curves.early_truncation_flag
|
||
np.testing.assert_array_equal(restored.perf_mask_stable, curves.perf_mask_stable)
|
||
|
||
for key in (
|
||
"roi_area_px",
|
||
"dark_area_px",
|
||
"pan_area_px",
|
||
"background_circle_area_px",
|
||
"valid_roi_all_px",
|
||
"valid_roi_perf_px",
|
||
"valid_pan_px",
|
||
"valid_background_circle_px",
|
||
"dark_component_count",
|
||
):
|
||
np.testing.assert_array_equal(getattr(restored, key), getattr(curves, key))
|
||
|
||
assert set(restored.curves) == set(CURVE_NAMES)
|
||
for name in CURVE_NAMES:
|
||
original = curves.curves[name]
|
||
back = restored.curves[name]
|
||
np.testing.assert_allclose(back.raw, original.raw, equal_nan=True)
|
||
np.testing.assert_allclose(back.candidate, original.candidate, equal_nan=True)
|
||
np.testing.assert_allclose(back.fitted, original.fitted, equal_nan=True)
|
||
np.testing.assert_allclose(back.processed, original.processed, equal_nan=True)
|
||
np.testing.assert_array_equal(back.valid_sample_mask, original.valid_sample_mask)
|
||
assert back.baseline_frames == original.baseline_frames
|
||
assert back.baseline_valid is original.baseline_valid
|
||
assert back.baseline_source == original.baseline_source
|
||
assert back.fit_baseline == pytest.approx(original.fit_baseline, nan_ok=True)
|
||
|
||
|
||
def test_roundtrip_keeps_nan_as_nan() -> None:
|
||
"""NaN 原样往返：既不被填成 0，也不被插值掉。"""
|
||
restored = unpack_tic_curves(pack_tic_curves(_make_curves()))
|
||
for name in CURVE_NAMES:
|
||
assert np.isnan(restored.curves[name].raw[2])
|
||
|
||
|
||
def test_invalid_baseline_roundtrips_as_nan_and_minus_one() -> None:
|
||
"""基线无效那条：均值 / 标准差是 NaN，帧区间 ``(-1, -1)``，valid 为 False。
|
||
|
||
``0.0`` 是一个能一路算下去、把下游全部算错的合法数值；NaN 不是。
|
||
"""
|
||
restored = unpack_tic_curves(pack_tic_curves(_make_curves()))
|
||
pan = restored.curves["roi_pan"]
|
||
assert np.isnan(pan.baseline_mean)
|
||
assert np.isnan(pan.baseline_sd)
|
||
assert pan.baseline_frames == (-1, -1)
|
||
assert pan.baseline_valid is False
|
||
|
||
|
||
# Phase 3: 存档层 schema —— 直接在 npz 原始存档上断言键名 / dtype / shape
|
||
|
||
|
||
def test_archive_schema_keys_dtypes_shapes() -> None:
|
||
"""打开 pack 产出的原始 npz 存档，逐一断言键集 / 存储 dtype / shape。
|
||
|
||
与对象层往返互补：往返只验 ``unpack(pack(x)) == x``，发现不了 pack 写入存档时的
|
||
dtype / shape 漂移。
|
||
"""
|
||
archive_bytes = pack_tic_curves(_make_curves())
|
||
with np.load(io.BytesIO(archive_bytes), allow_pickle=False) as archive:
|
||
expected_keys = {
|
||
"frame_indices",
|
||
"time_video_s",
|
||
"injection_offset_s",
|
||
"time_origin",
|
||
"fps",
|
||
"config_version",
|
||
"algorithm_version",
|
||
"curves_schema_version",
|
||
"reference_valid",
|
||
"early_truncation_flag",
|
||
"perf_mask_stable",
|
||
"roi_area_px",
|
||
"dark_area_px",
|
||
"pan_area_px",
|
||
"background_circle_area_px",
|
||
"valid_roi_all_px",
|
||
"valid_roi_perf_px",
|
||
"valid_pan_px",
|
||
"valid_background_circle_px",
|
||
"dark_component_count",
|
||
}
|
||
for name in CURVE_NAMES:
|
||
expected_keys |= {
|
||
f"{name}_raw",
|
||
f"{name}_candidate",
|
||
f"{name}_fitted",
|
||
f"{name}_processed",
|
||
f"{name}_valid_sample_mask",
|
||
f"{name}_duplicate_sample_mask",
|
||
f"{name}_baseline_mean",
|
||
f"{name}_baseline_sd",
|
||
f"{name}_baseline_frames",
|
||
f"{name}_baseline_valid",
|
||
f"{name}_fit_success",
|
||
f"{name}_fit_failure_reason",
|
||
f"{name}_curve_model",
|
||
f"{name}_smoothing_method",
|
||
f"{name}_baseline_source",
|
||
f"{name}_fit_parameters",
|
||
f"{name}_fit_quality",
|
||
f"{name}_fit_interval_s",
|
||
f"{name}_multi_peak_flag",
|
||
f"{name}_anchor_times_s",
|
||
}
|
||
assert set(archive.files) == expected_keys
|
||
|
||
assert archive["frame_indices"].dtype == np.int32
|
||
assert archive["frame_indices"].shape == (_T,)
|
||
assert archive["time_video_s"].dtype == np.float32
|
||
assert archive["time_video_s"].shape == (_T,)
|
||
assert archive["injection_offset_s"].dtype == np.float32
|
||
assert archive["injection_offset_s"].shape == (1,)
|
||
assert archive["fps"].dtype == np.float32
|
||
assert archive["fps"].shape == (1,)
|
||
assert archive["reference_valid"].dtype == np.int8
|
||
assert archive["early_truncation_flag"].dtype == np.int8
|
||
for key in ("roi_area_px", "pan_area_px", "dark_component_count"):
|
||
assert archive[key].dtype == np.int32
|
||
assert archive[key].shape == (_T,)
|
||
for name in CURVE_NAMES:
|
||
assert archive[f"{name}_raw"].dtype == np.float32
|
||
assert archive[f"{name}_raw"].shape == (_T,)
|
||
assert archive[f"{name}_candidate"].dtype == np.float32
|
||
assert archive[f"{name}_fitted"].dtype == np.float32
|
||
assert archive[f"{name}_processed"].dtype == np.float32
|
||
assert archive[f"{name}_baseline_mean"].dtype == np.float32
|
||
assert archive[f"{name}_baseline_mean"].shape == (1,)
|
||
assert archive[f"{name}_baseline_sd"].shape == (1,)
|
||
assert archive[f"{name}_baseline_frames"].dtype == np.int32
|
||
assert archive[f"{name}_baseline_frames"].shape == (2,)
|
||
assert archive[f"{name}_baseline_valid"].dtype == np.int8
|
||
assert archive[f"{name}_fit_parameters"].dtype == np.float64
|
||
assert archive[f"{name}_fit_parameters"].shape == (5,)
|
||
|
||
|
||
# Phase 4: 不存的那三样确实不在文件里
|
||
|
||
|
||
def test_derivable_quantities_are_not_stored() -> None:
|
||
"""``S(t)``、``t_injection``、无灌注比例都不进文件（各自的理由见 TicCurves docstring）。"""
|
||
with np.load(io.BytesIO(pack_tic_curves(_make_curves())), allow_pickle=False) as a:
|
||
keys = set(a.files)
|
||
assert not {key for key in keys if "baseline_subtracted" in key}
|
||
assert "time_injection_s" not in keys
|
||
assert not {key for key in keys if "non_perfused" in key}
|
||
|
||
|
||
# Phase 5: 从 TicResult 构造
|
||
|
||
|
||
def _tic_result():
|
||
"""跑一次真实的 ``compute_tic_for_video``（合成帧），拿一个 TicResult。"""
|
||
height, width, frame_count = 60, 80, 4
|
||
roi_polygon = np.array(
|
||
[[30.0, 10.0], [70.0, 10.0], [70.0, 40.0], [30.0, 40.0]], dtype=np.float32
|
||
)
|
||
frames: list[np.ndarray] = []
|
||
for _ in range(frame_count):
|
||
frame = np.zeros((height, width, 3), dtype=np.uint8)
|
||
frame[10:40, 30:70] = 100
|
||
frame[10:30, 5:25] = 50
|
||
frames.append(frame)
|
||
background = np.zeros((height, width), dtype=bool)
|
||
background[12:28, 8:22] = True
|
||
return compute_tic_for_video(
|
||
source_frames=frames,
|
||
roi_polygons=[roi_polygon for _ in range(frame_count)],
|
||
dark_polygons_per_frame=[[] for _ in range(frame_count)],
|
||
reference_mask=background,
|
||
fps=4.0,
|
||
prompt_cfg=_CONFIG.prompt_source,
|
||
pan_cfg=_CONFIG.pan_source,
|
||
tic_cfg=_CONFIG.tic_analysis,
|
||
frame_times_s=None,
|
||
)
|
||
|
||
|
||
def test_from_tic_result_roundtrips() -> None:
|
||
"""``from_tic_result`` 认得行键，产物能原样打包再解回来。"""
|
||
result = _tic_result()
|
||
curves = from_tic_result(result, fps=4.0, injection_offset_s=1.25, config_version="test-v1")
|
||
restored = unpack_tic_curves(pack_tic_curves(curves))
|
||
|
||
assert restored.time_origin == "injection"
|
||
assert restored.injection_offset_s == pytest.approx(1.25)
|
||
assert restored.frame_indices.tolist() == [0, 1, 2, 3]
|
||
np.testing.assert_allclose(restored.time_video_s, [0.0, 0.25, 0.5, 0.75], atol=1e-6)
|
||
assert set(restored.curves) == set(CURVE_NAMES)
|
||
assert restored.reference_valid is result.reference_valid
|
||
np.testing.assert_allclose(
|
||
restored.curves["roi_all"].raw,
|
||
[row["roi_all_raw"] for row in result.rows],
|
||
equal_nan=True,
|
||
)
|
||
|
||
|
||
def test_from_tic_result_without_injection_offset_writes_nan_not_zero() -> None:
|
||
"""推注偏移取不到时写 NaN、时间零点记「视频起点」——**不填 0**。
|
||
|
||
填 0 等于断言「视频和推注同时开始」，而那是我们没有的信息（design/0030 决策三）。
|
||
"""
|
||
curves = from_tic_result(
|
||
_tic_result(), fps=4.0, injection_offset_s=None, config_version="test-v1"
|
||
)
|
||
restored = unpack_tic_curves(pack_tic_curves(curves))
|
||
assert np.isnan(restored.injection_offset_s)
|
||
assert restored.time_origin == "video_start"
|
||
|
||
|
||
def test_from_tic_result_rejects_non_positive_fps() -> None:
|
||
with pytest.raises(ValueError):
|
||
from_tic_result(
|
||
_tic_result(),
|
||
fps=0.0,
|
||
injection_offset_s=None,
|
||
config_version="test-v1",
|
||
)
|
||
|
||
|
||
# Phase 6: 防御 —— 坏字节 / 长度不一致 / 禁 pickle
|
||
|
||
|
||
def test_unpack_bad_bytes_raises() -> None:
|
||
with pytest.raises((ValueError, OSError)):
|
||
unpack_tic_curves(b"this-is-not-a-valid-npz-archive")
|
||
|
||
|
||
def test_unpack_rejects_old_schema_before_reading_new_keys() -> None:
|
||
packed = pack_tic_curves(_make_curves())
|
||
with np.load(io.BytesIO(packed), allow_pickle=False) as archive:
|
||
payload = {key: archive[key] for key in archive.files}
|
||
payload["curves_schema_version"] = np.array("1")
|
||
buffer = io.BytesIO()
|
||
np.savez_compressed(buffer, **payload)
|
||
|
||
with pytest.raises(ValueError, match="只支持 curves schema 3"):
|
||
unpack_tic_curves(buffer.getvalue())
|
||
|
||
|
||
def test_pack_rejects_non_monotonic_time_axis() -> None:
|
||
curves = _make_curves()
|
||
object.__setattr__(
|
||
curves,
|
||
"time_video_s",
|
||
np.asarray([0.0, 0.1, 0.1, 0.3, 0.4], dtype=np.float32),
|
||
)
|
||
with pytest.raises(ValueError, match="time_video_s"):
|
||
pack_tic_curves(curves)
|
||
|
||
|
||
def test_pack_rejects_mismatched_column_length() -> None:
|
||
curves = _make_curves()
|
||
object.__setattr__(curves, "roi_area_px", np.arange(_T + 2, dtype=np.int32))
|
||
with pytest.raises(ValueError):
|
||
pack_tic_curves(curves)
|
||
|
||
|
||
def test_pack_rejects_mismatched_curve_length() -> None:
|
||
curves = _make_curves()
|
||
short = _series(0.0, valid=False)
|
||
object.__setattr__(short, "raw", np.zeros(2, dtype=np.float32))
|
||
curves.curves["roi_all"] = short
|
||
with pytest.raises(ValueError):
|
||
pack_tic_curves(curves)
|
||
|
||
|
||
def test_unpack_rejects_pickled_object_payload() -> None:
|
||
"""构造真实含 object pickle 的恶意 .npz，解包须被拒（``allow_pickle=False``）。"""
|
||
buffer = io.BytesIO()
|
||
np.savez_compressed(buffer, roi_all_raw=np.array([{"evil": 1}], dtype=object))
|
||
with pytest.raises((ValueError, OSError, KeyError)):
|
||
unpack_tic_curves(buffer.getvalue())
|
||
|
||
|
||
def test_pickle_payload_cannot_smuggle_through_unpack() -> None:
|
||
"""即便 pickle 字节伪装，unpack 内部 ``allow_pickle=False`` 也不得执行反序列化。"""
|
||
with pytest.raises((ValueError, OSError, KeyError)):
|
||
unpack_tic_curves(pickle.dumps({"evil": "code"}))
|