From 020f05f8eb0b5fa0f9f8b117053c2d79d7b1ca84 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Wed, 29 Apr 2026 17:13:00 -0500 Subject: [PATCH] feat(hrdps): grid_tasks source column, Canadian mask, HrdpsGridWorker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stage 3 (Elixir foundation) of the HRDPS plan. Lands the scaffolding that lets the Rust prop-grid-rs HRDPS branch be wired up cleanly when it ships, without further DB or Elixir-side changes. What ships: - `grid_tasks.source` column (default 'hrrr', backfill is a no-op). Replaces the (run_time, forecast_hour, kind) unique index with a 4-column key (..., source) so HRRR and HRDPS analysis/forecast rows for the same cycle coexist. - `Grid.hrdps_only_points/0` — Canadian cells inside HRDPS bbox (49-60°N, -141 to -52°W) but outside HRRR's CONUS bbox. Disjoint from `conus_points/0` by construction so the two grids never double-write the same (lat, lon). - `Grid.point_source/1` — `:hrrr | :hrdps | :neither` classification for a (lat, lon) pair. Routes per-QSO enrichment to the right NWP source. - `GridTaskEnqueuer.seed_with_analysis/2` and `seed/2` accept a `:source` option; defaults to "hrrr" so existing callers are unchanged. - `HrdpsGridWorker` cycle seeder mirroring PropagationGridWorker's shape but for HRDPS's 4×/day cadence. `:hrdps` queue added to base Oban config. What's deferred: - The Rust `prop-grid-rs` worker doesn't yet branch on `task.source`. Until that ships, `HrdpsGridWorker` stays out of runtime.exs cron — it would just produce rows that the Rust worker mishandles as HRRR. Module doc explicitly notes the dormant state. - Map overlay extension and FreshnessMonitor HRDPS staleness deferred to follow-up sessions once real data flows. Activation path documented in the updated plan file. Coverage cap recorded as Stage 8 decision: 60°N for v1 (SRTM stops there). Arctic CDEM deferred to a follow-up plan. --- config/config.exs | 8 +- lib/microwaveprop/propagation/grid.ex | 67 +++++++++++++- .../propagation/grid_task_enqueuer.ex | 34 ++++--- .../workers/hrdps_grid_worker.ex | 90 +++++++++++++++++++ ...0260429220736_add_source_to_grid_tasks.exs | 22 +++++ .../propagation/grid_task_enqueuer_test.exs | 38 ++++++++ test/microwaveprop/propagation/grid_test.exs | 57 ++++++++++++ .../workers/hrdps_grid_worker_test.exs | 66 ++++++++++++++ 8 files changed, 369 insertions(+), 13 deletions(-) create mode 100644 lib/microwaveprop/workers/hrdps_grid_worker.ex create mode 100644 priv/repo/migrations/20260429220736_add_source_to_grid_tasks.exs create mode 100644 test/microwaveprop/workers/hrdps_grid_worker_test.exs diff --git a/config/config.exs b/config/config.exs index f3871b2a..b5c3af61 100644 --- a/config/config.exs +++ b/config/config.exs @@ -85,7 +85,13 @@ config :microwaveprop, Oban, ionosphere: 1, space_weather: 1, mechanism: 4, - contact_import: 4 + contact_import: 4, + # HRDPS Canadian propagation chain seeder. Single slot — the worker + # only inserts grid_tasks rows, so concurrent execution buys nothing + # and uniqueness is enforced by the (run_time, kind, source) index. + # Dormant until the Rust prop-grid-rs HRDPS branch ships and the + # cron entry in runtime.exs is uncommented. + hrdps: 1 ], plugins: [ {Oban.Plugins.Pruner, max_age: 3600 * 24}, diff --git a/lib/microwaveprop/propagation/grid.ex b/lib/microwaveprop/propagation/grid.ex index 8b79e760..41a51d03 100644 --- a/lib/microwaveprop/propagation/grid.ex +++ b/lib/microwaveprop/propagation/grid.ex @@ -1,5 +1,19 @@ defmodule Microwaveprop.Propagation.Grid do - @moduledoc "CONUS grid definition for propagation scoring. 0.125 degree resolution (~14 km)." + @moduledoc """ + Propagation scoring grid definitions at 0.125° (~14 km). + + Two regions, disjoint by construction: + + * `conus_points/0` — HRRR-sourced CONUS cells (lat 25-50, lon -125 to -66). + * `hrdps_only_points/0` — Canadian cells covered by HRDPS but **not** by + HRRR. Bounded north at 60°N because SRTM elevation data stops at 60°N + (Stage 8 of the HRDPS plan defers Arctic coverage). The mask is a + rectangular bbox minus the HRRR overlap; over-water cells (Hudson Bay, + Great Lakes spillover) get scored and ignored downstream. + + `point_source/1` returns `:hrrr | :hrdps | :neither` so callers can + route fetch/score work to the correct NWP source. + """ @lat_min 25.0 @lat_max 50.0 @@ -7,6 +21,11 @@ defmodule Microwaveprop.Propagation.Grid do @lon_max -66.0 @step 0.125 + @hrdps_lat_min 49.0 + @hrdps_lat_max 60.0 + @hrdps_lon_min -141.0 + @hrdps_lon_max -52.0 + @doc "Returns all grid points as `{lat, lon}` tuples covering CONUS at 0.125 degree spacing." @spec conus_points() :: [{float(), float()}] def conus_points do @@ -66,6 +85,52 @@ defmodule Microwaveprop.Propagation.Grid do } end + @doc """ + Canadian-only grid points. Cells inside the HRDPS bbox (49-60°N, + -141 to -52°W) but outside HRRR's CONUS bbox. Disjoint from + `conus_points/0` by construction so the two grids never double-write + the same `(lat, lon)`. + """ + @spec hrdps_only_points() :: [{float(), float()}] + def hrdps_only_points do + for lat <- float_range(@hrdps_lat_min, @hrdps_lat_max, @step), + lon <- float_range(@hrdps_lon_min, @hrdps_lon_max, @step), + not in_conus_bbox?(lat, lon) do + {Float.round(lat, 3), Float.round(lon, 3)} + end + end + + @doc """ + Returns the source NWP model for a `{lat, lon}` point: + + * `:hrrr` — inside the CONUS bbox; route to HrrrClient. + * `:hrdps` — inside the Canadian extent (excluding HRRR overlap); + route to HrdpsClient. + * `:neither` — outside both. Callers should fall back to the + IEMRE/RAOB enrichment paths. + """ + @spec point_source({float(), float()} | %{lat: number(), lon: number()}) :: + :hrrr | :hrdps | :neither + def point_source({lat, lon}), do: classify(lat, lon) + def point_source(%{lat: lat, lon: lon}), do: classify(lat, lon) + + defp classify(lat, lon) do + cond do + in_conus_bbox?(lat, lon) -> :hrrr + in_hrdps_bbox?(lat, lon) -> :hrdps + true -> :neither + end + end + + defp in_conus_bbox?(lat, lon) do + lat >= @lat_min and lat <= @lat_max and lon >= @lon_min and lon <= @lon_max + end + + defp in_hrdps_bbox?(lat, lon) do + lat >= @hrdps_lat_min and lat <= @hrdps_lat_max and + lon >= @hrdps_lon_min and lon <= @hrdps_lon_max + end + defp float_range(start, stop, step) do count = round((stop - start) / step) + 1 Enum.map(0..(count - 1), fn i -> start + i * step end) diff --git a/lib/microwaveprop/propagation/grid_task_enqueuer.ex b/lib/microwaveprop/propagation/grid_task_enqueuer.ex index 238cd957..e0a8d234 100644 --- a/lib/microwaveprop/propagation/grid_task_enqueuer.ex +++ b/lib/microwaveprop/propagation/grid_task_enqueuer.ex @@ -36,9 +36,17 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do Seed one kind='analysis' row (f00) plus 18 kind='forecast' rows (f01..f18) for `run_time`. This is the Phase 3 Stream A cutover shape: Rust owns the entire chain end-to-end. + + ## Options + + * `:source` — `"hrrr"` (default) or `"hrdps"`. The Rust worker branches + on this column to pick the right URL pattern + decoder. HRDPS rows + are queued only once the Rust HRDPS branch ships (deferred plan + stage 4); until then no caller passes `:source`. """ - @spec seed_with_analysis(DateTime.t()) :: {:ok, non_neg_integer()} | {:error, term()} - def seed_with_analysis(%DateTime{} = run_time) do + @spec seed_with_analysis(DateTime.t(), keyword()) :: {:ok, non_neg_integer()} | {:error, term()} + def seed_with_analysis(%DateTime{} = run_time, opts \\ []) do + source = Keyword.get(opts, :source, "hrrr") run_time = DateTime.truncate(run_time, :second) now = DateTime.truncate(DateTime.utc_now(), :microsecond) @@ -50,6 +58,7 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do status: "queued", attempt: 0, kind: "analysis", + source: source, claimed_at: nil, completed_at: nil, error: nil, @@ -67,6 +76,7 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do status: "queued", attempt: 0, kind: "forecast", + source: source, claimed_at: nil, completed_at: nil, error: nil, @@ -75,16 +85,17 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do } end - do_insert([analysis_row | forecast_rows], run_time) + do_insert([analysis_row | forecast_rows], run_time, source) end @doc """ Legacy: seed only kind='forecast' rows (f01..f18). Retained for tooling that needs to re-seed the forecast lane without touching - the analysis row. `seed_with_analysis/1` is the production path. + the analysis row. `seed_with_analysis/2` is the production path. """ - @spec seed(DateTime.t()) :: {:ok, non_neg_integer()} | {:error, term()} - def seed(%DateTime{} = run_time) do + @spec seed(DateTime.t(), keyword()) :: {:ok, non_neg_integer()} | {:error, term()} + def seed(%DateTime{} = run_time, opts \\ []) do + source = Keyword.get(opts, :source, "hrrr") run_time = DateTime.truncate(run_time, :second) now = DateTime.truncate(DateTime.utc_now(), :microsecond) @@ -98,6 +109,7 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do status: "queued", attempt: 0, kind: "forecast", + source: source, claimed_at: nil, completed_at: nil, error: nil, @@ -106,7 +118,7 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do } end - do_insert(rows, run_time) + do_insert(rows, run_time, source) end @doc """ @@ -152,17 +164,17 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do %{requeued: requeued_n, failed: failed_n} end - defp do_insert(rows, run_time) do + defp do_insert(rows, run_time, source) do {count, _} = Repo.insert_all("grid_tasks", rows, on_conflict: :nothing, - conflict_target: [:run_time, :forecast_hour, :kind] + conflict_target: [:run_time, :forecast_hour, :kind, :source] ) if count > 0 do - Logger.info("GridTaskEnqueuer: seeded #{count} grid_tasks for run_time=#{run_time}") + Logger.info("GridTaskEnqueuer: seeded #{count} grid_tasks (source=#{source}) for run_time=#{run_time}") else - Logger.info("GridTaskEnqueuer: run_time=#{run_time} already seeded") + Logger.info("GridTaskEnqueuer: run_time=#{run_time} source=#{source} already seeded") end {:ok, count} diff --git a/lib/microwaveprop/workers/hrdps_grid_worker.ex b/lib/microwaveprop/workers/hrdps_grid_worker.ex new file mode 100644 index 00000000..9723fb29 --- /dev/null +++ b/lib/microwaveprop/workers/hrdps_grid_worker.ex @@ -0,0 +1,90 @@ +defmodule Microwaveprop.Workers.HrdpsGridWorker do + @moduledoc """ + Cycle seed worker for the HRDPS Canadian propagation chain. Mirrors + `Microwaveprop.Workers.PropagationGridWorker` but seeds `grid_tasks` + rows with `source: "hrdps"` so the Rust worker routes them through + the HRDPS fetch path. + + ## Cadence + + HRDPS runs 4×/day at 00/06/12/18Z and publishes f000 ~3-4h after each + cycle. This worker is intended to fire at HH+5 of each cycle hour + (05:35Z, 11:35Z, 17:35Z, 23:35Z) — see plan stage 6 for the cron entry. + + ## Activation + + **Not yet wired into runtime.exs cron.** The Rust prop-grid-rs worker + needs an HRDPS branch (plan stage 4) before it can drain + `source: "hrdps"` rows correctly. Until then this module is dormant + scaffolding — building blocks ready for the cron wire-up once the + Rust counterpart ships. + """ + + use Oban.Worker, + queue: :hrdps, + priority: 0, + max_attempts: 5, + # Deduplicate seed jobs in a 6-hour window matching the cycle cadence. + unique: [period: 6 * 3600, states: [:available, :scheduled, :executing, :retryable]] + + alias Microwaveprop.Instrument + alias Microwaveprop.Propagation.GridTaskEnqueuer + alias Microwaveprop.Weather.HrdpsClient + + require Logger + + @impl Oban.Worker + def perform(%Oban.Job{args: args}) when args == %{} do + Instrument.span([:hrdps, :grid_worker, :perform], %{}, fn -> + seed_chain() + end) + end + + defp seed_chain do + # HRDPS shares grid_tasks with HRRR; the same reclaim sweep covers + # both because reclaim_stale_running/0 doesn't filter on source. + _ = GridTaskEnqueuer.reclaim_stale_running() + + run_time = pick_run_time(DateTime.utc_now()) + + Logger.info("HrdpsGrid: seeding chain run_time=#{run_time} source=hrdps") + + case GridTaskEnqueuer.seed_with_analysis(run_time, source: "hrdps") do + {:ok, count} -> + :telemetry.execute( + [:microwaveprop, :hrdps, :grid_worker, :seeded], + %{queued_tasks: count}, + %{run_time: run_time} + ) + + :ok + + {:error, reason} -> + :telemetry.execute( + [:microwaveprop, :hrdps, :grid_worker, :exception], + %{}, + %{run_time: run_time, reason: inspect(reason)} + ) + + {:error, reason} + end + end + + @doc """ + Pick the freshest published HRDPS cycle. Probes `now-4h` (just past + typical publish latency) first; falls back to `now-10h` (one cycle + earlier) if the fresher cycle isn't yet on the datamart. + + Public so tests can exercise selection logic without going through Oban. + """ + @spec pick_run_time(DateTime.t()) :: DateTime.t() + def pick_run_time(%DateTime{} = now) do + fresh = now |> DateTime.add(-4, :hour) |> HrdpsClient.nearest_hrdps_cycle() |> DateTime.truncate(:second) + + if HrdpsClient.cycle_available?(fresh) do + fresh + else + now |> DateTime.add(-10, :hour) |> HrdpsClient.nearest_hrdps_cycle() |> DateTime.truncate(:second) + end + end +end diff --git a/priv/repo/migrations/20260429220736_add_source_to_grid_tasks.exs b/priv/repo/migrations/20260429220736_add_source_to_grid_tasks.exs new file mode 100644 index 00000000..85773030 --- /dev/null +++ b/priv/repo/migrations/20260429220736_add_source_to_grid_tasks.exs @@ -0,0 +1,22 @@ +defmodule Microwaveprop.Repo.Migrations.AddSourceToGridTasks do + use Ecto.Migration + + # `source` splits grid_tasks between HRRR-derived (CONUS) and HRDPS- + # derived (Canada) work. Rust workers branch on this column to pick the + # right URL pattern + decoder. Default 'hrrr' so existing rows keep + # their meaning; HRDPS rows arrive only once HrdpsGridWorker is added + # to the cron (deferred until the Rust HRDPS branch ships). + def change do + alter table(:grid_tasks) do + add :source, :string, null: false, default: "hrrr" + end + + # Replace (run_time, forecast_hour, kind) uniqueness with a 4-column + # key so HRRR and HRDPS analysis/forecast rows for the same cycle can + # coexist. The index also speeds up `claim_next` filters. + drop_if_exists unique_index(:grid_tasks, [:run_time, :forecast_hour, :kind]) + create unique_index(:grid_tasks, [:run_time, :forecast_hour, :kind, :source]) + + create index(:grid_tasks, [:source]) + end +end diff --git a/test/microwaveprop/propagation/grid_task_enqueuer_test.exs b/test/microwaveprop/propagation/grid_task_enqueuer_test.exs index b5850626..05f01245 100644 --- a/test/microwaveprop/propagation/grid_task_enqueuer_test.exs +++ b/test/microwaveprop/propagation/grid_task_enqueuer_test.exs @@ -51,6 +51,44 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuerTest do assert f0_count == 0 end + describe "source-aware seeding" do + test "seed/2 defaults to source='hrrr'" do + run_time = ~U[2026-04-29 12:00:00Z] + assert {:ok, 18} = GridTaskEnqueuer.seed(run_time) + + sources = + Repo.all(from(g in "grid_tasks", where: g.run_time == ^run_time, select: g.source)) + + assert Enum.uniq(sources) == ["hrrr"] + end + + test "seed_with_analysis/2 with source='hrdps' tags every row" do + run_time = ~U[2026-04-29 12:00:00Z] + assert {:ok, 19} = GridTaskEnqueuer.seed_with_analysis(run_time, source: "hrdps") + + sources = + Repo.all(from(g in "grid_tasks", where: g.run_time == ^run_time, select: g.source)) + + assert length(sources) == 19 + assert Enum.uniq(sources) == ["hrdps"] + end + + test "HRRR and HRDPS chains for the same run_time coexist (4-column unique key)" do + run_time = ~U[2026-04-29 18:00:00Z] + assert {:ok, 19} = GridTaskEnqueuer.seed_with_analysis(run_time, source: "hrrr") + assert {:ok, 19} = GridTaskEnqueuer.seed_with_analysis(run_time, source: "hrdps") + + total = Repo.one(from(g in "grid_tasks", where: g.run_time == ^run_time, select: count())) + assert total == 38 + end + + test "reseeding the same run_time + source is idempotent" do + run_time = ~U[2026-04-29 06:00:00Z] + assert {:ok, 19} = GridTaskEnqueuer.seed_with_analysis(run_time, source: "hrdps") + assert {:ok, 0} = GridTaskEnqueuer.seed_with_analysis(run_time, source: "hrdps") + end + end + describe "reclaim_stale_running/1" do test "flips stale-running rows back to queued and clears claimed_at" do now = DateTime.truncate(DateTime.utc_now(), :second) diff --git a/test/microwaveprop/propagation/grid_test.exs b/test/microwaveprop/propagation/grid_test.exs index cd4c6d47..e64ad0d2 100644 --- a/test/microwaveprop/propagation/grid_test.exs +++ b/test/microwaveprop/propagation/grid_test.exs @@ -114,4 +114,61 @@ defmodule Microwaveprop.Propagation.GridTest do assert length(points) == spec.lon_count * spec.lat_count end end + + describe "hrdps_only_points/0" do + test "covers cells north of HRRR's lat_max" do + points = Grid.hrdps_only_points() + # At least one cell strictly north of HRRR's lat_max=50 + assert Enum.any?(points, fn {lat, _lon} -> lat > 50.0 end) + end + + test "is disjoint from conus_points/0" do + conus = MapSet.new(Grid.conus_points()) + hrdps = MapSet.new(Grid.hrdps_only_points()) + assert MapSet.disjoint?(conus, hrdps), "hrdps_only_points must not overlap conus_points" + end + + test "all points within HRDPS bbox" do + Enum.each(Grid.hrdps_only_points(), fn {lat, lon} -> + assert lat >= 49.0 and lat <= 60.0 + assert lon >= -141.0 and lon <= -52.0 + end) + end + + test "points are on 0.125 degree grid" do + Enum.each(Grid.hrdps_only_points(), fn {lat, lon} -> + assert Float.round(lat * 8, 0) == lat * 8 + assert Float.round(lon * 8, 0) == lon * 8 + end) + end + + test "no point is inside the CONUS bounding box" do + Enum.each(Grid.hrdps_only_points(), fn {lat, lon} -> + in_conus = lat >= 25.0 and lat <= 50.0 and lon >= -125.0 and lon <= -66.0 + refute in_conus, "point #{lat},#{lon} is inside CONUS bbox" + end) + end + end + + describe "point_source/1" do + test "classifies cells inside CONUS as :hrrr" do + assert Grid.point_source({32.9, -97.0}) == :hrrr + assert Grid.point_source({49.0, -98.0}) == :hrrr + end + + test "classifies Canadian cells outside HRRR as :hrdps" do + assert Grid.point_source({53.32, -60.42}) == :hrdps + assert Grid.point_source({50.125, -97.0}) == :hrdps + end + + test "classifies points outside both bboxes as :neither" do + assert Grid.point_source({-30.0, 150.0}) == :neither + assert Grid.point_source({65.0, -90.0}) == :neither + end + + test "accepts %{lat: _, lon: _} maps as well as tuples" do + assert Grid.point_source(%{lat: 32.9, lon: -97.0}) == :hrrr + assert Grid.point_source(%{lat: 53.32, lon: -60.42}) == :hrdps + end + end end diff --git a/test/microwaveprop/workers/hrdps_grid_worker_test.exs b/test/microwaveprop/workers/hrdps_grid_worker_test.exs new file mode 100644 index 00000000..3890aa7c --- /dev/null +++ b/test/microwaveprop/workers/hrdps_grid_worker_test.exs @@ -0,0 +1,66 @@ +defmodule Microwaveprop.Workers.HrdpsGridWorkerTest do + use Microwaveprop.DataCase, async: false + use Oban.Testing, repo: Microwaveprop.Repo + + import Ecto.Query + + alias Microwaveprop.Repo + alias Microwaveprop.Workers.HrdpsGridWorker + + setup do + original = Application.get_env(:microwaveprop, :hrdps_cycle_available_fn) + + on_exit(fn -> + if original, + do: Application.put_env(:microwaveprop, :hrdps_cycle_available_fn, original), + else: Application.delete_env(:microwaveprop, :hrdps_cycle_available_fn) + end) + end + + describe "pick_run_time/1" do + test "returns the now-4h cycle when it's published" do + Application.put_env(:microwaveprop, :hrdps_cycle_available_fn, fn _dt -> true end) + + now = ~U[2026-04-29 16:30:00Z] + run_time = HrdpsGridWorker.pick_run_time(now) + + # now - 4h = 12:30Z; snap down to cycle = 12:00Z + assert run_time == ~U[2026-04-29 12:00:00Z] + end + + test "falls back to the previous cycle when fresh isn't yet published" do + Application.put_env(:microwaveprop, :hrdps_cycle_available_fn, fn _dt -> false end) + + now = ~U[2026-04-29 16:30:00Z] + run_time = HrdpsGridWorker.pick_run_time(now) + + # now - 10h = 06:30Z; snap down to cycle = 06:00Z + assert run_time == ~U[2026-04-29 06:00:00Z] + end + end + + describe "perform/1" do + test "seeds 19 grid_tasks rows tagged with source='hrdps'" do + Application.put_env(:microwaveprop, :hrdps_cycle_available_fn, fn _dt -> true end) + + assert :ok = perform_job(HrdpsGridWorker, %{}) + + sources_count = + Repo.one(from(g in "grid_tasks", where: g.source == "hrdps", select: count())) + + assert sources_count == 19 + end + + test "is idempotent — re-running the worker for the same cycle inserts nothing extra" do + Application.put_env(:microwaveprop, :hrdps_cycle_available_fn, fn _dt -> true end) + + assert :ok = perform_job(HrdpsGridWorker, %{}) + assert :ok = perform_job(HrdpsGridWorker, %{}) + + total = + Repo.one(from(g in "grid_tasks", where: g.source == "hrdps", select: count())) + + assert total == 19 + end + end +end