diff --git a/config/dev.exs b/config/dev.exs index ff6b67fa..0bf69a28 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -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 diff --git a/lib/microwaveprop/weather/hrrr_client.ex b/lib/microwaveprop/weather/hrrr_client.ex index 46fd6155..606e065e 100644 --- a/lib/microwaveprop/weather/hrrr_client.ex +++ b/lib/microwaveprop/weather/hrrr_client.ex @@ -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)) diff --git a/lib/mix/tasks/hrrr_backfill.ex b/lib/mix/tasks/hrrr_backfill.ex new file mode 100644 index 00000000..04c9dbc9 --- /dev/null +++ b/lib/mix/tasks/hrrr_backfill.ex @@ -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