From ffcd3d3f60fe9202364089bf48463b554c9e87fa Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 14 Apr 2026 16:14:43 -0500 Subject: [PATCH] Restore point_detail factor breakdown via persisted f00 profiles MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The propagation_scores → binary files cutover dropped the factors JSONB column, which left point_detail returning factors: %{} and broke the analysis breakdown popup on map clicks. Persist the enriched f00 grid_data to /data/scores/profiles/{iso}.etf.gz and rescore on demand at click time so the factor block renders again. --- lib/microwaveprop/propagation.ex | 35 +++- .../propagation/profiles_file.ex | 179 ++++++++++++++++++ .../workers/propagation_grid_worker.ex | 30 ++- .../propagation/profiles_file_test.exs | 102 ++++++++++ test/microwaveprop/propagation_test.exs | 39 +++- 5 files changed, 375 insertions(+), 10 deletions(-) create mode 100644 lib/microwaveprop/propagation/profiles_file.ex create mode 100644 test/microwaveprop/propagation/profiles_file_test.exs diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index 9af42072..d8b1eb2a 100644 --- a/lib/microwaveprop/propagation.ex +++ b/lib/microwaveprop/propagation.ex @@ -3,6 +3,7 @@ defmodule Microwaveprop.Propagation do alias Microwaveprop.Propagation.BandConfig alias Microwaveprop.Propagation.Grid + alias Microwaveprop.Propagation.ProfilesFile alias Microwaveprop.Propagation.ScoreCache alias Microwaveprop.Propagation.Scorer alias Microwaveprop.Propagation.ScoresFile @@ -187,9 +188,13 @@ defmodule Microwaveprop.Propagation do def prune_old_scores do cutoff = DateTime.add(DateTime.utc_now(), -2, :hour) file_deleted = ScoresFile.prune_older_than(cutoff) + profiles_deleted = ProfilesFile.prune_older_than(cutoff) - if file_deleted > 0 do - Logger.info("PropagationScores: pruned #{file_deleted} old score files (before #{cutoff})") + if file_deleted + profiles_deleted > 0 do + Logger.info( + "PropagationScores: pruned #{file_deleted} old score files + " <> + "#{profiles_deleted} profile files (before #{cutoff})" + ) end :ok @@ -374,16 +379,34 @@ defmodule Microwaveprop.Propagation do lat: snapped_lat, lon: snapped_lon, score: score, - # factors were retired with the propagation_scores table - # — the JS popup iterates an empty map and just doesn't - # render the factor breakdown. - factors: %{}, + factors: factors_for(band_mhz, time, snapped_lat, snapped_lon), valid_time: time } end end end + # Rebuild the factor breakdown for a clicked grid cell by rescoring + # the persisted HRRR profile. Forecast hours don't persist profiles + # (see PropagationGridWorker.process_forecast_hour) so they just get + # an empty map, which the JS popup tolerates by omitting the + # breakdown block. + defp factors_for(band_mhz, valid_time, lat, lon) do + case ProfilesFile.read_point(valid_time, lat, lon) do + nil -> + %{} + + profile -> + profile + |> score_grid_point(valid_time, lat, lon) + |> Enum.find(fn r -> r.band_mhz == band_mhz end) + |> case do + %{factors: factors} when is_map(factors) -> factors + _ -> %{} + end + end + end + @doc "Get the latest valid_time across all bands." @spec latest_valid_time() :: DateTime.t() | nil def latest_valid_time do diff --git a/lib/microwaveprop/propagation/profiles_file.ex b/lib/microwaveprop/propagation/profiles_file.ex new file mode 100644 index 00000000..850be117 --- /dev/null +++ b/lib/microwaveprop/propagation/profiles_file.ex @@ -0,0 +1,179 @@ +defmodule Microwaveprop.Propagation.ProfilesFile do + @moduledoc """ + On-disk store for the raw, enriched HRRR grid_data that + `PropagationGridWorker` produces for the f00 analysis hour. One + compressed ETF file per `valid_time` lands at + `{base_dir}/profiles/{iso}.etf.gz`. Used by `Propagation.point_detail/4` + to rebuild the factor breakdown for a clicked cell without a database + round trip. + + Only f00 is persisted — forecast hours intentionally skip the factor + breakdown on the map to keep the scoring+upsert phase under a minute + (see `Propagation.compute_scores/3`). + + ## Atomic writes + + Files are written through `rename(2)`: `path.tmp.` → rename to + the final path. On the shared NFS mount the rename is atomic, so a + concurrent reader sees either the old file, the new file, or nothing — + never a partial write. + """ + + alias Microwaveprop.Propagation.Grid + + require Logger + + @doc "Base directory the profile store lives under." + @spec base_dir() :: String.t() + def base_dir do + Path.join( + Application.get_env(:microwaveprop, :propagation_scores_dir, "/data/scores"), + "profiles" + ) + end + + @doc "Absolute path for the profile file covering `valid_time`." + @spec path_for(DateTime.t()) :: String.t() + def path_for(%DateTime{} = valid_time) do + iso = valid_time |> DateTime.truncate(:second) |> DateTime.to_iso8601() + Path.join(base_dir(), "#{iso}.etf.gz") + end + + @doc """ + Persist the enriched grid_data (`%{{lat, lon} => profile}`) for + `valid_time`. Writes compressed ETF; lat/lon keys are rounded to the + grid step so later point lookups can snap user-provided coordinates. + """ + @spec write!(DateTime.t(), %{{float(), float()} => map()}) :: :ok + def write!(%DateTime{} = valid_time, grid_data) when is_map(grid_data) do + path = path_for(valid_time) + File.mkdir_p!(Path.dirname(path)) + + snapped = + Map.new(grid_data, fn {{lat, lon}, profile} -> + {{Float.round(lat, 3), Float.round(lon, 3)}, profile} + end) + + binary = :erlang.term_to_binary(snapped, [:compressed]) + + tmp = path <> ".tmp." <> unique_suffix() + File.write!(tmp, binary, [:binary]) + File.rename!(tmp, path) + :ok + end + + @doc """ + Read a single profile for `(valid_time, lat, lon)`. Returns `nil` if + the file is missing or the point has no profile. Input lat/lon are + snapped to the nearest grid cell before lookup. + """ + @spec read_point(DateTime.t(), float(), float()) :: map() | nil + def read_point(%DateTime{} = valid_time, lat, lon) do + case File.read(path_for(valid_time)) do + {:ok, binary} -> + {snapped_lat, snapped_lon} = snap(lat, lon) + + binary + |> :erlang.binary_to_term([:safe]) + |> Map.get({snapped_lat, snapped_lon}) + + {:error, _} -> + nil + end + end + + @doc """ + Delete profile files whose valid_time is strictly before `cutoff`. + Returns the number of files removed. + """ + @spec prune_older_than(DateTime.t()) :: non_neg_integer() + def prune_older_than(%DateTime{} = cutoff) do + cutoff_unix = DateTime.to_unix(cutoff) + + base_dir() + |> list_profile_files() + |> Enum.reduce(0, fn {path, valid_time_unix}, acc -> + if valid_time_unix < cutoff_unix do + _ = File.rm(path) + acc + 1 + else + acc + end + end) + end + + @doc """ + Keep only profile files whose valid_time is inside the closed window + `[run_time, run_time + max_forecast_hour * 3600]`. Mirrors + `ScoresFile.retain_window/2` and is called at the end of a + `PropagationGridWorker` chain. + """ + @spec retain_window(DateTime.t(), non_neg_integer()) :: non_neg_integer() + def retain_window(%DateTime{} = run_time, max_forecast_hour) when max_forecast_hour >= 0 do + lo = DateTime.to_unix(run_time) + hi = lo + max_forecast_hour * 3600 + + base_dir() + |> list_profile_files() + |> Enum.reduce(0, fn {path, valid_time_unix}, acc -> + if valid_time_unix < lo or valid_time_unix > hi do + _ = File.rm(path) + acc + 1 + else + acc + end + end) + end + + @doc """ + Latest valid_time of any persisted profile, or nil if the store is + empty. + """ + @spec latest_valid_time() :: DateTime.t() | nil + def latest_valid_time do + case list_profile_files(base_dir()) do + [] -> + nil + + files -> + files + |> Enum.map(fn {_path, unix} -> unix end) + |> Enum.max() + |> DateTime.from_unix!() + end + end + + defp snap(lat, lon) do + step = Grid.step() + snapped_lat = Float.round(Float.round(lat / step) * step, 3) + snapped_lon = Float.round(Float.round(lon / step) * step, 3) + {snapped_lat, snapped_lon} + end + + defp list_profile_files(dir) do + case File.ls(dir) do + {:ok, entries} -> + for entry <- entries, + valid_time_unix = parse_valid_time(entry), + valid_time_unix != nil do + {Path.join(dir, entry), valid_time_unix} + end + + _ -> + [] + end + end + + defp parse_valid_time(filename) do + with [_, iso] <- Regex.run(~r/^(.+)\.etf\.gz$/, filename), + {:ok, dt, _} <- DateTime.from_iso8601(iso) do + DateTime.to_unix(dt) + else + _ -> nil + end + end + + defp unique_suffix do + "#{System.system_time(:nanosecond)}.#{:erlang.unique_integer([:positive])}" + end +end diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex index 61be3439..3683803d 100644 --- a/lib/microwaveprop/workers/propagation_grid_worker.ex +++ b/lib/microwaveprop/workers/propagation_grid_worker.ex @@ -26,6 +26,7 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do alias Microwaveprop.Propagation alias Microwaveprop.Propagation.BandConfig alias Microwaveprop.Propagation.Grid + alias Microwaveprop.Propagation.ProfilesFile alias Microwaveprop.Propagation.ScoreCache alias Microwaveprop.Propagation.ScoresFile alias Microwaveprop.Weather @@ -134,10 +135,14 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do # chain didn't happen to overwrite (files are keyed by # valid_time, so an "old f00" from a prior run escapes the # natural last-writer-wins path). - dropped = ScoresFile.retain_window(run_time, @max_forecast_hour) + dropped_scores = ScoresFile.retain_window(run_time, @max_forecast_hour) + dropped_profiles = ProfilesFile.retain_window(run_time, @max_forecast_hour) - if dropped > 0 do - Logger.info("PropagationGrid: discarded #{dropped} leftover score files from prior chain") + if dropped_scores + dropped_profiles > 0 do + Logger.info( + "PropagationGrid: discarded #{dropped_scores} leftover score files + " <> + "#{dropped_profiles} profile files from prior chain" + ) end Logger.info("PropagationGrid: chain complete for run_time=#{run_time}") @@ -193,6 +198,14 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do :erlang.garbage_collect() + # Persist the fully-enriched f00 grid_data so point_detail can + # rebuild the factor breakdown for a clicked cell by re-running + # the scorer against the stored profile. Forecast hours skip + # factors in compute_scores, so we skip them here too. + if forecast_hour == 0 do + persist_f00_profiles(grid_data, valid_time) + end + # Build weather cache rows from in-memory grid_data — avoids the ~20s # JSONB round trip that was crashing the worker before f01–f18 could run. rows = Weather.build_grid_cache_rows(grid_data, valid_time) @@ -223,6 +236,17 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do end end + defp persist_f00_profiles(grid_data, valid_time) do + timed("profiles", fn -> + try do + ProfilesFile.write!(valid_time, grid_data) + rescue + e -> + Logger.warning("PropagationGrid: profiles write failed: #{inspect(e)}") + end + end) + end + defp warm_cache(valid_time) do Enum.each(BandConfig.all_bands(), fn band -> Propagation.warm_cache_and_broadcast(band.freq_mhz, valid_time) diff --git a/test/microwaveprop/propagation/profiles_file_test.exs b/test/microwaveprop/propagation/profiles_file_test.exs new file mode 100644 index 00000000..f15937a5 --- /dev/null +++ b/test/microwaveprop/propagation/profiles_file_test.exs @@ -0,0 +1,102 @@ +defmodule Microwaveprop.Propagation.ProfilesFileTest do + # async: false — tests mutate the global Application env to redirect + # the scores dir to a private tmp tree. + use ExUnit.Case, async: false + + alias Microwaveprop.Propagation.ProfilesFile + + setup do + dir = + Path.join( + System.tmp_dir!(), + "profiles_file_test_#{System.unique_integer([:positive])}" + ) + + File.mkdir_p!(dir) + + prev = Application.get_env(:microwaveprop, :propagation_scores_dir) + Application.put_env(:microwaveprop, :propagation_scores_dir, dir) + + on_exit(fn -> + File.rm_rf!(dir) + + if prev do + Application.put_env(:microwaveprop, :propagation_scores_dir, prev) + else + Application.delete_env(:microwaveprop, :propagation_scores_dir) + end + end) + + %{dir: dir} + end + + describe "path_for/1" do + test "nests under base_dir/profiles/.etf.gz", %{dir: dir} do + assert ProfilesFile.path_for(~U[2026-04-14 18:00:00Z]) == + Path.join([dir, "profiles", "2026-04-14T18:00:00Z.etf.gz"]) + end + end + + describe "write!/2 + read_point/3" do + test "round-trips a point and snaps lat/lon to the 0.125° grid" do + valid_time = ~U[2026-04-14 18:00:00Z] + + grid_data = %{ + {32.75, -97.125} => %{surface_temp_c: 25.0, surface_dewpoint_c: 15.0, wind_u: 3.0, wind_v: -1.0}, + {25.0, -125.0} => %{surface_temp_c: 10.0} + } + + ProfilesFile.write!(valid_time, grid_data) + + # Exact match + assert %{surface_temp_c: 25.0, wind_u: 3.0} = + ProfilesFile.read_point(valid_time, 32.75, -97.125) + + # Nudged within snap tolerance still resolves to the same cell + assert %{surface_temp_c: 25.0} = + ProfilesFile.read_point(valid_time, 32.7, -97.1) + + # A point with no stored profile returns nil + assert ProfilesFile.read_point(valid_time, 40.0, -80.0) == nil + end + + test "returns nil for a missing valid_time" do + assert ProfilesFile.read_point(~U[2026-04-14 18:00:00Z], 32.75, -97.125) == nil + end + end + + describe "prune_older_than/1" do + test "deletes files with valid_time strictly before the cutoff" do + old = ~U[2026-04-14 10:00:00Z] + keep = ~U[2026-04-14 18:00:00Z] + + ProfilesFile.write!(old, %{{25.0, -125.0} => %{surface_temp_c: 1.0}}) + ProfilesFile.write!(keep, %{{25.0, -125.0} => %{surface_temp_c: 2.0}}) + + assert 1 == ProfilesFile.prune_older_than(~U[2026-04-14 12:00:00Z]) + assert ProfilesFile.read_point(old, 25.0, -125.0) == nil + assert %{surface_temp_c: 2.0} = ProfilesFile.read_point(keep, 25.0, -125.0) + end + end + + describe "retain_window/2" do + test "keeps only files whose valid_time is inside [run_time, run_time + max_fh * 3600]" do + run_time = ~U[2026-04-14 18:00:00Z] + + inside_f00 = ~U[2026-04-14 18:00:00Z] + inside_f10 = ~U[2026-04-15 04:00:00Z] + after_window = ~U[2026-04-15 13:00:00Z] + before_window = ~U[2026-04-14 10:00:00Z] + + for vt <- [inside_f00, inside_f10, after_window, before_window] do + ProfilesFile.write!(vt, %{{25.0, -125.0} => %{surface_temp_c: 1.0}}) + end + + assert 2 == ProfilesFile.retain_window(run_time, 18) + assert %{surface_temp_c: 1.0} = ProfilesFile.read_point(inside_f00, 25.0, -125.0) + assert %{surface_temp_c: 1.0} = ProfilesFile.read_point(inside_f10, 25.0, -125.0) + assert ProfilesFile.read_point(after_window, 25.0, -125.0) == nil + assert ProfilesFile.read_point(before_window, 25.0, -125.0) == nil + end + end +end diff --git a/test/microwaveprop/propagation_test.exs b/test/microwaveprop/propagation_test.exs index 69fbc7fd..a5cdd22a 100644 --- a/test/microwaveprop/propagation_test.exs +++ b/test/microwaveprop/propagation_test.exs @@ -2,6 +2,7 @@ defmodule Microwaveprop.PropagationTest do use Microwaveprop.DataCase, async: false alias Microwaveprop.Propagation + alias Microwaveprop.Propagation.ProfilesFile alias Microwaveprop.Propagation.ScoreCache alias Microwaveprop.Propagation.ScoresFile @@ -193,7 +194,7 @@ defmodule Microwaveprop.PropagationTest do end describe "point_detail/4" do - test "returns the score from the on-disk file with an empty factors map" do + test "returns the score from the on-disk file with an empty factors map when no profile is persisted" do valid_time = ~U[2026-07-15 13:00:00Z] Propagation.replace_scores( @@ -205,6 +206,42 @@ defmodule Microwaveprop.PropagationTest do Propagation.point_detail(10_000, 25.0, -125.0, valid_time) end + test "rebuilds the factor breakdown from a persisted f00 profile" do + valid_time = ~U[2026-07-15 13:00:00Z] + lat = 32.75 + lon = -97.125 + + profile = %{ + surface_temp_c: 25.0, + surface_dewpoint_c: 18.0, + surface_pressure_mb: 1013.0, + hpbl_m: 500.0, + wind_u: 3.0, + wind_v: 2.0, + cloud_cover_pct: 15.0, + precip_mm: 0.0, + profile: [ + %{"pres" => 1000.0, "tmpc" => 25.0, "dwpc" => 18.0, "hght" => 100.0}, + %{"pres" => 975.0, "tmpc" => 22.0, "dwpc" => 15.0, "hght" => 350.0}, + %{"pres" => 950.0, "tmpc" => 19.0, "dwpc" => 10.0, "hght" => 600.0} + ] + } + + ProfilesFile.write!(valid_time, %{{lat, lon} => profile}) + + Propagation.replace_scores( + [%{lat: lat, lon: lon, valid_time: valid_time, band_mhz: 10_000, score: 71, factors: nil}], + valid_time + ) + + detail = Propagation.point_detail(10_000, lat, lon, valid_time) + assert detail.score == 71 + assert is_map(detail.factors) + # The algorithm populates every weighted factor; humidity is the + # dominant input for 10 GHz so it must be present. + assert Map.has_key?(detail.factors, :humidity) + end + test "returns nil when the file doesn't exist for the requested point" do assert Propagation.point_detail(10_000, 25.0, -125.0, ~U[2026-07-15 13:00:00Z]) == nil end