154 lines
6 KiB
Python
154 lines
6 KiB
Python
"""The per-band availability model: the archive's own damage, as data.
|
|
|
|
What must hold: states are per (satellite, band, slot); reading outside the era is
|
|
MISSING; windows can be centred on a real fill case; the damage-pattern
|
|
distribution counts slots, not bands; the dual-gap policy is the renderer's,
|
|
applied per band; and an old slot-level model is refused, not misread.
|
|
"""
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from suvi import availability, paths
|
|
|
|
BANDS = len(availability.BANDS)
|
|
BASE = 1715400000 // paths.CADENCE * paths.CADENCE
|
|
|
|
|
|
def model(count=600, satellites=(16, 18)):
|
|
states = np.full((len(satellites), BANDS, count), availability.GOOD, dtype=np.uint8)
|
|
return availability.Availability(states=states, t0=BASE, cadence=paths.CADENCE,
|
|
satellites=satellites)
|
|
|
|
|
|
def test_states_must_be_per_band():
|
|
with pytest.raises(ValueError, match="bands"):
|
|
availability.Availability(states=np.zeros((2, 10), np.uint8), t0=0, cadence=240,
|
|
satellites=(16, 18))
|
|
|
|
|
|
def test_reading_outside_the_era_or_an_unknown_satellite_is_missing():
|
|
m = model(count=10)
|
|
assert (m.state(16, -1) == availability.MISSING).all()
|
|
assert (m.state(16, 10) == availability.MISSING).all()
|
|
assert (m.state(17, 3) == availability.MISSING).all()
|
|
assert m.usable(16, 3).all()
|
|
|
|
|
|
def test_pristine_needs_every_band_of_every_satellite():
|
|
m = model(count=5)
|
|
m.states[1, 4, 2] = availability.BAD
|
|
m.states[0, 0, 3] = availability.MISSING
|
|
assert m.pristine().tolist() == [True, True, False, False, True]
|
|
|
|
|
|
def test_windows_for_a_satellite_centre_on_its_damaged_slots():
|
|
m = model()
|
|
m.states[0, 1, [10, 400]] = availability.BAD
|
|
m.states[1, :, 200] = availability.MISSING # the other satellite's damage
|
|
rng = np.random.default_rng(0)
|
|
centres = {m.sample_window(rng, 16).centre for _ in range(50)}
|
|
assert centres == {10, 400}
|
|
assert {m.sample_window(rng, 18).centre for _ in range(10)} == {200}
|
|
|
|
|
|
def test_a_satellite_with_no_damage_cannot_centre_a_window():
|
|
with pytest.raises(ValueError):
|
|
model().sample_window(np.random.default_rng(0), 16)
|
|
|
|
|
|
def test_damage_patterns_count_slots_and_their_band_sets():
|
|
m = model(count=10)
|
|
m.states[0, 1, 0:6] = availability.BAD # 131 alone, six slots
|
|
m.states[1, :, 0:2] = availability.MISSING # whole slots, two
|
|
m.states[1, [0, 4], 5] = availability.BAD # 94+284, one
|
|
masks, odds = m.damage_patterns()
|
|
found = {tuple(np.flatnonzero(mask)): p for mask, p in zip(masks, odds)}
|
|
assert found == pytest.approx({(1,): 6 / 9, (0, 1, 2, 3, 4, 5): 2 / 9, (0, 4): 1 / 9})
|
|
|
|
|
|
def test_fractions_report_bands_and_whole_slots():
|
|
m = model(count=4)
|
|
m.states[0, 0, 0] = availability.BAD
|
|
rates = m.fractions()[16]
|
|
assert rates["bad"] == pytest.approx(1 / (4 * BANDS))
|
|
assert rates["pristine"] == pytest.approx(3 / 4)
|
|
|
|
|
|
def test_save_and_load_round_trip(tmp_path):
|
|
m = model(count=7)
|
|
m.states[1, 2, 3] = availability.BAD
|
|
m.save(str(tmp_path / "a.npz"))
|
|
back = availability.Availability.load(str(tmp_path / "a.npz"))
|
|
np.testing.assert_array_equal(back.states, m.states)
|
|
assert (back.t0, back.cadence, back.satellites) == (m.t0, m.cadence, m.satellites)
|
|
|
|
|
|
def test_a_slot_level_model_is_refused(tmp_path):
|
|
np.savez_compressed(str(tmp_path / "old.npz"), states=np.zeros((2, 10), np.uint8),
|
|
t0=np.int64(0), cadence=np.int64(240),
|
|
satellites=np.asarray([16, 18]))
|
|
with pytest.raises(ValueError, match="profile_archive"):
|
|
availability.Availability.load(str(tmp_path / "old.npz"))
|
|
|
|
|
|
# ------------------------------------------------------------------ dual gaps
|
|
|
|
|
|
def test_dual_gap_measures_the_run_containing_the_centre_per_band():
|
|
m = model()
|
|
m.states[:, 0, 200:300] = availability.MISSING # 94A on neither satellite
|
|
window = availability.TraceWindow(m, 250)
|
|
assert window.dual_gap() == 100
|
|
assert availability.TraceWindow(m, 299).dual_gap() == 100
|
|
assert availability.TraceWindow(m, 100).dual_gap() == 0
|
|
|
|
|
|
def test_dual_gap_is_zero_while_one_satellite_has_every_band():
|
|
m = model()
|
|
m.states[0, :, :] = availability.MISSING
|
|
assert availability.TraceWindow(m, 300).dual_gap() == 0
|
|
|
|
|
|
def test_dual_gap_takes_the_longest_band():
|
|
m = model()
|
|
m.states[:, 0, 290:310] = availability.MISSING
|
|
m.states[:, 3, 250:350] = availability.BAD
|
|
assert availability.TraceWindow(m, 300).dual_gap() == 100
|
|
|
|
|
|
def test_dual_gap_reads_beyond_the_era_as_dark_and_caps_the_walk():
|
|
m = model()
|
|
m.states[:, 2, :] = availability.BAD
|
|
assert availability.TraceWindow(m, 300).dual_gap(limit=50) == 51
|
|
assert availability.TraceWindow(m, 0).dual_gap(limit=50) == 51
|
|
|
|
|
|
def test_a_band_split_between_satellites_is_no_dual_gap():
|
|
"""2024-03-03: GOES-16 had no 94/131 and GOES-18 no 304, all day."""
|
|
m = model()
|
|
m.states[0, [0, 1], :] = availability.MISSING
|
|
m.states[1, 5, :] = availability.MISSING
|
|
assert availability.TraceWindow(m, 300).dual_gap() == 0
|
|
|
|
|
|
# ------------------------------------------------------------------ helpers
|
|
|
|
|
|
def test_run_lengths_finds_contiguous_runs():
|
|
states = np.array([2, 2, 0, 0, 0, 2, 1, 1, 2, 2, 2])
|
|
np.testing.assert_array_equal(availability.run_lengths(states, 2), [2, 1, 3])
|
|
np.testing.assert_array_equal(availability.run_lengths(states, 1), [2])
|
|
np.testing.assert_array_equal(availability.run_lengths(states, 0), [3])
|
|
|
|
|
|
def test_unfillable_times_flags_only_overlong_dual_gaps():
|
|
times = list(range(0, 100 * 240, 240))
|
|
usable16 = set(times) - set(times[10:40])
|
|
usable18 = set(times) - set(times[20:30])
|
|
assert availability.unfillable_times(times, {16: usable16, 18: usable18},
|
|
limit=5) == set(times[20:30])
|
|
assert availability.unfillable_times(times, {16: usable16, 18: usable18},
|
|
limit=10) == set()
|
|
assert availability.unfillable_times(times, {16: set(times), 18: set()},
|
|
limit=5) == set()
|