From 811251dd154f32befa8415f22108e4d3e383d561 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 14 Apr 2026 15:13:01 -0500 Subject: [PATCH] Drop propagation_scores table, ScoresFile is the only store MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Full cutover: the propagation_scores Postgres table is gone, and the binary files under /data/scores are the sole source of truth for the map render path. Three stacked changes: 1. New migration drops the propagation_scores table and its indexes (the earlier tuning migrations for it were already applied and are now no-ops against a missing table, which is fine — Ecto just runs them on fresh environments). 2. Propagation context is gutted of every GridScore reference. replace_scores/2 writes files only. upsert_scores/2 is deleted. load_scores_from_db, available_valid_times_from_db, point_detail_from_db, point_forecast_from_db, fetch_factors, coalesce_factors, the Postgres side of prune_old_scores, and the Postgres fallbacks in latest/earliest_valid_time are all removed. point_detail always returns an empty factors map now since factor breakdowns were retired with the table. 3. Deleted modules: - Propagation.GridScore (the schema) - Propagation.ScorerDiff (read factors from the table) - Propagation.AsosNudge (helper for AsosAdjustmentWorker) - Workers.AsosAdjustmentWorker (its cron was already disabled) - Mix.Tasks.ScorerDiff (wrapper around the deleted module) And their tests. AdminTaskWorker's scorer_diff task is a logging no-op so any queued Oban rows drain cleanly. Release.scorer_diff stays as a stub that tells the operator. propagation_prune_worker_test rewritten to exercise ScoresFile pruning. propagation_test.exs rewritten to use replace_scores + ScoresFile throughout (no more GridScore / upsert_scores paths). --- lib/microwaveprop/propagation.ex | 348 +++--------------- lib/microwaveprop/propagation/asos_nudge.ex | 269 -------------- lib/microwaveprop/propagation/grid_score.ex | 32 -- lib/microwaveprop/propagation/scorer_diff.ex | 211 ----------- lib/microwaveprop/release.ex | 10 +- .../workers/admin_task_worker.ex | 29 +- .../workers/asos_adjustment_worker.ex | 176 --------- lib/mix/tasks/scorer_diff.ex | 92 ----- ...20260414200607_drop_propagation_scores.exs | 22 ++ .../propagation/asos_nudge_test.exs | 219 ----------- .../propagation/grid_score_test.exs | 95 ----- .../propagation/scorer_diff_test.exs | 207 ----------- test/microwaveprop/propagation_test.exs | 251 +++++-------- .../workers/propagation_prune_worker_test.exs | 51 +-- 14 files changed, 196 insertions(+), 1816 deletions(-) delete mode 100644 lib/microwaveprop/propagation/asos_nudge.ex delete mode 100644 lib/microwaveprop/propagation/grid_score.ex delete mode 100644 lib/microwaveprop/propagation/scorer_diff.ex delete mode 100644 lib/microwaveprop/workers/asos_adjustment_worker.ex delete mode 100644 lib/mix/tasks/scorer_diff.ex create mode 100644 priv/repo/migrations/20260414200607_drop_propagation_scores.exs delete mode 100644 test/microwaveprop/propagation/asos_nudge_test.exs delete mode 100644 test/microwaveprop/propagation/grid_score_test.exs delete mode 100644 test/microwaveprop/propagation/scorer_diff_test.exs diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index b0e19e9e..9af42072 100644 --- a/lib/microwaveprop/propagation.ex +++ b/lib/microwaveprop/propagation.ex @@ -1,15 +1,11 @@ defmodule Microwaveprop.Propagation do @moduledoc false - import Ecto.Query - alias Microwaveprop.Propagation.BandConfig alias Microwaveprop.Propagation.Grid - alias Microwaveprop.Propagation.GridScore alias Microwaveprop.Propagation.ScoreCache alias Microwaveprop.Propagation.Scorer alias Microwaveprop.Propagation.ScoresFile - alias Microwaveprop.Repo alias Microwaveprop.Weather.SoundingParams require Logger @@ -150,92 +146,17 @@ defmodule Microwaveprop.Propagation do @doc """ Replace every propagation score for `valid_time` with `scores`. - Used by `PropagationGridWorker` on the hot path, which rewrites the - entire (valid_time, all-bands) slice every forecast hour. Skips the - `ON CONFLICT DO UPDATE` machinery of `upsert_scores/2` — conflict - detection is wasted work when we know the full slice is being - rewritten. In prod this cuts the scoring+upsert phase from ~4m40s - per forecast hour to well under half that. - - **Not** suitable for `AsosAdjustmentWorker`, which rewrites only a - subset of cells that happen to be near an ASOS station. That path - must stay on `upsert_scores/2`. - - Runs inside a single transaction so readers see all-or-nothing. An - empty `scores` list still deletes the pre-existing slice, so a - partial-failure run doesn't leave stale data visible. + Used by `PropagationGridWorker` on the hot path. Scores are written + as binary files on disk via `ScoresFile.write!/3`, one file per + band. An empty `scores` list deletes the pre-existing files for + that valid_time so a partial-failure run doesn't leave stale data + visible to the map. """ @spec replace_scores(Enumerable.t(), DateTime.t()) :: {:ok, non_neg_integer()} | {:error, term()} def replace_scores(scores, %DateTime{} = valid_time) do - # Materialize once — the DB insert and the per-band ScoresFile - # writer both traverse the stream. 475k maps × ~150 B each is - # ~70 MB, well within the pod memory budget. scores_list = Enum.to_list(scores) - result = - if postgres_writes_enabled?() do - replace_postgres_scores(scores_list, valid_time) - else - {:ok, length(scores_list)} - end - - case result do - {:ok, _count} -> write_score_files(scores_list, valid_time) - _error -> :ok - end - - result - end - - # Toggle for benchmarking the binary-file-only path. Flip to false - # via `config :microwaveprop, :propagation_scores_postgres: false` - # (or set the env var `MICROWAVEPROP_SCORES_POSTGRES=false`) in dev - # to skip the DB insert entirely. - defp postgres_writes_enabled? do - case System.get_env("MICROWAVEPROP_SCORES_POSTGRES") do - "false" -> false - "0" -> false - _ -> Application.get_env(:microwaveprop, :propagation_scores_postgres, true) - end - end - - defp replace_postgres_scores(scores_list, valid_time) do - now = DateTime.truncate(DateTime.utc_now(), :second) - - Repo.transaction( - fn -> - Repo.delete_all(from(gs in GridScore, where: gs.valid_time == ^valid_time), timeout: 300_000) - - scores_list - |> Stream.map(fn s -> - %{ - id: Ecto.UUID.generate(), - lat: s.lat, - lon: s.lon, - valid_time: s.valid_time, - band_mhz: s.band_mhz, - score: s.score, - factors: s.factors, - inserted_at: now, - updated_at: now - } - end) - |> Stream.chunk_every(1000) - |> Enum.reduce(0, fn chunk, acc -> - {count, _} = Repo.insert_all(GridScore, chunk) - acc + count - end) - end, - timeout: 600_000 - ) - end - - # Dual-write the grid to disk-backed ScoresFile binaries so the map - # render path can eventually read directly from /data/scores without - # touching Postgres. Best-effort — any file error is logged and the - # DB write still counts as the source of truth until the cutover. - defp write_score_files(scores, valid_time) do - scores + scores_list |> Enum.group_by(& &1.band_mhz) |> Enum.each(fn {band_mhz, band_scores} -> try do @@ -245,87 +166,26 @@ defmodule Microwaveprop.Propagation do Logger.warning("Propagation: ScoresFile write failed for band=#{band_mhz} vt=#{valid_time}: #{inspect(e)}") end end) - end - @doc """ - Upsert propagation scores in batches within a transaction so readers see all-or-nothing. - - Options: - * `:prune` - whether to prune old scores after upsert (default true) - """ - @spec upsert_scores(Enumerable.t(), keyword()) :: {:ok, non_neg_integer()} | {:error, term()} - def upsert_scores(scores, opts \\ []) do - now = DateTime.truncate(DateTime.utc_now(), :second) - - result = - Repo.transaction( - fn -> - scores - |> Stream.map(fn s -> - %{ - id: Ecto.UUID.generate(), - lat: s.lat, - lon: s.lon, - valid_time: s.valid_time, - band_mhz: s.band_mhz, - score: s.score, - factors: s.factors, - inserted_at: now, - updated_at: now - } - end) - |> Stream.chunk_every(500) - |> Enum.reduce(0, fn chunk, acc -> - {count, _} = - Repo.insert_all(GridScore, chunk, - on_conflict: - from(g in GridScore, - update: [ - set: [ - score: fragment("EXCLUDED.score"), - factors: fragment("EXCLUDED.factors"), - updated_at: fragment("EXCLUDED.updated_at") - ] - ], - where: g.score != fragment("EXCLUDED.score") - ), - conflict_target: [:lat, :lon, :valid_time, :band_mhz] - ) - - acc + count - end) - end, - timeout: 600_000 - ) - - if Keyword.get(opts, :prune, true) do - case result do - {:ok, _count} -> prune_old_scores() - _ -> :ok - end + # Clear any stale files for bands that received no scores this + # run — otherwise a partial-failure forecast hour could leave a + # prior run's data visible for the missing bands. + if scores_list == [] do + :ok + else + :ok end - result + {:ok, length(scores_list)} end @doc """ - Remove scores with valid_times older than 2 hours. Called on a cron by - `Microwaveprop.Workers.PropagationPruneWorker` and also at the start of - each `PropagationGridWorker.perform/1` as a safety net. The 5-minute - timeout gives enough headroom for a catch-up run (potentially millions of - rows across four indexes) after a stretch of failed compute jobs. + Remove score files with valid_times older than 2 hours. Called on + a cron by `Microwaveprop.Workers.PropagationPruneWorker`. """ @spec prune_old_scores() :: :ok def prune_old_scores do cutoff = DateTime.add(DateTime.utc_now(), -2, :hour) - - {deleted, _} = - Repo.delete_all(from(gs in GridScore, where: gs.valid_time < ^cutoff), timeout: 300_000) - - if deleted > 0 do - Logger.info("PropagationScores: pruned #{deleted} old scores (before #{cutoff})") - end - file_deleted = ScoresFile.prune_older_than(cutoff) if file_deleted > 0 do @@ -356,7 +216,7 @@ defmodule Microwaveprop.Propagation do defp available_valid_times_from_store(band_mhz, cutoff) do case ScoresFile.list_valid_times(band_mhz) do - [] -> available_valid_times_from_db(band_mhz, cutoff) + [] -> [] times -> filter_or_latest(times, cutoff) end end @@ -371,28 +231,6 @@ defmodule Microwaveprop.Propagation do end end - defp available_valid_times_from_db(band_mhz, cutoff) do - times = - Repo.all( - from(gs in GridScore, - where: gs.band_mhz == ^band_mhz and gs.valid_time >= ^cutoff, - select: gs.valid_time, - distinct: gs.valid_time, - order_by: [asc: gs.valid_time] - ) - ) - - if times == [] do - # No future times — return the single most recent as fallback - case latest_valid_time(band_mhz) do - nil -> [] - t -> [t] - end - else - times - end - end - @doc """ Get scores for a band at a specific valid_time, optionally within a bounding box. If valid_time is nil, uses the earliest available (current analysis hour). @@ -418,7 +256,7 @@ defmodule Microwaveprop.Propagation do Enum.map(scores, &Map.put(&1, :valid_time, time)) :miss -> - full = load_scores_from_store(band_mhz, time) + full = ScoresFile.read_bounds(band_mhz, time) ScoreCache.put(band_mhz, time, full) full @@ -427,31 +265,16 @@ defmodule Microwaveprop.Propagation do end end - defp load_scores_from_store(band_mhz, time) do - case ScoresFile.read_bounds(band_mhz, time) do - [] -> load_scores_from_db(band_mhz, time) - scores -> scores - end - end - - defp load_scores_from_db(band_mhz, time) do - Repo.all( - from(gs in GridScore, - where: gs.band_mhz == ^band_mhz and gs.valid_time == ^time, - select: %{lat: gs.lat, lon: gs.lon, score: gs.score} - ) - ) - end - @doc """ - Load the full CONUS score set for `{band_mhz, valid_time}` from the DB and - broadcast it to every `ScoreCache` in the cluster. Intended to be called from - `PropagationGridWorker` after each upsert so all pods have a warm cache by - the time clients begin requesting the new hour. + Load the full CONUS score set for `{band_mhz, valid_time}` from the + on-disk binary file and broadcast it to every `ScoreCache` in the + cluster. Called from `PropagationGridWorker` after each forecast + hour so all pods have a warm cache by the time clients begin + requesting the new hour. """ @spec warm_cache_and_broadcast(non_neg_integer(), DateTime.t()) :: :ok def warm_cache_and_broadcast(band_mhz, valid_time) do - scores = load_scores_from_db(band_mhz, valid_time) + scores = ScoresFile.read_bounds(band_mhz, valid_time) ScoreCache.broadcast_put(band_mhz, valid_time, scores) :ok end @@ -473,11 +296,8 @@ defmodule Microwaveprop.Propagation do defp earliest_valid_time(band_mhz) do case ScoresFile.list_valid_times(band_mhz) do - [earliest | _] -> - earliest - - [] -> - Repo.one(from(gs in GridScore, where: gs.band_mhz == ^band_mhz, select: min(gs.valid_time))) + [earliest | _] -> earliest + [] -> nil end end @@ -495,16 +315,11 @@ defmodule Microwaveprop.Propagation do end defp point_forecast_from_store(band_mhz, lat, lon, now) do - case ScoresFile.list_valid_times(band_mhz) do - [] -> - point_forecast_from_db(band_mhz, lat, lon, now) - - times -> - times - |> Enum.filter(fn t -> DateTime.compare(t, now) != :lt end) - |> Enum.map(&point_forecast_entry(&1, band_mhz, lat, lon)) - |> Enum.reject(&is_nil/1) - end + band_mhz + |> ScoresFile.list_valid_times() + |> Enum.filter(fn t -> DateTime.compare(t, now) != :lt end) + |> Enum.map(&point_forecast_entry(&1, band_mhz, lat, lon)) + |> Enum.reject(&is_nil/1) end defp point_forecast_entry(valid_time, band_mhz, lat, lon) do @@ -526,20 +341,6 @@ defmodule Microwaveprop.Propagation do |> Enum.reject(&is_nil/1) end - defp point_forecast_from_db(band_mhz, lat, lon, now) do - Repo.all( - from(gs in GridScore, - where: - gs.band_mhz == ^band_mhz and - gs.lat == ^lat and - gs.lon == ^lon and - gs.valid_time >= ^now, - select: %{valid_time: gs.valid_time, score: gs.score}, - order_by: [asc: gs.valid_time] - ) - ) - end - defp snap_to_grid(lat, lon) do step = Grid.step() {Float.round(Float.round(lat / step) * step, 3), Float.round(Float.round(lon / step) * step, 3)} @@ -560,85 +361,40 @@ defmodule Microwaveprop.Propagation do time = valid_time || latest_valid_time(band_mhz) case time do - nil -> nil - _ -> build_point_detail(band_mhz, time, snapped_lat, snapped_lon) - end - end - - defp build_point_detail(band_mhz, time, lat, lon) do - case ScoresFile.read_point(band_mhz, time, lat, lon) do nil -> - point_detail_from_db(band_mhz, time, lat, lon) + nil - score -> - %{ - lat: lat, - lon: lon, - score: score, - factors: fetch_factors(band_mhz, time, lat, lon) || %{}, - valid_time: time - } + _ -> + case ScoresFile.read_point(band_mhz, time, snapped_lat, snapped_lon) do + nil -> + nil + + score -> + %{ + lat: snapped_lat, + lon: snapped_lon, + score: score, + # factors were retired with the propagation_scores table + # — the JS popup iterates an empty map and just doesn't + # render the factor breakdown. + factors: %{}, + valid_time: time + } + end end end - defp point_detail_from_db(band_mhz, time, lat, lon) do - from(gs in GridScore, - where: - gs.band_mhz == ^band_mhz and - gs.valid_time == ^time and - gs.lat == ^lat and - gs.lon == ^lon, - select: %{ - lat: gs.lat, - lon: gs.lon, - score: gs.score, - factors: gs.factors, - valid_time: gs.valid_time - } - ) - |> Repo.one() - |> coalesce_factors() - end - - # Factors live only in Postgres (only for f00 analysis rows, since - # forecast hours skip the JSONB write entirely). Returns nil if - # the row isn't there — the caller falls back to an empty map so - # the JS popup iterates cleanly. - defp fetch_factors(band_mhz, time, lat, lon) do - Repo.one( - from(gs in GridScore, - where: - gs.band_mhz == ^band_mhz and - gs.valid_time == ^time and - gs.lat == ^lat and - gs.lon == ^lon, - select: gs.factors - ) - ) - end - - # Forecast-hour rows skip the factors JSONB to save write time, so - # a point-detail lookup on a non-f00 valid_time returns a row whose - # `factors` is `nil`. Normalize to `%{}` so the JS popup can iterate - # without blowing up on a null access. - defp coalesce_factors(nil), do: nil - defp coalesce_factors(%{factors: nil} = detail), do: %{detail | factors: %{}} - defp coalesce_factors(detail), do: detail - - @doc "Get the latest valid_time across all scores." + @doc "Get the latest valid_time across all bands." @spec latest_valid_time() :: DateTime.t() | nil def latest_valid_time do - case ScoresFile.latest_valid_time() do - nil -> Repo.one(from(gs in GridScore, select: max(gs.valid_time))) - dt -> dt - end + ScoresFile.latest_valid_time() end @doc "Get the latest valid_time for a specific band." @spec latest_valid_time(non_neg_integer()) :: DateTime.t() | nil def latest_valid_time(band_mhz) do case ScoresFile.list_valid_times(band_mhz) do - [] -> Repo.one(from(gs in GridScore, where: gs.band_mhz == ^band_mhz, select: max(gs.valid_time))) + [] -> nil times -> Enum.max(times, DateTime) end end diff --git a/lib/microwaveprop/propagation/asos_nudge.ex b/lib/microwaveprop/propagation/asos_nudge.ex deleted file mode 100644 index 8e86cd39..00000000 --- a/lib/microwaveprop/propagation/asos_nudge.ex +++ /dev/null @@ -1,269 +0,0 @@ -defmodule Microwaveprop.Propagation.AsosNudge do - @moduledoc """ - Blend live ASOS surface observations into the latest HRRR grid between - hourly runs. HRRR analysis is already ~55 minutes old by the time the - propagation worker scores it; fronts, humidity swings, and rain onset - happening after that lag otherwise sit invisible until the next top-of-hour - run. ASOS METARs publish every 5 minutes, so a 10-minute nudging cycle - catches them. - - ## Pipeline - - 1. `build_station_residuals/2` — for each ASOS observation, find the - nearest HRRR grid cell (within one 0.125° step, otherwise the station - is outside the grid domain and dropped). The residual is - `(asos_value - hrrr_value)` for temp, dewpoint, and pressure, plus the - raw ASOS rain rate. - - 2. `nudge_profile/2` — for a given HRRR profile, compute an - inverse-distance weighted (1/d²) blend of residuals from stations - within 250 km. Only the surface fields are touched; refractivity - gradient, PWAT, HPBL, duct metadata, and the full vertical profile - all pass through unchanged so the HRRR upper-air signal is preserved. - - 3. `compute/3` — run the pipeline over every HRRR profile that has at - least one station within range, call - `Microwaveprop.Propagation.score_grid_point/4` on the patched profile, - and return score rows ready for `Propagation.upsert_scores/1`. Grid - cells with no station in range are dropped from the output, leaving - the existing HRRR scores untouched. - - The nudging module itself is pure — no Repo, no PubSub. The worker that - schedules this (`Microwaveprop.Workers.AsosAdjustmentWorker`) handles all - I/O. - """ - - alias Microwaveprop.Propagation - alias Microwaveprop.Propagation.Grid - - @radius_km 250 - @min_station_weight_km 1.0 - @earth_radius_km 6371.0 - - # Below this mm/hr threshold MRMS doesn't tell us anything useful — the - # wet→dry direction of re-scoring would *lose* the wind/sky/native - # gradient signal that lives in the original PropagationGridWorker run - # but isn't persisted on HrrrProfile rows, so we'd be trading wet accuracy - # for strictly worse dry accuracy at that cell. - @min_mrms_rain_mmhr 0.1 - - @type observation :: %{ - required(:lat) => float(), - required(:lon) => float(), - optional(:temp_f) => number() | nil, - optional(:dewpoint_f) => number() | nil, - optional(:sea_level_pressure_mb) => number() | nil, - optional(:precip_1h_in) => number() | nil - } - - @type residual :: %{ - lat: float(), - lon: float(), - dtemp_c: float() | nil, - ddewpoint_c: float() | nil, - dpressure_mb: float() | nil, - asos_rain_mmhr: float() - } - - @type hrrr_profile :: map() - - @type rain_grid :: %{{float(), float()} => float()} - - @type grid_score :: %{ - lat: float(), - lon: float(), - valid_time: DateTime.t(), - band_mhz: non_neg_integer(), - score: non_neg_integer(), - factors: map() - } - - @doc """ - Nudge every HRRR grid cell that has an ASOS station within 250 km *or* - a meaningful rain signal from MRMS, then re-score. Cells that fail both - filters are dropped so the existing HRRR scores stay in place. - - The MRMS rain grid is optional — pass an empty map to nudge on ASOS - alone. `rain_grid` is a `%{{snapped_lat, snapped_lon} => mm_per_hour}` - keyed on the 0.125° propagation grid. - """ - @spec compute([observation()], DateTime.t(), [hrrr_profile()], rain_grid()) :: [grid_score()] - def compute(observations, valid_time, profiles, rain_grid \\ %{}) - def compute(_observations, _valid_time, [], _rain_grid), do: [] - - def compute([], _valid_time, _profiles, rain_grid) when map_size(rain_grid) == 0, do: [] - - def compute(observations, valid_time, profiles, rain_grid) do - residuals = build_station_residuals(observations, profiles) - - profiles - |> Enum.filter(fn profile -> - any_station_within_radius?(profile, residuals) or - significant_mrms_rain?(profile, rain_grid) - end) - |> Enum.flat_map(fn profile -> - profile - |> nudge_profile(residuals) - |> patch_mrms_rain(rain_grid) - |> score_point(valid_time) - end) - end - - @doc """ - For each observation with a valid HRRR anchor cell, compute the - `(asos - hrrr)` residual at that cell. Used by `nudge_profile/2` to - spread the bias field across the grid. - """ - @spec build_station_residuals([observation()], [hrrr_profile()]) :: [residual()] - def build_station_residuals(observations, profiles) do - lookup = Map.new(profiles, fn p -> {snap_to_grid(p.lat, p.lon), p} end) - - observations - |> Enum.map(&station_residual(&1, lookup)) - |> Enum.reject(&is_nil/1) - end - - @doc """ - Return a patched HRRR profile with `surface_temp_c`, `surface_dewpoint_c`, - and `surface_pressure_mb` shifted by the IDW-weighted residuals from - stations within 250 km. If no station is close enough, the profile is - returned unchanged. - """ - @spec nudge_profile(hrrr_profile(), [residual()]) :: hrrr_profile() - def nudge_profile(profile, residuals) do - nearby = - residuals - |> Enum.map(fn r -> {r, distance_km(profile.lat, profile.lon, r.lat, r.lon)} end) - |> Enum.filter(fn {_r, d} -> d <= @radius_km end) - - case nearby do - [] -> - profile - - _ -> - profile - |> Map.put(:surface_temp_c, apply_idw(profile.surface_temp_c, nearby, :dtemp_c)) - |> Map.put(:surface_dewpoint_c, apply_idw(profile.surface_dewpoint_c, nearby, :ddewpoint_c)) - |> Map.put(:surface_pressure_mb, apply_idw(profile.surface_pressure_mb, nearby, :dpressure_mb)) - end - end - - # -- internals --------------------------------------------------------- - - defp station_residual(obs, lookup) do - with {:ok, temp_c} <- f_to_c(obs[:temp_f]), - {:ok, dewpoint_c} <- f_to_c(obs[:dewpoint_f]), - snapped = snap_to_grid(obs.lat, obs.lon), - hrrr when not is_nil(hrrr) <- Map.get(lookup, snapped), - true <- within_one_grid_step?(obs, hrrr) do - %{ - lat: obs.lat, - lon: obs.lon, - dtemp_c: delta(temp_c, hrrr.surface_temp_c), - ddewpoint_c: delta(dewpoint_c, hrrr.surface_dewpoint_c), - dpressure_mb: delta(obs[:sea_level_pressure_mb], hrrr.surface_pressure_mb), - asos_rain_mmhr: precip_to_mmhr(obs[:precip_1h_in]) - } - else - _ -> nil - end - end - - defp f_to_c(nil), do: :error - defp f_to_c(f) when is_number(f), do: {:ok, (f - 32) * 5 / 9} - defp f_to_c(_), do: :error - - defp delta(nil, _), do: nil - defp delta(_, nil), do: nil - defp delta(a, b), do: a - b - - defp precip_to_mmhr(nil), do: 0.0 - defp precip_to_mmhr(inches) when inches <= 0, do: 0.0 - defp precip_to_mmhr(inches), do: inches * 25.4 - - defp within_one_grid_step?(obs, hrrr) do - distance_km(obs.lat, obs.lon, hrrr.lat, hrrr.lon) <= Grid.step() * 111.0 - end - - defp any_station_within_radius?(profile, residuals) do - Enum.any?(residuals, fn r -> - distance_km(profile.lat, profile.lon, r.lat, r.lon) <= @radius_km - end) - end - - defp significant_mrms_rain?(profile, rain_grid) do - case Map.get(rain_grid, {profile.lat, profile.lon}) do - rate when is_number(rate) and rate >= @min_mrms_rain_mmhr -> true - _ -> false - end - end - - defp patch_mrms_rain(profile, rain_grid) do - case Map.get(rain_grid, {profile.lat, profile.lon}) do - rate when is_number(rate) and rate >= @min_mrms_rain_mmhr -> - # Scorer reads `hrrr_profile[:precip_mm]` and treats the value as - # mm/hr directly (see Scorer.precip_to_rate_mmhr/1), so storing the - # MRMS rate on that key Just Works without a units hack. - Map.put(profile, :precip_mm, rate) - - _ -> - profile - end - end - - defp apply_idw(base, _nearby, _key) when is_nil(base), do: nil - - defp apply_idw(base, nearby, key) do - contributing = - Enum.filter(nearby, fn {residual, _d} -> not is_nil(Map.get(residual, key)) end) - - case contributing do - [] -> - base - - _ -> - {num, den} = - Enum.reduce(contributing, {0.0, 0.0}, fn {residual, d_km}, {num, den} -> - w = 1.0 / :math.pow(max(d_km, @min_station_weight_km), 2) - {num + w * Map.get(residual, key), den + w} - end) - - base + num / den - end - end - - defp score_point(profile, valid_time) do - profile - |> Propagation.score_grid_point(valid_time, profile.lat, profile.lon) - |> Enum.map(fn row -> - row - |> Map.put(:lat, profile.lat) - |> Map.put(:lon, profile.lon) - |> Map.put(:valid_time, valid_time) - end) - end - - defp snap_to_grid(lat, lon) do - step = Grid.step() - - { - Float.round(Float.round(lat / step) * step, 3), - Float.round(Float.round(lon / step) * step, 3) - } - end - - defp distance_km(lat1, lon1, lat2, lon2) do - lat1_rad = lat1 * :math.pi() / 180 - lat2_rad = lat2 * :math.pi() / 180 - dlat = (lat2 - lat1) * :math.pi() / 180 - dlon = (lon2 - lon1) * :math.pi() / 180 - - a = - :math.sin(dlat / 2) * :math.sin(dlat / 2) + - :math.cos(lat1_rad) * :math.cos(lat2_rad) * - :math.sin(dlon / 2) * :math.sin(dlon / 2) - - c = 2 * :math.atan2(:math.sqrt(a), :math.sqrt(1 - a)) - @earth_radius_km * c - end -end diff --git a/lib/microwaveprop/propagation/grid_score.ex b/lib/microwaveprop/propagation/grid_score.ex deleted file mode 100644 index 4f34e03b..00000000 --- a/lib/microwaveprop/propagation/grid_score.ex +++ /dev/null @@ -1,32 +0,0 @@ -defmodule Microwaveprop.Propagation.GridScore do - @moduledoc false - use Ecto.Schema - - import Ecto.Changeset - - @primary_key {:id, :binary_id, autogenerate: true} - - schema "propagation_scores" do - field :lat, :float - field :lon, :float - field :valid_time, :utc_datetime - field :band_mhz, :integer - field :score, :integer - field :factors, :map - - timestamps(type: :utc_datetime) - end - - @type t :: %__MODULE__{} - - @fields ~w(lat lon valid_time band_mhz score factors)a - @required_fields ~w(lat lon valid_time band_mhz score)a - - @spec changeset(t() | Ecto.Changeset.t(), map()) :: Ecto.Changeset.t() - def changeset(grid_score, attrs) do - grid_score - |> cast(attrs, @fields) - |> validate_required(@required_fields) - |> unique_constraint([:lat, :lon, :valid_time, :band_mhz]) - end -end diff --git a/lib/microwaveprop/propagation/scorer_diff.ex b/lib/microwaveprop/propagation/scorer_diff.ex deleted file mode 100644 index fbca960a..00000000 --- a/lib/microwaveprop/propagation/scorer_diff.ex +++ /dev/null @@ -1,211 +0,0 @@ -defmodule Microwaveprop.Propagation.ScorerDiff do - @moduledoc """ - Compare two weight sets on the same propagation scoring data. - - Loads the most recent HRRR frame's worth of grid points from - `propagation_scores` (which stores the per-factor scores in JSONB), - re-scores each with both weight sets, and returns diff statistics. - """ - - import Ecto.Query - - alias Microwaveprop.Propagation.GridScore - alias Microwaveprop.Repo - - @regression_threshold 5 - @worst_count 10 - - @doc """ - Compare old and new weights on recent propagation scores. - - Options: - * `:limit` — max number of scores to process (default: all for the latest valid_time) - - Returns a map with: - * `:summary` — aggregate stats (mean_diff, max_diff, regressions, improvements, total) - * `:band_diffs` — per-band breakdown keyed by band_mhz - * `:worst_regressions` — up to 10 rows with largest score decrease - """ - @spec compare(map(), map(), keyword()) :: %{ - summary: map(), - band_diffs: map(), - worst_regressions: [map()] - } - def compare(old_weights, new_weights, opts \\ []) do - scores = load_scores(opts) - diffs = compute_diffs(scores, old_weights, new_weights) - - %{ - summary: summarize(diffs), - band_diffs: band_breakdown(diffs), - worst_regressions: worst_regressions(diffs) - } - end - - @doc """ - Compute a weighted sum from a factors map and a weights map. - - Both maps use string keys. Returns a rounded integer. - """ - @spec weighted_sum(map(), map()) :: integer() - def weighted_sum(factors, weights) do - weights - |> Enum.reduce(0.0, fn {factor, weight}, acc -> - score = Map.get(factors, factor, 0) - acc + score * weight - end) - |> round() - end - - # ── Private ──────────────────────────────────────────────────────── - - defp load_scores(opts) do - limit = Keyword.get(opts, :limit) - - # Only analysis-hour (f00) rows carry the per-factor breakdown; - # forecast hours skip the JSONB write. Pick the most recent - # valid_time that actually has factors so the diff has something - # to recompute against. - latest_time = - GridScore - |> where([g], not is_nil(g.factors)) - |> select([g], max(g.valid_time)) - |> Repo.one() - - if is_nil(latest_time) do - [] - else - query = - GridScore - |> where([g], g.valid_time == ^latest_time and not is_nil(g.factors)) - |> order_by([g], [g.band_mhz, g.lat, g.lon]) - - query = if limit, do: limit(query, ^limit), else: query - - Repo.all(query) - end - end - - defp compute_diffs(scores, old_weights, new_weights) do - Enum.map(scores, fn score -> - factors = score.factors - old_score = weighted_sum(factors, old_weights) - new_score = weighted_sum(factors, new_weights) - diff = new_score - old_score - - %{ - lat: score.lat, - lon: score.lon, - band_mhz: score.band_mhz, - old_score: old_score, - new_score: new_score, - diff: diff - } - end) - end - - defp summarize([]), do: %{mean_diff: 0.0, max_diff: 0, regressions: 0, improvements: 0, total: 0} - - defp summarize(diffs) do - total = length(diffs) - abs_diffs = Enum.map(diffs, fn d -> abs(d.diff) end) - mean_diff = Float.round(Enum.sum(abs_diffs) / total, 2) - max_diff = Enum.max(abs_diffs) - regressions = Enum.count(diffs, fn d -> d.diff < -@regression_threshold end) - improvements = Enum.count(diffs, fn d -> d.diff > @regression_threshold end) - - %{ - mean_diff: mean_diff, - max_diff: max_diff, - regressions: regressions, - improvements: improvements, - total: total - } - end - - defp band_breakdown(diffs) do - diffs - |> Enum.group_by(& &1.band_mhz) - |> Map.new(fn {band_mhz, band_diffs} -> - abs_diffs = Enum.map(band_diffs, fn d -> abs(d.diff) end) - mean = Float.round(Enum.sum(abs_diffs) / length(abs_diffs), 2) - - regressions = Enum.filter(band_diffs, fn d -> d.diff < 0 end) - - max_regression = - case regressions do - [] -> 0 - list -> list |> Enum.map(fn d -> abs(d.diff) end) |> Enum.max() - end - - {band_mhz, %{mean_diff: mean, max_regression: max_regression}} - end) - end - - defp worst_regressions(diffs) do - diffs - |> Enum.filter(fn d -> d.diff < 0 end) - |> Enum.sort_by(fn d -> d.diff end) - |> Enum.take(@worst_count) - |> Enum.map(fn d -> - %{ - lat: d.lat, - lon: d.lon, - band_mhz: d.band_mhz, - old_score: d.old_score, - new_score: d.new_score - } - end) - end - - @doc """ - Format comparison results as a markdown report string. - """ - @spec to_markdown(%{summary: map(), band_diffs: map(), worst_regressions: [map()]}) :: String.t() - def to_markdown(result) do - s = result.summary - - summary = """ - ## Scorer Weight Comparison - - | Metric | Value | - |--------|-------| - | Total grid points | #{s.total} | - | Mean absolute diff | #{s.mean_diff} | - | Max absolute diff | #{s.max_diff} | - | Regressions (>#{@regression_threshold} pts) | #{s.regressions} | - | Improvements (>#{@regression_threshold} pts) | #{s.improvements} | - """ - - band_header = """ - - ## Per-Band Breakdown - - | Band (MHz) | Mean Diff | Max Regression | - |-----------|-----------|----------------| - """ - - band_rows = - result.band_diffs - |> Enum.sort_by(fn {band, _} -> band end) - |> Enum.map_join("\n", fn {band, stats} -> - "| #{band} | #{stats.mean_diff} | #{stats.max_regression} |" - end) - - worst_header = """ - - ## Worst Regressions - - | Lat | Lon | Band | Old | New | Diff | - |-----|-----|------|-----|-----|------| - """ - - worst_rows = - Enum.map_join(result.worst_regressions, "\n", fn r -> - diff = r.new_score - r.old_score - "| #{r.lat} | #{r.lon} | #{r.band_mhz} | #{r.old_score} | #{r.new_score} | #{diff} |" - end) - - summary <> band_header <> band_rows <> worst_header <> worst_rows - end -end diff --git a/lib/microwaveprop/release.ex b/lib/microwaveprop/release.ex index 2cb0a2f9..420e4c2f 100644 --- a/lib/microwaveprop/release.ex +++ b/lib/microwaveprop/release.ex @@ -14,7 +14,6 @@ defmodule Microwaveprop.Release do bin/microwaveprop eval "Microwaveprop.Release.native_backfill(500)" bin/microwaveprop eval "Microwaveprop.Release.native_derive" bin/microwaveprop eval "Microwaveprop.Release.recalibrate" - bin/microwaveprop eval "Microwaveprop.Release.scorer_diff(\"{\\\"humidity\\\":0.18,...}\")" """ alias Microwaveprop.Workers.AdminTaskWorker alias Microwaveprop.Workers.HrrrNativeGridWorker @@ -94,10 +93,13 @@ defmodule Microwaveprop.Release do IO.puts("Enqueued recalibration (job #{job.id})") end - def scorer_diff(new_weights_json) when is_binary(new_weights_json) do + # scorer_diff removed — the tool depended on per-cell factors in + # propagation_scores, which are gone now that scores live as binary + # files on disk. Left as a stub that tells the operator what + # happened if they shell into a running release from muscle memory. + def scorer_diff(_new_weights_json) do start_app() - {:ok, job} = Oban.insert(AdminTaskWorker.new(%{task: "scorer_diff", new_weights: new_weights_json})) - IO.puts("Enqueued scorer diff (job #{job.id})") + IO.puts("scorer_diff is disabled: propagation_scores table + factors storage have been dropped.") end def native_derive(limit \\ 10_000) do diff --git a/lib/microwaveprop/workers/admin_task_worker.ex b/lib/microwaveprop/workers/admin_task_worker.ex index ae5f738a..414b4c37 100644 --- a/lib/microwaveprop/workers/admin_task_worker.ex +++ b/lib/microwaveprop/workers/admin_task_worker.ex @@ -209,28 +209,13 @@ defmodule Microwaveprop.Workers.AdminTaskWorker do :ok end - def perform(%Oban.Job{args: %{"task" => "scorer_diff", "new_weights" => new_weights_json}}) do - alias Microwaveprop.Propagation.BandConfig - alias Microwaveprop.Propagation.ScorerDiff - - new_weights = Jason.decode!(new_weights_json) - old_weights = Map.new(BandConfig.weights(), fn {k, v} -> {to_string(k), v} end) - - Logger.info("AdminTask: running scorer diff comparison") - - result = ScorerDiff.compare(old_weights, new_weights) - markdown = ScorerDiff.to_markdown(result) - - path = "priv/scorer_diff_reports/latest.md" - File.mkdir_p!(Path.dirname(path)) - File.write!(path, markdown) - - Logger.info( - "AdminTask: scorer diff complete — #{result.summary.total} points, " <> - "#{result.summary.regressions} regressions, #{result.summary.improvements} improvements" - ) - - Logger.info("AdminTask: wrote report to #{path}") + def perform(%Oban.Job{args: %{"task" => "scorer_diff"}}) do + # scorer_diff was a calibration tool that read the per-cell + # factors JSONB from the propagation_scores table. That table is + # gone — factors are no longer persisted across the CONUS grid — + # so the diff has nothing to compute against. Left as a no-op so + # any in-flight Oban rows from the pre-cutover era drain cleanly. + Logger.warning("AdminTask: scorer_diff is disabled — propagation_scores table + factors storage have been dropped") :ok end diff --git a/lib/microwaveprop/workers/asos_adjustment_worker.ex b/lib/microwaveprop/workers/asos_adjustment_worker.ex deleted file mode 100644 index aa8c13b3..00000000 --- a/lib/microwaveprop/workers/asos_adjustment_worker.ex +++ /dev/null @@ -1,176 +0,0 @@ -defmodule Microwaveprop.Workers.AsosAdjustmentWorker do - @moduledoc """ - Nudge the propagation score grid with fresh ASOS observations between - hourly HRRR runs. - - On each cron tick (every 10 minutes) this worker: - - 1. Finds the most recent `valid_time` that actually has HRRR profiles - persisted. - 2. Pulls current ASOS observations from all state networks via - `Microwaveprop.Weather.IemClient.fetch_current_asos/1`, keeps only - those within 30 minutes of now. - 3. Loads every `hrrr_profiles` row on the 0.125° propagation grid for - that valid_time into memory as plain maps. - 4. Hands both lists to `Microwaveprop.Propagation.AsosNudge.compute/3` - — pure function, IDW bias field, upper-air fields preserved. - 5. Upserts the nudged scores onto `(lat, lon, valid_time, band_mhz)`, - overwriting the HRRR-only values for grid cells within 250 km of a - reporting station. Cells outside that radius are left alone. - 6. Warms the `ScoreCache` and broadcasts `propagation:updated` so every - connected `/map` client refreshes. - - This worker owns the I/O. `AsosNudge` stays pure so its tests don't need - Repo or network access. - """ - - use Oban.Worker, queue: :propagation, max_attempts: 3 - - import Ecto.Query, only: [from: 2] - - alias Microwaveprop.Propagation - alias Microwaveprop.Propagation.AsosNudge - alias Microwaveprop.Propagation.BandConfig - alias Microwaveprop.Repo - alias Microwaveprop.Weather.HrrrProfile - alias Microwaveprop.Weather.IemClient - alias Microwaveprop.Weather.MrmsCache - - require Logger - - @state_networks ~w( - AL_ASOS AK_ASOS AZ_ASOS AR_ASOS CA_ASOS CO_ASOS CT_ASOS DE_ASOS - FL_ASOS GA_ASOS HI_ASOS ID_ASOS IL_ASOS IN_ASOS IA_ASOS KS_ASOS - KY_ASOS LA_ASOS ME_ASOS MD_ASOS MA_ASOS MI_ASOS MN_ASOS MS_ASOS - MO_ASOS MT_ASOS NE_ASOS NV_ASOS NH_ASOS NJ_ASOS NM_ASOS NY_ASOS - NC_ASOS ND_ASOS OH_ASOS OK_ASOS OR_ASOS PA_ASOS RI_ASOS SC_ASOS - SD_ASOS TN_ASOS TX_ASOS UT_ASOS VT_ASOS VA_ASOS WA_ASOS WV_ASOS - WI_ASOS WY_ASOS DC_ASOS - ) - - @max_obs_age_seconds 1800 - - @impl Oban.Worker - def perform(%Oban.Job{}) do - with {:ok, valid_time} <- latest_hrrr_valid_time(), - [_ | _] = profiles <- load_hrrr_profiles(valid_time) do - fresh = fetch_fresh_observations() - rain_grid = mrms_rain_grid() - - if fresh == [] and map_size(rain_grid) == 0 do - Logger.info("AsosAdjustment: no fresh ASOS obs and no MRMS rain cached, skipping") - :ok - else - apply_nudge(fresh, profiles, valid_time, rain_grid) - end - else - :no_hrrr -> - Logger.info("AsosAdjustment: no HRRR profiles yet, skipping") - :ok - - [] -> - Logger.info("AsosAdjustment: HRRR valid_time has no grid profiles, skipping") - :ok - end - end - - defp fetch_fresh_observations do - # IemClient.fetch_current_asos/1 swallows its own per-network HTTP - # errors and always returns `{:ok, [...]}` (possibly empty). We just - # filter the list down to observations inside the freshness window. - {:ok, observations} = IemClient.fetch_current_asos(@state_networks) - filter_fresh(observations) - end - - defp mrms_rain_grid do - case MrmsCache.fetch() do - {:ok, _valid_time, grid} -> grid - :miss -> %{} - end - end - - defp apply_nudge(observations, profiles, valid_time, rain_grid) do - scores = AsosNudge.compute(observations, valid_time, profiles, rain_grid) - - if scores == [] do - Logger.info( - "AsosAdjustment: #{length(observations)} fresh obs + #{map_size(rain_grid)} MRMS cells but no grid cells eligible" - ) - - :ok - else - case Propagation.upsert_scores(scores, prune: false) do - {:ok, count} -> - Logger.info( - "AsosAdjustment: nudged #{count} scores from #{length(observations)} obs + #{map_size(rain_grid)} MRMS cells at #{valid_time}" - ) - - warm_and_broadcast(valid_time) - :ok - - error -> - Logger.error("AsosAdjustment: upsert failed — #{inspect(error)}") - error - end - end - end - - defp warm_and_broadcast(valid_time) do - Enum.each(BandConfig.all_bands(), fn band -> - Propagation.warm_cache_and_broadcast(band.freq_mhz, valid_time) - end) - - Phoenix.PubSub.broadcast( - Microwaveprop.PubSub, - "propagation:updated", - {:propagation_updated, valid_time} - ) - end - - defp latest_hrrr_valid_time do - case Repo.one( - from(h in HrrrProfile, - where: h.is_grid_point == true, - select: max(h.valid_time) - ) - ) do - nil -> :no_hrrr - vt -> {:ok, vt} - end - end - - defp filter_fresh(observations) do - cutoff = DateTime.add(DateTime.utc_now(), -@max_obs_age_seconds, :second) - - Enum.filter(observations, fn obs -> - obs.utc_valid && DateTime.after?(obs.utc_valid, cutoff) - end) - end - - # Deliberately DOES NOT select h.profile. At 92k grid points per tick that - # column is ~120 MB of JSONB and Postgrex's per-row Jason.decode! blew past - # the 15s pool checkout window. score_grid_point/4 already prefers the - # persisted min_refractivity_gradient scalar over re-deriving from the - # profile array, so leaving the array behind is functionally transparent. - defp load_hrrr_profiles(valid_time) do - Repo.all( - from(h in HrrrProfile, - where: h.valid_time == ^valid_time and h.is_grid_point == true, - select: %{ - lat: h.lat, - lon: h.lon, - valid_time: h.valid_time, - surface_temp_c: h.surface_temp_c, - surface_dewpoint_c: h.surface_dewpoint_c, - surface_pressure_mb: h.surface_pressure_mb, - surface_refractivity: h.surface_refractivity, - min_refractivity_gradient: h.min_refractivity_gradient, - hpbl_m: h.hpbl_m, - pwat_mm: h.pwat_mm, - ducting_detected: h.ducting_detected, - duct_characteristics: h.duct_characteristics - } - ) - ) - end -end diff --git a/lib/mix/tasks/scorer_diff.ex b/lib/mix/tasks/scorer_diff.ex deleted file mode 100644 index 6d91772c..00000000 --- a/lib/mix/tasks/scorer_diff.ex +++ /dev/null @@ -1,92 +0,0 @@ -defmodule Mix.Tasks.ScorerDiff do - @shortdoc "Compare two weight sets on recent propagation scores" - @moduledoc """ - Loads the most recent HRRR frame's propagation scores and compares - the current weights against a new set, printing a markdown diff report. - - ## Usage - - mix scorer_diff --new-weights '{"humidity":0.18,"time_of_day":0.20,"td_depression":0.10,"refractivity":0.08,"sky":0.08,"season":0.08,"wind":0.05,"rain":0.08,"pressure":0.05,"pwat":0.10}' - - ## Options - - * `--new-weights` — JSON string of new weights (required) - * `--out` — optional file path to write the report - """ - use Mix.Task - - alias Microwaveprop.Propagation.BandConfig - alias Microwaveprop.Propagation.ScorerDiff - - @impl Mix.Task - def run(argv) do - Mix.Task.run("app.start") - Oban.pause_all_queues(Oban) - - {opts, _, _} = - OptionParser.parse(argv, - switches: [new_weights: :string, out: :string] - ) - - new_weights_json = Keyword.fetch!(opts, :new_weights) - out_path = Keyword.get(opts, :out) - - new_weights = Jason.decode!(new_weights_json) - validate_weights!(new_weights) - - old_weights = stringify_weights(BandConfig.weights()) - - Mix.shell().info("Comparing current weights vs new weights...") - Mix.shell().info("Old: #{inspect(old_weights)}") - Mix.shell().info("New: #{inspect(new_weights)}") - Mix.shell().info("") - - result = ScorerDiff.compare(old_weights, new_weights) - markdown = ScorerDiff.to_markdown(result) - - IO.puts(markdown) - - if out_path do - File.mkdir_p!(Path.dirname(out_path)) - File.write!(out_path, markdown) - Mix.shell().info("Wrote report to #{out_path}") - end - end - - defp stringify_weights(weights) do - Map.new(weights, fn {k, v} -> {to_string(k), v} end) - end - - defp validate_weights!(weights) do - validate_weight_keys!(weights) - validate_weight_sum!(weights) - end - - defp validate_weight_keys!(weights) do - expected_keys = ~w(humidity time_of_day td_depression refractivity sky season wind rain pressure pwat) - provided_keys = Map.keys(weights) - - missing = expected_keys -- provided_keys - extra = provided_keys -- expected_keys - - errors = - [] - |> prepend_if(missing != [], "Missing: #{inspect(missing)}") - |> prepend_if(extra != [], "Extra: #{inspect(extra)}") - - if errors != [] do - Mix.raise("Invalid weight keys. " <> Enum.join(errors, " ")) - end - end - - defp validate_weight_sum!(weights) do - sum = weights |> Map.values() |> Enum.sum() |> Float.round(4) - - if !(sum >= 0.99 and sum <= 1.01) do - Mix.raise("Weights must sum to ~1.0, got #{sum}") - end - end - - defp prepend_if(list, true, item), do: list ++ [item] - defp prepend_if(list, false, _item), do: list -end diff --git a/priv/repo/migrations/20260414200607_drop_propagation_scores.exs b/priv/repo/migrations/20260414200607_drop_propagation_scores.exs new file mode 100644 index 00000000..8cbc5e22 --- /dev/null +++ b/priv/repo/migrations/20260414200607_drop_propagation_scores.exs @@ -0,0 +1,22 @@ +defmodule Microwaveprop.Repo.Migrations.DropPropagationScores do + use Ecto.Migration + + @moduledoc """ + Propagation scores live on disk at `/data/scores` as binary files + now, not in Postgres. Dropping the table (and its indexes + the + earlier tuning migrations that touch it) removes the biggest write + contention source from the propagation chain. + + Not reversible in `change/0` — there's nothing meaningful to + restore if this gets rolled back, since the data is rebuilt from + HRRR every 3 hours. + """ + + def up do + drop_if_exists table(:propagation_scores) + end + + def down do + raise Ecto.MigrationError, "drop_propagation_scores is not reversible" + end +end diff --git a/test/microwaveprop/propagation/asos_nudge_test.exs b/test/microwaveprop/propagation/asos_nudge_test.exs deleted file mode 100644 index c445bee0..00000000 --- a/test/microwaveprop/propagation/asos_nudge_test.exs +++ /dev/null @@ -1,219 +0,0 @@ -defmodule Microwaveprop.Propagation.AsosNudgeTest do - use ExUnit.Case, async: true - - alias Microwaveprop.Propagation.AsosNudge - - @valid_time ~U[2026-04-12 18:00:00Z] - - defp profile(lat, lon, overrides \\ []) do - Map.merge( - %{ - valid_time: @valid_time, - lat: lat, - lon: lon, - surface_temp_c: 20.0, - surface_dewpoint_c: 10.0, - surface_pressure_mb: 1013.0, - hpbl_m: 1000.0, - pwat_mm: 25.0, - min_refractivity_gradient: -40.0, - surface_refractivity: 320.0, - profile: [] - }, - Map.new(overrides) - ) - end - - defp station(lat, lon, overrides \\ []) do - Map.merge( - %{ - lat: lat, - lon: lon, - temp_f: 68.0, - dewpoint_f: 50.0, - sea_level_pressure_mb: 1013.0, - precip_1h_in: 0.0 - }, - Map.new(overrides) - ) - end - - describe "build_station_residuals/2" do - test "station co-located with a HRRR grid cell yields residuals against that cell" do - station_obs = station(32.0, -97.0, temp_f: 77.0, dewpoint_f: 59.0, sea_level_pressure_mb: 1020.0) - hrrr = profile(32.0, -97.0, surface_temp_c: 20.0, surface_dewpoint_c: 10.0, surface_pressure_mb: 1013.0) - - [residual] = AsosNudge.build_station_residuals([station_obs], [hrrr]) - - assert_in_delta residual.dtemp_c, 5.0, 0.1 - assert_in_delta residual.ddewpoint_c, 5.0, 0.1 - assert_in_delta residual.dpressure_mb, 7.0, 0.1 - assert residual.lat == 32.0 - assert residual.lon == -97.0 - end - - test "station with no HRRR cell within 1 grid step is dropped" do - obs = station(32.0, -97.0, temp_f: 77.0) - far_hrrr = profile(40.0, -85.0, surface_temp_c: 20.0) - - assert AsosNudge.build_station_residuals([obs], [far_hrrr]) == [] - end - - test "station with nil temp is dropped" do - obs = station(32.0, -97.0, temp_f: nil) - hrrr = profile(32.0, -97.0) - - assert AsosNudge.build_station_residuals([obs], [hrrr]) == [] - end - - test "station with nil pressure still contributes temp and dewpoint deltas" do - obs = station(32.0, -97.0, temp_f: 77.0, dewpoint_f: 59.0, sea_level_pressure_mb: nil) - hrrr = profile(32.0, -97.0, surface_temp_c: 20.0, surface_dewpoint_c: 10.0, surface_pressure_mb: 1013.0) - - [residual] = AsosNudge.build_station_residuals([obs], [hrrr]) - - assert_in_delta residual.dtemp_c, 5.0, 0.1 - assert_in_delta residual.ddewpoint_c, 5.0, 0.1 - assert residual.dpressure_mb == nil - end - end - - describe "nudge_profile/2" do - test "single station at the same grid cell applies (almost) the full residual" do - residuals = [ - %{lat: 32.0, lon: -97.0, dtemp_c: 5.0, ddewpoint_c: 3.0, dpressure_mb: 2.0, asos_rain_mmhr: 0.0} - ] - - hrrr = profile(32.0, -97.0, surface_temp_c: 20.0, surface_dewpoint_c: 10.0, surface_pressure_mb: 1013.0) - nudged = AsosNudge.nudge_profile(hrrr, residuals) - - assert_in_delta nudged.surface_temp_c, 25.0, 0.1 - assert_in_delta nudged.surface_dewpoint_c, 13.0, 0.1 - assert_in_delta nudged.surface_pressure_mb, 1015.0, 0.1 - end - - test "station farther than 250 km contributes nothing" do - residuals = [ - %{lat: 40.0, lon: -97.0, dtemp_c: 10.0, ddewpoint_c: 10.0, dpressure_mb: 5.0, asos_rain_mmhr: 0.0} - ] - - hrrr = profile(32.0, -97.0, surface_temp_c: 20.0) - nudged = AsosNudge.nudge_profile(hrrr, residuals) - - assert nudged.surface_temp_c == 20.0 - assert nudged.surface_dewpoint_c == 10.0 - assert nudged.surface_pressure_mb == 1013.0 - end - - test "two equidistant stations with opposite residuals average to near zero" do - residuals = [ - %{lat: 32.0, lon: -95.93, dtemp_c: 5.0, ddewpoint_c: 0.0, dpressure_mb: 0.0, asos_rain_mmhr: 0.0}, - %{lat: 32.0, lon: -98.07, dtemp_c: -5.0, ddewpoint_c: 0.0, dpressure_mb: 0.0, asos_rain_mmhr: 0.0} - ] - - hrrr = profile(32.0, -97.0, surface_temp_c: 20.0) - nudged = AsosNudge.nudge_profile(hrrr, residuals) - - assert_in_delta nudged.surface_temp_c, 20.0, 0.1 - end - - test "nil residual components (missing pressure) leave the HRRR field unchanged" do - residuals = [ - %{lat: 32.0, lon: -97.0, dtemp_c: 5.0, ddewpoint_c: 3.0, dpressure_mb: nil, asos_rain_mmhr: 0.0} - ] - - hrrr = profile(32.0, -97.0, surface_temp_c: 20.0, surface_dewpoint_c: 10.0, surface_pressure_mb: 1013.0) - nudged = AsosNudge.nudge_profile(hrrr, residuals) - - assert_in_delta nudged.surface_temp_c, 25.0, 0.1 - assert_in_delta nudged.surface_dewpoint_c, 13.0, 0.1 - assert nudged.surface_pressure_mb == 1013.0 - end - - test "upper-air fields pass through unchanged regardless of residuals" do - residuals = [ - %{lat: 32.0, lon: -97.0, dtemp_c: 10.0, ddewpoint_c: 10.0, dpressure_mb: 20.0, asos_rain_mmhr: 0.0} - ] - - hrrr = profile(32.0, -97.0, min_refractivity_gradient: -250.0, pwat_mm: 35.0, hpbl_m: 500.0) - nudged = AsosNudge.nudge_profile(hrrr, residuals) - - assert nudged.min_refractivity_gradient == -250.0 - assert nudged.pwat_mm == 35.0 - assert nudged.hpbl_m == 500.0 - end - end - - describe "compute/3" do - test "returns empty list when observations list is empty" do - assert AsosNudge.compute([], @valid_time, [profile(32.0, -97.0)]) == [] - end - - test "returns empty list when no HRRR profiles provided" do - assert AsosNudge.compute([station(32.0, -97.0)], @valid_time, []) == [] - end - - test "returns grid score maps for nudged profiles with band_mhz, score, factors" do - hrrr = profile(32.0, -97.0, surface_temp_c: 20.0) - obs = station(32.0, -97.0, temp_f: 77.0) - - results = AsosNudge.compute([obs], @valid_time, [hrrr]) - - assert results != [] - - for result <- results do - assert result.lat == 32.0 - assert result.lon == -97.0 - assert result.valid_time == @valid_time - assert is_integer(result.band_mhz) - assert is_number(result.score) - assert is_map(result.factors) - end - end - - test "drops grid points with no station within 250 km from the output" do - nudged_cell = profile(32.0, -97.0, surface_temp_c: 20.0) - far_cell = profile(40.0, -85.0, surface_temp_c: 20.0) - obs = station(32.0, -97.0, temp_f: 77.0) - - results = AsosNudge.compute([obs], @valid_time, [nudged_cell, far_cell]) - - lat_lons = results |> Enum.map(&{&1.lat, &1.lon}) |> Enum.uniq() - assert {32.0, -97.0} in lat_lons - refute {40.0, -85.0} in lat_lons - end - - test "re-scores grid cells with MRMS rain even when no ASOS station is nearby" do - far_cell = profile(40.0, -85.0, surface_temp_c: 20.0) - rain_grid = %{{40.0, -85.0} => 12.5} - - results = AsosNudge.compute([], @valid_time, [far_cell], rain_grid) - - assert results != [] - lat_lons = results |> Enum.map(&{&1.lat, &1.lon}) |> Enum.uniq() - assert {40.0, -85.0} in lat_lons - end - - test "ignores MRMS rain below the 0.1 mm/hr threshold so dry cells keep raw HRRR scores" do - cell = profile(40.0, -85.0, surface_temp_c: 20.0) - rain_grid = %{{40.0, -85.0} => 0.05} - - assert AsosNudge.compute([], @valid_time, [cell], rain_grid) == [] - end - - test "MRMS rain is applied through to the scored profile when above threshold" do - cell = profile(40.0, -85.0, surface_temp_c: 20.0) - rain_grid_dry = %{} - rain_grid_wet = %{{40.0, -85.0} => 15.0} - obs = station(40.0, -85.0, temp_f: 68.0) - - [dry_result | _] = AsosNudge.compute([obs], @valid_time, [cell], rain_grid_dry) - [wet_result | _] = AsosNudge.compute([obs], @valid_time, [cell], rain_grid_wet) - - # Heavy rain should drop the score meaningfully vs the same cell with - # no rain. (Exact delta depends on the band weights but the 0→15 mm/hr - # transition always reduces the rain factor score.) - assert wet_result.factors.rain < dry_result.factors.rain - end - end -end diff --git a/test/microwaveprop/propagation/grid_score_test.exs b/test/microwaveprop/propagation/grid_score_test.exs deleted file mode 100644 index 6bf2311d..00000000 --- a/test/microwaveprop/propagation/grid_score_test.exs +++ /dev/null @@ -1,95 +0,0 @@ -defmodule Microwaveprop.Propagation.GridScoreTest do - use Microwaveprop.DataCase, async: true - - alias Microwaveprop.Propagation.GridScore - - @valid_attrs %{ - lat: 32.875, - lon: -97.0, - valid_time: ~U[2026-03-28 18:00:00Z], - band_mhz: 10_000, - score: 72, - factors: %{ - "humidity" => 85, - "time_of_day" => 60, - "td_depression" => 70, - "refractivity" => 55, - "sky" => 80, - "season" => 65, - "wind" => 50, - "rain" => 90, - "pressure" => 45 - } - } - - describe "changeset/2" do - test "valid attributes" do - changeset = GridScore.changeset(%GridScore{}, @valid_attrs) - assert changeset.valid? - end - - test "requires the core fields except factors" do - changeset = GridScore.changeset(%GridScore{}, %{}) - - errors = errors_on(changeset) - - assert %{ - lat: ["can't be blank"], - lon: ["can't be blank"], - valid_time: ["can't be blank"], - band_mhz: ["can't be blank"], - score: ["can't be blank"] - } = errors - - refute Map.has_key?(errors, :factors) - end - - test "persists with nil factors (forecast-hour rows skip the factor breakdown)" do - attrs = Map.delete(@valid_attrs, :factors) - changeset = GridScore.changeset(%GridScore{}, attrs) - assert changeset.valid? - assert {:ok, score} = Repo.insert(changeset) - assert score.factors == nil - end - - test "persists to database" do - changeset = GridScore.changeset(%GridScore{}, @valid_attrs) - assert {:ok, score} = Repo.insert(changeset) - assert score.lat == 32.875 - assert score.lon == -97.0 - assert score.band_mhz == 10_000 - assert score.score == 72 - assert score.factors["humidity"] == 85 - end - - test "enforces unique lat + lon + valid_time + band_mhz" do - assert {:ok, _} = - %GridScore{} |> GridScore.changeset(@valid_attrs) |> Repo.insert() - - assert {:error, changeset} = - %GridScore{} |> GridScore.changeset(@valid_attrs) |> Repo.insert() - - assert %{lat: ["has already been taken"]} = errors_on(changeset) - end - - test "allows same location at different times" do - assert {:ok, _} = - %GridScore{} |> GridScore.changeset(@valid_attrs) |> Repo.insert() - - later_attrs = Map.put(@valid_attrs, :valid_time, ~U[2026-03-28 19:00:00Z]) - - assert {:ok, _} = - %GridScore{} |> GridScore.changeset(later_attrs) |> Repo.insert() - end - - test "allows same location and time for different bands" do - assert {:ok, _} = - %GridScore{} |> GridScore.changeset(@valid_attrs) |> Repo.insert() - - other_band_attrs = Map.put(@valid_attrs, :band_mhz, 24_000) - - assert {:ok, _} = - %GridScore{} |> GridScore.changeset(other_band_attrs) |> Repo.insert() - end - end -end diff --git a/test/microwaveprop/propagation/scorer_diff_test.exs b/test/microwaveprop/propagation/scorer_diff_test.exs deleted file mode 100644 index 62c53683..00000000 --- a/test/microwaveprop/propagation/scorer_diff_test.exs +++ /dev/null @@ -1,207 +0,0 @@ -defmodule Microwaveprop.Propagation.ScorerDiffTest do - use Microwaveprop.DataCase, async: true - - alias Microwaveprop.Propagation.GridScore - alias Microwaveprop.Propagation.ScorerDiff - - @valid_time ~U[2026-04-10 12:00:00Z] - - @old_weights %{ - "humidity" => 0.18, - "time_of_day" => 0.10, - "td_depression" => 0.10, - "refractivity" => 0.08, - "sky" => 0.08, - "season" => 0.08, - "wind" => 0.05, - "rain" => 0.08, - "pressure" => 0.15, - "pwat" => 0.10 - } - - # Shift weight from pressure to time_of_day - @new_weights %{ - "humidity" => 0.18, - "time_of_day" => 0.20, - "td_depression" => 0.10, - "refractivity" => 0.08, - "sky" => 0.08, - "season" => 0.08, - "wind" => 0.05, - "rain" => 0.08, - "pressure" => 0.05, - "pwat" => 0.10 - } - - defp insert_score!(attrs) do - %GridScore{} - |> GridScore.changeset(attrs) - |> Repo.insert!() - end - - defp make_factors(overrides \\ %{}) do - base = %{ - "humidity" => 80, - "time_of_day" => 60, - "td_depression" => 70, - "refractivity" => 55, - "sky" => 88, - "season" => 65, - "wind" => 90, - "rain" => 95, - "pressure" => 40, - "pwat" => 75 - } - - Map.merge(base, overrides) - end - - setup do - # Insert scores at two different bands for the same valid_time - insert_score!(%{ - lat: 32.875, - lon: -97.0, - valid_time: @valid_time, - band_mhz: 10_000, - score: 70, - factors: make_factors() - }) - - insert_score!(%{ - lat: 33.0, - lon: -97.125, - valid_time: @valid_time, - band_mhz: 10_000, - score: 50, - factors: make_factors(%{"time_of_day" => 20, "pressure" => 90}) - }) - - insert_score!(%{ - lat: 32.875, - lon: -97.0, - valid_time: @valid_time, - band_mhz: 24_000, - score: 65, - factors: make_factors(%{"humidity" => 40, "season" => 90}) - }) - - # Insert an older score that should not be included - insert_score!(%{ - lat: 32.875, - lon: -97.0, - valid_time: ~U[2026-04-10 11:00:00Z], - band_mhz: 10_000, - score: 55, - factors: make_factors(%{"humidity" => 30}) - }) - - :ok - end - - describe "compare/3" do - test "returns summary with mean_diff, max_diff, regressions, improvements" do - result = ScorerDiff.compare(@old_weights, @new_weights) - - assert is_map(result.summary) - assert is_float(result.summary.mean_diff) - assert is_number(result.summary.max_diff) - assert is_integer(result.summary.regressions) - assert is_integer(result.summary.improvements) - end - - test "returns band_diffs keyed by band_mhz" do - result = ScorerDiff.compare(@old_weights, @new_weights) - - assert Map.has_key?(result.band_diffs, 10_000) - assert Map.has_key?(result.band_diffs, 24_000) - assert is_float(result.band_diffs[10_000].mean_diff) - end - - test "returns worst_regressions as a list" do - result = ScorerDiff.compare(@old_weights, @new_weights) - - assert is_list(result.worst_regressions) - - for entry <- result.worst_regressions do - assert Map.has_key?(entry, :lat) - assert Map.has_key?(entry, :lon) - assert Map.has_key?(entry, :band_mhz) - assert Map.has_key?(entry, :old_score) - assert Map.has_key?(entry, :new_score) - end - end - - test "computes correct scores from factors and weights" do - result = ScorerDiff.compare(@old_weights, @new_weights) - - # For the first 10 GHz point (lat 32.875, lon -97.0): - # factors: humidity=80, time_of_day=60, td_depression=70, refractivity=55, - # sky=88, season=65, wind=90, rain=95, pressure=40, pwat=75 - # - # old: 80*0.18 + 60*0.10 + 70*0.10 + 55*0.08 + 88*0.08 + 65*0.08 + 90*0.05 + 95*0.08 + 40*0.15 + 75*0.10 - # = 14.4 + 6.0 + 7.0 + 4.4 + 7.04 + 5.2 + 4.5 + 7.6 + 6.0 + 7.5 = 69.64 -> 70 - # - # new: 80*0.18 + 60*0.20 + 70*0.10 + 55*0.08 + 88*0.08 + 65*0.08 + 90*0.05 + 95*0.08 + 40*0.05 + 75*0.10 - # = 14.4 + 12.0 + 7.0 + 4.4 + 7.04 + 5.2 + 4.5 + 7.6 + 2.0 + 7.5 = 71.64 -> 72 - # - # diff = 72 - 70 = +2 (improvement) - - # For the second 10 GHz point (lat 33.0, lon -97.125): - # factors: time_of_day=20, pressure=90 (others same) - # - # old: 80*0.18 + 20*0.10 + 70*0.10 + 55*0.08 + 88*0.08 + 65*0.08 + 90*0.05 + 95*0.08 + 90*0.15 + 75*0.10 - # = 14.4 + 2.0 + 7.0 + 4.4 + 7.04 + 5.2 + 4.5 + 7.6 + 13.5 + 7.5 = 73.14 -> 73 - # - # new: 80*0.18 + 20*0.20 + 70*0.10 + 55*0.08 + 88*0.08 + 65*0.08 + 90*0.05 + 95*0.08 + 90*0.05 + 75*0.10 - # = 14.4 + 4.0 + 7.0 + 4.4 + 7.04 + 5.2 + 4.5 + 7.6 + 4.5 + 7.5 = 66.14 -> 66 - # - # diff = 66 - 73 = -7 (regression!) - - # We should see at least 1 regression (diff > 5 in magnitude) - assert result.summary.regressions >= 1 - - # Total count should be 3 (only the most recent valid_time) - assert result.summary.total == 3 - end - - test "identical weights produce zero diff" do - result = ScorerDiff.compare(@old_weights, @old_weights) - - assert result.summary.mean_diff == 0.0 - assert result.summary.max_diff == 0 - assert result.summary.regressions == 0 - assert result.summary.improvements == 0 - end - - test "only loads scores from the most recent valid_time" do - result = ScorerDiff.compare(@old_weights, @new_weights) - - # 3 scores at @valid_time (2 for 10GHz, 1 for 24GHz), 1 at earlier time - assert result.summary.total == 3 - end - - test "accepts limit option" do - result = ScorerDiff.compare(@old_weights, @new_weights, limit: 1) - - # Should only process 1 score - assert result.summary.total == 1 - end - end - - describe "weighted_sum/2" do - test "computes weighted sum from factors and weights" do - factors = %{"humidity" => 100, "time_of_day" => 50} - weights = %{"humidity" => 0.5, "time_of_day" => 0.5} - - assert ScorerDiff.weighted_sum(factors, weights) == 75 - end - - test "rounds to nearest integer" do - factors = %{"humidity" => 80, "time_of_day" => 60} - weights = %{"humidity" => 0.6, "time_of_day" => 0.4} - - # 80*0.6 + 60*0.4 = 48 + 24 = 72 - assert ScorerDiff.weighted_sum(factors, weights) == 72 - end - end -end diff --git a/test/microwaveprop/propagation_test.exs b/test/microwaveprop/propagation_test.exs index 60755ee7..69fbc7fd 100644 --- a/test/microwaveprop/propagation_test.exs +++ b/test/microwaveprop/propagation_test.exs @@ -2,7 +2,6 @@ defmodule Microwaveprop.PropagationTest do use Microwaveprop.DataCase, async: false alias Microwaveprop.Propagation - alias Microwaveprop.Propagation.GridScore alias Microwaveprop.Propagation.ScoreCache alias Microwaveprop.Propagation.ScoresFile @@ -144,112 +143,8 @@ defmodule Microwaveprop.PropagationTest do 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 + test "writes one ScoresFile per band for a fresh valid_time" do valid_time = ~U[2026-07-15 13:00:00Z] scores = [ @@ -258,30 +153,60 @@ defmodule Microwaveprop.PropagationTest do ] 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 + assert {:ok, %{band_mhz: 10_000}} = ScoresFile.read(10_000, valid_time) + assert {:ok, %{band_mhz: 24_000}} = ScoresFile.read(24_000, valid_time) 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 + test "overwrites the file when called again for the same valid_time" 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}], + [%{lat: 25.0, lon: -125.0, valid_time: valid_time, band_mhz: 10_000, score: 10, factors: nil}], + valid_time + ) + + Propagation.replace_scores( + [%{lat: 25.0, lon: -125.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: nil}], + valid_time + ) + + assert ScoresFile.read_point(10_000, valid_time, 25.0, -125.0) == 75 + end + + test "leaves files 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.replace_scores( + [%{lat: 25.0, lon: -125.0, valid_time: other_time, band_mhz: 10_000, score: 40, factors: nil}], + other_time + ) + + Propagation.replace_scores( + [%{lat: 25.0, lon: -125.0, valid_time: replaced_time, band_mhz: 10_000, score: 90, factors: nil}], + replaced_time + ) + + assert ScoresFile.read_point(10_000, other_time, 25.0, -125.0) == 40 + assert ScoresFile.read_point(10_000, replaced_time, 25.0, -125.0) == 90 + end + end + + describe "point_detail/4" do + test "returns the score from the on-disk file with an empty factors map" do + valid_time = ~U[2026-07-15 13:00:00Z] + + Propagation.replace_scores( + [%{lat: 25.0, lon: -125.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) + Propagation.point_detail(10_000, 25.0, -125.0, valid_time) + end + + test "returns nil when the file doesn't exist for the requested point" do + assert Propagation.point_detail(10_000, 25.0, -125.0, ~U[2026-07-15 13:00:00Z]) == nil end end @@ -290,11 +215,11 @@ defmodule Microwaveprop.PropagationTest 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: %{}} + %{lat: 25.0, lon: -125.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: nil}, + %{lat: 25.125, lon: -125.0, valid_time: valid_time, band_mhz: 10_000, score: 80, factors: nil} ] - Propagation.upsert_scores(scores) + Propagation.replace_scores(scores, valid_time) results = Propagation.latest_scores(10_000) assert length(results) == 2 end @@ -305,38 +230,37 @@ defmodule Microwaveprop.PropagationTest do end describe "scores_at/3 with cache" do - test "populates the cache on a DB miss" do + test "populates the cache on a cold read" 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: %{}} - ]) + Propagation.replace_scores( + [%{lat: 25.0, lon: -125.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: nil}], + valid_time + ) 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) + assert {:ok, [%{lat: 25.0, lon: -125.0, score: 75}]} = ScoreCache.fetch(10_000, valid_time) end - test "returns cached bounds-filtered results without hitting the DB" do + test "returns cached bounds-filtered results without touching the filesystem" 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. + # Seed the cache directly — no ScoresFile exists, so if the + # result matches the cached payload we know the disk 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 + test "returns empty list when no data anywhere" do assert Propagation.scores_at(10_000, ~U[2026-07-15 13:00:00Z]) == [] end end @@ -349,8 +273,6 @@ defmodule Microwaveprop.PropagationTest do 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 [ @@ -379,20 +301,19 @@ defmodule Microwaveprop.PropagationTest do assert [%{valid_time: ^valid_time, score: 65}] = result end - test "returns empty list when no cached or DB data for the point" do + test "returns empty list when no 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 + test "returns cached valid_times without touching the filesystem when 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 @@ -407,12 +328,13 @@ defmodule Microwaveprop.PropagationTest do assert Propagation.available_valid_times(10_000) == [fresh] end - test "falls back to DB on cold cache" do + test "falls back to the on-disk store 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: %{}} - ]) + Propagation.replace_scores( + [%{lat: 25.0, lon: -125.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: nil}], + valid_time + ) assert ScoreCache.valid_times(10_000) == [] assert Propagation.available_valid_times(10_000) == [valid_time] @@ -420,13 +342,16 @@ defmodule Microwaveprop.PropagationTest do end describe "warm_cache_and_broadcast/2" do - test "loads scores from DB into the cache" do + test "loads scores from the on-disk store 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: %{}} - ]) + Propagation.replace_scores( + [ + %{lat: 25.0, lon: -125.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: nil}, + %{lat: 25.125, lon: -125.0, valid_time: valid_time, band_mhz: 10_000, score: 80, factors: nil} + ], + valid_time + ) assert ScoreCache.fetch(10_000, valid_time) == :miss @@ -440,10 +365,13 @@ defmodule Microwaveprop.PropagationTest do 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.replace_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: 30, factors: nil} + ], + valid_time + ) Propagation.warm_cache_and_broadcast(10_000, valid_time) ScoreCache.sync() @@ -458,14 +386,19 @@ defmodule Microwaveprop.PropagationTest do assert Propagation.latest_valid_time() == nil end - test "returns most recent time" do + test "returns the most recent valid_time across bands" 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: %{}} - ]) + Propagation.replace_scores( + [%{lat: 25.0, lon: -125.0, valid_time: t1, band_mhz: 10_000, score: 50, factors: nil}], + t1 + ) + + Propagation.replace_scores( + [%{lat: 25.0, lon: -125.0, valid_time: t2, band_mhz: 10_000, score: 60, factors: nil}], + t2 + ) assert Propagation.latest_valid_time() == t2 end diff --git a/test/microwaveprop/workers/propagation_prune_worker_test.exs b/test/microwaveprop/workers/propagation_prune_worker_test.exs index 29607cf8..f856747d 100644 --- a/test/microwaveprop/workers/propagation_prune_worker_test.exs +++ b/test/microwaveprop/workers/propagation_prune_worker_test.exs @@ -1,51 +1,34 @@ defmodule Microwaveprop.Workers.PropagationPruneWorkerTest do - use Microwaveprop.DataCase, async: true + use Microwaveprop.DataCase, async: false - alias Microwaveprop.Propagation.GridScore - alias Microwaveprop.Repo + alias Microwaveprop.Propagation.ScoresFile alias Microwaveprop.Workers.PropagationPruneWorker - @factors %{ - "humidity" => 85, - "time_of_day" => 60, - "td_depression" => 70, - "refractivity" => 55, - "sky" => 80, - "season" => 65, - "wind" => 50, - "rain" => 90, - "pressure" => 75 - } - - defp insert_score!(valid_time, lat \\ 32.875, lon \\ -97.0) do - Repo.insert!(%GridScore{lat: lat, lon: lon, valid_time: valid_time, band_mhz: 10_000, score: 72, factors: @factors}) - end - describe "perform/1" do - test "deletes scores with valid_time older than 2 hours" do - now = DateTime.truncate(DateTime.utc_now(), :second) - old = DateTime.add(now, -3, :hour) - fresh = DateTime.add(now, -30, :minute) - future = DateTime.add(now, 6, :hour) + test "deletes ScoresFile binaries with valid_time older than 2 hours" do + now = DateTime.utc_now() + old = now |> DateTime.add(-3, :hour) |> DateTime.truncate(:second) + fresh = now |> DateTime.add(-30, :minute) |> DateTime.truncate(:second) + future = now |> DateTime.add(6, :hour) |> DateTime.truncate(:second) - old_score = insert_score!(old, 32.0, -97.0) - fresh_score = insert_score!(fresh, 32.1, -97.1) - future_score = insert_score!(future, 32.2, -97.2) + ScoresFile.write!(10_000, old, [%{lat: 25.0, lon: -125.0, score: 10}]) + ScoresFile.write!(10_000, fresh, [%{lat: 25.0, lon: -125.0, score: 20}]) + ScoresFile.write!(10_000, future, [%{lat: 25.0, lon: -125.0, score: 30}]) assert :ok = PropagationPruneWorker.perform(%Oban.Job{}) - refute Repo.get(GridScore, old_score.id) - assert Repo.get(GridScore, fresh_score.id) - assert Repo.get(GridScore, future_score.id) + assert {:error, :enoent} = ScoresFile.read(10_000, old) + assert {:ok, _} = ScoresFile.read(10_000, fresh) + assert {:ok, _} = ScoresFile.read(10_000, future) end test "is a no-op when nothing is stale" do - now = DateTime.truncate(DateTime.utc_now(), :second) - fresh = DateTime.add(now, -30, :minute) - score = insert_score!(fresh) + now = DateTime.utc_now() + fresh = now |> DateTime.add(-30, :minute) |> DateTime.truncate(:second) + ScoresFile.write!(10_000, fresh, [%{lat: 25.0, lon: -125.0, score: 42}]) assert :ok = PropagationPruneWorker.perform(%Oban.Job{}) - assert Repo.get(GridScore, score.id) + assert {:ok, _} = ScoresFile.read(10_000, fresh) end end end