Add HRRR GRIB2 caching and backfill task for finer pressure levels
- Cache raw GRIB2 byte ranges to ~/hrrr in dev (never deleted) - mix hrrr_backfill re-fetches QSO-linked HRRR profiles with 13 levels - Supports --all (re-fetch everything) and --limit N flags - Groups by HRRR hour to batch requests, 500ms rate limit - AWS archive has full HRRR history, same URL pattern as live feed
This commit is contained in:
parent
bdaf74f5ef
commit
ff0ca89646
3 changed files with 206 additions and 0 deletions
|
|
@ -92,6 +92,9 @@ config :microwaveprop, load_ml_model: true
|
|||
# Use local SRTM1 tiles for elevation lookups instead of the Open-Meteo API
|
||||
config :microwaveprop, srtm_tiles_dir: Path.expand("~/srtm/tiles")
|
||||
|
||||
# Cache raw HRRR GRIB2 files locally (dev only — do not delete)
|
||||
config :microwaveprop, hrrr_cache_dir: Path.expand("~/hrrr")
|
||||
|
||||
# Freshness monitor watches for stale propagation scores
|
||||
config :microwaveprop, start_freshness_monitor: true
|
||||
|
||||
|
|
|
|||
|
|
@ -298,6 +298,27 @@ defmodule Microwaveprop.Weather.HrrrClient do
|
|||
defp download_grib_ranges(_url, []), do: {:ok, <<>>}
|
||||
|
||||
defp download_grib_ranges(url, ranges) do
|
||||
# Check local cache first (dev only)
|
||||
cache_key = url_to_cache_key(url, ranges)
|
||||
|
||||
case read_cache(cache_key) do
|
||||
{:ok, binary} ->
|
||||
Logger.info("HRRR cache hit: #{cache_key}")
|
||||
{:ok, binary}
|
||||
|
||||
:miss ->
|
||||
case download_grib_ranges_remote(url, ranges) do
|
||||
{:ok, binary} = ok ->
|
||||
write_cache(cache_key, binary)
|
||||
ok
|
||||
|
||||
error ->
|
||||
error
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
defp download_grib_ranges_remote(url, ranges) do
|
||||
# Merge adjacent/overlapping ranges to reduce HTTP requests
|
||||
merged = merge_ranges(ranges)
|
||||
|
||||
|
|
@ -336,6 +357,39 @@ defmodule Microwaveprop.Weather.HrrrClient do
|
|||
end
|
||||
end
|
||||
|
||||
defp hrrr_cache_dir, do: Application.get_env(:microwaveprop, :hrrr_cache_dir)
|
||||
|
||||
defp url_to_cache_key(url, ranges) do
|
||||
# e.g. hrrr.20240922_conus_hrrr.t18z.wrfprsf00_r42.grib2
|
||||
uri = URI.parse(url)
|
||||
path_part = uri.path |> String.trim_leading("/") |> String.replace("/", "_")
|
||||
range_hash = ranges |> :erlang.phash2() |> Integer.to_string(16)
|
||||
"#{path_part}_r#{range_hash}.grib2"
|
||||
end
|
||||
|
||||
defp read_cache(key) do
|
||||
case hrrr_cache_dir() do
|
||||
nil -> :miss
|
||||
dir ->
|
||||
path = Path.join(dir, key)
|
||||
case File.read(path) do
|
||||
{:ok, binary} -> {:ok, binary}
|
||||
{:error, _} -> :miss
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
defp write_cache(key, binary) do
|
||||
case hrrr_cache_dir() do
|
||||
nil -> :ok
|
||||
dir ->
|
||||
File.mkdir_p!(dir)
|
||||
path = Path.join(dir, key)
|
||||
File.write!(path, binary)
|
||||
Logger.info("HRRR cached: #{path} (#{byte_size(binary)} bytes)")
|
||||
end
|
||||
end
|
||||
|
||||
defp merge_ranges(ranges) do
|
||||
ranges
|
||||
|> Enum.sort_by(&elem(&1, 0))
|
||||
|
|
|
|||
149
lib/mix/tasks/hrrr_backfill.ex
Normal file
149
lib/mix/tasks/hrrr_backfill.ex
Normal file
|
|
@ -0,0 +1,149 @@
|
|||
defmodule Mix.Tasks.HrrrBackfill do
|
||||
@moduledoc """
|
||||
Re-fetches HRRR profiles for all QSO-linked contacts with the current
|
||||
(finer) pressure level set. Replaces existing profiles with updated data.
|
||||
|
||||
Usage:
|
||||
mix hrrr_backfill # backfill all contacts missing HRRR or with < 13 levels
|
||||
mix hrrr_backfill --all # re-fetch all contacts regardless of level count
|
||||
mix hrrr_backfill --limit 100 # process at most 100 hours
|
||||
"""
|
||||
|
||||
use Mix.Task
|
||||
|
||||
import Ecto.Query
|
||||
|
||||
alias Microwaveprop.Radio
|
||||
alias Microwaveprop.Radio.Contact
|
||||
alias Microwaveprop.Repo
|
||||
alias Microwaveprop.Weather
|
||||
alias Microwaveprop.Weather.HrrrClient
|
||||
alias Microwaveprop.Weather.HrrrProfile
|
||||
alias Microwaveprop.Weather.SoundingParams
|
||||
|
||||
require Logger
|
||||
|
||||
@min_levels 13
|
||||
|
||||
@impl Mix.Task
|
||||
def run(args) do
|
||||
Mix.Task.run("app.start")
|
||||
|
||||
{opts, _, _} = OptionParser.parse(args, switches: [all: :boolean, limit: :integer])
|
||||
refetch_all? = Keyword.get(opts, :all, false)
|
||||
limit = Keyword.get(opts, :limit, nil)
|
||||
|
||||
# Get all distinct (rounded_hour, path_points) combinations for contacts
|
||||
contacts =
|
||||
from(c in Contact,
|
||||
where: not is_nil(c.pos1) and c.qso_timestamp >= ^~U[2014-01-01 00:00:00Z],
|
||||
select: %{id: c.id, pos1: c.pos1, pos2: c.pos2, qso_timestamp: c.qso_timestamp},
|
||||
order_by: [desc: c.qso_timestamp]
|
||||
)
|
||||
|> Repo.all()
|
||||
|
||||
# Group by rounded HRRR hour to batch requests
|
||||
by_hour =
|
||||
contacts
|
||||
|> Enum.flat_map(fn c ->
|
||||
hour = HrrrClient.nearest_hrrr_hour(c.qso_timestamp)
|
||||
|
||||
Radio.contact_path_points(c)
|
||||
|> Enum.map(fn {lat, lon} ->
|
||||
{rlat, rlon} = Weather.round_to_hrrr_grid(lat, lon)
|
||||
{hour, {rlat, rlon}}
|
||||
end)
|
||||
end)
|
||||
|> Enum.group_by(fn {hour, _} -> hour end, fn {_, point} -> point end)
|
||||
|> Enum.map(fn {hour, points} -> {hour, Enum.uniq(points)} end)
|
||||
|> Enum.sort_by(fn {hour, _} -> hour end, {:desc, DateTime})
|
||||
|
||||
# Filter to only hours needing backfill
|
||||
by_hour =
|
||||
if refetch_all? do
|
||||
by_hour
|
||||
else
|
||||
Enum.filter(by_hour, fn {hour, points} ->
|
||||
needs_backfill?(hour, points)
|
||||
end)
|
||||
end
|
||||
|
||||
by_hour = if limit, do: Enum.take(by_hour, limit), else: by_hour
|
||||
|
||||
total = length(by_hour)
|
||||
total_points = Enum.reduce(by_hour, 0, fn {_, pts}, acc -> acc + length(pts) end)
|
||||
Logger.info("HRRR backfill: #{total} hours, #{total_points} points to process")
|
||||
|
||||
by_hour
|
||||
|> Enum.with_index(1)
|
||||
|> Enum.each(fn {{hour, points}, idx} ->
|
||||
Logger.info("HRRR backfill [#{idx}/#{total}]: #{hour} (#{length(points)} points)")
|
||||
|
||||
case HrrrClient.fetch_grid(points, hour) do
|
||||
{:ok, grid_data} ->
|
||||
Enum.each(grid_data, fn {{lat, lon}, data} ->
|
||||
store_profile(lat, lon, hour, data)
|
||||
end)
|
||||
|
||||
Logger.info("HRRR backfill [#{idx}/#{total}]: saved #{map_size(grid_data)} profiles")
|
||||
|
||||
{:error, reason} ->
|
||||
Logger.warning("HRRR backfill [#{idx}/#{total}]: failed - #{inspect(reason)}")
|
||||
end
|
||||
|
||||
# Rate limit to avoid overwhelming NOAA
|
||||
if idx < total, do: Process.sleep(500)
|
||||
end)
|
||||
|
||||
Logger.info("HRRR backfill complete")
|
||||
end
|
||||
|
||||
defp needs_backfill?(hour, points) do
|
||||
# Check if any point for this hour is missing or has fewer than @min_levels
|
||||
sample = hd(points)
|
||||
{lat, lon} = sample
|
||||
|
||||
case Repo.one(
|
||||
from(h in HrrrProfile,
|
||||
where: h.lat == ^lat and h.lon == ^lon and h.valid_time == ^hour,
|
||||
select: fragment("array_length(profile, 1)")
|
||||
)
|
||||
) do
|
||||
nil -> true
|
||||
count when count < @min_levels -> true
|
||||
_ -> false
|
||||
end
|
||||
end
|
||||
|
||||
defp store_profile(lat, lon, valid_time, data) do
|
||||
params = SoundingParams.derive(data.profile)
|
||||
|
||||
attrs =
|
||||
%{
|
||||
valid_time: valid_time,
|
||||
lat: lat,
|
||||
lon: lon,
|
||||
run_time: data.run_time,
|
||||
profile: data.profile,
|
||||
hpbl_m: data.hpbl_m,
|
||||
pwat_mm: data.pwat_mm,
|
||||
surface_temp_c: data.surface_temp_c,
|
||||
surface_dewpoint_c: data.surface_dewpoint_c,
|
||||
surface_pressure_mb: data.surface_pressure_mb
|
||||
}
|
||||
|> maybe_add_derived(params)
|
||||
|
||||
Weather.upsert_hrrr_profile(attrs)
|
||||
end
|
||||
|
||||
defp maybe_add_derived(attrs, nil), do: attrs
|
||||
|
||||
defp maybe_add_derived(attrs, params) do
|
||||
Map.merge(attrs, %{
|
||||
surface_refractivity: params.surface_refractivity,
|
||||
min_refractivity_gradient: params.min_refractivity_gradient,
|
||||
ducting_detected: params.ducting_detected,
|
||||
duct_characteristics: params.duct_characteristics
|
||||
})
|
||||
end
|
||||
end
|
||||
Loading…
Add table
Reference in a new issue