"""约束 Lognormal TIC 拟合的纯算法单元测试（照搬旧项目 test_tic_fitting.py）。""" import numpy as np import pytest from cadx.ceus import ( CeusConfig, LognormalParameters, TicFittingConfig, evaluate_lognormal, fit_lognormal_curve, ) from cadx.ceus._tic_fitting import _select_primary_peak # pyright: ignore[reportPrivateUsage] from conftest import load_ceus_section # 配置不设默认值：旧测试靠默认值构造的配置，改为在 config/ceus.yaml 的取值上 model_copy。 _CONFIG = CeusConfig.model_validate(load_ceus_section()) def _fit( times: np.ndarray, values: np.ndarray, *, duplicates: np.ndarray | None = None, baseline: float | None = 30.0, noise: float | None = 0.2, cfg: TicFittingConfig | None = None, ): return fit_lognormal_curve( times, values, baseline=baseline, baseline_noise=noise, baseline_end_index=20, duplicate_mask=(np.zeros(times.size, dtype=bool) if duplicates is None else duplicates), sample_mask=None, cfg=cfg or _CONFIG.tic_analysis.fitting, anchor_cfg=_CONFIG.tic_analysis.anchor_frames, ) def test_recovers_clean_lognormal_and_continuous_anchors() -> None: times = np.linspace(0.0, 60.0, 601) truth = LognormalParameters( area=400.0, mu=2.2, sigma=0.5, arrival_offset_s=8.0, baseline=30.0, ) result = _fit(times, evaluate_lognormal(times, truth)) assert result.success is True assert result.parameters is not None assert result.r2 == pytest.approx(1.0, abs=1e-8) assert result.parameters.arrival_offset_s == pytest.approx(8.0, abs=0.05) assert result.parameters.sigma == pytest.approx(0.5, abs=0.02) assert result.d1_time_s is not None assert result.d1_time_s < result.d2_time_s < result.d3_time_s < result.d4_time_s # pyright: ignore[reportOperatorIssue] np.testing.assert_allclose(result.processed, result.fitted) def test_soft_l1_fit_survives_single_compression_spike() -> None: times = np.linspace(0.0, 60.0, 601) truth = LognormalParameters(400.0, 2.2, 0.5, 8.0, 30.0) values = evaluate_lognormal(times, truth) values[180] += 80.0 result = _fit(times, values) assert result.success is True assert result.parameters is not None assert result.d3_time_s == pytest.approx( truth.arrival_offset_s + np.exp(truth.mu - truth.sigma**2), abs=0.25 ) def test_duplicate_samples_are_excluded_but_keep_original_mapping() -> None: times = np.linspace(0.0, 60.0, 601) truth = LognormalParameters(400.0, 2.2, 0.5, 8.0, 30.0) values = evaluate_lognormal(times, truth) duplicates = np.zeros(times.size, dtype=bool) duplicates[100:102] = True result = _fit(times, values, duplicates=duplicates) assert result.success is True np.testing.assert_array_equal(result.duplicate_sample_mask, duplicates) assert not result.valid_sample_mask[100:102].any() assert result.processed.shape == values.shape def test_missing_observed_baseline_estimates_prior_but_rejects_flat_curve() -> None: times = np.linspace(0.0, 60.0, 601) values = np.full(times.size, 30.0) result = _fit(times, values, baseline=None, noise=None) assert result.success is False assert result.failure_reason == "credible_peak_not_found" assert result.baseline_source == "estimated" assert np.isnan(result.processed).all() assert np.isnan(result.fitted).all() def test_left_truncated_curve_uses_estimated_baseline_and_still_fits() -> None: times = np.linspace(0.0, 50.0, 501) truth = LognormalParameters(700.0, 2.7, 0.65, -4.0, 25.0) result = _fit( times, evaluate_lognormal(times, truth), baseline=None, noise=None, ) assert result.success is True assert result.baseline_source == "estimated" assert result.parameters is not None assert result.d3_time_s is not None def test_explicit_sample_mask_excludes_unstable_perf_frame() -> None: times = np.linspace(0.0, 60.0, 601) truth = LognormalParameters(400.0, 2.2, 0.5, 8.0, 30.0) values = evaluate_lognormal(times, truth) values[180] += 100.0 sample_mask = np.ones(times.size, dtype=bool) sample_mask[180] = False result = fit_lognormal_curve( times, values, baseline=30.0, baseline_noise=0.2, baseline_end_index=20, duplicate_mask=np.zeros(times.size, dtype=bool), sample_mask=sample_mask, cfg=_CONFIG.tic_analysis.fitting, anchor_cfg=_CONFIG.tic_analysis.anchor_frames, ) assert result.success is True assert result.valid_sample_mask[180] is np.False_ def test_quality_failure_keeps_rejected_fit_separate_from_processed() -> None: rng = np.random.default_rng(9) times = np.linspace(0.0, 60.0, 601) truth = LognormalParameters(160.0, 2.0, 0.55, 8.0, 30.0) values = evaluate_lognormal(times, truth) + rng.normal(0.0, 18.0, times.size) cfg = _CONFIG.tic_analysis.fitting.model_copy(update={"minimum_r2": 0.95, "maximum_nrmse": 0.1}) result = _fit(times, values, noise=1.0, cfg=cfg) assert result.success is False assert result.failure_reason in { "fit_quality_failed", "credible_peak_not_found", "model_peak_on_boundary", "parameter_on_boundary", } assert np.isnan(result.processed).all() if result.failure_reason in { "fit_quality_failed", "model_peak_on_boundary", "parameter_on_boundary", }: assert result.parameters is not None assert np.isfinite(result.fitted).all() assert np.isfinite(result.r2) assert np.isfinite(result.nrmse) def test_primary_peak_uses_local_finite_run_and_rejects_single_sample_spike() -> None: grid = np.arange(12.0) candidate = np.asarray([0.0, 0.0, 10.0, 0.0, np.nan, np.nan, 0.0, 2.5, 5.0, 2.5, 0.0, 0.0]) selected = _select_primary_peak([2, 8], grid, candidate, baseline=0.0) assert selected == 8 def test_two_boluses_mark_multi_peak_and_fit_first_pass() -> None: times = np.linspace(0.0, 70.0, 701) first = evaluate_lognormal(times, LognormalParameters(300.0, 2.0, 0.4, 5.0, 30.0)) second = evaluate_lognormal(times, LognormalParameters(260.0, 1.8, 0.4, 28.0, 0.0)) result = _fit(times, first + second) assert result.multi_peak_flag is True assert result.fit_end_s < 34.0 def test_repeated_shallow_motion_peaks_do_not_truncate_first_bolus() -> None: times = np.linspace(0.0, 70.0, 701) truth = LognormalParameters(400.0, 2.4, 0.75, 5.0, 30.0) broad_bolus = evaluate_lognormal(times, truth) motion = np.where(times > 8.0, 2.0 * np.sin(times * 2.5), 0.0) result = _fit(times, broad_bolus + motion) assert result.multi_peak_flag is False assert result.fit_end_s == pytest.approx(times[-1]) assert result.success is True def test_truncated_washout_does_not_use_fit_boundary_as_d4() -> None: full_times = np.linspace(0.0, 60.0, 601) truth = LognormalParameters(400.0, 2.2, 0.5, 8.0, 30.0) full_values = evaluate_lognormal(full_times, truth) keep = full_times <= 17.0 result = _fit(full_times[keep], full_values[keep]) assert result.success is True assert result.d3_time_s is not None assert result.d4_time_s is None def test_narrow_bolus_keeps_analytic_d4_in_long_video() -> None: times = np.linspace(0.0, 90.0, 901) truth = LognormalParameters(400.0, 2.2, 0.15, 8.0, 30.0) result = _fit(times, evaluate_lognormal(times, truth)) sigma_squared = truth.sigma**2 wash_out_factor = (sigma_squared - np.sqrt(sigma_squared**2 + 4.0 * sigma_squared)) / 2.0 expected_d4 = truth.arrival_offset_s + np.exp(truth.mu - sigma_squared - wash_out_factor) assert result.success is True assert result.d4_time_s == pytest.approx(expected_d4, abs=0.05)