From ed67efb256818a7b194be0920b196ac76bad58e1 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sun, 12 Apr 2026 14:49:20 -0500 Subject: [PATCH] Add MRMS rain mosaic, fix beacons crash, fix UTC clock flash MRMS ---- Layer the NOAA MRMS PrecipRate product onto the score grid so rain fade updates every 2 minutes instead of every hour alongside HRRR. New modules: - Microwaveprop.Weather.MrmsClient: fetches the latest .grib2.gz off the NCEP mirror (Req auto-decompresses so no gunzip step), writes the raw GRIB2 to a temp file, and calls the existing wgrib2 wrapper with the 0.125 propagation grid spec to get interpolated cells. Returns a %{{lat, lon} => mm_per_hour} map with missing-value sentinels dropped. - Microwaveprop.Weather.MrmsCache: ETS-backed GenServer mirroring ScoreCache/GridCache. Caches a single "current" entry keyed by valid_time with PubSub broadcast so peer nodes stay in sync and only the Oban leader pays the fetch + regrid cost. - Microwaveprop.Workers.MrmsFetchWorker: cron every 2 minutes, short- circuits when the cached valid_time already matches the newest file. Microwaveprop.Propagation.AsosNudge.compute/4 now takes an optional rain_grid. When a cell has MRMS rain >= 0.1 mm/hr it gets patched onto the HRRR profile's `precip_mm` field (the scorer already reads it there) and the cell is re-scored even with no ASOS station nearby. Cells with MRMS rain below the threshold aren't touched so dry cells keep their raw HRRR scores (which have the wind/sky/native-gradient signal that isn't persisted on HrrrProfile rows and would otherwise be lost). AsosAdjustmentWorker pulls MrmsCache on every tick and passes the grid through to AsosNudge.compute/4. Also skips the IemClient error branch that can never happen and handles the ASOS-empty + MRMS-empty case explicitly. MrmsCache wired into the supervision tree; MrmsFetchWorker cron entry added to config.exs and dev.exs. Four new AsosNudge cases cover MRMS-only re-scoring, threshold gating, and the wet/dry score delta. Beacons 500 ----------- Beacon.format_freq/1 and format_mw/1 crashed on whole-number floats (e.g. 24192.0) because `frac == 0.0` could become false under float rounding while `trim_trailing_zeros/1` stripped the decimal point, leaving a 1-element list that couldn't be destructured as [_, frac]. Shared format_number/1 helper handles integer input directly and pattern-matches both the "int-only" and "int + frac" shapes. Added stream_data property tests covering the whole microwave range for both integers and floats to catch this class of bug before prod. UTC clock flash --------------- The /weather and /map UTC clocks were empty until the JS hook mounted post-WebSocket, producing a several-second blank spot on initial load and a clobber risk on sidebar re-renders. Mount now computes a server-rendered `initial_utc_clock` string and the template seeds the element with that plus `phx-update="ignore"` so LiveView morphdom won't overwrite what the hook writes. --- config/config.exs | 1 + config/dev.exs | 1 + lib/microwaveprop/application.ex | 1 + lib/microwaveprop/beacons/beacon.ex | 44 ++--- lib/microwaveprop/propagation/asos_nudge.ex | 57 +++++- lib/microwaveprop/weather/mrms_cache.ex | 80 ++++++++ lib/microwaveprop/weather/mrms_client.ex | 181 ++++++++++++++++++ .../workers/asos_adjustment_worker.ex | 46 +++-- .../workers/mrms_fetch_worker.ex | 41 ++++ lib/microwaveprop_web/live/map_live.ex | 6 + .../live/weather_map_live.ex | 5 + test/microwaveprop/beacons_test.exs | 93 +++++++++ .../propagation/asos_nudge_test.exs | 33 ++++ .../weather/mrms_client_test.exs | 72 +++++++ 14 files changed, 616 insertions(+), 45 deletions(-) create mode 100644 lib/microwaveprop/weather/mrms_cache.ex create mode 100644 lib/microwaveprop/weather/mrms_client.ex create mode 100644 lib/microwaveprop/workers/mrms_fetch_worker.ex create mode 100644 test/microwaveprop/weather/mrms_client_test.exs diff --git a/config/config.exs b/config/config.exs index 92aca30a..358950c5 100644 --- a/config/config.exs +++ b/config/config.exs @@ -65,6 +65,7 @@ config :microwaveprop, Oban, {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker}, {"*/5 * * * *", Microwaveprop.Commercial.PollWorker}, {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker}, + {"*/2 * * * *", Microwaveprop.Workers.MrmsFetchWorker}, {"*/10 * * * *", Microwaveprop.Workers.AsosAdjustmentWorker}, {"*/15 * * * *", Microwaveprop.Workers.PropagationPruneWorker} ]} diff --git a/config/dev.exs b/config/dev.exs index 523449f6..c1394f64 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -88,6 +88,7 @@ config :microwaveprop, Oban, {Oban.Plugins.Cron, crontab: [ {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker}, + {"*/2 * * * *", Microwaveprop.Workers.MrmsFetchWorker}, {"*/10 * * * *", Microwaveprop.Workers.AsosAdjustmentWorker}, {"*/15 * * * *", Microwaveprop.Workers.PropagationPruneWorker}, {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker} diff --git a/lib/microwaveprop/application.ex b/lib/microwaveprop/application.ex index c5f27a37..1e79cf78 100644 --- a/lib/microwaveprop/application.ex +++ b/lib/microwaveprop/application.ex @@ -17,6 +17,7 @@ defmodule Microwaveprop.Application do Microwaveprop.Cache, Microwaveprop.Propagation.ScoreCache, Microwaveprop.Weather.GridCache, + Microwaveprop.Weather.MrmsCache, Microwaveprop.Weather.NexradCache, {Oban, Application.fetch_env!(:microwaveprop, Oban)}, Microwaveprop.RepoListener, diff --git a/lib/microwaveprop/beacons/beacon.ex b/lib/microwaveprop/beacons/beacon.ex index df00a7e1..8eb414f0 100644 --- a/lib/microwaveprop/beacons/beacon.ex +++ b/lib/microwaveprop/beacons/beacon.ex @@ -56,18 +56,7 @@ defmodule Microwaveprop.Beacons.Beacon do def format_mw(nil), do: "" def format_mw(mw) when is_number(mw) do - int_part = trunc(mw) - frac = mw - int_part - - formatted_int = add_commas(int_part) - - if frac == 0.0 do - formatted_int - else - decimal = mw |> :erlang.float_to_binary(decimals: 3) |> trim_trailing_zeros() - [_int, frac_part] = String.split(decimal, ".") - "#{formatted_int}.#{frac_part}" - end + format_number(mw) end def format_mw(mw), do: to_string(mw) @@ -77,22 +66,29 @@ defmodule Microwaveprop.Beacons.Beacon do def format_freq(nil), do: "" def format_freq(mhz) when is_number(mhz) do - int_part = trunc(mhz) - frac = mhz - int_part - - formatted_int = add_commas(int_part) - - if frac == 0.0 do - formatted_int - else - decimal = mhz |> :erlang.float_to_binary(decimals: 3) |> trim_trailing_zeros() - [_int, frac_part] = String.split(decimal, ".") - "#{formatted_int}.#{frac_part}" - end + format_number(mhz) end def format_freq(mhz), do: to_string(mhz) + # Shared formatter for format_mw/1 and format_freq/1. Unconditionally + # renders to 3 decimal places, trims trailing zeros, and handles the + # edge case where `trim_trailing_zeros/1` strips the decimal point + # entirely (e.g. 24192.0 → "24192.000" → "24192", which previously + # crashed the `[_int, frac_part] = ...` pattern match when + # `frac != 0.0` due to float rounding error). + defp format_number(n) when is_integer(n), do: add_commas(n) + + defp format_number(n) when is_float(n) do + formatted_int = n |> trunc() |> add_commas() + decimal = n |> :erlang.float_to_binary(decimals: 3) |> trim_trailing_zeros() + + case String.split(decimal, ".") do + [_int_only] -> formatted_int + [_int, frac_part] -> "#{formatted_int}.#{frac_part}" + end + end + defp add_commas(int) do int |> Integer.to_string() diff --git a/lib/microwaveprop/propagation/asos_nudge.ex b/lib/microwaveprop/propagation/asos_nudge.ex index 0a30ee3e..8e86cd39 100644 --- a/lib/microwaveprop/propagation/asos_nudge.ex +++ b/lib/microwaveprop/propagation/asos_nudge.ex @@ -40,6 +40,13 @@ defmodule Microwaveprop.Propagation.AsosNudge do @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(), @@ -60,6 +67,8 @@ defmodule Microwaveprop.Propagation.AsosNudge do @type hrrr_profile :: map() + @type rain_grid :: %{{float(), float()} => float()} + @type grid_score :: %{ lat: float(), lon: float(), @@ -70,22 +79,32 @@ defmodule Microwaveprop.Propagation.AsosNudge do } @doc """ - Nudge every HRRR grid cell that has an ASOS station within 250 km and - re-score it. Cells outside that radius are dropped so the existing HRRR - scores stay in place. - """ - @spec compute([observation()], DateTime.t(), [hrrr_profile()]) :: [grid_score()] - def compute([], _valid_time, _profiles), do: [] - def compute(_observations, _valid_time, []), do: [] + 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. - def compute(observations, valid_time, profiles) do + 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(&any_station_within_radius?(&1, residuals)) + |> 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 @@ -172,6 +191,26 @@ defmodule Microwaveprop.Propagation.AsosNudge do 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 diff --git a/lib/microwaveprop/weather/mrms_cache.ex b/lib/microwaveprop/weather/mrms_cache.ex new file mode 100644 index 00000000..cbf8865e --- /dev/null +++ b/lib/microwaveprop/weather/mrms_cache.ex @@ -0,0 +1,80 @@ +defmodule Microwaveprop.Weather.MrmsCache do + @moduledoc """ + Node-local ETS cache of the latest MRMS PrecipRate grid, regridded onto + the 0.125° propagation grid. Mirrors the pattern used by + `Microwaveprop.Propagation.ScoreCache` and + `Microwaveprop.Weather.GridCache` — a single GenServer owns the table, + callers read directly from ETS, and `broadcast_put/2` fans updates out + to peer nodes via PubSub. + + Only one node runs `Microwaveprop.Workers.MrmsFetchWorker` per cron + tick (Oban Pro Smart engine picks a leader), so the other nodes get + the grid via the `"mrms:cache"` topic and stay in sync without each + one paying the 1 MB fetch + wgrib2 regrid cost. + """ + use GenServer + + alias Phoenix.PubSub + + @table :mrms_cache + @topic "mrms:cache" + @pubsub Microwaveprop.PubSub + + @type rain_grid :: %{{float(), float()} => float()} + + @spec start_link(keyword()) :: GenServer.on_start() + def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__) + + @doc "Return the cached `{valid_time, rain_grid}` or `:miss`." + @spec fetch() :: {:ok, DateTime.t(), rain_grid()} | :miss + def fetch do + case :ets.lookup(@table, :current) do + [{_, valid_time, grid}] -> {:ok, valid_time, grid} + [] -> :miss + end + end + + @doc "Current MRMS valid_time if anything is cached." + @spec valid_time() :: DateTime.t() | nil + def valid_time do + case fetch() do + {:ok, vt, _} -> vt + :miss -> nil + end + end + + @spec put(DateTime.t(), rain_grid()) :: :ok + def put(valid_time, grid) do + :ets.insert(@table, {:current, valid_time, grid}) + :ok + end + + @doc "Insert locally and broadcast to peer nodes." + @spec broadcast_put(DateTime.t(), rain_grid()) :: :ok + def broadcast_put(valid_time, grid) do + put(valid_time, grid) + PubSub.broadcast(@pubsub, @topic, {:mrms_cache_refresh, valid_time, grid}) + :ok + end + + @spec clear() :: :ok + def clear do + :ets.delete_all_objects(@table) + :ok + end + + @impl true + def init(_opts) do + :ets.new(@table, [:set, :named_table, :public, read_concurrency: true]) + PubSub.subscribe(@pubsub, @topic) + {:ok, %{}} + end + + @impl true + def handle_info({:mrms_cache_refresh, valid_time, grid}, state) do + put(valid_time, grid) + {:noreply, state} + end + + def handle_info(_msg, state), do: {:noreply, state} +end diff --git a/lib/microwaveprop/weather/mrms_client.ex b/lib/microwaveprop/weather/mrms_client.ex new file mode 100644 index 00000000..c36244d2 --- /dev/null +++ b/lib/microwaveprop/weather/mrms_client.ex @@ -0,0 +1,181 @@ +defmodule Microwaveprop.Weather.MrmsClient do + @moduledoc """ + Fetch NOAA MRMS `PrecipRate` surface rain-rate mosaics and interpolate + them onto the same 0.125° propagation grid we score against. MRMS is + ~1 km resolution and updates every 2 minutes, so between hourly HRRR runs + it's the freshest rain signal we can get over CONUS. + + The module is pure transport + decode: fetch, gunzip, run wgrib2 to + regrid, return `%{{lat, lon} => rain_rate_mmhr}`. Caching and cron + scheduling live in `Microwaveprop.Weather.MrmsCache` and + `Microwaveprop.Workers.MrmsFetchWorker`. + """ + + alias Microwaveprop.Propagation.Grid + alias Microwaveprop.Weather.Grib2.Wgrib2 + + require Logger + + @base_url "https://mrms.ncep.noaa.gov/2D/PrecipRate/" + @match_pattern ":PrecipRate:" + @missing_sentinel -3.0 + + @type rain_grid :: %{{float(), float()} => float()} + @type listing_entry :: %{filename: String.t(), valid_time: DateTime.t()} + + @doc """ + List the most recent MRMS PrecipRate files published on the NCEP index. + Returns entries sorted newest-first so callers can just take the head. + """ + @spec list_latest(non_neg_integer()) :: {:ok, [listing_entry()]} | {:error, term()} + def list_latest(limit \\ 10) do + case Req.get(@base_url, req_options()) do + {:ok, %{status: 200, body: body}} -> + entries = + body + |> parse_listing() + |> Enum.sort_by(& &1.valid_time, {:desc, DateTime}) + |> Enum.take(limit) + + {:ok, entries} + + {:ok, %{status: status}} -> + {:error, {:http_status, status}} + + {:error, reason} -> + {:error, reason} + end + end + + @doc """ + Fetch a single MRMS file by filename, decompress, regrid to the 0.125° + propagation grid, and return `{valid_time, %{{lat, lon} => rain_rate_mmhr}}`. + + Missing-data sentinels (-3) are dropped from the returned grid so callers + can test membership directly. + """ + @spec fetch_and_decode(String.t()) :: {:ok, DateTime.t(), rain_grid()} | {:error, term()} + def fetch_and_decode(filename) do + with {:ok, valid_time} <- parse_filename(filename), + {:ok, grib_bytes} <- download(filename), + {:ok, tmp_path} <- write_temp(grib_bytes), + {:ok, grid} <- regrid(tmp_path) do + File.rm(tmp_path) + {:ok, valid_time, grid} + end + end + + @doc """ + Convenience: list the latest files, grab the newest that isn't already + cached, and return the decoded grid. `known_valid_time` is whatever the + caller already has — if the latest file matches it we short-circuit. + """ + @spec fetch_latest(DateTime.t() | nil) :: + {:ok, DateTime.t(), rain_grid()} | {:up_to_date, DateTime.t()} | {:error, term()} + def fetch_latest(known_valid_time \\ nil) do + case list_latest(1) do + {:ok, [latest | _]} -> + if known_valid_time && DateTime.compare(latest.valid_time, known_valid_time) != :gt do + {:up_to_date, known_valid_time} + else + fetch_and_decode(latest.filename) + end + + {:ok, []} -> + {:error, :no_files_listed} + + other -> + other + end + end + + # -- internals (public-but-undocumented for tests) -------------------- + + @doc false + def parse_listing(html) do + ~r/MRMS_PrecipRate_00\.00_(\d{8}-\d{6})\.grib2\.gz/ + |> Regex.scan(html) + |> Enum.uniq() + |> Enum.flat_map(fn [full, stamp] -> + case parse_stamp(stamp) do + {:ok, dt} -> [%{filename: full, valid_time: dt}] + _ -> [] + end + end) + end + + defp parse_filename(filename) do + case Regex.run(~r/MRMS_PrecipRate_00\.00_(\d{8}-\d{6})\.grib2\.gz/, filename) do + [_, stamp] -> parse_stamp(stamp) + _ -> {:error, :bad_filename} + end + end + + defp parse_stamp(stamp) do + with <> <- + stamp, + {:ok, naive} <- + NaiveDateTime.new(si(y), si(m), si(d), si(h), si(mi), si(s)) do + {:ok, DateTime.from_naive!(naive, "Etc/UTC")} + else + _ -> {:error, :bad_stamp} + end + end + + defp si(s), do: String.to_integer(s) + + defp download(filename) do + # Req auto-decompresses gzip responses, so the body we receive is + # already the raw GRIB2 binary (starts with "GRIB" magic) even though + # the URL ends in .gz. No gunzip step needed. + url = @base_url <> filename + + case Req.get(url, req_options()) do + {:ok, %{status: 200, body: body}} when is_binary(body) -> + {:ok, body} + + {:ok, %{status: status}} -> + {:error, {:http_status, status}} + + {:error, reason} -> + {:error, reason} + end + end + + defp write_temp(grib_bytes) do + path = Path.join(System.tmp_dir!(), "mrms_preciprate_#{System.unique_integer([:positive])}.grib2") + File.write!(path, grib_bytes) + {:ok, path} + end + + defp regrid(tmp_path) do + case Wgrib2.extract_grid_from_file(tmp_path, @match_pattern, Grid.wgrib2_grid_spec()) do + {:ok, cells} -> {:ok, flatten_rain_grid(cells)} + error -> error + end + end + + @doc false + def flatten_rain_grid(cells) do + Enum.reduce(cells, %{}, fn {{lat, lon}, fields}, acc -> + case extract_preciprate(fields) do + nil -> acc + rate -> Map.put(acc, {lat, lon}, rate) + end + end) + end + + defp extract_preciprate(fields) do + Enum.find_value(fields, fn + {"PrecipRate" <> _, value} when is_number(value) and value > @missing_sentinel and value >= 0 -> + value + + _ -> + nil + end) + end + + defp req_options do + [receive_timeout: 120_000, retry: :safe_transient, max_retries: 3] + end +end diff --git a/lib/microwaveprop/workers/asos_adjustment_worker.ex b/lib/microwaveprop/workers/asos_adjustment_worker.ex index 61ffb47a..885aca94 100644 --- a/lib/microwaveprop/workers/asos_adjustment_worker.ex +++ b/lib/microwaveprop/workers/asos_adjustment_worker.ex @@ -34,6 +34,7 @@ defmodule Microwaveprop.Workers.AsosAdjustmentWorker do alias Microwaveprop.Repo alias Microwaveprop.Weather.HrrrProfile alias Microwaveprop.Weather.IemClient + alias Microwaveprop.Weather.MrmsCache require Logger @@ -52,36 +53,57 @@ defmodule Microwaveprop.Workers.AsosAdjustmentWorker do @impl Oban.Worker def perform(%Oban.Job{}) do with {:ok, valid_time} <- latest_hrrr_valid_time(), - {:ok, observations} <- IemClient.fetch_current_asos(@state_networks), - [_ | _] = fresh <- filter_fresh(observations), [_ | _] = profiles <- load_hrrr_profiles(valid_time) do - apply_nudge(fresh, profiles, valid_time) + 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: no fresh ASOS data, skipping") - :ok - - {:error, reason} -> - Logger.warning("AsosAdjustment: aborted — #{inspect(reason)}") + Logger.info("AsosAdjustment: HRRR valid_time has no grid profiles, skipping") :ok end end - defp apply_nudge(observations, profiles, valid_time) do - scores = AsosNudge.compute(observations, valid_time, profiles) + 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 but no grid cells within range") + 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 at #{valid_time}") + 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 diff --git a/lib/microwaveprop/workers/mrms_fetch_worker.ex b/lib/microwaveprop/workers/mrms_fetch_worker.ex new file mode 100644 index 00000000..5212f1a0 --- /dev/null +++ b/lib/microwaveprop/workers/mrms_fetch_worker.ex @@ -0,0 +1,41 @@ +defmodule Microwaveprop.Workers.MrmsFetchWorker do + @moduledoc """ + Every ~2 minutes, pull the newest NOAA MRMS PrecipRate grid and cache + the regridded version (0.125° CONUS) in `Microwaveprop.Weather.MrmsCache`. + `Microwaveprop.Workers.AsosAdjustmentWorker` reads that cache on its + own tick to overlay real radar rain rates onto the score grid. + + Runs on the `propagation` queue so it shares the scoring node's crontab + and isn't competing with the heavier HRRR pipeline for slots. + """ + use Oban.Worker, queue: :propagation, max_attempts: 2 + + alias Microwaveprop.Weather.MrmsCache + alias Microwaveprop.Weather.MrmsClient + + require Logger + + @impl Oban.Worker + def perform(%Oban.Job{}) do + case MrmsClient.fetch_latest(MrmsCache.valid_time()) do + {:ok, valid_time, grid} -> + MrmsCache.broadcast_put(valid_time, grid) + + Logger.info("MrmsFetch: cached #{map_size(grid)} cells at #{valid_time} (#{count_rainy(grid)} with rain)") + + :ok + + {:up_to_date, valid_time} -> + Logger.debug("MrmsFetch: cache already has #{valid_time}") + :ok + + {:error, reason} -> + Logger.warning("MrmsFetch: skipped — #{inspect(reason)}") + :ok + end + end + + defp count_rainy(grid) do + Enum.count(grid, fn {_, v} -> v > 0 end) + end +end diff --git a/lib/microwaveprop_web/live/map_live.ex b/lib/microwaveprop_web/live/map_live.ex index 8a6ffd13..0d7e8acf 100644 --- a/lib/microwaveprop_web/live/map_live.ex +++ b/lib/microwaveprop_web/live/map_live.ex @@ -20,6 +20,8 @@ defmodule MicrowavepropWeb.MapLive do Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "propagation:updated") end + socket = assign(socket, :initial_utc_clock, Calendar.strftime(DateTime.utc_now(), "%H:%M UTC")) + case LiveStash.recover_state(socket) do {:recovered, socket} -> {:ok, socket} @@ -398,8 +400,10 @@ defmodule MicrowavepropWeb.MapLive do
+ {@initial_utc_clock}