Graham McIntire
690a3523e2
First step of the disk-backed scores migration from the DuckDB plan doc. Ended up shipping the raw-binary variant instead of Parquet because the data is disposable after ~2h — the ecosystem benefits of Parquet only pay off for long-lived datasets, and the binary path has zero new dependencies. Microwaveprop.Propagation.ScoresFile writes one file per (band_mhz, valid_time) tuple under the configured scores dir, default /data/scores in prod and priv/dev_scores in dev. Layout is a 33-byte header plus a dense n_rows × n_cols uint8 array (255 is the no-data sentinel). The whole CONUS grid serializes to ~93 KB per band, and writes use the temp-then-rename pattern so NFSv4 concurrent readers never see a partial file. Propagation.replace_scores/2 now materializes the score stream once and dual-writes: Postgres on the primary path, then one ScoresFile per band as a best-effort follow-up (any file error is logged but doesn't fail the DB write, so we can verify the file path in prod before cutting readers over). Propagation.prune_old_scores/0 also clears expired score files so the existing 15-minute prune cron covers both storage layers. Dev configuration points at priv/dev_scores/, added to .gitignore. Test configuration points at a per-run tmp directory.
473 lines
17 KiB
Elixir
473 lines
17 KiB
Elixir
defmodule Microwaveprop.PropagationTest do
|
|
use Microwaveprop.DataCase, async: false
|
|
|
|
alias Microwaveprop.Propagation
|
|
alias Microwaveprop.Propagation.GridScore
|
|
alias Microwaveprop.Propagation.ScoreCache
|
|
alias Microwaveprop.Propagation.ScoresFile
|
|
|
|
setup do
|
|
ScoreCache.clear()
|
|
:ok
|
|
end
|
|
|
|
describe "score_grid_point/4" do
|
|
test "scores a single point for all bands" do
|
|
hrrr_profile = %{
|
|
surface_temp_c: 25.0,
|
|
surface_dewpoint_c: 18.0,
|
|
surface_pressure_mb: 1013.0,
|
|
hpbl_m: 500.0,
|
|
wind_u: 3.0,
|
|
wind_v: 2.0,
|
|
cloud_cover_pct: 15.0,
|
|
precip_mm: 0.0,
|
|
profile: [
|
|
%{"pres" => 1000.0, "tmpc" => 25.0, "dwpc" => 18.0, "hght" => 100.0},
|
|
%{"pres" => 975.0, "tmpc" => 22.0, "dwpc" => 15.0, "hght" => 350.0},
|
|
%{"pres" => 950.0, "tmpc" => 19.0, "dwpc" => 10.0, "hght" => 600.0}
|
|
]
|
|
}
|
|
|
|
valid_time = ~U[2026-07-15 13:00:00Z]
|
|
results = Propagation.score_grid_point(hrrr_profile, valid_time, 33.0, -97.0)
|
|
|
|
assert length(results) == 19
|
|
|
|
Enum.each(results, fn result ->
|
|
assert result.score >= 0 and result.score <= 100
|
|
assert is_map(result.factors)
|
|
assert is_integer(result.band_mhz)
|
|
end)
|
|
end
|
|
|
|
test "NEXRAD reflectivity drops the 24 GHz rain score when HRRR precip_mm is 0" do
|
|
# Same profile, once without NEXRAD and once with a heavy-rain dBZ. The
|
|
# 24 GHz rain-score factor must drop because NEXRAD can report rain that
|
|
# HRRR's hourly precip accumulation hasn't caught yet.
|
|
base_profile = %{
|
|
surface_temp_c: 25.0,
|
|
surface_dewpoint_c: 18.0,
|
|
surface_pressure_mb: 1013.0,
|
|
hpbl_m: 500.0,
|
|
wind_u: 3.0,
|
|
wind_v: 2.0,
|
|
cloud_cover_pct: 15.0,
|
|
precip_mm: 0.0,
|
|
profile: [
|
|
%{"pres" => 1000.0, "tmpc" => 25.0, "dwpc" => 18.0, "hght" => 100.0},
|
|
%{"pres" => 950.0, "tmpc" => 19.0, "dwpc" => 10.0, "hght" => 600.0}
|
|
]
|
|
}
|
|
|
|
valid_time = ~U[2026-07-15 13:00:00Z]
|
|
|
|
without = Propagation.score_grid_point(base_profile, valid_time, 33.0, -97.0)
|
|
|
|
with_nexrad =
|
|
base_profile
|
|
|> Map.put(:nexrad_max_reflectivity_dbz, 45.0)
|
|
|> Propagation.score_grid_point(valid_time, 33.0, -97.0)
|
|
|
|
rain_dry = Enum.find(without, &(&1.band_mhz == 24_000)).factors.rain
|
|
rain_wet = Enum.find(with_nexrad, &(&1.band_mhz == 24_000)).factors.rain
|
|
|
|
assert rain_wet < rain_dry
|
|
end
|
|
|
|
test "uses a persisted min_refractivity_gradient scalar without decoding the profile array" do
|
|
# AsosAdjustmentWorker loads 92k HRRR profile rows per 10-min tick and
|
|
# can't afford to pull the ~1 KB JSONB `profile` column for each one —
|
|
# Postgrex's Jason.decode! per row blew past the 15s pool checkout.
|
|
# hrrr_profiles persists min_refractivity_gradient as a scalar at
|
|
# ingestion time, so score_grid_point must honour it when the profile
|
|
# array is absent instead of trying to re-derive from a missing list.
|
|
hrrr_profile_without_array = %{
|
|
surface_temp_c: 25.0,
|
|
surface_dewpoint_c: 18.0,
|
|
surface_pressure_mb: 1013.0,
|
|
hpbl_m: 500.0,
|
|
pwat_mm: 28.0,
|
|
wind_u: 3.0,
|
|
wind_v: 2.0,
|
|
cloud_cover_pct: 15.0,
|
|
precip_mm: 0.0,
|
|
min_refractivity_gradient: -400.0
|
|
}
|
|
|
|
valid_time = ~U[2026-07-15 13:00:00Z]
|
|
results = Propagation.score_grid_point(hrrr_profile_without_array, valid_time, 33.0, -97.0)
|
|
|
|
assert length(results) == 19
|
|
|
|
Enum.each(results, fn result ->
|
|
assert result.score >= 0 and result.score <= 100
|
|
end)
|
|
|
|
result_10g = Enum.find(results, &(&1.band_mhz == 10_000))
|
|
# A strong negative refractivity gradient (< -300 N/km) should leave
|
|
# the refractivity factor visibly positive; if score_grid_point had
|
|
# silently used 0.0 we'd see the neutral baseline instead.
|
|
assert result_10g.factors.refractivity > 55
|
|
end
|
|
|
|
test "NEXRAD is ignored when HRRR precip_mm already reports heavier rain" do
|
|
base_profile = %{
|
|
surface_temp_c: 25.0,
|
|
surface_dewpoint_c: 18.0,
|
|
surface_pressure_mb: 1013.0,
|
|
hpbl_m: 500.0,
|
|
wind_u: 3.0,
|
|
wind_v: 2.0,
|
|
cloud_cover_pct: 15.0,
|
|
# 30 mm/hr → far above the ~3 mm/hr NEXRAD gives for 30 dBZ, so HRRR wins.
|
|
precip_mm: 30.0,
|
|
profile: [
|
|
%{"pres" => 1000.0, "tmpc" => 25.0, "dwpc" => 18.0, "hght" => 100.0},
|
|
%{"pres" => 950.0, "tmpc" => 19.0, "dwpc" => 10.0, "hght" => 600.0}
|
|
]
|
|
}
|
|
|
|
valid_time = ~U[2026-07-15 13:00:00Z]
|
|
|
|
with_light_nexrad =
|
|
base_profile
|
|
|> Map.put(:nexrad_max_reflectivity_dbz, 30.0)
|
|
|> Propagation.score_grid_point(valid_time, 33.0, -97.0)
|
|
|
|
without = Propagation.score_grid_point(base_profile, valid_time, 33.0, -97.0)
|
|
|
|
rain_no_nx = Enum.find(without, &(&1.band_mhz == 24_000)).factors.rain
|
|
rain_with_nx = Enum.find(with_light_nexrad, &(&1.band_mhz == 24_000)).factors.rain
|
|
|
|
assert rain_with_nx == rain_no_nx
|
|
end
|
|
end
|
|
|
|
describe "upsert_scores/1" do
|
|
test "inserts scores" do
|
|
valid_time = ~U[2026-07-15 13:00:00Z]
|
|
|
|
scores = [
|
|
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: %{humidity: 90}},
|
|
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 24_000, score: 60, factors: %{humidity: 40}}
|
|
]
|
|
|
|
assert {:ok, 2} = Propagation.upsert_scores(scores)
|
|
assert Repo.aggregate(GridScore, :count) == 2
|
|
end
|
|
|
|
test "upserts on conflict" do
|
|
valid_time = ~U[2026-07-15 13:00:00Z]
|
|
|
|
scores = [
|
|
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: %{}}
|
|
]
|
|
|
|
Propagation.upsert_scores(scores)
|
|
Propagation.upsert_scores([%{hd(scores) | score: 80}])
|
|
|
|
assert Repo.aggregate(GridScore, :count) == 1
|
|
end
|
|
end
|
|
|
|
describe "replace_scores/2" do
|
|
test "inserts all rows when the valid_time slice is empty" do
|
|
valid_time = ~U[2026-07-15 13:00:00Z]
|
|
|
|
scores = [
|
|
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: %{humidity: 90}},
|
|
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 24_000, score: 60, factors: %{humidity: 40}}
|
|
]
|
|
|
|
assert {:ok, 2} = Propagation.replace_scores(scores, valid_time)
|
|
assert Repo.aggregate(GridScore, :count) == 2
|
|
end
|
|
|
|
test "replaces pre-existing rows for the same valid_time" do
|
|
valid_time = ~U[2026-07-15 13:00:00Z]
|
|
|
|
Propagation.upsert_scores([
|
|
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 10, factors: %{}},
|
|
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 24_000, score: 20, factors: %{}}
|
|
])
|
|
|
|
assert {:ok, 2} =
|
|
Propagation.replace_scores(
|
|
[
|
|
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: %{}},
|
|
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 24_000, score: 60, factors: %{}}
|
|
],
|
|
valid_time
|
|
)
|
|
|
|
scores = GridScore |> Repo.all() |> Enum.sort_by(& &1.band_mhz)
|
|
assert [%{score: 75, band_mhz: 10_000}, %{score: 60, band_mhz: 24_000}] = scores
|
|
end
|
|
|
|
test "leaves rows for other valid_times untouched" do
|
|
other_time = ~U[2026-07-15 12:00:00Z]
|
|
replaced_time = ~U[2026-07-15 13:00:00Z]
|
|
|
|
Propagation.upsert_scores([
|
|
%{lat: 35.0, lon: -97.0, valid_time: other_time, band_mhz: 10_000, score: 40, factors: %{}}
|
|
])
|
|
|
|
Propagation.replace_scores(
|
|
[%{lat: 35.0, lon: -97.0, valid_time: replaced_time, band_mhz: 10_000, score: 90, factors: %{}}],
|
|
replaced_time
|
|
)
|
|
|
|
assert Repo.aggregate(GridScore, :count) == 2
|
|
assert Repo.one(from gs in GridScore, where: gs.valid_time == ^other_time, select: gs.score) == 40
|
|
assert Repo.one(from gs in GridScore, where: gs.valid_time == ^replaced_time, select: gs.score) == 90
|
|
end
|
|
|
|
test "deletes pre-existing rows for the valid_time even when the new list is empty" do
|
|
# A partial run that produced zero scores (e.g. HRRR fetch failed
|
|
# for every band) should still clear stale rows so the map doesn't
|
|
# show data from a previous forecast.
|
|
valid_time = ~U[2026-07-15 13:00:00Z]
|
|
|
|
Propagation.upsert_scores([
|
|
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 10, factors: %{}}
|
|
])
|
|
|
|
assert {:ok, 0} = Propagation.replace_scores([], valid_time)
|
|
assert Repo.aggregate(GridScore, :count) == 0
|
|
end
|
|
|
|
test "accepts nil factors (forecast-hour chain steps skip the JSONB write)" do
|
|
valid_time = ~U[2026-07-15 13:00:00Z]
|
|
|
|
scores = [
|
|
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: nil},
|
|
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 24_000, score: 60, factors: nil}
|
|
]
|
|
|
|
assert {:ok, 2} = Propagation.replace_scores(scores, valid_time)
|
|
assert [nil, nil] = Repo.all(from gs in GridScore, select: gs.factors)
|
|
end
|
|
|
|
test "dual-writes one ScoresFile per band alongside the DB insert" do
|
|
valid_time = ~U[2026-07-15 13:00:00Z]
|
|
|
|
scores = [
|
|
%{lat: 25.0, lon: -125.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: nil},
|
|
%{lat: 25.0, lon: -125.0, valid_time: valid_time, band_mhz: 24_000, score: 60, factors: nil}
|
|
]
|
|
|
|
assert {:ok, 2} = Propagation.replace_scores(scores, valid_time)
|
|
|
|
# One binary file per band lands on disk at the configured path.
|
|
assert {:ok, grid_10g} =
|
|
ScoresFile.read(10_000, valid_time)
|
|
|
|
assert {:ok, grid_24g} =
|
|
ScoresFile.read(24_000, valid_time)
|
|
|
|
assert grid_10g.band_mhz == 10_000
|
|
assert grid_24g.band_mhz == 24_000
|
|
end
|
|
end
|
|
|
|
describe "point_detail/4 with forecast-hour rows" do
|
|
test "coalesces nil factors to an empty map so the JS popup renders" do
|
|
valid_time = ~U[2026-07-15 13:00:00Z]
|
|
|
|
Propagation.replace_scores(
|
|
[%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 42, factors: nil}],
|
|
valid_time
|
|
)
|
|
|
|
assert %{score: 42, factors: %{}} =
|
|
Propagation.point_detail(10_000, 35.0, -97.0, valid_time)
|
|
end
|
|
end
|
|
|
|
describe "latest_scores/1" do
|
|
test "returns latest scores for a band" do
|
|
valid_time = ~U[2026-07-15 13:00:00Z]
|
|
|
|
scores = [
|
|
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: %{}},
|
|
%{lat: 36.0, lon: -96.0, valid_time: valid_time, band_mhz: 10_000, score: 80, factors: %{}}
|
|
]
|
|
|
|
Propagation.upsert_scores(scores)
|
|
results = Propagation.latest_scores(10_000)
|
|
assert length(results) == 2
|
|
end
|
|
|
|
test "returns empty list when no data" do
|
|
assert Propagation.latest_scores(10_000) == []
|
|
end
|
|
end
|
|
|
|
describe "scores_at/3 with cache" do
|
|
test "populates the cache on a DB miss" do
|
|
valid_time = ~U[2026-07-15 13:00:00Z]
|
|
|
|
Propagation.upsert_scores([
|
|
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: %{}}
|
|
])
|
|
|
|
assert ScoreCache.fetch(10_000, valid_time) == :miss
|
|
_ = Propagation.scores_at(10_000, valid_time)
|
|
assert {:ok, [%{lat: 35.0, lon: -97.0, score: 75}]} = ScoreCache.fetch(10_000, valid_time)
|
|
end
|
|
|
|
test "returns cached bounds-filtered results without hitting the DB" do
|
|
valid_time = ~U[2026-07-15 13:00:00Z]
|
|
|
|
# Seed the cache directly — DB is empty, so if the result matches the
|
|
# cached payload we know the DB was not consulted.
|
|
ScoreCache.put(10_000, valid_time, [
|
|
%{lat: 32.0, lon: -97.0, score: 75},
|
|
%{lat: 40.0, lon: -74.0, score: 50}
|
|
])
|
|
|
|
assert Repo.aggregate(GridScore, :count) == 0
|
|
|
|
bounds = %{"south" => 30.0, "north" => 35.0, "west" => -100.0, "east" => -95.0}
|
|
result = Propagation.scores_at(10_000, valid_time, bounds)
|
|
|
|
assert length(result) == 1
|
|
assert [%{lat: 32.0, lon: -97.0, score: 75}] = result
|
|
end
|
|
|
|
test "returns empty list when no data in cache or DB" do
|
|
assert Propagation.scores_at(10_000, ~U[2026-07-15 13:00:00Z]) == []
|
|
end
|
|
end
|
|
|
|
describe "point_forecast/3 with cache" do
|
|
test "returns forecast timeline from cached scores when warm" do
|
|
t1 = DateTime.truncate(DateTime.add(DateTime.utc_now(), 3600, :second), :second)
|
|
t2 = DateTime.add(t1, 3600, :second)
|
|
|
|
ScoreCache.put(10_000, t1, [%{lat: 32.875, lon: -97.0, score: 70}])
|
|
ScoreCache.put(10_000, t2, [%{lat: 32.875, lon: -97.0, score: 75}])
|
|
|
|
assert Repo.aggregate(GridScore, :count) == 0
|
|
|
|
result = Propagation.point_forecast(10_000, 32.875, -97.0)
|
|
|
|
assert [
|
|
%{valid_time: ^t1, score: 70},
|
|
%{valid_time: ^t2, score: 75}
|
|
] = result
|
|
end
|
|
|
|
test "filters past valid_times out" do
|
|
past = DateTime.utc_now() |> DateTime.add(-3600, :second) |> DateTime.truncate(:second)
|
|
future = DateTime.utc_now() |> DateTime.add(3600, :second) |> DateTime.truncate(:second)
|
|
|
|
ScoreCache.put(10_000, past, [%{lat: 32.875, lon: -97.0, score: 60}])
|
|
ScoreCache.put(10_000, future, [%{lat: 32.875, lon: -97.0, score: 80}])
|
|
|
|
result = Propagation.point_forecast(10_000, 32.875, -97.0)
|
|
assert [%{valid_time: ^future, score: 80}] = result
|
|
end
|
|
|
|
test "snaps coordinates to the nearest grid point" do
|
|
valid_time = DateTime.truncate(DateTime.add(DateTime.utc_now(), 3600, :second), :second)
|
|
ScoreCache.put(10_000, valid_time, [%{lat: 32.875, lon: -97.0, score: 65}])
|
|
|
|
# Off-grid query snaps to 32.875, -97.0
|
|
result = Propagation.point_forecast(10_000, 32.88, -96.98)
|
|
assert [%{valid_time: ^valid_time, score: 65}] = result
|
|
end
|
|
|
|
test "returns empty list when no cached or DB data for the point" do
|
|
assert Propagation.point_forecast(10_000, 32.875, -97.0) == []
|
|
end
|
|
end
|
|
|
|
describe "available_valid_times/1 with cache" do
|
|
test "returns cached valid_times without hitting the DB when cache is warm" do
|
|
t1 = DateTime.truncate(DateTime.utc_now(), :second)
|
|
t2 = DateTime.add(t1, 3600, :second)
|
|
|
|
ScoreCache.put(10_000, t1, [])
|
|
ScoreCache.put(10_000, t2, [])
|
|
|
|
assert Repo.aggregate(GridScore, :count) == 0
|
|
assert Propagation.available_valid_times(10_000) == [t1, t2]
|
|
end
|
|
|
|
test "filters cached times older than the 1-hour cutoff" do
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
stale = DateTime.add(now, -7200, :second)
|
|
fresh = DateTime.add(now, 1800, :second)
|
|
|
|
ScoreCache.put(10_000, stale, [])
|
|
ScoreCache.put(10_000, fresh, [])
|
|
|
|
assert Propagation.available_valid_times(10_000) == [fresh]
|
|
end
|
|
|
|
test "falls back to DB on cold cache" do
|
|
valid_time = DateTime.truncate(DateTime.utc_now(), :second)
|
|
|
|
Propagation.upsert_scores([
|
|
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: %{}}
|
|
])
|
|
|
|
assert ScoreCache.valid_times(10_000) == []
|
|
assert Propagation.available_valid_times(10_000) == [valid_time]
|
|
end
|
|
end
|
|
|
|
describe "warm_cache_and_broadcast/2" do
|
|
test "loads scores from DB into the cache" do
|
|
valid_time = ~U[2026-07-15 13:00:00Z]
|
|
|
|
Propagation.upsert_scores([
|
|
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: %{}},
|
|
%{lat: 36.0, lon: -96.0, valid_time: valid_time, band_mhz: 10_000, score: 80, factors: %{}}
|
|
])
|
|
|
|
assert ScoreCache.fetch(10_000, valid_time) == :miss
|
|
|
|
Propagation.warm_cache_and_broadcast(10_000, valid_time)
|
|
ScoreCache.sync()
|
|
|
|
assert {:ok, scores} = ScoreCache.fetch(10_000, valid_time)
|
|
assert length(scores) == 2
|
|
end
|
|
|
|
test "only warms the requested band" do
|
|
valid_time = ~U[2026-07-15 13:00:00Z]
|
|
|
|
Propagation.upsert_scores([
|
|
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: %{}},
|
|
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 24_000, score: 30, factors: %{}}
|
|
])
|
|
|
|
Propagation.warm_cache_and_broadcast(10_000, valid_time)
|
|
ScoreCache.sync()
|
|
|
|
assert {:ok, [_]} = ScoreCache.fetch(10_000, valid_time)
|
|
assert ScoreCache.fetch(24_000, valid_time) == :miss
|
|
end
|
|
end
|
|
|
|
describe "latest_valid_time/0" do
|
|
test "returns nil when empty" do
|
|
assert Propagation.latest_valid_time() == nil
|
|
end
|
|
|
|
test "returns most recent time" do
|
|
t1 = ~U[2026-07-15 12:00:00Z]
|
|
t2 = ~U[2026-07-15 13:00:00Z]
|
|
|
|
Propagation.upsert_scores([
|
|
%{lat: 35.0, lon: -97.0, valid_time: t1, band_mhz: 10_000, score: 50, factors: %{}},
|
|
%{lat: 35.0, lon: -97.0, valid_time: t2, band_mhz: 10_000, score: 60, factors: %{}}
|
|
])
|
|
|
|
assert Propagation.latest_valid_time() == t2
|
|
end
|
|
end
|
|
end
|