diff --git a/config/runtime.exs b/config/runtime.exs index bd1b0599..8353ead8 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -210,6 +210,11 @@ if config_env() == :prod do # are keyed by (band, valid_time) so last-writer-wins gives # newer analysis data naturally. {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker}, + # 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 + # (25 jobs per run, 100/day) into the :gefs queue. + {"30 5,11,17,23 * * *", Microwaveprop.Workers.GefsFetchWorker}, # MRMS refreshes the PrecipRate cache that AsosAdjustmentWorker # overlays onto the score grid. Both must run together — an # ASOS nudge without a fresh MRMS frame reverts to HRRR-only diff --git a/lib/microwaveprop/weather/gefs_client.ex b/lib/microwaveprop/weather/gefs_client.ex index b2546e67..b3dab9a6 100644 --- a/lib/microwaveprop/weather/gefs_client.ex +++ b/lib/microwaveprop/weather/gefs_client.ex @@ -148,8 +148,8 @@ defmodule Microwaveprop.Weather.GefsClient do surface_dewpoint_c: sfc_dewpoint_c, surface_pressure_mb: sfc_pressure_mb, pwat_mm: parsed["PWAT:entire atmosphere (considered as a single layer)"], - wind_u_mps: parsed["UGRD:10 m above ground"], - wind_v_mps: parsed["VGRD:10 m above ground"], + wind_u: parsed["UGRD:10 m above ground"], + wind_v: parsed["VGRD:10 m above ground"], cloud_cover_pct: parsed["TCDC:entire atmosphere"], precip_mm: parsed["APCP:surface"], profile: profile diff --git a/lib/microwaveprop/workers/gefs_fetch_worker.ex b/lib/microwaveprop/workers/gefs_fetch_worker.ex index 74a4b2eb..782ed223 100644 --- a/lib/microwaveprop/workers/gefs_fetch_worker.ex +++ b/lib/microwaveprop/workers/gefs_fetch_worker.ex @@ -16,6 +16,7 @@ defmodule Microwaveprop.Workers.GefsFetchWorker do max_attempts: 5, unique: [period: 3600, states: [:available, :scheduled, :executing, :retryable]] + alias Microwaveprop.Propagation alias Microwaveprop.Weather alias Microwaveprop.Weather.GefsClient alias Microwaveprop.Weather.SoundingParams @@ -35,9 +36,8 @@ defmodule Microwaveprop.Workers.GefsFetchWorker do case GefsClient.fetch_grid_profiles(run_date, run_hour, fh) do {:ok, profiles} -> - attrs = Enum.map(profiles, &build_profile_attrs(run_time, fh, &1)) - {count, _} = Weather.upsert_gefs_profiles_batch(attrs) - Logger.info("GefsFetch: stored #{count}/#{length(attrs)} profiles for fh=#{fh}") + valid_time = run_time |> DateTime.add(fh * 3600, :second) |> DateTime.truncate(:second) + score_and_persist(profiles, run_time, fh, valid_time) :ok {:error, :wgrib2_not_available} -> @@ -52,8 +52,54 @@ defmodule Microwaveprop.Workers.GefsFetchWorker do end end + def perform(%Oban.Job{args: args}) when args == %{} do + seed_extended_horizon() + end + def perform(%Oban.Job{args: _args}), do: {:cancel, :invalid_args} + @doc """ + Forecast hours enqueued per GEFS run: f024 through f168 at the + 6-hour cadence GEFS publishes at. Covers Day 2-7 — the window HRRR + can't reach. Starts at f024 so the near-term (f00-f18) stays + HRRR-owned and the handoff lines up on the f018 boundary. + """ + @spec extended_horizon_hours() :: [non_neg_integer()] + def extended_horizon_hours, do: Enum.to_list(24..168//6) + + @doc """ + Enqueues one fetch job per extended-horizon forecast hour for the + most recent GEFS run that should already be published on NOMADS + (run publication lags ~3-4 hours behind the run time; the seeder + targets runs 5+ hours in the past for safety). + """ + @spec seed_extended_horizon() :: :ok + def seed_extended_horizon do + run_time = most_recent_available_run(DateTime.utc_now()) + + Logger.info("GefsFetch: seeding extended-horizon chain for run_time=#{run_time}") + + Enum.each(extended_horizon_hours(), fn fh -> + %{"run_time" => DateTime.to_iso8601(run_time), "forecast_hour" => fh} + |> new() + |> Oban.insert() + end) + + :ok + end + + @doc """ + Picks the most-recent 00/06/12/18Z GEFS run whose file set should be + fully published given an ~5-hour publication lag. Exposed so tests + can pin a deterministic "now". + """ + @spec most_recent_available_run(DateTime.t()) :: DateTime.t() + def most_recent_available_run(%DateTime{} = now) do + now + |> DateTime.add(-5 * 3600, :second) + |> GefsClient.nearest_run() + end + @doc """ Converts a `GefsClient.build_profile/1` result (already tagged with `lat` and `lon`) into a full `gefs_profiles` row attr map. Exposed @@ -75,8 +121,8 @@ defmodule Microwaveprop.Workers.GefsFetchWorker do surface_dewpoint_c: profile.surface_dewpoint_c, surface_pressure_mb: profile.surface_pressure_mb, pwat_mm: profile[:pwat_mm], - wind_u_mps: profile[:wind_u_mps], - wind_v_mps: profile[:wind_v_mps], + wind_u_mps: profile[:wind_u] || profile[:wind_u_mps], + wind_v_mps: profile[:wind_v] || profile[:wind_v_mps], cloud_cover_pct: profile[:cloud_cover_pct], precip_mm: profile[:precip_mm] } @@ -95,6 +141,68 @@ defmodule Microwaveprop.Workers.GefsFetchWorker do end end + # Runs the propagation scorer across the fetched GEFS grid and + # persists scores to the `.ntms` file for this `valid_time` so the + # extended-horizon outlook picks them up via the usual + # `Propagation.scores_at/3` path. Also upserts raw profiles to + # `gefs_profiles` for future per-contact extended-horizon lookups. + defp score_and_persist(profiles, run_time, fh, valid_time) do + grid_data = profiles_to_grid(profiles) + + scores = + grid_data + |> Task.async_stream( + fn {{lat, lon}, profile} -> + profile + |> Propagation.score_grid_point(valid_time, lat, lon) + |> Enum.map(fn r -> + %{ + lat: lat, + lon: lon, + valid_time: valid_time, + band_mhz: r.band_mhz, + score: r.score, + factors: nil + } + end) + end, + max_concurrency: System.schedulers_online() * 2, + timeout: 30_000 + ) + |> Stream.flat_map(fn + {:ok, results} -> results + {:exit, _reason} -> [] + end) + |> Enum.to_list() + + case Propagation.replace_scores(scores, valid_time) do + {:ok, count} -> + Logger.info("GefsFetch: wrote #{count} scores for fh=#{fh} @ #{valid_time}") + broadcast_updated(valid_time) + + error -> + Logger.error("GefsFetch: replace_scores failed fh=#{fh}: #{inspect(error)}") + end + + attrs = Enum.map(profiles, &build_profile_attrs(run_time, fh, &1)) + {count, _} = Weather.upsert_gefs_profiles_batch(attrs) + Logger.info("GefsFetch: upserted #{count}/#{length(attrs)} profile rows for fh=#{fh}") + end + + defp profiles_to_grid(profiles) do + Map.new(profiles, fn profile -> + {{profile.lat, profile.lon}, profile} + end) + end + + defp broadcast_updated(valid_time) do + Phoenix.PubSub.broadcast( + Microwaveprop.PubSub, + "propagation:updated", + {:propagation_updated, [valid_time]} + ) + end + defp validate_run_hour(hour) when hour in @valid_run_hours, do: :ok defp validate_run_hour(_), do: {:error, :invalid_run_hour} diff --git a/test/microwaveprop/weather/gefs_client_test.exs b/test/microwaveprop/weather/gefs_client_test.exs index 571657eb..84044955 100644 --- a/test/microwaveprop/weather/gefs_client_test.exs +++ b/test/microwaveprop/weather/gefs_client_test.exs @@ -131,8 +131,8 @@ defmodule Microwaveprop.Weather.GefsClientTest do test "carries PWAT, winds, cloud cover, precip through untouched" do p = GefsClient.build_profile(@raw) assert p.pwat_mm == 25.4 - assert p.wind_u_mps == 3.0 - assert p.wind_v_mps == -1.5 + assert p.wind_u == 3.0 + assert p.wind_v == -1.5 assert p.cloud_cover_pct == 62.0 assert p.precip_mm == 0.0 end diff --git a/test/microwaveprop/workers/gefs_fetch_worker_test.exs b/test/microwaveprop/workers/gefs_fetch_worker_test.exs index 62c568b3..b09718d3 100644 --- a/test/microwaveprop/workers/gefs_fetch_worker_test.exs +++ b/test/microwaveprop/workers/gefs_fetch_worker_test.exs @@ -6,9 +6,9 @@ defmodule Microwaveprop.Workers.GefsFetchWorkerTest do alias Microwaveprop.Workers.GefsFetchWorker describe "perform/1" do - test "rejects args missing run_time or forecast_hour" do - job = %Oban.Job{args: %{}} - assert {:cancel, _} = GefsFetchWorker.perform(job) + test "cancels on malformed (non-empty, not a seed) args" do + args = %{"some_unrelated" => "garbage"} + assert {:cancel, _} = GefsFetchWorker.perform(%Oban.Job{args: args}) end test "cancels when run_hour is not a valid GEFS cycle" do @@ -71,6 +71,68 @@ defmodule Microwaveprop.Workers.GefsFetchWorkerTest do end end + describe "extended_horizon_hours/0" do + test "spans f024 through f168 at 6-hour cadence (25 entries)" do + hours = GefsFetchWorker.extended_horizon_hours() + assert List.first(hours) == 24 + assert List.last(hours) == 168 + assert length(hours) == 25 + assert Enum.all?(hours, &(rem(&1, 6) == 0)) + end + end + + describe "most_recent_available_run/1" do + test "returns the run cycle 5 or more hours earlier" do + # 15:00 UTC → 10:00 UTC → snaps to 06Z + assert GefsFetchWorker.most_recent_available_run(~U[2026-04-18 15:00:00Z]) == + ~U[2026-04-18 06:00:00Z] + end + + test "rolls across midnight when now is near 04Z" do + # 04:00 UTC → 23:00 the day before → snaps to 18Z previous day + assert GefsFetchWorker.most_recent_available_run(~U[2026-04-18 04:00:00Z]) == + ~U[2026-04-17 18:00:00Z] + end + end + + describe "scoring interop" do + alias Microwaveprop.Propagation + alias Microwaveprop.Weather.GefsClient + + test "a GEFS-shaped profile produces non-empty band scores" do + raw = %{ + "TMP:2 m above ground" => 293.15, + "RH:2 m above ground" => 50.0, + "PRES:surface" => 101_325.0, + "PWAT:entire atmosphere (considered as a single layer)" => 20.0, + "UGRD:10 m above ground" => 3.0, + "VGRD:10 m above ground" => -1.5, + "TCDC:entire atmosphere" => 40.0, + "APCP:surface" => 0.0, + "TMP:1000 mb" => 293.15, + "RH:1000 mb" => 50.0, + "HGT:1000 mb" => 110.0, + "TMP:925 mb" => 288.15, + "RH:925 mb" => 40.0, + "HGT:925 mb" => 780.0, + "TMP:850 mb" => 283.15, + "RH:850 mb" => 35.0, + "HGT:850 mb" => 1500.0 + } + + scores = raw |> GefsClient.build_profile() |> Propagation.score_grid_point(~U[2026-04-20 18:00:00Z], 32.9, -97.0) + + assert is_list(scores) + assert scores != [] + + score = hd(scores) + assert is_integer(score.band_mhz) + assert is_integer(score.score) + assert score.score >= 0 + assert score.score <= 100 + end + end + describe "storing profiles" do test "round-trips a full batch through the context" do run_time = ~U[2026-04-18 12:00:00Z]