CADx-4/tests/test_ceus_tic_plot.py

"""TIC 折线图单元测试：缺测不许画成零（照搬旧项目 test_ceus_tic_plot.py）。"""

from typing import Any

import cv2
import numpy as np
import pytest
from matplotlib.axes import Axes

from cadx.ceus import (
    CURVE_NAMES,
    BaselineResult,
    LognormalFitResult,
    TicPlotMarker,
    TicResult,
    render_tic_png,
)

_ROI_ALL_BGR = (184, 99, 37)
_ROI_PERF_BGR = (0, 103, 212)
_ROI_PERF_RAW_BGR = (194, 219, 245)
_ROI_PAN_BGR = (237, 58, 124)
_ROI_PAN_RAW_BGR = (251, 208, 224)
_MANUAL_AT_BGR = (30, 30, 220)

_FRAME_COUNT = 21
# 去暗区曲线中间这一段没采到（NaN），落在时间轴 35%~65% 之间。
_GAP = range(7, 14)


def _invalid_baselines() -> dict[str, BaselineResult]:
    """四条曲线都给一个无效基线——这张图只读 ``rows``，基线怎么样它不关心。"""
    return {
        name: BaselineResult(mean=None, sd=None, frame_start=-1, frame_end=-1, valid=False)
        for name in CURVE_NAMES
    }


def _result(
    *,
    perf_all_nan: bool,
    perf_gap: bool,
    pan_available: bool = False,
    alternate_background: bool = False,
    perf_fit_success: bool | None = None,
    pan_fit_success: bool | None = None,
    pan_fitted_all_nan: bool = False,
    perf_processed_value: float | None = None,
    perf_singleton: bool = False,
) -> TicResult:
    """造一份 ``roi_perf`` 可缺失或断开的 ``TicResult``。"""
    rows: list[dict[str, Any]] = []
    for index in range(_FRAME_COUNT):
        perf = float("nan") if perf_all_nan else 90.0 + 10.0 * (index % 2)
        if perf_singleton:
            perf = 95.0 if index == _FRAME_COUNT // 2 else float("nan")
        if perf_gap and index in _GAP:
            perf = float("nan")
        values = {
            "roi_all": 10.0 * index,
            "roi_perf": perf,
            "roi_pan": 180.0 + index if pan_available else float("nan"),
            "background_circle": (150.0 - index if alternate_background else 40.0 + index),
        }
        row: dict[str, Any] = {"frame_index": index, "time_sec": float(index)}
        for name, value in values.items():
            row[f"{name}_raw"] = value
            row[f"{name}_processed"] = (
                perf_processed_value
                if name == "roi_perf" and perf_processed_value is not None
                else value
            )
        rows.append(row)

    fits: dict[str, LognormalFitResult] = {}
    for name in CURVE_NAMES:
        values = np.asarray([row[f"{name}_processed"] for row in rows], dtype=float)
        success = bool(np.isfinite(values).any())
        if name == "roi_perf" and perf_fit_success is not None:
            success = perf_fit_success
        if name == "roi_pan" and pan_fit_success is not None:
            success = pan_fit_success
        fitted = values.copy()
        if name == "roi_pan" and pan_fitted_all_nan:
            fitted[:] = np.nan
        fits[name] = LognormalFitResult(
            success=success,
            failure_reason="" if success else "no_data",
            algorithm_version="lognormal-v2",
            curve_model="lognormal",
            smoothing_method="",
            baseline_source="observed",
            candidate=values.copy(),
            fitted=fitted,
            processed=values.copy(),
            valid_sample_mask=np.isfinite(values),
            duplicate_sample_mask=np.zeros(_FRAME_COUNT, dtype=bool),
            parameters=None,
            r2=float("nan"),
            nrmse=float("nan"),
            residual_noise_ratio=float("nan"),
            fit_start_s=0.0,
            fit_end_s=float(_FRAME_COUNT - 1),
            multi_peak_flag=False,
            d1_time_s=None,
            d2_time_s=None,
            d3_time_s=None,
            d4_time_s=None,
        )
    return TicResult(
        rows=rows,
        parameters=None,
        baselines=_invalid_baselines(),
        fits=fits,
        early_truncation_flag=False,
        reference_valid=pan_available,
        pan_detected_fraction=1.0 if pan_available else 0.0,
    )


def _decode(png: bytes) -> np.ndarray:
    canvas = cv2.imdecode(np.frombuffer(png, dtype=np.uint8), cv2.IMREAD_COLOR)
    assert canvas is not None, "渲染出来的字节解不回一张图"
    return canvas


def _render(*, perf_all_nan: bool, perf_gap: bool) -> np.ndarray:
    return _decode(
        render_tic_png(_result(perf_all_nan=perf_all_nan, perf_gap=perf_gap), markers=())
    )


def _color_pixels(canvas: np.ndarray, color: tuple[int, int, int]) -> np.ndarray:
    difference = np.abs(canvas.astype(np.int16) - np.asarray(color, dtype=np.int16))
    return np.all(difference <= 8, axis=2)


def test_all_nan_curve_is_not_drawn_at_zero() -> None:
    """``roi_perf`` 整条 NaN 时不画；同一条有数据时画得出来。

    第二句是要害：只断言「没画」的话，一个「什么都不画」的实现照样能过。
    """
    absent = _color_pixels(_render(perf_all_nan=True, perf_gap=False), _ROI_PERF_BGR)
    present = _color_pixels(_render(perf_all_nan=False, perf_gap=False), _ROI_PERF_BGR)

    assert int(absent.sum()) == 0, "整条 NaN 的曲线被画出来了——多半是被当成了 0"
    assert int(present.sum()) > 0, "同一条曲线有数据时反而没画出来"


def test_nan_gap_breaks_the_polyline_instead_of_dipping_to_zero() -> None:
    canvas = _render(perf_all_nan=False, perf_gap=True)
    perf_pixels = _color_pixels(canvas, _ROI_PERF_BGR)
    height, width = canvas.shape[:2]
    inside_gap = perf_pixels[
        int(0.08 * height) : int(0.62 * height),
        int(0.45 * width) : int(0.55 * width),
    ]

    assert int(inside_gap.sum()) == 0, "缺口那一段仍有去暗区曲线像素——折线没有断开"
    assert int(perf_pixels.sum()) > 0, "整条曲线都没画出来——断开断过头了"


def test_curves_with_data_are_still_drawn() -> None:
    """有数据的曲线照画不误（防止「遇 NaN 就整条放弃」）。"""
    canvas = _render(perf_all_nan=True, perf_gap=True)
    assert int(_color_pixels(canvas, _ROI_ALL_BGR).sum()) > 0


def test_isolated_raw_sample_is_visible() -> None:
    result = _result(
        perf_all_nan=True,
        perf_gap=False,
        perf_fit_success=False,
        perf_singleton=True,
    )
    canvas = _decode(render_tic_png(result, markers=()))
    height, width = canvas.shape[:2]
    plot_center = _color_pixels(canvas, _ROI_PERF_RAW_BGR)[
        int(0.2 * height) : int(0.65 * height),
        int(0.45 * width) : int(0.55 * width),
    ]

    assert int(plot_center.sum()) > 0


def test_rejected_fit_keeps_raw_but_hides_processed_curve() -> None:
    result = _result(
        perf_all_nan=False,
        perf_gap=False,
        perf_fit_success=False,
        perf_processed_value=170.0,
    )
    canvas = _decode(render_tic_png(result, markers=()))

    assert int(_color_pixels(canvas, _ROI_PERF_RAW_BGR).sum()) > 0
    assert int(_color_pixels(canvas, _ROI_PERF_BGR).sum()) == 0


def test_pan_reference_raw_and_fit_are_drawn_when_available() -> None:
    result = _result(
        perf_all_nan=False,
        perf_gap=False,
        pan_available=True,
    )
    canvas = _decode(render_tic_png(result, markers=()))

    assert int(_color_pixels(canvas, _ROI_PAN_RAW_BGR).sum()) > 0
    assert int(_color_pixels(canvas, _ROI_PAN_BGR).sum()) > 0


def test_missing_pan_reference_is_not_drawn_at_zero() -> None:
    canvas = _decode(render_tic_png(_result(perf_all_nan=False, perf_gap=False), markers=()))

    assert int(_color_pixels(canvas, _ROI_PAN_RAW_BGR).sum()) == 0
    assert int(_color_pixels(canvas, _ROI_PAN_BGR).sum()) == 0


def test_rejected_pan_fit_draws_diagnostic_fitted_curve() -> None:
    result = _result(
        perf_all_nan=False,
        perf_gap=False,
        pan_available=True,
        pan_fit_success=False,
    )
    canvas = _decode(render_tic_png(result, markers=()))

    assert int(_color_pixels(canvas, _ROI_PAN_RAW_BGR).sum()) > 0
    assert int(_color_pixels(canvas, _ROI_PAN_BGR).sum()) > 0


def test_rejected_pan_fit_is_dashed_and_labeled(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    plotted_styles: list[dict[str, Any]] = []
    original_plot = Axes.plot  # pyright: ignore[reportUnknownVariableType, reportUnknownMemberType]

    def capture_plot(self: Axes, *args: Any, **kwargs: Any) -> Any:
        plotted_styles.append(kwargs)
        return original_plot(self, *args, **kwargs)

    monkeypatch.setattr(Axes, "plot", capture_plot)
    render_tic_png(
        _result(
            perf_all_nan=False,
            perf_gap=False,
            pan_available=True,
            pan_fit_success=False,
        ),
        markers=(),
    )

    reference_fit = next(
        style
        for style in plotted_styles
        if style.get("label") == "roi_pan reference Lognormal fit [QC rejected]"
    )
    assert reference_fit["linestyle"] == "--"


def test_rejected_pan_fit_without_finite_model_does_not_fabricate_line() -> None:
    result = _result(
        perf_all_nan=False,
        perf_gap=False,
        pan_available=True,
        pan_fit_success=False,
        pan_fitted_all_nan=True,
    )
    canvas = _decode(render_tic_png(result, markers=()))

    assert int(_color_pixels(canvas, _ROI_PAN_RAW_BGR).sum()) > 0
    assert int(_color_pixels(canvas, _ROI_PAN_BGR).sum()) == 0


def test_background_does_not_change_tic_plot() -> None:
    regular = _result(perf_all_nan=False, perf_gap=False)
    changed = _result(
        perf_all_nan=False,
        perf_gap=False,
        alternate_background=True,
    )
    regular_canvas = _decode(render_tic_png(regular, markers=()))
    changed_canvas = _decode(render_tic_png(changed, markers=()))

    np.testing.assert_array_equal(regular_canvas, changed_canvas)


def test_plot_uses_review_ready_canvas_size() -> None:
    canvas = _render(perf_all_nan=False, perf_gap=False)

    assert canvas.shape == (800, 1600, 3)


def test_review_marker_is_drawn_without_changing_curve_data() -> None:
    result = _result(perf_all_nan=False, perf_gap=False)
    png = render_tic_png(
        result,
        markers=(TicPlotMarker("manual AT", 10.0, _MANUAL_AT_BGR),),
    )
    canvas = _decode(png)

    assert int(_color_pixels(canvas, _MANUAL_AT_BGR).sum()) > 0
    assert int(_color_pixels(canvas, _ROI_PERF_BGR).sum()) > 0


def test_marker_outside_observed_time_is_not_attached_to_boundary() -> None:
    result = _result(perf_all_nan=False, perf_gap=False)
    png = render_tic_png(
        result,
        markers=(TicPlotMarker("manual AT", 100.0, _MANUAL_AT_BGR),),
    )
    canvas = _decode(png)

    assert int(_color_pixels(canvas, _MANUAL_AT_BGR).sum()) == 0