"""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"}))