285 lines
13 KiB
Python
285 lines
13 KiB
Python
"""Scoring detectors on real labelled defects.
|
|
|
|
The numbers this produces decide which detectors production trusts, so the
|
|
properties pinned here are the arithmetic (recall per category, false positives on
|
|
clean and flare frames kept apart), the combination rule, the treatment of
|
|
abstentions and neutral labels, and that `measure` gives the series detectors
|
|
real context on both satellites.
|
|
"""
|
|
|
|
import json
|
|
import math
|
|
import os
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
import detector_eval as ev
|
|
from conftest import solar_disc, write_fits
|
|
from suvi import db, paths
|
|
|
|
T0 = 1704067200
|
|
|
|
|
|
def frame(sat=16, band=171, t=T0, **verdicts):
|
|
out = {"satellite": sat, "wavelength": band, "t_start": t, "far_field": 0.03}
|
|
for name in ev.MEASURED:
|
|
out[name] = ["good", None]
|
|
for name, value in verdicts.items():
|
|
out[name] = value
|
|
return out
|
|
|
|
|
|
# ------------------------------------------------------------------- clustering
|
|
|
|
|
|
def label(t, band=171, sat=16, category="normal"):
|
|
return {"satellite": sat, "wavelength": band, "t_start": t, "category": category}
|
|
|
|
|
|
def test_clusters_split_on_gaps_and_bands_and_share_satellites():
|
|
labels = [label(T0), label(T0 + 240, sat=18), label(T0 + 30 * 240),
|
|
label(T0, band=304)]
|
|
groups = ev.clusters(labels, gap=24)
|
|
shapes = [(band, [r["t_start"] for r in members]) for band, members in groups]
|
|
assert shapes == [(171, [T0, T0 + 240]), (171, [T0 + 30 * 240]), (304, [T0])]
|
|
|
|
|
|
# ---------------------------------------------------------------------- scoring
|
|
|
|
|
|
def test_expand_resolves_aliases_and_combinations():
|
|
assert ev.expand("pair+temporal_v1") == ["header_v1", "disc_v1", "temporal_v1"]
|
|
assert ev.expand("pair_old") == ["header_v1_old", "disc_v1"]
|
|
|
|
|
|
def test_unknown_terms_fail_loudly():
|
|
with pytest.raises(KeyError, match="unknown detector"):
|
|
ev.says_bad(frame(), "header_v9", {})
|
|
|
|
|
|
@pytest.mark.parametrize("verdict,expected", [
|
|
(["bad", "x"], True), (["good", None], False), (["unknown", "?"], False), (None, False),
|
|
])
|
|
def test_only_an_explicit_bad_counts(verdict, expected):
|
|
assert ev.says_bad(frame(temporal_v1=verdict), "temporal_v1", {}) is expected
|
|
|
|
|
|
def test_far_field_threshold_is_relative_to_the_band_baseline():
|
|
base = {171: 0.03}
|
|
assert ev.says_bad(frame(far_field=0.13), "far_field@4", base)
|
|
assert not ev.says_bad(frame(far_field=0.11), "far_field@4", base)
|
|
assert not ev.says_bad(frame(band=94, far_field=9.0), "far_field@4", base) # no baseline
|
|
assert not ev.says_bad(frame(far_field=None), "far_field@4", base)
|
|
|
|
|
|
def test_temporal_v2_is_judged_from_stored_scores_against_current_limits(monkeypatch):
|
|
f = frame(temporal_v2_scores={"brightness": 0.8, "structure": 1.0})
|
|
assert ev.says_bad(f, "temporal_v2", {})
|
|
monkeypatch.setitem(ev.detectors.TEMPORAL2_LIMITS, "brightness", 0.9)
|
|
assert not ev.says_bad(f, "temporal_v2", {})
|
|
assert not ev.says_bad(frame(temporal_v2_scores=None), "temporal_v2", {})
|
|
|
|
|
|
def test_baselines_come_from_normal_frames_only():
|
|
frames = [frame(t=T0 + i, far_field=v) for i, v in enumerate((1.0, 2.0, 3.0, 99.0))]
|
|
labels = {(16, 171, T0 + i): c for i, c in enumerate(["normal"] * 3 + ["bar"])}
|
|
assert ev.far_field_baselines(frames, labels) == {171: 2.0}
|
|
|
|
|
|
def test_score_counts_recall_per_category_and_false_positives_apart():
|
|
frames = [frame(t=T0, disc_v1=["bad", "x"]), # bar caught by disc
|
|
frame(t=T0 + 240), # bar missed
|
|
frame(t=T0 + 480, header_v1=["bad", "x"]), # normal wrongly rejected
|
|
frame(t=T0 + 720), # flare kept
|
|
frame(t=T0 + 960, header_v1=["bad", "x"])] # neutral: never scored
|
|
labels = {(16, 171, T0): "bar", (16, 171, T0 + 240): "bar",
|
|
(16, 171, T0 + 480): "normal", (16, 171, T0 + 720): "flare_normal",
|
|
(16, 171, T0 + 960): "header_only"}
|
|
table = ev.score(frames, labels, ["pair", "disc_v1"], {})
|
|
assert table["pair"] == {"bar": (1, 2), "normal": (1, 1), "flare_normal": (0, 1)}
|
|
assert table["disc_v1"]["normal"] == (0, 1)
|
|
recall, fpr, flare = ev.summary(table["pair"])
|
|
assert (recall, fpr, flare) == (0.5, 1.0, 0.0)
|
|
|
|
|
|
def test_summary_of_an_empty_group_is_nan_not_zero():
|
|
recall, fpr, flare = ev.summary({"bar": (1, 1)})
|
|
assert recall == 1.0 and math.isnan(fpr) and math.isnan(flare)
|
|
|
|
|
|
def test_format_table_prints_every_row(capsys):
|
|
table = {"pair": {"bar": (1, 2), "normal": (0, 4)}, "disc_v1": {"bar": (0, 2), "normal": (0, 4)}}
|
|
text = ev.format_table(table)
|
|
assert "pair" in text and "disc_v1" in text and "0.500" in text
|
|
|
|
|
|
# ---------------------------------------------------------------------- measure
|
|
|
|
|
|
def write_frame(root, satellite, wavelength, when, image, **headers):
|
|
name = paths.FrameName(satellite=satellite, wavelength=wavelength, t_start=when,
|
|
t_end=when + 240, version="1-0-2")
|
|
write_fits(os.path.join(str(root), name.relpath()), image,
|
|
headers={"WAVELNTH": wavelength, **headers})
|
|
return name
|
|
|
|
|
|
def test_measure_runs_every_detector_with_series_context(archive, tmp_path):
|
|
"""A bright bar in the middle of a quiet stream, with the other satellite clean."""
|
|
conn = db.connect(str(tmp_path / "index.sqlite"))
|
|
rng = np.random.default_rng(0)
|
|
for satellite in (16, 18):
|
|
for i in range(13):
|
|
when = T0 + i * 240
|
|
image = solar_disc(peak=1.0) * (1 + 0.01 * rng.standard_normal())
|
|
if (satellite, i) == (16, 6):
|
|
image = image.copy()
|
|
image[40:100, :] = 20.0 # saturated bar across the top
|
|
name = write_frame(archive, satellite, 171, when, image.astype(np.float32))
|
|
db.upsert_frame(conn, name, name.relpath(), size_bytes=1_500_000)
|
|
conn.commit()
|
|
labels = tmp_path / "labels.json"
|
|
labels.write_text(json.dumps({"labels": [label(T0 + 6 * 240, category="bar"),
|
|
label(T0 + 3 * 240, category="normal")]}))
|
|
out = tmp_path / "eval.json"
|
|
assert ev.main(["measure", "--labels", str(labels), "--out", str(out), "--workers", "1",
|
|
"--db", str(tmp_path / "index.sqlite")]) == 0
|
|
frames = {f["t_start"]: f for f in json.loads(out.read_text())["frames"]}
|
|
bar, clean = frames[T0 + 6 * 240], frames[T0 + 3 * 240]
|
|
for name in ev.MEASURED:
|
|
assert name in bar
|
|
assert bar["far_field"] > 100 * clean["far_field"]
|
|
assert bar["temporal_v1"][0] == "bad"
|
|
assert ev.says_bad(bar, "temporal_v2", {}) and not ev.says_bad(clean, "temporal_v2", {})
|
|
assert clean["temporal_v1"][0] == "good"
|
|
assert bar["crosssat_v1"] is not None and bar["legacy"] is None
|
|
|
|
assert ev.main(["score", "--labels", str(labels), "--features", str(out),
|
|
"--rows", "pair,far_field@4,temporal_v1", "--misses", "pair"]) == 0
|
|
|
|
|
|
# ---------------------------------------------------------------- stored runs
|
|
|
|
|
|
def stored_index(tmp_path, verdicts):
|
|
"""An index whose one judgment run gave each (t, band) the verdict named."""
|
|
conn = db.connect(str(tmp_path / "index.sqlite"))
|
|
run = db.create_detector_run(conn, "judge", {"x": 1})
|
|
rows = []
|
|
for (t, band), verdict in verdicts.items():
|
|
name = paths.FrameName(satellite=16, wavelength=band, t_start=t, t_end=t + 240,
|
|
version="1-0-2")
|
|
frame_id = db.upsert_frame(conn, name, name.relpath(), size_bytes=1)
|
|
rows.append((frame_id, verdict, f"why {t}" if verdict == "bad" else None, None, 0))
|
|
db.record_detections(conn, run, rows)
|
|
conn.commit()
|
|
return conn, run
|
|
|
|
|
|
STORED = {(T0, 171): "bad", (T0 + 240, 171): "good", (T0 + 480, 171): "bad",
|
|
(T0 + 720, 171): "good", (T0 + 960, 171): "bad"}
|
|
STORED_LABELS = [label(T0, category="bar"), label(T0 + 240, category="ghost"),
|
|
label(T0 + 480, category="normal"),
|
|
label(T0 + 720, category="flare_normal"),
|
|
label(T0 + 960, category="header_only"), # neutral: not scored
|
|
label(T0 + 1200, category="bar")] # never judged by the run
|
|
|
|
|
|
def test_a_stored_run_is_scored_from_its_verdicts(tmp_path):
|
|
conn, run = stored_index(tmp_path, STORED)
|
|
counts, missed, rejected = ev.stored_table(conn, run, STORED_LABELS)
|
|
assert counts == {"bar": (1, 1), "ghost": (0, 1), "normal": (1, 1),
|
|
"flare_normal": (0, 1)}
|
|
assert [r["category"] for r, _ in missed] == ["ghost"]
|
|
assert [(r["category"], why) for r, why in rejected] == [("normal", f"why {T0 + 480}")]
|
|
|
|
|
|
def test_a_stored_run_ignores_other_runs_verdicts(tmp_path):
|
|
conn, run = stored_index(tmp_path, STORED)
|
|
other = db.create_detector_run(conn, "judge", {"x": 2})
|
|
counts, _, _ = ev.stored_table(conn, other, STORED_LABELS)
|
|
assert counts == {}
|
|
|
|
|
|
def test_the_stored_command_prints_the_table_and_both_lists(tmp_path, capsys):
|
|
conn, run = stored_index(tmp_path, STORED)
|
|
labels = tmp_path / "labels.json"
|
|
labels.write_text(json.dumps({"labels": STORED_LABELS}))
|
|
assert ev.main(["stored", "--labels", str(labels), "--run", str(run),
|
|
"--db", str(tmp_path / "index.sqlite")]) == 0
|
|
out = capsys.readouterr().out
|
|
assert f"judgment run {run}: 4 labelled frames judged" in out
|
|
assert "defects passed: 1" in out and "clean frames rejected: 1" in out
|
|
assert f"why {T0 + 480}" in out
|
|
|
|
|
|
# ---------------------------------------------------------- dry runs and sheets
|
|
|
|
|
|
def test_score_verdicts_counts_and_lists_by_category():
|
|
labels = [label(T0, category="bar"), label(T0 + 240, category="normal"),
|
|
label(T0 + 480, category="header_only"), label(T0 + 720, category="ghost")]
|
|
verdicts = {(16, 171, T0): ("bad", "x"), (16, 171, T0 + 240): ("bad", "flare"),
|
|
(16, 171, T0 + 480): ("bad", "y")}
|
|
counts, missed, rejected = ev.score_verdicts(verdicts, labels)
|
|
assert counts == {"bar": (1, 1), "normal": (1, 1)}
|
|
assert missed == [] and [r["t_start"] for r, _ in rejected] == [T0 + 240]
|
|
|
|
|
|
def test_changes_list_frames_whose_verdict_moved_and_skip_unjudged_ones():
|
|
verdicts = {(16, 171, T0): ("bad", "relative_v2", "dim"),
|
|
(16, 171, T0 + 240): ("good", None, None),
|
|
(16, 171, T0 + 480): ("bad", "pointing_v1", "off")}
|
|
reference = {(16, 171, T0): ("good", None), (16, 171, T0 + 240): ("bad", "temporal_v2: x")}
|
|
newly_bad, newly_good = ev.changes(verdicts, reference)
|
|
assert newly_bad == [((16, 171, T0), "relative_v2: dim")]
|
|
assert newly_good == [((16, 171, T0 + 240), "was temporal_v2: x")]
|
|
|
|
|
|
def measured_archive(archive, tmp_path):
|
|
"""An index with five 171A frames measured, the fourth at half brightness, and
|
|
a reference run that called every frame good."""
|
|
import filter_FITS
|
|
from test_filter_judge import write_frame
|
|
|
|
conn = db.connect(str(tmp_path / "index.sqlite"))
|
|
ids = [write_frame(conn, archive, 171, i, solar_disc(peak=1.0) * gain)
|
|
for i, gain in enumerate([1.0, 1.0, 1.0, 0.5, 1.0, 1.0, 1.0])]
|
|
conn.commit()
|
|
assert filter_FITS.main(["--db", str(tmp_path / "index.sqlite"), "measure",
|
|
"--start", str(T0), "--end", str(T0 + 86400),
|
|
"--workers", "1"]) == 0
|
|
reference = db.create_detector_run(conn, "judge", {"reference": True})
|
|
db.record_detections(conn, reference, [(i, "good", None, None, 0) for i in ids])
|
|
conn.commit()
|
|
return conn, ids, reference
|
|
|
|
|
|
def test_dry_run_judges_in_memory_and_lists_what_changed(archive, tmp_path, capsys):
|
|
conn, ids, reference = measured_archive(archive, tmp_path)
|
|
labels = tmp_path / "labels.json"
|
|
labels.write_text(json.dumps({"labels": [label(T0 + 3 * 240, category="gain")]}))
|
|
out = tmp_path / "changes.json"
|
|
runs_before = conn.execute("SELECT COUNT(*) FROM detector_run").fetchone()[0]
|
|
assert ev.main(["dry-run", "--labels", str(labels), "--start", str(T0),
|
|
"--end", str(T0 + 86400), "--reference", str(reference),
|
|
"--out", str(out), "--db", str(tmp_path / "index.sqlite")]) == 0
|
|
assert conn.execute("SELECT COUNT(*) FROM detector_run").fetchone()[0] == runs_before
|
|
rows = json.loads(out.read_text())
|
|
assert [r[2] for r in rows["newly_bad"]] == ["2024-01-01 00:12"]
|
|
assert rows["newly_bad"][0][3].split(":")[0] in ("temporal_v2", "relative_v2")
|
|
assert "defects passed: 0" in capsys.readouterr().out
|
|
|
|
|
|
def test_sheet_draws_each_row_between_its_good_neighbours(archive, tmp_path):
|
|
from PIL import Image
|
|
|
|
conn, ids, reference = measured_archive(archive, tmp_path)
|
|
rows = tmp_path / "rows.json"
|
|
rows.write_text(json.dumps({"newly_bad": [[16, 171, "2024-01-01 00:12", "dim"]]}))
|
|
out = tmp_path / "sheet.jpg"
|
|
assert ev.main(["sheet", "--rows", str(rows), "--run", str(reference), "--out", str(out),
|
|
"--db", str(tmp_path / "index.sqlite")]) == 0
|
|
with Image.open(out) as image:
|
|
assert image.size == (3 * ev.SHEET_TILE, ev.SHEET_TILE + 40)
|