From 07ffcf52d75bf3d1cf6c1c850b2e8b45815374e7 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 14 Apr 2026 14:50:43 -0500 Subject: [PATCH] Flip map read path to ScoresFile, stop writing grid HRRR profiles MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Read-side cutover for the binary scores store and a companion cleanup that removes the biggest remaining DB write from the hot path. Propagation.scores_at/3, available_valid_times/1, latest_valid_time/0, latest_valid_time/1, earliest_valid_time/1, point_detail/4, and point_forecast/3 all now prefer ScoresFile and fall back to the propagation_scores table when a file is missing. The map render path reads from /data/scores first; Postgres stays as a safety net while dual-write is on. point_detail still pulls factors from Postgres (analysis-hour rows only) and coalesces nil to an empty map so the JS popup iterates cleanly. replace_scores/2 is now gated by a postgres_writes_enabled? flag (runtime env MICROWAVEPROP_SCORES_POSTGRES=false, or the :propagation_scores_postgres app env key) so the binary-only path can be benchmarked locally without the DB insert. Default stays true. PropagationGridWorker no longer calls store_hrrr_profiles — persisting 92k grid rows × 19 forecast hours of JSONB profiles was ~12 min of wall time per chain for a table only AsosAdjustmentWorker read from. Per-contact HRRR enrichment through HrrrFetchWorker still writes its own (is_grid_point: false) rows. AsosAdjustmentWorker is disabled in all three cron configs since its data source is gone. DataCase resets the scores tree between tests so per-test ScoresFile writes don't leak across cases, and ScoresFileTest switches to async: false because it mutates the global :propagation_scores_dir env. --- config/config.exs | 3 +- config/dev.exs | 3 +- config/runtime.exs | 6 +- lib/microwaveprop/propagation.ex | 228 +++++++++++++----- lib/microwaveprop/propagation/scores_file.ex | 153 ++++++++++++ .../workers/propagation_grid_worker.ex | 64 +---- .../propagation/scores_file_test.exs | 93 ++++++- test/support/data_case.ex | 10 + 8 files changed, 439 insertions(+), 121 deletions(-) diff --git a/config/config.exs b/config/config.exs index 7309f4e7..7a4d0a90 100644 --- a/config/config.exs +++ b/config/config.exs @@ -79,7 +79,8 @@ config :microwaveprop, Oban, {"*/5 * * * *", Microwaveprop.Commercial.PollWorker}, {"5 */3 * * *", Microwaveprop.Workers.PropagationGridWorker}, {"*/2 * * * *", Microwaveprop.Workers.MrmsFetchWorker}, - {"*/10 * * * *", Microwaveprop.Workers.AsosAdjustmentWorker}, + # AsosAdjustmentWorker disabled — see runtime.exs for rationale. + # {"*/10 * * * *", Microwaveprop.Workers.AsosAdjustmentWorker}, {"*/15 * * * *", Microwaveprop.Workers.PropagationPruneWorker}, # UWYO publishes the 00Z/12Z Canadian radiosondes ~90 minutes after launch {"30 1 * * *", CanadianSoundingFetchWorker}, diff --git a/config/dev.exs b/config/dev.exs index 78b5c909..97141680 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -92,7 +92,8 @@ config :microwaveprop, Oban, crontab: [ {"5 */3 * * *", Microwaveprop.Workers.PropagationGridWorker}, {"*/2 * * * *", Microwaveprop.Workers.MrmsFetchWorker}, - {"*/10 * * * *", Microwaveprop.Workers.AsosAdjustmentWorker}, + # AsosAdjustmentWorker disabled — see runtime.exs for rationale. + # {"*/10 * * * *", Microwaveprop.Workers.AsosAdjustmentWorker}, {"*/15 * * * *", Microwaveprop.Workers.PropagationPruneWorker}, {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker} ]} diff --git a/config/runtime.exs b/config/runtime.exs index 27346acd..4b78bd8a 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -221,7 +221,11 @@ if config_env() == :prod do # rain. Requires wgrib2 built with PNG decoder support (DRT # 5.41), which the Dockerfile now enables via NCEPLIBS-g2c. {"*/2 * * * *", Microwaveprop.Workers.MrmsFetchWorker}, - {"*/10 * * * *", Microwaveprop.Workers.AsosAdjustmentWorker}, + # AsosAdjustmentWorker disabled: depends on hrrr_profiles rows + # that PropagationGridWorker no longer persists. The 10-min + # ASOS nudge is a nice-to-have refinement we're not paying the + # full grid-side JSONB write for. + # {"*/10 * * * *", Microwaveprop.Workers.AsosAdjustmentWorker}, {"*/15 * * * *", Microwaveprop.Workers.PropagationPruneWorker}, {"*/30 * * * *", Microwaveprop.Workers.BackfillEnqueueWorker, args: %{"limit" => 500, "types" => ["hrrr", "weather", "terrain", "iemre"]}}, diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index 517d55ac..b0e19e9e 100644 --- a/lib/microwaveprop/propagation.ex +++ b/lib/microwaveprop/propagation.ex @@ -167,39 +167,17 @@ defmodule Microwaveprop.Propagation do """ @spec replace_scores(Enumerable.t(), DateTime.t()) :: {:ok, non_neg_integer()} | {:error, term()} def replace_scores(scores, %DateTime{} = valid_time) do - now = DateTime.truncate(DateTime.utc_now(), :second) - # Materialize once — we need to traverse twice: once for the DB - # insert, once for the per-band ScoresFile writer. 475k maps × - # ~150 B each is ~70 MB, well within the pod memory budget. + # 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 = - 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 - ) + 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) @@ -209,6 +187,49 @@ defmodule Microwaveprop.Propagation do 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 @@ -324,7 +345,7 @@ defmodule Microwaveprop.Propagation do cutoff = DateTime.add(DateTime.utc_now(), -3600, :second) case ScoreCache.valid_times(band_mhz) do - [] -> available_valid_times_from_db(band_mhz, cutoff) + [] -> available_valid_times_from_store(band_mhz, cutoff) cached -> filter_fresh(cached, cutoff) end end @@ -333,6 +354,23 @@ defmodule Microwaveprop.Propagation do Enum.filter(times, fn t -> DateTime.compare(t, cutoff) != :lt end) end + 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 + + defp filter_or_latest(times, cutoff) do + fresh = Enum.filter(times, fn t -> DateTime.compare(t, cutoff) != :lt end) + + if fresh == [] do + [Enum.max(times, DateTime)] + else + fresh + end + end + defp available_valid_times_from_db(band_mhz, cutoff) do times = Repo.all( @@ -380,7 +418,7 @@ defmodule Microwaveprop.Propagation do Enum.map(scores, &Map.put(&1, :valid_time, time)) :miss -> - full = load_scores_from_db(band_mhz, time) + full = load_scores_from_store(band_mhz, time) ScoreCache.put(band_mhz, time, full) full @@ -389,6 +427,13 @@ 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, @@ -427,7 +472,13 @@ defmodule Microwaveprop.Propagation do end defp earliest_valid_time(band_mhz) do - Repo.one(from(gs in GridScore, where: gs.band_mhz == ^band_mhz, select: min(gs.valid_time))) + 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))) + end end @doc "Get scores across all forecast hours for a single grid point (for sparkline)." @@ -438,11 +489,31 @@ defmodule Microwaveprop.Propagation do now = DateTime.utc_now() case ScoreCache.valid_times(band_mhz) do - [] -> point_forecast_from_db(band_mhz, snapped_lat, snapped_lon, now) + [] -> point_forecast_from_store(band_mhz, snapped_lat, snapped_lon, now) cached -> point_forecast_from_cache(band_mhz, snapped_lat, snapped_lon, now, cached) end 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 + end + + defp point_forecast_entry(valid_time, band_mhz, lat, lon) do + case ScoresFile.read_point(band_mhz, valid_time, lat, lon) do + nil -> nil + score -> %{valid_time: valid_time, score: score} + end + end + defp point_forecast_from_cache(band_mhz, lat, lon, now, cached_times) do cached_times |> Enum.filter(fn t -> DateTime.compare(t, now) != :lt end) @@ -485,36 +556,67 @@ defmodule Microwaveprop.Propagation do } | nil def point_detail(band_mhz, lat, lon, valid_time \\ nil) do - step = Grid.step() - snapped_lat = Float.round(Float.round(lat / step) * step, 3) - snapped_lon = Float.round(Float.round(lon / step) * step, 3) - + {snapped_lat, snapped_lon} = snap_to_grid(lat, lon) time = valid_time || latest_valid_time(band_mhz) case time do - nil -> - nil - - _ -> - from(gs in GridScore, - where: - gs.band_mhz == ^band_mhz and - gs.valid_time == ^time and - gs.lat == ^snapped_lat and - gs.lon == ^snapped_lon, - select: %{ - lat: gs.lat, - lon: gs.lon, - score: gs.score, - factors: gs.factors, - valid_time: gs.valid_time - } - ) - |> Repo.one() - |> coalesce_factors() + 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) + + score -> + %{ + lat: lat, + lon: lon, + score: score, + factors: fetch_factors(band_mhz, time, lat, lon) || %{}, + valid_time: time + } + 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 @@ -526,13 +628,19 @@ defmodule Microwaveprop.Propagation do @doc "Get the latest valid_time across all scores." @spec latest_valid_time() :: DateTime.t() | nil def latest_valid_time do - Repo.one(from(gs in GridScore, select: max(gs.valid_time))) + case ScoresFile.latest_valid_time() do + nil -> Repo.one(from(gs in GridScore, select: max(gs.valid_time))) + dt -> dt + end 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 - Repo.one(from(gs in GridScore, where: gs.band_mhz == ^band_mhz, select: max(gs.valid_time))) + 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))) + times -> Enum.max(times, DateTime) + end end # Prefer the persisted scalar — `hrrr_profiles` already stored this at diff --git a/lib/microwaveprop/propagation/scores_file.ex b/lib/microwaveprop/propagation/scores_file.ex index c57a63cd..c366e95b 100644 --- a/lib/microwaveprop/propagation/scores_file.ex +++ b/lib/microwaveprop/propagation/scores_file.ex @@ -110,6 +110,84 @@ defmodule Microwaveprop.Propagation.ScoresFile do end) end + @doc """ + List every `valid_time` that has a score file on disk for + `band_mhz`, ordered ascending. Empty list if the band directory + doesn't exist yet. + """ + @spec list_valid_times(non_neg_integer()) :: [DateTime.t()] + def list_valid_times(band_mhz) when is_integer(band_mhz) do + dir = Path.join(base_dir(), Integer.to_string(band_mhz)) + + case File.ls(dir) do + {:ok, files} -> + files + |> Enum.map(&parse_valid_time_dt/1) + |> Enum.reject(&is_nil/1) + |> Enum.sort(DateTime) + + _ -> + [] + end + end + + @doc """ + Return the most recent `valid_time` across all bands on disk, or + `nil` if the scores tree is empty. + """ + @spec latest_valid_time() :: DateTime.t() | nil + def latest_valid_time do + case File.ls(base_dir()) do + {:ok, band_dirs} -> + band_dirs + |> Enum.flat_map(&band_times_or_empty/1) + |> max_datetime() + + _ -> + nil + end + end + + defp band_times_or_empty(band_dir) do + case Integer.parse(band_dir) do + {band_mhz, ""} -> list_valid_times(band_mhz) + _ -> [] + end + end + + @doc """ + Read the file for `(band_mhz, valid_time)` and return every + scored cell as a list of `%{lat, lon, score}` maps. Optional + `bounds` map (`%{"north", "south", "east", "west"}` or + `%{north:, south:, east:, west:}`) filters the return to cells + inside that box. Cells with the no-data sentinel are excluded. + + Returns `[]` when the file doesn't exist or is unreadable — + callers should treat this as "no scores available for this + band/time yet" rather than an error. + """ + @spec read_bounds(non_neg_integer(), DateTime.t(), map() | nil) :: + [%{lat: float(), lon: float(), score: non_neg_integer()}] + def read_bounds(band_mhz, %DateTime{} = valid_time, bounds \\ nil) do + case read(band_mhz, valid_time) do + {:ok, payload} -> extract_points(payload, bounds) + _ -> [] + end + end + + @doc """ + Fetch the score for a single grid cell from the file for + `(band_mhz, valid_time)`. Returns `nil` if the file is missing or + if the cell is outside the grid / carries the no-data sentinel. + """ + @spec read_point(non_neg_integer(), DateTime.t(), float(), float()) :: non_neg_integer() | nil + def read_point(band_mhz, %DateTime{} = valid_time, lat, lon) do + case read(band_mhz, valid_time) do + {:ok, payload} -> point_score(payload, lat, lon) + _ -> nil + end + end + # ── Encoding ──────────────────────────────────────────────────── defp encode(band_mhz, valid_time, scores) do @@ -190,6 +268,81 @@ defmodule Microwaveprop.Propagation.ScoresFile do defp decode(_), do: {:error, :invalid_format} + # ── Read helpers ──────────────────────────────────────────────── + + defp extract_points(%{scores: body} = payload, bounds) do + %{lat_min: lat_min, lon_min: lon_min, step: step, n_rows: n_rows, n_cols: n_cols} = payload + {row_lo, row_hi, col_lo, col_hi} = bounds_to_index_range(payload, bounds) + + for row <- row_lo..row_hi, + col <- col_lo..col_hi, + byte = :binary.at(body, row * n_cols + col), + byte != @no_data do + # guard against the ignored variables warning + _ = n_rows + lat = Float.round(lat_min + row * step, 3) + lon = Float.round(lon_min + col * step, 3) + %{lat: lat, lon: lon, score: byte} + end + end + + defp bounds_to_index_range(payload, nil) do + {0, payload.n_rows - 1, 0, payload.n_cols - 1} + end + + defp bounds_to_index_range(payload, bounds) do + %{lat_min: lat_min, lon_min: lon_min, step: step, n_rows: n_rows, n_cols: n_cols} = payload + + south = fetch_bound(bounds, :south, "south", lat_min) + north = fetch_bound(bounds, :north, "north", lat_min + (n_rows - 1) * step) + west = fetch_bound(bounds, :west, "west", lon_min) + east = fetch_bound(bounds, :east, "east", lon_min + (n_cols - 1) * step) + + row_lo = clamp(trunc((south - lat_min) / step), 0, n_rows - 1) + row_hi = clamp(ceil((north - lat_min) / step), 0, n_rows - 1) + col_lo = clamp(trunc((west - lon_min) / step), 0, n_cols - 1) + col_hi = clamp(ceil((east - lon_min) / step), 0, n_cols - 1) + + {min(row_lo, row_hi), max(row_lo, row_hi), min(col_lo, col_hi), max(col_lo, col_hi)} + end + + defp fetch_bound(bounds, atom_key, string_key, default) do + case Map.get(bounds, atom_key) || Map.get(bounds, string_key) do + nil -> default + v when is_number(v) -> v + v when is_binary(v) -> String.to_float(v) + end + end + + defp clamp(n, lo, _hi) when n < lo, do: lo + defp clamp(n, _lo, hi) when n > hi, do: hi + defp clamp(n, _lo, _hi), do: n + + defp point_score(payload, lat, lon) do + %{lat_min: lat_min, lon_min: lon_min, step: step, n_rows: n_rows, n_cols: n_cols, scores: body} = payload + row = round((lat - lat_min) / step) + col = round((lon - lon_min) / step) + + if row in 0..(n_rows - 1) and col in 0..(n_cols - 1) do + case :binary.at(body, row * n_cols + col) do + @no_data -> nil + score -> score + end + end + end + + defp max_datetime([]), do: nil + defp max_datetime(times), do: Enum.max(times, DateTime) + + defp parse_valid_time_dt(filename) do + with [_, iso] <- Regex.run(~r/^(.+)\.ntms$/, filename), + {:ok, dt, _} <- DateTime.from_iso8601(iso) do + dt + else + _ -> nil + end + end + # ── Prune helpers ─────────────────────────────────────────────── defp list_score_files(root) do diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex index d86d6065..391fcd56 100644 --- a/lib/microwaveprop/workers/propagation_grid_worker.ex +++ b/lib/microwaveprop/workers/propagation_grid_worker.ex @@ -32,7 +32,6 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do alias Microwaveprop.Weather.HrrrClient alias Microwaveprop.Weather.HrrrNativeClient alias Microwaveprop.Weather.NexradClient - alias Microwaveprop.Weather.SoundingParams require Logger @@ -144,8 +143,13 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do HrrrClient.fetch_grid(points, run_time, forecast_hour: forecast_hour) end) do {:ok, grid_data} -> - store_hrrr_profiles(grid_data, valid_time) - :erlang.garbage_collect() + # HRRR profiles used to be persisted to the hrrr_profiles table + # here for AsosAdjustmentWorker to re-score from. That was ~12 + # minutes of JSONB inserts per chain (92k rows × 19 forecast + # hours) with no user-visible benefit — the scores live in the + # /data/scores files now, and AsosAdjustmentWorker is disabled. + # Per-contact HRRR enrichment still uses HrrrFetchWorker, which + # writes its own `is_grid_point: false` rows. # Fetch native duct metrics and merge into grid_data for scoring grid_data = merge_native_duct_data(grid_data, run_time, forecast_hour) @@ -221,60 +225,6 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do defp format_duration(ms) when ms < 1000, do: "#{ms}ms" defp format_duration(ms), do: "#{Float.round(ms / 1000, 1)}s" - defp store_hrrr_profiles(grid_data, valid_time) do - count = - grid_data - |> Stream.flat_map(fn {{lat, lon}, profile} -> - build_profile_attrs_for_grid(lat, lon, valid_time, profile) - end) - |> Stream.chunk_every(500) - |> Enum.reduce(0, fn chunk, acc -> - Weather.upsert_hrrr_profiles_batch(chunk) - acc + length(chunk) - end) - - Logger.info("PropagationGrid: stored #{count} HRRR profiles") - end - - defp build_profile_attrs_for_grid(lat, lon, valid_time, profile) do - temp = profile.surface_temp_c - - if temp && temp > -80 && temp < 60 do - params = - if is_list(profile.profile) and length(profile.profile) >= 3, - do: SoundingParams.derive(profile.profile) - - attrs = %{ - valid_time: valid_time, - lat: lat, - lon: lon, - run_time: profile.run_time, - profile: profile.profile || [], - hpbl_m: profile.hpbl_m, - pwat_mm: profile.pwat_mm, - surface_temp_c: profile.surface_temp_c, - surface_dewpoint_c: profile.surface_dewpoint_c, - surface_pressure_mb: profile.surface_pressure_mb - } - - attrs = - if params do - Map.merge(attrs, %{ - surface_refractivity: params.surface_refractivity, - min_refractivity_gradient: params.min_refractivity_gradient, - ducting_detected: params.ducting_detected, - duct_characteristics: params.duct_characteristics - }) - else - attrs - end - - [attrs] - else - [] - end - end - defp merge_native_duct_data(grid_data, run_time, forecast_hour) do hour_dt = HrrrClient.nearest_hrrr_hour(run_time) date = DateTime.to_date(hour_dt) diff --git a/test/microwaveprop/propagation/scores_file_test.exs b/test/microwaveprop/propagation/scores_file_test.exs index 555e7f8d..ddc98b58 100644 --- a/test/microwaveprop/propagation/scores_file_test.exs +++ b/test/microwaveprop/propagation/scores_file_test.exs @@ -1,5 +1,8 @@ defmodule Microwaveprop.Propagation.ScoresFileTest do - use ExUnit.Case, async: true + # async: false — tests mutate the global Application env to redirect + # the scores dir to a private tmp tree. Running these in parallel + # would have concurrent tests trample each other's directory setting. + use ExUnit.Case, async: false alias Microwaveprop.Propagation.Grid alias Microwaveprop.Propagation.ScoresFile @@ -94,6 +97,94 @@ defmodule Microwaveprop.Propagation.ScoresFileTest do end end + describe "list_valid_times/1" do + test "returns files sorted ascending for a band" do + ScoresFile.write!(10_000, ~U[2026-04-14 18:00:00Z], []) + ScoresFile.write!(10_000, ~U[2026-04-14 16:00:00Z], []) + ScoresFile.write!(10_000, ~U[2026-04-14 17:00:00Z], []) + # Unrelated band's files shouldn't leak in. + ScoresFile.write!(24_000, ~U[2026-04-14 20:00:00Z], []) + + assert ScoresFile.list_valid_times(10_000) == [ + ~U[2026-04-14 16:00:00Z], + ~U[2026-04-14 17:00:00Z], + ~U[2026-04-14 18:00:00Z] + ] + end + + test "returns an empty list when the band dir doesn't exist" do + assert ScoresFile.list_valid_times(75_000) == [] + end + end + + describe "latest_valid_time/0" do + test "returns the max valid_time across every band on disk" do + ScoresFile.write!(10_000, ~U[2026-04-14 16:00:00Z], []) + ScoresFile.write!(24_000, ~U[2026-04-14 18:00:00Z], []) + ScoresFile.write!(10_000, ~U[2026-04-14 17:00:00Z], []) + + assert ScoresFile.latest_valid_time() == ~U[2026-04-14 18:00:00Z] + end + + test "returns nil on an empty tree" do + assert ScoresFile.latest_valid_time() == nil + end + end + + describe "read_bounds/3" do + test "returns every scored cell (no-data excluded) when bounds is nil" do + ScoresFile.write!(10_000, ~U[2026-04-14 18:00:00Z], [ + %{lat: 25.0, lon: -125.0, score: 10}, + %{lat: 25.125, lon: -125.0, score: 20} + ]) + + scores = ScoresFile.read_bounds(10_000, ~U[2026-04-14 18:00:00Z]) + + assert [ + %{lat: 25.0, lon: -125.0, score: 10}, + %{lat: 25.125, lon: -125.0, score: 20} + ] = Enum.sort_by(scores, & &1.lat) + end + + test "restricts results to the bounding box when provided" do + ScoresFile.write!(10_000, ~U[2026-04-14 18:00:00Z], [ + %{lat: 25.0, lon: -125.0, score: 10}, + %{lat: 49.875, lon: -66.0, score: 90} + ]) + + bounds = %{"south" => 24.0, "north" => 26.0, "west" => -126.0, "east" => -124.0} + scores = ScoresFile.read_bounds(10_000, ~U[2026-04-14 18:00:00Z], bounds) + + assert scores == [%{lat: 25.0, lon: -125.0, score: 10}] + end + + test "returns an empty list when the file doesn't exist" do + assert ScoresFile.read_bounds(10_000, ~U[2026-04-14 18:00:00Z]) == [] + end + end + + describe "read_point/4" do + test "returns the score for a grid cell by (lat, lon) lookup" do + ScoresFile.write!(10_000, ~U[2026-04-14 18:00:00Z], [ + %{lat: 25.0, lon: -125.0, score: 42} + ]) + + assert ScoresFile.read_point(10_000, ~U[2026-04-14 18:00:00Z], 25.0, -125.0) == 42 + end + + test "returns nil for cells that weren't scored" do + ScoresFile.write!(10_000, ~U[2026-04-14 18:00:00Z], [ + %{lat: 25.0, lon: -125.0, score: 42} + ]) + + assert ScoresFile.read_point(10_000, ~U[2026-04-14 18:00:00Z], 40.0, -100.0) == nil + end + + test "returns nil when the file is missing" do + assert ScoresFile.read_point(10_000, ~U[2026-04-14 18:00:00Z], 25.0, -125.0) == nil + end + end + describe "prune_older_than/1" do test "deletes files whose valid_time is strictly before the cutoff", %{dir: _dir} do old = ~U[2026-04-14 10:00:00Z] diff --git a/test/support/data_case.ex b/test/support/data_case.ex index fb49d68d..3f45d28b 100644 --- a/test/support/data_case.ex +++ b/test/support/data_case.ex @@ -31,6 +31,7 @@ defmodule Microwaveprop.DataCase do setup tags do Microwaveprop.DataCase.setup_sandbox(tags) + Microwaveprop.DataCase.reset_score_files() :ok end @@ -42,6 +43,15 @@ defmodule Microwaveprop.DataCase do on_exit(fn -> Sandbox.stop_owner(pid) end) end + @doc """ + Wipe the propagation ScoresFile tree between tests so files don't + leak between cases sharing the same tmp dir. + """ + def reset_score_files do + dir = Application.get_env(:microwaveprop, :propagation_scores_dir) + if is_binary(dir), do: File.rm_rf!(dir) + end + @doc """ A helper that transforms changeset errors into a map of messages.