From 1da78a80e590d5527f2eb6215b717b85fc9f749d Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Wed, 29 Apr 2026 17:29:37 -0500 Subject: [PATCH] feat(hrdps): activate Canadian propagation chain end-to-end MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stages 6, 7, 10 of the HRDPS plan, plus the Elixir read-side merge that makes Rust's `.hrdps.prop` writes show up on /map. Read side: - `ScoresFile.path_for_hrdps/2` + `read_hrdps/2` — companions to the HRRR equivalents. Same decode path; different filename. - `ScoresFile.read_bounds/3` now reads HRRR + HRDPS files and concatenates the cells. Files are disjoint by construction (Grid.hrdps_only_points excludes CONUS) so no de-dup needed. - `ScoresFile.read_point/4` checks `.prop`, then `.hrdps.prop`, then legacy `.ntms` — first hit wins. Cells outside their owning region read as no-data in the other file's grid, so the order doesn't matter for correctness. - `Propagation.warm_cache_and_broadcast/2` reads both files and merges before broadcasting to the cluster ScoreCache. A single missing file (HRDPS pre-cycle, HRRR briefly absent) is now OK — the cache warms with whichever side is available. Activation: - runtime.exs cron: HRDPS seed worker fires at HH:35 of each cycle hour (05:35Z, 11:35Z, 17:35Z, 23:35Z) — 5h after cycle, 30 min staggered from HRRR's */15 schedule. - HrdpsGridWorker moduledoc updated: no longer dormant; Rust prop-grid-rs handles source='hrdps' rows. - map_live North America bbox extended to (24.5, 60.0, -141.0, -52.0) for the out-of-coverage banner check; visitors anywhere in CONUS or the Canadian extent now stay banner-free. Cleanup: - mix hrdps.probe deleted — its risks are retired and the production HrdpsClient covers the same surface. What's still in motion (not blocking the Canadian /map): - Per-valid_time profile + scalar artifacts for HRDPS (would let /weather show Canadian temperature/refractivity layers). Rust pipeline currently skips those so the HRRR-written companion files aren't clobbered — separate stage when /weather UX is decided. - FreshnessMonitor HRDPS staleness tracking — HRDPS has its own cron and a different cadence than HRRR; HRRR-stale logic doesn't apply cleanly. Defer until prod tells us what alarms we actually want. --- config/runtime.exs | 7 + lib/microwaveprop/propagation.ex | 41 +++-- lib/microwaveprop/propagation/scores_file.ex | 52 +++++- .../workers/hrdps_grid_worker.ex | 9 +- lib/microwaveprop_web/live/map_live.ex | 13 +- lib/mix/tasks/hrdps_probe.ex | 165 ------------------ .../propagation/scores_file_test.exs | 32 ++++ 7 files changed, 123 insertions(+), 196 deletions(-) delete mode 100644 lib/mix/tasks/hrdps_probe.ex diff --git a/config/runtime.exs b/config/runtime.exs index 5da21990..da5d8e78 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -247,6 +247,13 @@ if config_env() == :prod do # so duplicate fires for an already-seeded run_time are a single # idempotent INSERT batch. {"5,20,35,50 * * * *", Microwaveprop.Workers.PropagationGridWorker}, + # HRDPS Canadian propagation chain. ECCC publishes HRDPS f000 + # ~3-4h after each 00/06/12/18Z cycle, so seeding at HH:35 of + # each cycle hour (5h after cycle, 30 min offset from HRRR's + # */15 schedule) gives a comfortable margin before the f000 + # files land. The Rust prop-grid-rs HRDPS branch (run_chain_step_hrdps) + # drains the resulting source='hrdps' rows. + {"35 5,11,17,23 * * *", Microwaveprop.Workers.HrdpsGridWorker}, # GEFS runs publish ~3-4h after the 00/06/12/18Z cycle; seed the # Day 2-7 outlook 5h after each run to stay safely past NOMADS' # publication lag. Seeder enqueues f024-f168 at 6-hour cadence diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index c1e2db29..2bf742d0 100644 --- a/lib/microwaveprop/propagation.ex +++ b/lib/microwaveprop/propagation.ex @@ -345,27 +345,38 @@ defmodule Microwaveprop.Propagation do end @doc """ - 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. + Load the full North America score set for `{band_mhz, valid_time}` from + the on-disk binary files (HRRR `.prop` + HRDPS `.hrdps.prop`, merged) 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. - Returns `{:error, reason}` when the score file is missing or - corrupt — callers distinguish those from the successful empty-grid - case to avoid poisoning the cache with `[]` on a bad read. + Returns `{:error, :enoent}` only when neither file exists. A single + missing file (HRDPS pre-cycle, HRRR briefly absent) is OK — the cache + warms with whichever side is available. """ @spec warm_cache_and_broadcast(non_neg_integer(), DateTime.t()) :: :ok | {:error, :enoent | :invalid_format} def warm_cache_and_broadcast(band_mhz, valid_time) do - case ScoresFile.read(band_mhz, valid_time) do - {:ok, payload} -> - scores = ScoresFile.extract_points(payload, nil) - ScoreCache.broadcast_put(band_mhz, valid_time, scores) - :ok + hrrr = + case ScoresFile.read(band_mhz, valid_time) do + {:ok, payload} -> ScoresFile.extract_points(payload, nil) + _ -> nil + end - {:error, reason} -> - {:error, reason} + hrdps = + case ScoresFile.read_hrdps(band_mhz, valid_time) do + {:ok, payload} -> ScoresFile.extract_points(payload, nil) + _ -> nil + end + + case {hrrr, hrdps} do + {nil, nil} -> + {:error, :enoent} + + {h, c} -> + ScoreCache.broadcast_put(band_mhz, valid_time, (h || []) ++ (c || [])) + :ok end end diff --git a/lib/microwaveprop/propagation/scores_file.ex b/lib/microwaveprop/propagation/scores_file.ex index 9d5062df..e511a3eb 100644 --- a/lib/microwaveprop/propagation/scores_file.ex +++ b/lib/microwaveprop/propagation/scores_file.ex @@ -50,13 +50,25 @@ defmodule Microwaveprop.Propagation.ScoresFile do Application.get_env(:microwaveprop, :propagation_scores_dir, "/data/scores") end - @doc "Returns the full path for a `(band_mhz, valid_time)` score file." + @doc "Returns the full path for a `(band_mhz, valid_time)` HRRR score file." @spec path_for(non_neg_integer(), DateTime.t()) :: String.t() def path_for(band_mhz, %DateTime{} = valid_time) when is_integer(band_mhz) do iso = valid_time |> DateTime.truncate(:second) |> DateTime.to_iso8601() Path.join([base_dir(), Integer.to_string(band_mhz), "#{iso}.prop"]) end + @doc """ + Returns the path for the HRDPS-source companion score file. Sibling of + `path_for/2` — both files coexist for the same `(band, valid_time)`. + Read-side functions transparently merge the two so callers get the + union of CONUS + Canadian scored cells. + """ + @spec path_for_hrdps(non_neg_integer(), DateTime.t()) :: String.t() + def path_for_hrdps(band_mhz, %DateTime{} = valid_time) when is_integer(band_mhz) do + iso = valid_time |> DateTime.truncate(:second) |> DateTime.to_iso8601() + Path.join([base_dir(), Integer.to_string(band_mhz), "#{iso}.hrdps.prop"]) + end + @doc false @spec legacy_path_for(non_neg_integer(), DateTime.t()) :: String.t() def legacy_path_for(band_mhz, %DateTime{} = valid_time) when is_integer(band_mhz) do @@ -241,9 +253,35 @@ defmodule Microwaveprop.Propagation.ScoresFile do @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) - _ -> [] + hrrr_cells = + case read(band_mhz, valid_time) do + {:ok, payload} -> extract_points(payload, bounds) + _ -> [] + end + + hrdps_cells = + case read_hrdps(band_mhz, valid_time) do + {:ok, payload} -> extract_points(payload, bounds) + _ -> [] + end + + # The two grids are disjoint by construction (Grid.hrdps_only_points + # excludes CONUS), so concat without dedup. Order isn't load-bearing — + # callers downstream of /scores/cells don't depend on ordering. + hrrr_cells ++ hrdps_cells + end + + @doc """ + Read the HRDPS-source score file for `(band_mhz, valid_time)`. Same + decode shape as `read/2`; uses the `.hrdps.prop` extension. + """ + @spec read_hrdps(non_neg_integer(), DateTime.t()) :: + {:ok, map()} | {:error, :enoent | :invalid_format} + def read_hrdps(band_mhz, %DateTime{} = valid_time) when is_integer(band_mhz) do + case File.read(path_for_hrdps(band_mhz, valid_time)) do + {:ok, binary} -> decode(binary) + {:error, :enoent} -> {:error, :enoent} + {:error, other} -> {:error, other} end end @@ -261,7 +299,11 @@ defmodule Microwaveprop.Propagation.ScoresFile do @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 Enum.find_value( - [path_for(band_mhz, valid_time), legacy_path_for(band_mhz, valid_time)], + [ + path_for(band_mhz, valid_time), + path_for_hrdps(band_mhz, valid_time), + legacy_path_for(band_mhz, valid_time) + ], &open_and_read_point(&1, lat, lon) ) end diff --git a/lib/microwaveprop/workers/hrdps_grid_worker.ex b/lib/microwaveprop/workers/hrdps_grid_worker.ex index 9723fb29..65aeee02 100644 --- a/lib/microwaveprop/workers/hrdps_grid_worker.ex +++ b/lib/microwaveprop/workers/hrdps_grid_worker.ex @@ -13,11 +13,10 @@ defmodule Microwaveprop.Workers.HrdpsGridWorker do ## 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. + Active. The Rust prop-grid-rs worker handles `source: "hrdps"` rows via + `pipeline::run_chain_step_hrdps`, fetching from MSC Datamart, writing + scores to `//.hrdps.prop` (sibling of the HRRR + `.prop` file for the same cycle). Cron entry lives in `runtime.exs`. """ use Oban.Worker, diff --git a/lib/microwaveprop_web/live/map_live.ex b/lib/microwaveprop_web/live/map_live.ex index 16cae79e..8700c0ad 100644 --- a/lib/microwaveprop_web/live/map_live.ex +++ b/lib/microwaveprop_web/live/map_live.ex @@ -15,13 +15,14 @@ defmodule MicrowavepropWeb.MapLive do @default_center %{lat: 32.897, lon: -97.038} @default_zoom 7 - # Continental-US bounding box (excludes Alaska, Hawaii, Puerto Rico). Used - # to decide whether to show the "data is CONUS-only" banner to visitors - # whose Cloudflare geolocation falls outside the dataset's coverage. + # North American coverage bbox: HRRR (CONUS) + HRDPS (Canadian extent + # to 60°N — Arctic deferred until SRTM gap is filled). Used to decide + # whether to show the "out of coverage area" banner to visitors whose + # Cloudflare geolocation falls outside the combined dataset. @conus_lat_min 24.5 - @conus_lat_max 49.5 - @conus_lon_min -125.0 - @conus_lon_max -66.5 + @conus_lat_max 60.0 + @conus_lon_min -141.0 + @conus_lon_max -52.0 # How often to re-query oban_jobs for pipeline state while the map is # open. Short enough that the "Updating…" chip shows up within a page- diff --git a/lib/mix/tasks/hrdps_probe.ex b/lib/mix/tasks/hrdps_probe.ex deleted file mode 100644 index 4bf96ac1..00000000 --- a/lib/mix/tasks/hrdps_probe.ex +++ /dev/null @@ -1,165 +0,0 @@ -defmodule Mix.Tasks.Hrdps.Probe do - @shortdoc "Throwaway probe: fetch one HRDPS variable, decode at sample cities" - @moduledoc """ - One-shot probe for the Canadian HRDPS model. Fetches a single GRIB2 - file from MSC Datamart and uses `wgrib2 -lon` to extract values at - five Canadian cities, printing a table for visual sanity-checking. - - Designed as a wedge to retire risk before committing to a real - HrdpsClient: validates URL pattern, GRIB2 decode, and rotated-lat/lon - reprojection without touching the DB, Oban, or the score grid. - - Throwaway code. Will be deleted once the real client lands. - - mix hrdps.probe # latest cycle, f000, 2m temperature - mix hrdps.probe --hour 6 # f006 from latest cycle - mix hrdps.probe --cycle 20260429T12Z - - URL pattern (current as of 2026-04-29; ECCC reorganized away from the - old `/model_hrdps/` root to a date-prefixed structure): - - https://dd.weather.gc.ca/{YYYYMMDD}/WXO-DD/model_hrdps/continental/2.5km/ - {HH}/{FFF}/{YYYYMMDD}T{HH}Z_MSC_HRDPS_{VAR}_{LEVEL}_RLatLon0.0225_PT{FFF}H.grib2 - """ - use Mix.Task - - @datamart_base "https://dd.weather.gc.ca" - - # Variable / level slug for 2m temperature in HRDPS naming. HRRR-side - # equivalent is `TMP:2 m above ground`; HRDPS uses MSC's per-variable - # filename convention. - @variable_slug "TMP_AGL-2m" - - @cities [ - {"YYZ Toronto", 43.68, -79.63}, - {"YVR Vancouver", 49.19, -123.18}, - {"YYC Calgary", 51.11, -114.02}, - {"YOW Ottawa", 45.32, -75.67}, - {"YHZ Halifax", 44.88, -63.51} - ] - - @impl Mix.Task - def run(argv) do - {opts, _, _} = - OptionParser.parse(argv, - strict: [cycle: :string, hour: :integer], - aliases: [c: :cycle, h: :hour] - ) - - {:ok, _} = Application.ensure_all_started(:req) - - forecast_hour = Keyword.get(opts, :hour, 0) - cycle = opts |> Keyword.get(:cycle) |> resolve_cycle() - - url = build_url(cycle, forecast_hour) - IO.puts("Fetching: #{url}") - - {:ok, path} = download(url) - IO.puts("Downloaded #{path} (#{File.stat!(path).size} bytes)") - - print_inventory(path) - print_city_table(path, cycle, forecast_hour) - end - - # ----- Cycle resolution ----- - - defp resolve_cycle(nil), do: latest_published_cycle() - - defp resolve_cycle(<>) do - %{date: "#{year}#{month}#{day}", hour: hour} - end - - defp resolve_cycle(other) do - Mix.raise("Invalid --cycle #{inspect(other)} — expected YYYYMMDDTHHZ") - end - - # HRDPS publishes ~3-4h after each cycle (00/06/12/18Z). Walk back from - # "now - 4h" until we find a cycle whose root directory exists. - defp latest_published_cycle do - now = DateTime.utc_now() - candidates = for h <- 0..3, do: DateTime.add(now, -((4 + h * 6) * 3600), :second) - - candidates - |> Enum.map(&snap_to_cycle/1) - |> Enum.uniq() - |> Enum.find(&cycle_published?/1) - |> case do - nil -> Mix.raise("No published HRDPS cycle found in the last 24h") - cycle -> cycle - end - end - - defp snap_to_cycle(%DateTime{} = dt) do - cycle_hour = div(dt.hour, 6) * 6 - %{date: Calendar.strftime(dt, "%Y%m%d"), hour: String.pad_leading("#{cycle_hour}", 2, "0")} - end - - defp cycle_published?(%{date: date, hour: hour}) do - url = "#{@datamart_base}/#{date}/WXO-DD/model_hrdps/continental/2.5km/#{hour}/000/" - - case Req.head(url, retry: false, connect_options: [timeout: 5_000]) do - {:ok, %{status: 200}} -> true - _ -> false - end - end - - # ----- URL + download ----- - - defp build_url(%{date: date, hour: hour}, forecast_hour) do - fff = String.pad_leading("#{forecast_hour}", 3, "0") - filename = "#{date}T#{hour}Z_MSC_HRDPS_#{@variable_slug}_RLatLon0.0225_PT#{fff}H.grib2" - - "#{@datamart_base}/#{date}/WXO-DD/model_hrdps/continental/2.5km/#{hour}/#{fff}/#{filename}" - end - - defp download(url) do - path = Path.join(System.tmp_dir!(), "hrdps_probe_#{System.unique_integer([:positive])}.grib2") - - case Req.get(url, into: File.stream!(path), retry: false) do - {:ok, %{status: 200}} -> {:ok, path} - {:ok, %{status: status}} -> Mix.raise("Datamart returned HTTP #{status} for #{url}") - {:error, reason} -> Mix.raise("Fetch failed: #{inspect(reason)}") - end - end - - # ----- wgrib2 inspection ----- - - defp print_inventory(path) do - {output, 0} = System.cmd("wgrib2", [path], stderr_to_stdout: true) - IO.puts("\nGRIB2 inventory:") - IO.puts(" " <> String.trim(output)) - end - - defp print_city_table(path, cycle, forecast_hour) do - args = [path] ++ Enum.flat_map(@cities, fn {_, lat, lon} -> ["-lon", "#{lon}", "#{lat}"] end) - {output, 0} = System.cmd("wgrib2", args, stderr_to_stdout: true) - - values = parse_wgrib2_lon_output(output) - - IO.puts("\nHRDPS 2m temperature, cycle #{cycle.date}T#{cycle.hour}Z f#{pad3(forecast_hour)}:\n") - IO.puts(" city lat,lon temp") - IO.puts(" -------------------- -------------------- ------") - - @cities - |> Enum.zip(values) - |> Enum.each(fn {{name, lat, lon}, kelvin} -> - celsius = kelvin - 273.15 - - IO.puts(" #{String.pad_trailing(name, 20)} #{format_coord(lat, lon)} #{format_temp(celsius)}") - end) - end - - # wgrib2 -lon emits one line for the GRIB record with all `lon=…lat=…val=…` - # entries colon-separated. For five cities we get five `val=` matches. - defp parse_wgrib2_lon_output(output) do - ~r/val=([\-\d\.]+)/ - |> Regex.scan(output) - |> Enum.map(fn [_, v] -> String.to_float(v) end) - end - - defp format_coord(lat, lon), do: "#{:io_lib.format("~6.2f", [lat])},#{:io_lib.format("~7.2f", [lon])}" - - defp format_temp(celsius), do: "#{:io_lib.format("~5.1f", [celsius])}°C" - - defp pad3(n), do: String.pad_leading("#{n}", 3, "0") -end diff --git a/test/microwaveprop/propagation/scores_file_test.exs b/test/microwaveprop/propagation/scores_file_test.exs index 309b4552..3b294e66 100644 --- a/test/microwaveprop/propagation/scores_file_test.exs +++ b/test/microwaveprop/propagation/scores_file_test.exs @@ -253,6 +253,38 @@ defmodule Microwaveprop.Propagation.ScoresFileTest do 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 + + test "merges cells from a sibling .hrdps.prop file when present" do + # Write the HRRR file via the public writer, then copy it to the + # HRDPS path. Both files now contain the same cell — read_bounds + # should return both, doubling the cell list. (Production HRDPS + # files have a different bbox; this test only verifies the merge + # plumbing routes through both paths.) + vt = ~U[2026-04-14 18:00:00Z] + ScoresFile.write!(10_000, vt, [%{lat: 25.0, lon: -125.0, score: 42}]) + + hrrr_path = ScoresFile.path_for(10_000, vt) + hrdps_path = ScoresFile.path_for_hrdps(10_000, vt) + File.cp!(hrrr_path, hrdps_path) + + scores = ScoresFile.read_bounds(10_000, vt) + assert length(scores) == 2 + assert Enum.all?(scores, &(&1.score == 42)) + end + + test "returns HRDPS cells alone when only the .hrdps.prop file exists" do + vt = ~U[2026-04-14 18:00:00Z] + # Use write! to land a file at the HRRR path, then move it to + # the HRDPS path so only the HRDPS file remains. + ScoresFile.write!(10_000, vt, [%{lat: 25.0, lon: -125.0, score: 80}]) + + File.rename!( + ScoresFile.path_for(10_000, vt), + ScoresFile.path_for_hrdps(10_000, vt) + ) + + assert [%{score: 80}] = ScoresFile.read_bounds(10_000, vt) + end end describe "read_point/4" do