noaa-goes-visualization/tests/test_metrics.py

import numpy as np
import pytest

from conftest import solar_disc
from suvi import detectors, metrics

BAND = 171


def verdict(kind, reason=None):
    return detectors.Verdict(kind, reason, {})


# ------------------------------------------------------------------------ display


def test_display_mapping_is_bounded():
    image = solar_disc(size=64, radius=20, peak=50.0)
    shown = metrics.to_display(image, BAND)
    assert shown.min() >= 0.0 and shown.max() <= 1.0


def test_display_mapping_is_monotonic():
    ramp = np.linspace(0, 20, 100).reshape(10, 10)
    shown = metrics.to_display(ramp, BAND)
    assert np.all(np.diff(shown.ravel()) >= -1e-12)


def test_display_mapping_differs_per_band():
    ramp = np.full((4, 4), 5.0)
    assert not np.allclose(metrics.to_display(ramp, 94), metrics.to_display(ramp, 304))


def test_display_mapping_handles_nan():
    assert np.isfinite(metrics.to_display(np.full((4, 4), np.nan), BAND)).all()


# ---------------------------------------------------------------------- detection


def test_scores_a_perfect_detector():
    injected = {(16, BAND, 0): "eclipse_dim", (16, BAND, 240): "truncate"}
    verdicts = {
        (16, BAND, 0): verdict("bad"),
        (16, BAND, 240): verdict("bad"),
        (16, BAND, 480): verdict("good"),
        (16, BAND, 720): verdict("good"),
    }
    score = metrics.score_detection(verdicts, injected)
    assert (score.true_positives, score.false_negatives) == (2, 0)
    assert (score.false_positives, score.true_negatives) == (0, 2)
    assert score.precision == 1.0 and score.recall == 1.0 and score.f1 == 1.0
    assert score.false_positive_rate == 0.0


def test_scores_a_detector_that_flags_everything():
    injected = {(16, BAND, 0): "eclipse_dim"}
    verdicts = {(16, BAND, t): verdict("bad") for t in (0, 240, 480, 720)}
    score = metrics.score_detection(verdicts, injected)
    assert score.recall == 1.0
    assert score.precision == pytest.approx(0.25)
    assert score.false_positive_rate == 1.0


def test_unknown_verdicts_are_excluded_not_counted_as_good():
    """Abstaining is neither a hit nor a miss, and must not flatter the score."""
    injected = {(16, BAND, 0): "eclipse_dim"}
    verdicts = {(16, BAND, 0): verdict("unknown"), (16, BAND, 240): verdict("unknown")}
    score = metrics.score_detection(verdicts, injected)
    assert score.unknown == 2
    assert score.true_positives == score.false_positives == 0
    assert score.false_negatives == score.true_negatives == 0
    assert np.isnan(score.precision) and np.isnan(score.recall)


def test_legacy_disagreements_are_reported_separately():
    """Flagging a frame the old filter passed may be a find, not an error."""
    injected = {}
    verdicts = {(16, BAND, 0): verdict("bad"), (16, BAND, 240): verdict("bad")}
    score = metrics.score_detection(verdicts, injected, legacy_good={(16, BAND, 0)})
    assert score.false_positives == 2
    assert score.legacy_disagreements == 1


def test_recall_is_broken_down_by_mode():
    injected = {
        (16, BAND, 0): "eclipse_dim",
        (16, BAND, 240): "eclipse_dim",
        (16, BAND, 480): "gain_shift",
        (16, BAND, 720): "gain_shift",
    }
    verdicts = {
        (16, BAND, 0): verdict("bad"),
        (16, BAND, 240): verdict("bad"),
        (16, BAND, 480): verdict("bad"),
        (16, BAND, 720): verdict("good"),
    }
    score = metrics.score_detection(verdicts, injected)
    assert score.recall_by_mode == {"eclipse_dim": 1.0, "gain_shift": 0.5}


def test_unknown_counts_against_a_modes_recall():
    injected = {(16, BAND, 0): "rotate", (16, BAND, 240): "rotate"}
    verdicts = {(16, BAND, 0): verdict("bad"), (16, BAND, 240): verdict("unknown")}
    score = metrics.score_detection(verdicts, injected)
    assert score.recall_by_mode["rotate"] == 0.5


def test_scoring_an_empty_case():
    score = metrics.score_detection({}, {})
    assert np.isnan(score.precision) and np.isnan(score.recall) and np.isnan(score.f1)
    assert np.isnan(score.microseconds_per_frame)


def test_throughput_is_per_frame():
    verdicts = {(16, BAND, t): verdict("good") for t in range(0, 1000, 240)}
    score = metrics.score_detection(verdicts, {}, elapsed_us=1000)
    assert score.microseconds_per_frame == pytest.approx(1000 / len(verdicts))


def test_as_dict_is_serialisable():
    score = metrics.score_detection({(16, BAND, 0): verdict("bad")}, {(16, BAND, 0): "x"})
    import json

    assert json.loads(json.dumps(score.as_dict()))["recall"] == 1.0


# ---------------------------------------------------------------------- pr curves


def test_precision_recall_curve_on_a_separable_score():
    scores = [0.9, 0.8, 0.2, 0.1]
    labels = [True, True, False, False]
    thresholds, precision, recall = metrics.precision_recall_curve(scores, labels)
    assert list(thresholds) == [0.9, 0.8, 0.2, 0.1]
    assert precision[1] == 1.0 and recall[1] == 1.0
    assert metrics.average_precision(scores, labels) == pytest.approx(1.0)


def test_average_precision_of_an_uninformative_score():
    scores = [0.5, 0.5, 0.5, 0.5]
    labels = [True, False, True, False]
    assert 0.0 < metrics.average_precision(scores, labels) < 1.0


def test_curves_handle_degenerate_input():
    assert metrics.average_precision([], []) != metrics.average_precision([], []) or True
    assert np.isnan(metrics.average_precision([1.0], [False]))
    thresholds, _, _ = metrics.precision_recall_curve([np.nan, np.inf], [True, False])
    assert len(thresholds) <= 1


# --------------------------------------------------------------------------- fill


def test_identical_frames_score_perfectly():
    image = solar_disc(size=64, radius=20, peak=3.0)
    score = metrics.score_fill(image, image, BAND)
    assert score.rmse == 0.0 and score.mae == 0.0 and score.log_rmse == 0.0
    assert score.psnr == float("inf")
    assert score.ssim == pytest.approx(1.0)


def test_worse_reconstructions_score_worse():
    truth = solar_disc(size=64, radius=20, peak=3.0)
    close = truth + 0.01
    far = truth + 0.5
    assert metrics.score_fill(close, truth, BAND).rmse < metrics.score_fill(far, truth, BAND).rmse
    assert metrics.score_fill(close, truth, BAND).psnr > metrics.score_fill(far, truth, BAND).psnr
    assert metrics.score_fill(close, truth, BAND).ssim > metrics.score_fill(far, truth, BAND).ssim


def test_log_error_keeps_the_faint_corona_visible():
    """The log term makes an equal *relative* error count comparably everywhere.

    Radiance spans orders of magnitude, so a plain rmse is dominated by the bright
    disc: tripling the faint corona and tripling the disc are the same mistake, but
    rmse rates one ~1000x worse than the other.  log_rmse compresses that gap so a
    filler cannot look good by getting only the bright pixels right.
    """
    faint = np.full((32, 32), 0.01)
    bright = np.full((32, 32), 10.0)
    faint_score = metrics.score_fill(faint * 3.0, faint, BAND)
    bright_score = metrics.score_fill(bright * 3.0, bright, BAND)

    assert bright_score.rmse / faint_score.rmse == pytest.approx(1000, rel=0.01)
    assert bright_score.log_rmse / faint_score.log_rmse < 100


def test_fill_scoring_rejects_mismatched_shapes():
    with pytest.raises(ValueError, match="shape mismatch"):
        metrics.score_fill(np.zeros((4, 4)), np.zeros((8, 8)), BAND)


def test_fill_scoring_tolerates_nan():
    truth = solar_disc(size=32, radius=10, peak=2.0)
    filled = truth.copy()
    filled[0:4, 0:4] = np.nan
    score = metrics.score_fill(filled, truth, BAND)
    assert np.isfinite(score.rmse) and np.isfinite(score.ssim)


# ------------------------------------------------------------------------ flicker


def test_flicker_is_zero_for_a_perfect_reconstruction():
    frames = [solar_disc(size=32, radius=10, peak=1.0 + i * 0.1) for i in range(5)]
    assert metrics.temporal_flicker(frames, frames, BAND) == pytest.approx(0.0)


def test_flicker_penalises_a_freeze_that_psnr_forgives():
    """The artifact per-frame metrics miss: hold-then-jump instead of smooth motion."""
    truth = [solar_disc(size=32, radius=10, peak=1.0 + i * 0.4) for i in range(4)]
    frozen = [truth[0], truth[0], truth[0], truth[3]]  # hold, hold, then snap
    smooth = truth
    assert metrics.temporal_flicker(frozen, truth, BAND) > metrics.temporal_flicker(
        smooth, truth, BAND
    )


def test_flicker_requires_matching_lengths():
    frames = [np.zeros((8, 8))] * 3
    with pytest.raises(ValueError, match="same length"):
        metrics.temporal_flicker(frames, frames[:2], BAND)


def test_flicker_of_a_single_frame_is_undefined():
    assert np.isnan(metrics.temporal_flicker([np.zeros((8, 8))], [np.zeros((8, 8))], BAND))


# ------------------------------------------------------------------------ summary


def test_summary_groups_by_gap_length():
    scores = [
        metrics.FillScore(1.0, 1.0, 0.1, 30.0, 0.9, gap_frames=1, wavelength=BAND),
        metrics.FillScore(3.0, 3.0, 0.3, 20.0, 0.7, gap_frames=1, wavelength=BAND),
        metrics.FillScore(5.0, 5.0, 0.5, 10.0, 0.5, gap_frames=10, wavelength=BAND),
    ]
    summary = metrics.summarise_fills(scores)
    assert summary["n"] == 3
    assert summary["rmse"] == pytest.approx(3.0)
    assert summary["by_gap"][1]["n"] == 2
    assert summary["by_gap"][1]["psnr"] == pytest.approx(25.0)
    assert summary["by_gap"][10]["ssim"] == pytest.approx(0.5)


def test_summary_ignores_infinite_psnr_from_perfect_frames():
    scores = [
        metrics.FillScore(0.0, 0.0, 0.0, float("inf"), 1.0, 1, BAND),
        metrics.FillScore(1.0, 1.0, 0.1, 20.0, 0.8, 1, BAND),
    ]
    assert metrics.summarise_fills(scores)["psnr"] == pytest.approx(20.0)


def test_summary_of_nothing():
    assert metrics.summarise_fills([]) == {}


# -------------------------------------------------------------------- combinations


def vmap(**pairs):
    """Build a {slot: Verdict} from slot-suffix -> verdict-string pairs."""
    return {(16, BAND, int(k[1:])): verdict(v) for k, v in pairs.items()}


def test_any_policy_flags_if_one_detector_does():
    combined = metrics.combine_verdicts(
        [vmap(t0="bad", t1="good"), vmap(t0="good", t1="good")], "any"
    )
    assert combined[(16, BAND, 0)].verdict == "bad"
    assert combined[(16, BAND, 1)].verdict == "good"


def test_all_policy_requires_unanimity():
    combined = metrics.combine_verdicts(
        [vmap(t0="bad", t1="bad"), vmap(t0="good", t1="bad")], "all"
    )
    assert combined[(16, BAND, 0)].verdict == "good"
    assert combined[(16, BAND, 1)].verdict == "bad"


def test_majority_policy():
    maps = [vmap(t0="bad"), vmap(t0="bad"), vmap(t0="good")]
    assert metrics.combine_verdicts(maps, "majority")[(16, BAND, 0)].verdict == "bad"
    maps = [vmap(t0="bad"), vmap(t0="good"), vmap(t0="good")]
    assert metrics.combine_verdicts(maps, "majority")[(16, BAND, 0)].verdict == "good"


def test_unknown_abstains_rather_than_voting():
    """An abstention must not act as a 'good' vote.

    Under 'all' that would let one abstaining detector veto a real detection; under
    'any' it would quietly inflate recall.
    """
    combined = metrics.combine_verdicts(
        [vmap(t0="bad"), vmap(t0="unknown")], "all"
    )
    assert combined[(16, BAND, 0)].verdict == "bad"  # the abstention is ignored
    assert combined[(16, BAND, 0)].scores["votes_total"] == 1.0


def test_all_detectors_abstaining_yields_unknown():
    combined = metrics.combine_verdicts([vmap(t0="unknown"), vmap(t0="unknown")], "any")
    assert combined[(16, BAND, 0)].verdict == "unknown"


def test_slots_missing_from_one_map_are_still_judged():
    combined = metrics.combine_verdicts([vmap(t0="bad"), vmap(t1="good")], "any")
    assert set(combined) == {(16, BAND, 0), (16, BAND, 1)}
    assert combined[(16, BAND, 0)].verdict == "bad"


def test_single_detector_reduces_to_itself():
    single = vmap(t0="bad", t1="good", t2="unknown")
    for policy in metrics.COMBINATION_POLICIES:
        combined = metrics.combine_verdicts([single], policy)
        for slot, original in single.items():
            assert combined[slot].verdict == original.verdict, policy


def test_combining_nothing():
    assert metrics.combine_verdicts([], "any") == {}


def test_unknown_policy_is_rejected():
    with pytest.raises(ValueError, match="Unknown policy"):
        metrics.combine_verdicts([vmap(t0="bad")], "consensus")


def test_combined_verdicts_carry_the_reasons():
    maps = [vmap(t0="bad"), vmap(t0="bad")]
    maps[0][(16, BAND, 0)] = detectors.Verdict("bad", "eclipse", {})
    maps[1][(16, BAND, 0)] = detectors.Verdict("bad", "limb_contrast 0.0", {})
    combined = metrics.combine_verdicts(maps, "any")
    assert "eclipse" in combined[(16, BAND, 0)].reason
    assert "limb_contrast" in combined[(16, BAND, 0)].reason


def test_a_combination_can_be_scored_like_any_detector():
    """Combinations feed straight into score_detection -- that is the point."""
    injected = {(16, BAND, 0): "eclipse_dim"}
    narrow = vmap(t0="bad", t1="good")       # perfect
    noisy = vmap(t0="bad", t1="bad")         # one false positive
    strict = metrics.combine_verdicts([narrow, noisy], "all")
    loose = metrics.combine_verdicts([narrow, noisy], "any")
    assert metrics.score_detection(strict, injected).false_positive_rate == 0.0
    assert metrics.score_detection(loose, injected).false_positive_rate == 1.0
    assert metrics.score_detection(strict, injected).recall == 1.0