diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index 1429dcbb..e23556a1 100644 --- a/lib/microwaveprop/propagation.ex +++ b/lib/microwaveprop/propagation.ex @@ -288,8 +288,9 @@ defmodule Microwaveprop.Propagation do end @doc """ - Retains only score/profile/scalar files within the active 48-hour - forecast window, deleting everything older than `run_time`. Called + Retains score files through GEFS's 168-hour horizon and profile/scalar + files through HRRR's 48-hour horizon, deleting files older than + `run_time`. Called by `NotifyListener` after chain completion to keep `/data/scores` within bounds. @@ -297,11 +298,9 @@ defmodule Microwaveprop.Propagation do """ @spec retain_scores_window(DateTime.t()) :: :ok def retain_scores_window(%DateTime{} = run_time) do - max_fh = 48 - - scores_deleted = ScoresFile.retain_window(run_time, max_fh) - profiles_deleted = ProfilesFile.retain_window(run_time, max_fh) - scalars_deleted = ScalarFile.retain_window(run_time, max_fh) + scores_deleted = ScoresFile.retain_window(run_time, 168) + profiles_deleted = ProfilesFile.retain_window(run_time, 48) + scalars_deleted = ScalarFile.retain_window(run_time, 48) total = scores_deleted + profiles_deleted + scalars_deleted if total > 0 do @@ -442,46 +441,34 @@ defmodule Microwaveprop.Propagation do @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 - hrrr = - case ScoresFile.read(band_mhz, valid_time) do - {:ok, payload} -> ScoresFile.extract_points(payload, nil) - _ -> nil - end - - hrdps = - case ScoresFile.read_hrdps(band_mhz, valid_time) do - {:ok, payload} -> ScoresFile.extract_points(payload, nil) - _ -> nil - end + hrrr = read_score_points(&ScoresFile.read/2, band_mhz, valid_time) + hrdps = read_score_points(&ScoresFile.read_hrdps/2, band_mhz, valid_time) # Merge priority: HRRR > HRDPS > GEFS. GEFS provides extended-horizon # coverage beyond HRRR's 48h window. Within the f24-f48 overlap, cells # already present in HRRR/HRDPS are skipped so the coarser GEFS scores # don't override the higher-resolution ones. merged = (hrrr || []) ++ (hrdps || []) - merged_keys = MapSet.new(merged, fn %{lat: lat, lon: lon} -> {lat, lon} end) + gefs = read_score_points(&ScoresFile.read_gefs/2, band_mhz, valid_time) || [] + scores = ScoresFile.merge_preferred(merged, gefs) - gefs = - case ScoresFile.read_gefs(band_mhz, valid_time) do - {:ok, payload} -> - payload - |> ScoresFile.extract_points(nil) - |> Enum.reject(fn %{lat: lat, lon: lon} -> MapSet.member?(merged_keys, {lat, lon}) end) - - _ -> - [] - end - - case {hrrr, hrdps, gefs} do + case {hrrr, hrdps, scores} do {nil, nil, []} -> {:error, :enoent} - {h, c, g} -> - ScoreCache.broadcast_put(band_mhz, valid_time, (h || []) ++ (c || []) ++ g) + _ -> + ScoreCache.broadcast_put(band_mhz, valid_time, scores) :ok end end + defp read_score_points(reader, band_mhz, valid_time) do + case reader.(band_mhz, valid_time) do + {:ok, payload} -> ScoresFile.extract_points(payload, nil) + _ -> nil + end + end + defp filter_bounds(scores, nil), do: scores defp filter_bounds(scores, %{"south" => s, "north" => n, "west" => w, "east" => e}) do diff --git a/lib/microwaveprop/propagation/notify_listener.ex b/lib/microwaveprop/propagation/notify_listener.ex index 94d074c8..0f31072b 100644 --- a/lib/microwaveprop/propagation/notify_listener.ex +++ b/lib/microwaveprop/propagation/notify_listener.ex @@ -120,7 +120,7 @@ defmodule Microwaveprop.Propagation.NotifyListener do {past, future} = Propagation.hot_cache_window() ScoreCache.prune_outside_window(past, future) - # Sweep NFS scores to the active 48h window anchored on run_time so + # Sweep NFS files to their active forecast windows anchored on run_time so # stale forecast hours from the previous cycle don't accumulate. # Calling with valid_time would slide the window forward for each # completed forecast-hour task and delete the current run's earlier diff --git a/lib/microwaveprop/propagation/path_compute.ex b/lib/microwaveprop/propagation/path_compute.ex index 58e5e36f..05f7abf4 100644 --- a/lib/microwaveprop/propagation/path_compute.ex +++ b/lib/microwaveprop/propagation/path_compute.ex @@ -378,25 +378,33 @@ defmodule Microwaveprop.Propagation.PathCompute do end defp collect_profile_fields(profiles) do - Enum.reduce(profiles, {[], [], [], [], [], [], [], [], [], []}, fn p, {ts, ds, ps, gs, bs, ws, wus, wvs, scs, pms} -> - { - if(p.surface_temp_c == nil, do: ts, else: [p.surface_temp_c | ts]), - if(p.surface_dewpoint_c == nil, do: ds, else: [p.surface_dewpoint_c | ds]), - if(p.surface_pressure_mb == nil, do: ps, else: [p.surface_pressure_mb | ps]), - if(p.min_refractivity_gradient == nil, - do: gs, - else: [p.min_refractivity_gradient | gs] - ), - if(p.hpbl_m == nil, do: bs, else: [p.hpbl_m | bs]), - if(p.pwat_mm == nil, do: ws, else: [p.pwat_mm | ws]), - if(Map.get(p, :wind_u) == nil, do: wus, else: [Map.get(p, :wind_u) | wus]), - if(Map.get(p, :wind_v) == nil, do: wvs, else: [Map.get(p, :wind_v) | wvs]), - if(Map.get(p, :cloud_cover_pct) == nil, do: scs, else: [Map.get(p, :cloud_cover_pct) | scs]), - if(Map.get(p, :precip_mm) == nil, do: pms, else: [Map.get(p, :precip_mm) | pms]) - } + fields = [ + :surface_temp_c, + :surface_dewpoint_c, + :surface_pressure_mb, + :min_refractivity_gradient, + :hpbl_m, + :pwat_mm, + :wind_u, + :wind_v, + :cloud_cover_pct, + :precip_mm + ] + + profiles + |> Enum.reduce(Map.new(fields, &{&1, []}), fn profile, acc -> + Enum.reduce(fields, acc, fn field, field_acc -> + Map.update!(field_acc, field, &prepend_if_present(&1, Map.get(profile, field))) + end) + end) + |> then(fn values -> + List.to_tuple(Enum.map(fields, &Map.fetch!(values, &1))) end) end + defp prepend_if_present(values, nil), do: values + defp prepend_if_present(values, value), do: [value | values] + defp build_conditions( avg_temp_c, avg_dewpoint_c, @@ -406,28 +414,9 @@ defmodule Microwaveprop.Propagation.PathCompute do {pressures, gradients, bl_depths, pwats, wind_us, wind_vs, sky_covers, precip_mms}, native_duct ) do - wind_speed = - if wind_us != [] and wind_vs != [] do - wind_us - |> Enum.zip(wind_vs) - |> Enum.map(fn {u, v} -> Scorer.wind_speed_kts(u, v) end) - |> Enum.filter(& &1) - |> case do - [] -> nil - speeds -> Enum.max(speeds) - end - end - - sky_cover = if sky_covers == [], do: nil, else: Enum.max(sky_covers) - - rain_rate = - if precip_mms == [] do - 0.0 - else - precip_mms - |> Enum.map(&Scorer.precip_to_rate_mmhr/1) - |> Enum.max() - end + wind_speed = max_wind_speed(wind_us, wind_vs) + sky_cover = max_or_default(sky_covers, nil) + rain_rate = precip_mms |> Enum.map(&Scorer.precip_to_rate_mmhr/1) |> max_or_default(0.0) %{ abs_humidity: Scorer.absolute_humidity(avg_temp_c, avg_dewpoint_c), @@ -453,6 +442,17 @@ defmodule Microwaveprop.Propagation.PathCompute do } end + defp max_wind_speed(wind_us, wind_vs) do + wind_us + |> Enum.zip(wind_vs) + |> Enum.map(fn {u, v} -> Scorer.wind_speed_kts(u, v) end) + |> Enum.reject(&is_nil/1) + |> max_or_default(nil) + end + + defp max_or_default([], default), do: default + defp max_or_default(values, _default), do: Enum.max(values) + @doc false @spec compute_loss_budget(float(), float(), map(), map() | nil, map() | nil) :: map() def compute_loss_budget(dist_km, freq_ghz, band_config, terrain_result, conditions) do diff --git a/lib/microwaveprop/propagation/scores_file.ex b/lib/microwaveprop/propagation/scores_file.ex index 5a7d583d..ee885628 100644 --- a/lib/microwaveprop/propagation/scores_file.ex +++ b/lib/microwaveprop/propagation/scores_file.ex @@ -304,18 +304,31 @@ defmodule Microwaveprop.Propagation.ScoresFile do # GEFS fills cells that neither HRRR nor HRDPS cover. Within the # f24-f48 overlap window HRRR wins; beyond 48h GEFS is the only # source for extended-horizon outlooks. - merged_keys = MapSet.new(merged, fn %{lat: lat, lon: lon} -> {lat, lon} end) - gefs_cells = case read_gefs(band_mhz, valid_time) do {:ok, payload} -> extract_points(payload, bounds) _ -> [] end - extras = - Enum.reject(gefs_cells, fn %{lat: lat, lon: lon} -> MapSet.member?(merged_keys, {lat, lon}) end) + merge_preferred(merged, gefs_cells) + end - merged ++ extras + @doc """ + Merges score cells with `preferred` taking precedence over `fallback`. + + Coordinates are compared by canonical propagation-grid index rather than + raw floating-point equality, so equivalent cells from different decoders + cannot leak through because of insignificant coordinate drift. + """ + @spec merge_preferred([map()], [map()]) :: [map()] + def merge_preferred(preferred, fallback) do + preferred_keys = MapSet.new(preferred, &canonical_cell/1) + preferred ++ Enum.reject(fallback, &MapSet.member?(preferred_keys, canonical_cell(&1))) + end + + defp canonical_cell(%{lat: lat, lon: lon}) do + bounds = Grid.bounds() + {round((lat - bounds.lat_min) / Grid.step()), round((lon - bounds.lon_min) / Grid.step())} end @doc """ diff --git a/lib/microwaveprop/workers/gefs_fetch_worker.ex b/lib/microwaveprop/workers/gefs_fetch_worker.ex index fd433652..2a5a579d 100644 --- a/lib/microwaveprop/workers/gefs_fetch_worker.ex +++ b/lib/microwaveprop/workers/gefs_fetch_worker.ex @@ -39,7 +39,6 @@ defmodule Microwaveprop.Workers.GefsFetchWorker do {:ok, profiles} -> 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} -> Logger.warning("GefsFetch: wgrib2 missing on host — cancelling job") @@ -188,27 +187,30 @@ defmodule Microwaveprop.Workers.GefsFetchWorker do # Write via ScoresFile.write_gefs! so extended-horizon scores land in # `.gefs.prop` files — separate from HRRR's `.prop` files. Within the # f24-f48 overlap window HRRR wins; beyond 48h GEFS is the only source. - total = - scores - |> Enum.group_by(& &1.band_mhz) - |> Enum.reduce(0, fn {band_mhz, band_scores}, acc -> - try do - ScoresFile.write_gefs!(band_mhz, valid_time, band_scores) - acc + length(band_scores) - rescue - e -> - Logger.error("GefsFetch: ScoresFile.write_gefs! failed band=#{band_mhz} fh=#{fh}: #{inspect(e)}") + with {:ok, total} <- persist_scores(scores, valid_time) do + Logger.info("GefsFetch: wrote #{total} GEFS scores for fh=#{fh} @ #{valid_time}") + broadcast_updated(valid_time) - acc - end - 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}") + :ok + end + end - Logger.info("GefsFetch: wrote #{total} GEFS scores for fh=#{fh} @ #{valid_time}") - broadcast_updated(valid_time) - - 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}") + @doc false + @spec persist_scores([map()], DateTime.t()) :: {:ok, non_neg_integer()} | {:error, term()} + def persist_scores(scores, valid_time) do + scores + |> Enum.group_by(& &1.band_mhz) + |> Enum.reduce_while({:ok, 0}, fn {band_mhz, band_scores}, {:ok, total} -> + try do + ScoresFile.write_gefs!(band_mhz, valid_time, band_scores) + {:cont, {:ok, total + length(band_scores)}} + rescue + error -> {:halt, {:error, {:score_write_failed, band_mhz, error}}} + end + end) end defp profiles_to_grid(profiles) do diff --git a/test/microwaveprop/propagation/notify_listener_test.exs b/test/microwaveprop/propagation/notify_listener_test.exs index 308901ef..91991266 100644 --- a/test/microwaveprop/propagation/notify_listener_test.exs +++ b/test/microwaveprop/propagation/notify_listener_test.exs @@ -88,6 +88,16 @@ defmodule Microwaveprop.Propagation.NotifyListenerTest do assert ScoreCache.fetch(10_000, stale) == :miss end + test "retains GEFS score files through 168 hours" do + run_time = ~U[2026-04-21 00:00:00Z] + extended_time = DateTime.add(run_time, 168, :hour) + ScoresFile.write_gefs!(10_000, extended_time, sample_scores()) + + {:ok, _task_pid} = NotifyListener.handle_propagation_ready(run_time, run_time) + + assert File.exists?(ScoresFile.path_for_gefs(10_000, extended_time)) + end + test "kicks off async ScalarFile materialization for the new valid_time" do vt = ~U[2026-04-21 16:00:00Z] ScoresFile.write!(10_000, vt, sample_scores()) diff --git a/test/microwaveprop/propagation/scores_file_test.exs b/test/microwaveprop/propagation/scores_file_test.exs index c86cd26b..f2b42596 100644 --- a/test/microwaveprop/propagation/scores_file_test.exs +++ b/test/microwaveprop/propagation/scores_file_test.exs @@ -224,6 +224,20 @@ defmodule Microwaveprop.Propagation.ScoresFileTest do end describe "read_bounds/3" do + test "canonical grid cells from GEFS do not duplicate preferred cells" do + preferred = [%{lat: 32.875, lon: -97.0, score: 80}] + + fallback = [ + %{lat: 32.87500001, lon: -97.00000001, score: 20}, + %{lat: 33.0, lon: -97.0, score: 30} + ] + + assert ScoresFile.merge_preferred(preferred, fallback) == [ + %{lat: 32.875, lon: -97.0, score: 80}, + %{lat: 33.0, lon: -97.0, score: 30} + ] + end + 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}, @@ -335,6 +349,19 @@ defmodule Microwaveprop.Propagation.ScoresFileTest do end describe "retain_window/2" do + test "recognizes GEFS files through the extended horizon" do + run_time = ~U[2026-04-14 00:00:00Z] + inside = DateTime.add(run_time, 168, :hour) + outside = DateTime.add(run_time, 174, :hour) + + ScoresFile.write_gefs!(10_000, inside, []) + ScoresFile.write_gefs!(10_000, outside, []) + + assert ScoresFile.retain_window(run_time, 168) == 1 + assert File.exists?(ScoresFile.path_for_gefs(10_000, inside)) + refute File.exists?(ScoresFile.path_for_gefs(10_000, outside)) + end + test "deletes files outside [run_time, run_time + hours] across every band" do run_time = ~U[2026-04-14 16:00:00Z] diff --git a/test/microwaveprop/workers/gefs_fetch_worker_test.exs b/test/microwaveprop/workers/gefs_fetch_worker_test.exs index 41df119a..ec8abebd 100644 --- a/test/microwaveprop/workers/gefs_fetch_worker_test.exs +++ b/test/microwaveprop/workers/gefs_fetch_worker_test.exs @@ -3,6 +3,7 @@ defmodule Microwaveprop.Workers.GefsFetchWorkerTest do use ExUnitProperties alias Microwaveprop.Propagation + alias Microwaveprop.Propagation.ScoresFile alias Microwaveprop.Weather alias Microwaveprop.Weather.GefsClient alias Microwaveprop.Weather.GefsProfile @@ -84,6 +85,32 @@ defmodule Microwaveprop.Workers.GefsFetchWorkerTest do end end + describe "persist_scores/2" do + test "returns an error when any score file cannot be written" do + blocker = Path.join(System.tmp_dir!(), "gefs_write_blocker_#{System.unique_integer([:positive])}") + File.write!(blocker, "not a directory") + previous = Application.get_env(:microwaveprop, :propagation_scores_dir) + Application.put_env(:microwaveprop, :propagation_scores_dir, Path.join(blocker, "scores")) + + on_exit(fn -> + File.rm!(blocker) + + if previous do + Application.put_env(:microwaveprop, :propagation_scores_dir, previous) + else + Application.delete_env(:microwaveprop, :propagation_scores_dir) + end + end) + + scores = [%{band_mhz: 10_000, lat: 32.875, lon: -97.0, score: 50}] + + assert {:error, {:score_write_failed, 10_000, _}} = + GefsFetchWorker.persist_scores(scores, ~U[2026-04-19 12:00:00Z]) + + refute File.exists?(ScoresFile.path_for_gefs(10_000, ~U[2026-04-19 12:00:00Z])) + 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