Persist weather grid for every forecast hour via ProfilesFile
Extend the f00 profile persistence to cover every forecast hour (f00-f18) so /weather keeps working across pod restarts and shows forecast-hour atmospheric data without a fresh database. Route the Weather cold path (latest_grid_valid_time, load_weather_grid, weather_point_detail) through ProfilesFile first, with the legacy hrrr_profiles table as a last-resort historical fallback. Warm GridCache from the latest profile on app start so /weather is hot the moment a pod boots.
This commit is contained in:
parent
2b4756e921
commit
130e062054
7 changed files with 217 additions and 29 deletions
|
|
@ -35,6 +35,11 @@ defmodule Microwaveprop.Application do
|
||||||
# Run migrations after Repo is started
|
# Run migrations after Repo is started
|
||||||
Microwaveprop.Release.migrate()
|
Microwaveprop.Release.migrate()
|
||||||
|
|
||||||
|
# Warm GridCache from the latest persisted ProfilesFile so /weather
|
||||||
|
# has data immediately after a pod restart. Runs after the GridCache
|
||||||
|
# GenServer is up (it's in `children` above).
|
||||||
|
Microwaveprop.Weather.warm_grid_cache_from_latest_profile()
|
||||||
|
|
||||||
# Load ML model in dev/test only (Nx/Axon not compiled for prod)
|
# Load ML model in dev/test only (Nx/Axon not compiled for prod)
|
||||||
if Application.get_env(:microwaveprop, :load_ml_model, false) do
|
if Application.get_env(:microwaveprop, :load_ml_model, false) do
|
||||||
Microwaveprop.Propagation.load_ml_model()
|
Microwaveprop.Propagation.load_ml_model()
|
||||||
|
|
|
||||||
|
|
@ -62,6 +62,20 @@ defmodule Microwaveprop.Propagation.ProfilesFile do
|
||||||
:ok
|
:ok
|
||||||
end
|
end
|
||||||
|
|
||||||
|
@doc """
|
||||||
|
Read the full persisted grid_data for `valid_time`. Returns
|
||||||
|
`{:ok, %{{lat, lon} => profile}}` or `{:error, :enoent}` if the file
|
||||||
|
doesn't exist. Used by `Microwaveprop.Weather` to warm `GridCache`
|
||||||
|
on pod startup without touching the database.
|
||||||
|
"""
|
||||||
|
@spec read(DateTime.t()) :: {:ok, %{{float(), float()} => map()}} | {:error, :enoent}
|
||||||
|
def read(%DateTime{} = valid_time) do
|
||||||
|
case File.read(path_for(valid_time)) do
|
||||||
|
{:ok, binary} -> {:ok, :erlang.binary_to_term(binary, [:safe])}
|
||||||
|
{:error, _} -> {:error, :enoent}
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
@doc """
|
@doc """
|
||||||
Read a single profile for `(valid_time, lat, lon)`. Returns `nil` if
|
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
|
the file is missing or the point has no profile. Input lat/lon are
|
||||||
|
|
@ -69,13 +83,10 @@ defmodule Microwaveprop.Propagation.ProfilesFile do
|
||||||
"""
|
"""
|
||||||
@spec read_point(DateTime.t(), float(), float()) :: map() | nil
|
@spec read_point(DateTime.t(), float(), float()) :: map() | nil
|
||||||
def read_point(%DateTime{} = valid_time, lat, lon) do
|
def read_point(%DateTime{} = valid_time, lat, lon) do
|
||||||
case File.read(path_for(valid_time)) do
|
case read(valid_time) do
|
||||||
{:ok, binary} ->
|
{:ok, grid_data} ->
|
||||||
{snapped_lat, snapped_lon} = snap(lat, lon)
|
{snapped_lat, snapped_lon} = snap(lat, lon)
|
||||||
|
Map.get(grid_data, {snapped_lat, snapped_lon})
|
||||||
binary
|
|
||||||
|> :erlang.binary_to_term([:safe])
|
|
||||||
|> Map.get({snapped_lat, snapped_lon})
|
|
||||||
|
|
||||||
{:error, _} ->
|
{:error, _} ->
|
||||||
nil
|
nil
|
||||||
|
|
|
||||||
|
|
@ -3,6 +3,7 @@ defmodule Microwaveprop.Weather do
|
||||||
|
|
||||||
import Ecto.Query
|
import Ecto.Query
|
||||||
|
|
||||||
|
alias Microwaveprop.Propagation.ProfilesFile
|
||||||
alias Microwaveprop.Repo
|
alias Microwaveprop.Repo
|
||||||
alias Microwaveprop.Weather.Era5Profile
|
alias Microwaveprop.Weather.Era5Profile
|
||||||
alias Microwaveprop.Weather.GridCache
|
alias Microwaveprop.Weather.GridCache
|
||||||
|
|
@ -361,20 +362,26 @@ defmodule Microwaveprop.Weather do
|
||||||
|
|
||||||
@spec latest_grid_valid_time() :: DateTime.t() | nil
|
@spec latest_grid_valid_time() :: DateTime.t() | nil
|
||||||
def latest_grid_valid_time do
|
def latest_grid_valid_time do
|
||||||
case GridCache.latest_valid_time() do
|
cond do
|
||||||
%DateTime{} = vt ->
|
vt = GridCache.latest_valid_time() -> vt
|
||||||
vt
|
vt = ProfilesFile.latest_valid_time() -> vt
|
||||||
|
true -> latest_grid_valid_time_from_db()
|
||||||
nil ->
|
|
||||||
Repo.one(
|
|
||||||
from(h in HrrrProfile,
|
|
||||||
where: h.is_grid_point == true,
|
|
||||||
select: max(h.valid_time)
|
|
||||||
)
|
|
||||||
)
|
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
# Last-resort fallback for historical data sitting in the legacy
|
||||||
|
# hrrr_profiles table. PropagationGridWorker no longer writes
|
||||||
|
# grid-point rows there, so in steady state this returns nil and
|
||||||
|
# the ProfilesFile fallback above is the real source of truth.
|
||||||
|
defp latest_grid_valid_time_from_db do
|
||||||
|
Repo.one(
|
||||||
|
from(h in HrrrProfile,
|
||||||
|
where: h.is_grid_point == true,
|
||||||
|
select: max(h.valid_time)
|
||||||
|
)
|
||||||
|
)
|
||||||
|
end
|
||||||
|
|
||||||
@doc """
|
@doc """
|
||||||
Cache-only read for the /weather LiveView mount hot path. Returns whatever
|
Cache-only read for the /weather LiveView mount hot path. Returns whatever
|
||||||
is in `GridCache` for the latest valid_time and fires a deduped background
|
is in `GridCache` for the latest valid_time and fires a deduped background
|
||||||
|
|
@ -420,13 +427,24 @@ defmodule Microwaveprop.Weather do
|
||||||
rows
|
rows
|
||||||
|
|
||||||
:miss ->
|
:miss ->
|
||||||
full = load_weather_grid_from_db(latest_vt)
|
full = load_grid_rows_for(latest_vt)
|
||||||
GridCache.put(latest_vt, full)
|
GridCache.put(latest_vt, full)
|
||||||
filter_weather_bounds(full, bounds)
|
filter_weather_bounds(full, bounds)
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
# Build derived GridCache rows for a valid_time from whichever
|
||||||
|
# source has data: the persisted ProfilesFile first (hot path in
|
||||||
|
# steady state), then the legacy hrrr_profiles table (historical
|
||||||
|
# data only).
|
||||||
|
defp load_grid_rows_for(valid_time) do
|
||||||
|
case ProfilesFile.read(valid_time) do
|
||||||
|
{:ok, grid_data} -> build_grid_cache_rows(grid_data, valid_time)
|
||||||
|
{:error, _} -> load_weather_grid_from_db(valid_time)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
defp filter_weather_bounds(rows, nil), do: rows
|
defp filter_weather_bounds(rows, nil), do: rows
|
||||||
|
|
||||||
defp filter_weather_bounds(rows, %{"south" => s, "north" => n, "west" => w, "east" => e}) do
|
defp filter_weather_bounds(rows, %{"south" => s, "north" => n, "west" => w, "east" => e}) do
|
||||||
|
|
@ -495,11 +513,38 @@ defmodule Microwaveprop.Weather do
|
||||||
"""
|
"""
|
||||||
@spec warm_grid_cache_and_broadcast(DateTime.t()) :: :ok
|
@spec warm_grid_cache_and_broadcast(DateTime.t()) :: :ok
|
||||||
def warm_grid_cache_and_broadcast(valid_time) do
|
def warm_grid_cache_and_broadcast(valid_time) do
|
||||||
rows = load_weather_grid_from_db(valid_time)
|
rows = load_grid_rows_for(valid_time)
|
||||||
GridCache.broadcast_put(valid_time, rows)
|
GridCache.broadcast_put(valid_time, rows)
|
||||||
:ok
|
:ok
|
||||||
end
|
end
|
||||||
|
|
||||||
|
@doc """
|
||||||
|
Warm the local `GridCache` from the latest persisted `ProfilesFile`
|
||||||
|
on pod startup. Makes `/weather` usable immediately after a deploy
|
||||||
|
instead of waiting for the next hourly PropagationGridWorker run.
|
||||||
|
Local put only — every node reads the same NFS mount so no need to
|
||||||
|
broadcast.
|
||||||
|
"""
|
||||||
|
@spec warm_grid_cache_from_latest_profile() :: :ok
|
||||||
|
def warm_grid_cache_from_latest_profile do
|
||||||
|
case ProfilesFile.latest_valid_time() do
|
||||||
|
nil ->
|
||||||
|
:ok
|
||||||
|
|
||||||
|
valid_time ->
|
||||||
|
try do
|
||||||
|
rows = load_grid_rows_for(valid_time)
|
||||||
|
GridCache.put(valid_time, rows)
|
||||||
|
Logger.info("Weather: warmed GridCache from ProfilesFile for #{valid_time} (#{length(rows)} rows)")
|
||||||
|
rescue
|
||||||
|
e ->
|
||||||
|
Logger.warning("Weather: ProfilesFile warm failed: #{inspect(e)}")
|
||||||
|
end
|
||||||
|
|
||||||
|
:ok
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
@doc """
|
@doc """
|
||||||
Build derived weather grid cache rows directly from the in-memory HRRR
|
Build derived weather grid cache rows directly from the in-memory HRRR
|
||||||
`grid_data` that `PropagationGridWorker` already has after fetching a
|
`grid_data` that `PropagationGridWorker` already has after fetching a
|
||||||
|
|
@ -566,8 +611,30 @@ defmodule Microwaveprop.Weather do
|
||||||
snapped_lon = Float.round(Float.round(lon / step) * step, 3)
|
snapped_lon = Float.round(Float.round(lon / step) * step, 3)
|
||||||
|
|
||||||
case GridCache.fetch_point(valid_time, snapped_lat, snapped_lon) do
|
case GridCache.fetch_point(valid_time, snapped_lat, snapped_lon) do
|
||||||
{:ok, row} -> row
|
{:ok, row} ->
|
||||||
:miss -> weather_point_detail_from_db(valid_time, snapped_lat, snapped_lon)
|
row
|
||||||
|
|
||||||
|
:miss ->
|
||||||
|
case weather_point_detail_from_profiles(valid_time, snapped_lat, snapped_lon) do
|
||||||
|
nil -> weather_point_detail_from_db(valid_time, snapped_lat, snapped_lon)
|
||||||
|
row -> row
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
# Derive a single GridCache-shaped row from a persisted ProfilesFile
|
||||||
|
# entry for `(valid_time, lat, lon)`. Returns nil when the file
|
||||||
|
# doesn't exist or the point has no profile.
|
||||||
|
defp weather_point_detail_from_profiles(valid_time, snapped_lat, snapped_lon) do
|
||||||
|
case ProfilesFile.read_point(valid_time, snapped_lat, snapped_lon) do
|
||||||
|
nil ->
|
||||||
|
nil
|
||||||
|
|
||||||
|
profile ->
|
||||||
|
case build_grid_cache_rows(%{{snapped_lat, snapped_lon} => profile}, valid_time) do
|
||||||
|
[row] -> row
|
||||||
|
_ -> nil
|
||||||
|
end
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -198,13 +198,12 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
|
||||||
|
|
||||||
:erlang.garbage_collect()
|
:erlang.garbage_collect()
|
||||||
|
|
||||||
# Persist the fully-enriched f00 grid_data so point_detail can
|
# Persist the fully-enriched grid_data for every forecast hour
|
||||||
# rebuild the factor breakdown for a clicked cell by re-running
|
# so (a) /weather can show forecast-hour data after a pod
|
||||||
# the scorer against the stored profile. Forecast hours skip
|
# restart and (b) point_detail can rebuild the factor
|
||||||
# factors in compute_scores, so we skip them here too.
|
# breakdown for a clicked cell at any forecast hour by
|
||||||
if forecast_hour == 0 do
|
# re-running the scorer against the stored profile.
|
||||||
persist_f00_profiles(grid_data, valid_time)
|
persist_profiles(grid_data, valid_time)
|
||||||
end
|
|
||||||
|
|
||||||
# Build weather cache rows from in-memory grid_data — avoids the ~20s
|
# 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.
|
# JSONB round trip that was crashing the worker before f01–f18 could run.
|
||||||
|
|
@ -236,7 +235,7 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
defp persist_f00_profiles(grid_data, valid_time) do
|
defp persist_profiles(grid_data, valid_time) do
|
||||||
timed("profiles", fn ->
|
timed("profiles", fn ->
|
||||||
try do
|
try do
|
||||||
ProfilesFile.write!(valid_time, grid_data)
|
ProfilesFile.write!(valid_time, grid_data)
|
||||||
|
|
|
||||||
|
|
@ -37,6 +37,39 @@ defmodule Microwaveprop.Propagation.ProfilesFileTest do
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
describe "read/1" do
|
||||||
|
test "returns the full grid_data map round-tripped through ETF" 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},
|
||||||
|
{33.0, -97.0} => %{surface_temp_c: 26.0}
|
||||||
|
}
|
||||||
|
|
||||||
|
ProfilesFile.write!(valid_time, grid_data)
|
||||||
|
|
||||||
|
assert {:ok, round_tripped} = ProfilesFile.read(valid_time)
|
||||||
|
assert round_tripped[{32.75, -97.125}][:surface_temp_c] == 25.0
|
||||||
|
assert round_tripped[{33.0, -97.0}][:surface_temp_c] == 26.0
|
||||||
|
end
|
||||||
|
|
||||||
|
test "returns :enoent when no file exists" do
|
||||||
|
assert ProfilesFile.read(~U[2026-04-14 18:00:00Z]) == {:error, :enoent}
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
describe "latest_valid_time/0" do
|
||||||
|
test "returns the most recent persisted valid_time" do
|
||||||
|
assert ProfilesFile.latest_valid_time() == nil
|
||||||
|
|
||||||
|
for vt <- [~U[2026-04-14 10:00:00Z], ~U[2026-04-14 18:00:00Z], ~U[2026-04-14 14:00:00Z]] do
|
||||||
|
ProfilesFile.write!(vt, %{{25.0, -125.0} => %{surface_temp_c: 1.0}})
|
||||||
|
end
|
||||||
|
|
||||||
|
assert ProfilesFile.latest_valid_time() == ~U[2026-04-14 18:00:00Z]
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
describe "write!/2 + read_point/3" do
|
describe "write!/2 + read_point/3" do
|
||||||
test "round-trips a point and snaps lat/lon to the 0.125° grid" 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]
|
valid_time = ~U[2026-04-14 18:00:00Z]
|
||||||
|
|
|
||||||
|
|
@ -1,6 +1,7 @@
|
||||||
defmodule Microwaveprop.WeatherGridTest do
|
defmodule Microwaveprop.WeatherGridTest do
|
||||||
use Microwaveprop.DataCase, async: false
|
use Microwaveprop.DataCase, async: false
|
||||||
|
|
||||||
|
alias Microwaveprop.Propagation.ProfilesFile
|
||||||
alias Microwaveprop.Repo
|
alias Microwaveprop.Repo
|
||||||
alias Microwaveprop.Weather
|
alias Microwaveprop.Weather
|
||||||
alias Microwaveprop.Weather.GridCache
|
alias Microwaveprop.Weather.GridCache
|
||||||
|
|
@ -359,6 +360,33 @@ defmodule Microwaveprop.WeatherGridTest do
|
||||||
assert Weather.weather_point_detail(32.15, -97.40, now) == nil
|
assert Weather.weather_point_detail(32.15, -97.40, now) == nil
|
||||||
end
|
end
|
||||||
|
|
||||||
|
test "falls back to ProfilesFile when GridCache is cold and DB has no grid points" do
|
||||||
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||||
|
|
||||||
|
grid_data = %{
|
||||||
|
{32.125, -97.375} => %{
|
||||||
|
surface_temp_c: 25.0,
|
||||||
|
surface_dewpoint_c: 18.0,
|
||||||
|
surface_pressure_mb: 1013.0,
|
||||||
|
hpbl_m: 500.0,
|
||||||
|
pwat_mm: 30.0,
|
||||||
|
profile: [
|
||||||
|
%{"pres" => 1000.0, "tmpc" => 25.0, "dwpc" => 18.0, "hght" => 100.0},
|
||||||
|
%{"pres" => 850.0, "tmpc" => 15.0, "dwpc" => 10.0, "hght" => 1500.0},
|
||||||
|
%{"pres" => 500.0, "tmpc" => -15.0, "dwpc" => -25.0, "hght" => 5600.0}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
ProfilesFile.write!(now, grid_data)
|
||||||
|
|
||||||
|
result = Weather.weather_point_detail(32.15, -97.40, now)
|
||||||
|
|
||||||
|
assert result.lat == 32.125
|
||||||
|
assert result.temperature == 25.0
|
||||||
|
assert result.surface_pressure_mb == 1013.0
|
||||||
|
end
|
||||||
|
|
||||||
test "includes derived upper-air fields and strips raw profile data" do
|
test "includes derived upper-air fields and strips raw profile data" do
|
||||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||||
|
|
||||||
|
|
@ -403,6 +431,48 @@ defmodule Microwaveprop.WeatherGridTest do
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
describe "latest_grid_valid_time/0 + ProfilesFile fallback" do
|
||||||
|
test "uses ProfilesFile.latest_valid_time when GridCache and DB are both empty" do
|
||||||
|
vt = ~U[2026-04-14 18:00:00Z]
|
||||||
|
|
||||||
|
ProfilesFile.write!(vt, %{
|
||||||
|
{32.125, -97.375} => %{surface_temp_c: 20.0, surface_dewpoint_c: 15.0}
|
||||||
|
})
|
||||||
|
|
||||||
|
assert Weather.latest_grid_valid_time() == vt
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
describe "load_weather_grid/1 + ProfilesFile fallback" do
|
||||||
|
test "returns derived rows from ProfilesFile when GridCache is empty" do
|
||||||
|
vt = ~U[2026-04-14 18:00:00Z]
|
||||||
|
|
||||||
|
ProfilesFile.write!(vt, %{
|
||||||
|
{32.125, -97.375} => %{
|
||||||
|
surface_temp_c: 25.0,
|
||||||
|
surface_dewpoint_c: 18.0,
|
||||||
|
surface_pressure_mb: 1013.0,
|
||||||
|
hpbl_m: 500.0,
|
||||||
|
pwat_mm: 30.0
|
||||||
|
},
|
||||||
|
{32.250, -97.375} => %{
|
||||||
|
surface_temp_c: 26.0,
|
||||||
|
surface_dewpoint_c: 19.0,
|
||||||
|
surface_pressure_mb: 1012.0,
|
||||||
|
hpbl_m: 600.0,
|
||||||
|
pwat_mm: 32.0
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
bounds = %{"south" => 32.0, "north" => 33.0, "west" => -98.0, "east" => -97.0}
|
||||||
|
rows = Weather.load_weather_grid(bounds)
|
||||||
|
|
||||||
|
assert length(rows) == 2
|
||||||
|
assert Enum.any?(rows, &(&1.temperature == 25.0))
|
||||||
|
assert Enum.any?(rows, &(&1.temperature == 26.0))
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
defp insert_hrrr_profile(attrs) do
|
defp insert_hrrr_profile(attrs) do
|
||||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -37,6 +37,9 @@ defmodule MicrowavepropWeb.ConnCase do
|
||||||
|
|
||||||
setup tags do
|
setup tags do
|
||||||
Microwaveprop.DataCase.setup_sandbox(tags)
|
Microwaveprop.DataCase.setup_sandbox(tags)
|
||||||
|
Microwaveprop.DataCase.reset_score_files()
|
||||||
|
Microwaveprop.Weather.GridCache.clear()
|
||||||
|
Microwaveprop.Propagation.ScoreCache.clear()
|
||||||
{:ok, conn: Phoenix.ConnTest.build_conn()}
|
{:ok, conn: Phoenix.ConnTest.build_conn()}
|
||||||
end
|
end
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue