graham/prop
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Phase 1 tasks 1.1-1.5: HRRR native hybrid-sigma ingestion

Browse source 
- Spike docs at docs/research/hrrr_native_levels.md confirming files
  are on AWS S3 for 5+ years, 50 hybrid levels, and include TKE and
  SPFH needed for Phase 2 turbulence features. Architectural finding:
  per-point on-demand fetching is impractical (~530 MB/file), so
  the ingestion worker batches per (date, hour) instead.
- hrrr_native_profiles schema: arrays per level plus cached surface
  scalars and placeholder columns for Phase 2/4 derived fields.
  Strictly additive — the existing hrrr_profiles table is untouched.
- HrrrNativeClient: pure URL/message-list helpers, build_native_profile/1
  that turns a parsed wgrib2 map into the schema shape (TDD'd).
- Exposed HrrrClient.download_grib_ranges/2 so the native client
  reuses the existing parallel byte-range download + disk cache.
- HrrrNativeGridWorker: Oban worker keyed on {year, month, day, hour},
  unique at :infinity, pulls distinct (lat, lon) points from contacts
  in the ±30 min window, downloads the native grib2, extracts per
  point, bulk-upserts.
- mix hrrr_native_backfill --limit N enqueues the top-N hours by
  contact count.

Phase 1 gate still pending Task 1.6 (sanity-check backtest after
live data lands).
This commit is contained in:
Graham McIntire 2026-04-09 16:23:51 -05:00
parent 7630934bcb
commit 900685aa06
10 changed files with 943 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

131
docs/research/hrrr_native_levels.md Normal file
View file

@ -0,0 +1,131 @@
# HRRR Hybrid-Sigma ("Native") Levels — Spike Findings
**Date:** 2026-04-09
**Purpose:** Task 1.1 of `docs/plans/2026-04-09-propagation-modeling-improvements.md`
**Question:** Can we practically ingest HRRR native hybrid-sigma profiles to replace the 25 hPa pressure-level product for boundary-layer propagation features?
## TL;DR
**Yes — with one architecture change.** The files are available on AWS S3 going back at least 5 years and expose all the variables needed for Richardson number, θₑ, shear, and TKE computations on 50 hybrid-sigma levels. But the per-file download is too large to run on-demand per QSO point. Phase 1 should ingest native profiles in batch mode per HRRR run hour, not per QSO.
## File location
Same S3 bucket as the existing surface/pressure products:
```
https://noaa-hrrr-bdp-pds.s3.amazonaws.com/hrrr.<YYYYMMDD>/conus/hrrr.t<HH>z.wrfnatf00.grib2
https://noaa-hrrr-bdp-pds.s3.amazonaws.com/hrrr.<YYYYMMDD>/conus/hrrr.t<HH>z.wrfnatf00.grib2.idx
```
`wrfnatf00` = native levels, 0-hour forecast. Forecast hours `wrfnatf01` through `wrfnatf18` also exist.
## Historical availability
Spot-checked on 2026-04-09 by HEAD-ing the `.idx` URL:
| Date | HTTP status |
|---|---|
| 2026-03-09 (30d) | 200 |
| 2026-02-09 (60d) | 200 |
| 2026-01-09 (~90d) | 200 |
| 2025-12-09 (~120d) | 200 |
| 2024-04-09 (~1y) | 200 |
| 2023-04-09 (~2y) | 200 |
| 2022-04-09 (~3y) | 200 |
| 2021-04-09 (~5y) | 200 |
No rotation problem. We can backtest against years of historical native data if we want to, which unblocks Phase 6 (climatology) as well.
## What the file contains
The `.idx` file for one HRRR run has **1133 GRIB2 messages**. **1000 of them are on hybrid levels 1-50**, plus 133 surface/2m/10m/composite entries.
Each of the 50 hybrid levels carries these 20 variables:
```
PRES pressure (Pa)
HGT geopotential height (m)
TMP temperature (K)
SPFH specific humidity (kg/kg) ← derive Td, θ, θₑ from SPFH + TMP + PRES
UGRD u wind (m/s)
VGRD v wind (m/s)
VVEL vertical velocity (Pa/s)
TKE turbulent kinetic energy (m²/s²) ← **exactly the turbulence signal Phase 2 needs**
CLMR cloud water mixing ratio
CIMIXR cloud ice mixing ratio
RWMR rain water mixing ratio
SNMR snow mixing ratio
GRLE graupel mixing ratio
NCONCD number concentration of cloud droplets
NCCICE number concentration of cloud ice
SPNCR number concentration of rain
PMTF ???
PMTC ???
FRACCC convective cloud fraction
MASSDEN particulate mass density
```
The seven we care about for Phase 2–4 are **TMP, SPFH, HGT, UGRD, VGRD, TKE, PRES** × 50 levels = 350 messages out of the 1133.
Note: **there is no DPT on hybrid levels.** The pressure-level product has dewpoint directly; the native product only has specific humidity. We will compute `Td` from `(T, SPFH, P)` using the Bolton (1980) or Magnus-Tetens formulation. This is standard and cheap.
## Vertical spacing
Hybrid levels are defined by the HRRR vertical coordinate function; level 1 is closest to the surface. The actual geometric height of each level varies by grid point (because hybrid coordinates conform to terrain). Typical spacing for a plains grid point:
- Level 1: ~8 m AGL
- Level 2: ~25 m AGL
- Level 3: ~50 m AGL
- Level 5: ~130 m AGL
- Level 10: ~460 m AGL
- Level 20: ~1.7 km AGL
- Level 30: ~3.8 km AGL
- Level 50: ~19 km AGL
**The first ~5 hybrid levels alone span the range where most surface-based ducts live.** That's the whole point: 25 hPa pressure levels (~250 m spacing at the surface) completely miss anything thinner than ~200 m, and the native product gives us 30-50 m resolution in the same band.
(Actual per-level heights must be read from the HGT message for each grid point, not from a table. We store `heights_m` as `{:array, :float}` on `hrrr_native_profiles` for exactly this reason.)
## Download size — the architectural gotcha
Downloaded one full `wrfnatf00.grib2.idx` and walked the offsets for the seven essential variables on all 50 levels:
- 350 GRIB2 messages
- Total byte span of the essential messages: **~530 MB**
- Full native file: ~566 MB (so essentials are almost the whole file)
That's **per HRRR run hour**, for the full CONUS grid.
This matters because the plan's draft of Task 1.4 has a per-QSO `HrrrNativeFetchWorker`. 530 MB × 58k QSOs = 29 TB of re-downloads, which is absurd. **Byte-range requests don't save us** — GRIB2 messages can only be decoded whole, so we'd have to pull ~350 × 1.5 MB per point, and AWS doesn't support server-side subsetting on this bucket.
### What to do instead
Two workable architectures:
1. **Batch per run hour.** A single worker downloads one full `wrfnatf00.grib2` (~566 MB), extracts native profiles at *all* points of interest for that hour in one pass (the grid of contacts + whatever points the live scorer wants), and bulk-upserts them into `hrrr_native_profiles`. This is how the existing `PropagationGridWorker` pipeline works for the pressure-level file; we mirror it for native levels.
2. **Per-day local cache.** Download one file per `(date, hour)`, stash it on local disk for the duration of a backfill job, extract all points of interest for that hour, delete the file. This is probably only worth the bandwidth for dense backfills and can be folded into option 1.
**Plan correction:** the `HrrrNativeFetchWorker` in Task 1.4 should be renamed `HrrrNativeGridWorker` and take `{date, hour}` as its unique key, not `{lat, lon, valid_time}`. The backfill mix task in Task 1.5 iterates over historical (date, hour) pairs where we have contacts and enqueues one grid worker per hour.
## Bandwidth budget
If we backfill 1 year of hourly native grids for every hour where we have at least one QSO, and we have ~58k QSOs spread over 5+ years, conservatively ~15k distinct (date, hour) hours. 15k × 566 MB ≈ **8.5 TB**. That's a lot. We don't need it all — Phase 2 only needs enough data to backtest. Realistic first backfill:
- Top 500 distinct hours by QSO count → ~280 GB (one-time, finishes overnight)
- Extract only the lat/lon points where we have QSOs in that hour plus the propagation grid points
- Store derived profiles only (< 5 KB per profile), not the raw GRIB2
Storage for derived profiles: 500 hours × ~300 points/hour × 5 KB ≈ 750 MB in the DB. Fine.
## Proposed Phase 1 plan adjustments
1. Rename worker: `HrrrNativeGridWorker` instead of `HrrrNativeFetchWorker`; unique keys `{date, hour}`; downloads the full file once per run hour and bulk-processes points.
2. Keep `hrrr_native_profiles` schema as planned (per-point, per-valid_time), just populated in batch.
3. Task 1.5 backfill mix task now enqueues grid workers for the top-N (date, hour) hours by QSO density.
4. Feature functions in later phases look up `hrrr_native_profiles` the same way the existing scorer looks up `hrrr_profiles` — no API change on the consuming side.
5. Computing Td from SPFH is a small Phase 2 helper; budget ~half a day for it.
## Gate decision
**Gate pass.** Files are available historically, all required variables are present, vertical resolution is 5-30× finer than the pressure-level product near the surface, and the ingestion architecture is a known pattern (matches the existing grid worker). Proceed to Task 1.2 (schema) with the batch-mode correction.

13
lib/microwaveprop/weather/hrrr_client.ex
View file

@ -308,9 +308,18 @@ defmodule Microwaveprop.Weather.HrrrClient do
end
end
defp download_grib_ranges(_url, []), do: {:ok, <<>>}
@doc """
Downloads the given byte ranges from a GRIB2 URL and concatenates
them in offset order. Merges adjacent/overlapping ranges and runs
requests in parallel. Backed by a disk cache keyed on
`(url, ranges)` in dev.
defp download_grib_ranges(url, ranges) do
Public so the native-level client can reuse it rather than
duplicating the HTTP/parallelism/cache logic.
"""
def download_grib_ranges(_url, []), do: {:ok, <<>>}
def download_grib_ranges(url, ranges) do
# Check local cache first (dev only)
cache_key = url_to_cache_key(url, ranges)

126
lib/microwaveprop/weather/hrrr_native_client.ex Normal file
View file

@ -0,0 +1,126 @@
defmodule Microwaveprop.Weather.HrrrNativeClient do
@moduledoc """
Fetches HRRR native hybrid-sigma profiles from the AWS HRRR bucket.
This is the companion to `HrrrClient`, which works against the
surface and 25 hPa pressure-level products. The native file
(`wrfnatf00.grib2`) carries all variables on the 50 hybrid-sigma
levels native to the HRRR model grid. Vertical spacing near the
surface is ~10-50 m instead of the ~250 m the pressure-level
product gives us — crucial for resolving the ducts and
boundary-layer inversions discussed in
`docs/plans/2026-04-09-propagation-modeling-improvements.md`.
## Design: batch, not per-point
Each native-level HRRR file is ~566 MB. Essential variables (TMP,
SPFH, HGT, UGRD, VGRD, TKE, PRES on all 50 hybrid levels) span
~530 MB of that file. Per-point on-demand fetching is not viable.
Instead, the worker fetches the file once per `(date, hour)`,
extracts native profiles for every point of interest in one pass,
and bulk-inserts them.
See `docs/research/hrrr_native_levels.md` for the full analysis.
"""
alias Microwaveprop.Weather.HrrrClient
@native_levels 1..50
@native_variables ~w(TMP SPFH HGT UGRD VGRD TKE PRES)
@hrrr_base_default "https://noaa-hrrr-bdp-pds.s3.amazonaws.com"
defp hrrr_base, do: Application.get_env(:microwaveprop, :hrrr_base_url, @hrrr_base_default)
@doc "Number of native hybrid-sigma levels in HRRR (currently 50)."
def native_level_count, do: Enum.count(@native_levels)
@doc "The seven essential variables we extract on every native level."
def native_variables, do: @native_variables
@doc """
The list of `%{var:, level:}` messages we extract from every native
file. 7 vars × 50 levels = 350 messages, matching the spike in
Task 1.1.
"""
def native_messages do
for level <- @native_levels, var <- @native_variables do
%{var: var, level: "#{level} hybrid level"}
end
end
@doc """
Builds the AWS S3 URL for a native-level HRRR grib2 file.
## Examples
iex> Microwaveprop.Weather.HrrrNativeClient.hrrr_native_url(~D[2026-04-09], 12)
"https://noaa-hrrr-bdp-pds.s3.amazonaws.com/hrrr.20260409/conus/hrrr.t12z.wrfnatf00.grib2"
"""
def hrrr_native_url(date, hour, forecast_hour \\ 0) do
date_str = Calendar.strftime(date, "%Y%m%d")
hour_str = hour |> Integer.to_string() |> String.pad_leading(2, "0")
fh_str = forecast_hour |> Integer.to_string() |> String.pad_leading(2, "0")
"#{hrrr_base()}/hrrr.#{date_str}/conus/hrrr.t#{hour_str}z.wrfnatf#{fh_str}.grib2"
end
@doc """
Converts a parsed `%{"VAR:level" => value}` map into a
`HrrrNativeProfile`-shaped map with level arrays sorted by ascending
hybrid level (level 1 = surface).
This is the pure-function bit of the pipeline: the network/GRIB2
decoding lives elsewhere and just feeds `parsed` in. Isolating this
lets us unit-test every invariant we care about (array lengths,
ordering, surface scalar caching) without touching the network.
"""
def build_native_profile(parsed) when is_map(parsed) do
levels =
Enum.map(@native_levels, fn level ->
level_str = "#{level} hybrid level"
%{
level: level,
hgt: parsed["HGT:#{level_str}"],
tmp: parsed["TMP:#{level_str}"],
spfh: parsed["SPFH:#{level_str}"],
pres: parsed["PRES:#{level_str}"],
ugrd: parsed["UGRD:#{level_str}"],
vgrd: parsed["VGRD:#{level_str}"],
tke: parsed["TKE:#{level_str}"]
}
end)
|> Enum.reject(fn %{hgt: hgt, tmp: tmp} -> is_nil(hgt) or is_nil(tmp) end)
|> Enum.sort_by(& &1.hgt)
level_count = length(levels)
%{
level_count: level_count,
heights_m: Enum.map(levels, & &1.hgt),
temp_k: Enum.map(levels, & &1.tmp),
spfh: Enum.map(levels, & &1.spfh),
pressure_pa: Enum.map(levels, & &1.pres),
u_wind_ms: Enum.map(levels, & &1.ugrd),
v_wind_ms: Enum.map(levels, & &1.vgrd),
tke_m2s2: Enum.map(levels, & &1.tke),
surface_temp_k: parsed["TMP:surface"] || List.first(levels) |> safe_get(:tmp),
surface_spfh: parsed["SPFH:2 m above ground"] || List.first(levels) |> safe_get(:spfh),
surface_pressure_pa: parsed["PRES:surface"] || List.first(levels) |> safe_get(:pres)
}
end
defp safe_get(nil, _key), do: nil
defp safe_get(map, key), do: Map.get(map, key)
@doc """
Returns the list of byte ranges to download for the essentials in
one native HRRR file. Used by the (still-to-be-built) grid worker.
Wraps `HrrrClient.byte_ranges_for_messages/2` with our native
message list so callers don't have to know both.
"""
def essential_byte_ranges(idx_entries) do
HrrrClient.byte_ranges_for_messages(idx_entries, native_messages())
end
end

93
lib/microwaveprop/weather/hrrr_native_profile.ex Normal file
View file

@ -0,0 +1,93 @@
defmodule Microwaveprop.Weather.HrrrNativeProfile do
@moduledoc """
A single HRRR profile extracted on the native hybrid-sigma vertical
coordinate (50 levels), as opposed to `HrrrProfile` which is
interpolated to a coarse set of pressure levels.
This schema is intentionally additive: it does not replace
`HrrrProfile`. The legacy scorer keeps running against
`hrrr_profiles` until Phase 9 swaps it out.
Columns marked "derived (Phase 2/4)" are written by the
boundary-layer turbulence and duct-geometry phases — they stay
nullable so the ingestion worker can populate just the raw profile
columns and a later `mix hrrr_native_derive` task fills the rest in.
"""
use Ecto.Schema
import Ecto.Changeset
@primary_key {:id, :binary_id, autogenerate: true}
@foreign_key_type :binary_id
schema "hrrr_native_profiles" do
field :valid_time, :utc_datetime
field :run_time, :utc_datetime
field :lat, :float
field :lon, :float
# Raw native-level columns, level 1 (surface) first, strictly
# monotone in height.
field :level_count, :integer
field :heights_m, {:array, :float}
field :temp_k, {:array, :float}
field :spfh, {:array, :float}
field :pressure_pa, {:array, :float}
field :u_wind_ms, {:array, :float}
field :v_wind_ms, {:array, :float}
field :tke_m2s2, {:array, :float}
# Cached surface scalars — convenient for backtests that only need
# the lowest level without unpacking the whole profile.
field :surface_temp_k, :float
field :surface_spfh, :float
field :surface_pressure_pa, :float
# Derived (Phase 2/4).
field :inversion_top_m, :float
field :bulk_richardson, :float
field :theta_e_jump_k, :float
field :shear_at_top_ms, :float
field :ducts, {:array, :map}
field :best_duct_band_ghz, :float
timestamps(type: :utc_datetime)
end
@required ~w(valid_time lat lon)a
@optional ~w(
run_time level_count heights_m temp_k spfh pressure_pa
u_wind_ms v_wind_ms tke_m2s2
surface_temp_k surface_spfh surface_pressure_pa
inversion_top_m bulk_richardson theta_e_jump_k shear_at_top_ms
ducts best_duct_band_ghz
)a
def changeset(profile, attrs) do
profile
|> cast(attrs, @required ++ @optional)
|> validate_required(@required)
|> validate_array_lengths()
|> unique_constraint([:lat, :lon, :valid_time])
end
# All level arrays must have the same length, matching level_count.
defp validate_array_lengths(changeset) do
level_count = get_field(changeset, :level_count)
arrays =
[:heights_m, :temp_k, :spfh, :pressure_pa, :u_wind_ms, :v_wind_ms, :tke_m2s2]
|> Enum.map(&{&1, get_field(changeset, &1)})
|> Enum.reject(fn {_k, v} -> is_nil(v) end)
expected = level_count || (arrays |> Enum.map(fn {_, v} -> length(v) end) |> Enum.max(fn -> 0 end))
Enum.reduce(arrays, changeset, fn {key, values}, acc ->
if length(values) == expected do
acc
else
add_error(acc, key, "length #{length(values)} does not match expected #{expected}")
end
end)
end
end

164
lib/microwaveprop/workers/hrrr_native_grid_worker.ex Normal file
View file

@ -0,0 +1,164 @@
defmodule Microwaveprop.Workers.HrrrNativeGridWorker do
@moduledoc """
Fetches one hour's worth of HRRR native hybrid-sigma profiles for
every point of interest (contact location) and bulk-inserts them
into `hrrr_native_profiles`.
Jobs are unique on `{year, month, day, hour}` so a backfill sweep
that enqueues duplicate hours collapses automatically.
This is the batch companion to the existing `HrrrFetchWorker`. The
native-level product is ~566 MB per run hour, so per-point
on-demand fetching is impractical — see
`docs/research/hrrr_native_levels.md`. Instead, each job grabs one
file once and extracts native profiles for every point of interest
in a single pass.
"""
use Oban.Worker,
queue: :hrrr,
max_attempts: 3,
unique: [
period: :infinity,
states: [:available, :scheduled, :executing, :retryable],
keys: [:year, :month, :day, :hour]
]
import Ecto.Query
alias Microwaveprop.Radio.Contact
alias Microwaveprop.Repo
alias Microwaveprop.Weather.Grib2.Extractor
alias Microwaveprop.Weather.HrrrClient
alias Microwaveprop.Weather.HrrrNativeClient
alias Microwaveprop.Weather.HrrrNativeProfile
require Logger
@impl Oban.Worker
def perform(%Oban.Job{args: args}) do
%{"year" => year, "month" => month, "day" => day, "hour" => hour} = args
{:ok, date} = Date.new(year, month, day)
{:ok, valid_time} = DateTime.new(date, Time.new!(hour, 0, 0), "Etc/UTC")
points = points_of_interest_for_hour(valid_time)
cond do
points == [] ->
Logger.info("HrrrNativeGridWorker: no points for #{valid_time}, skipping")
:ok
already_ingested?(points, valid_time) ->
Logger.info("HrrrNativeGridWorker: #{valid_time} already ingested, skipping")
:ok
true ->
fetch_and_upsert(date, hour, valid_time, points)
end
end
@doc false
def points_of_interest_for_hour(valid_time) do
time_start = DateTime.add(valid_time, -1800, :second)
time_end = DateTime.add(valid_time, 1800, :second)
Contact
|> where([c], not is_nil(c.pos1))
|> where([c], c.qso_timestamp >= ^time_start and c.qso_timestamp <= ^time_end)
|> select([c], c.pos1)
|> Repo.all()
|> Enum.flat_map(fn pos ->
case {pos["lat"], pos["lon"] || pos["lng"]} do
{lat, lon} when is_number(lat) and is_number(lon) -> [{snap(lat), snap(lon)}]
_ -> []
end
end)
|> Enum.uniq()
end
defp snap(x), do: Float.round(x * 1.0, 3)
defp already_ingested?(points, valid_time) do
# Short-circuit: if we already have a profile for every point at
# this valid_time, the worker is a no-op. Any missing point
# triggers a full download (partial fills are rare; the cost of
# re-downloading for a handful of gaps is acceptable).
count =
HrrrNativeProfile
|> where([p], p.valid_time == ^valid_time)
|> where(
[p],
fragment("ROW(?, ?) IN (SELECT * FROM UNNEST(?::float[], ?::float[]))",
p.lat,
p.lon,
^Enum.map(points, &elem(&1, 0)),
^Enum.map(points, &elem(&1, 1))
)
)
|> select([p], count(p.id))
|> Repo.one()
count >= length(points)
end
defp fetch_and_upsert(date, hour, valid_time, points) do
url = HrrrNativeClient.hrrr_native_url(date, hour)
idx_url = url <> ".idx"
with {:ok, idx_text} <- fetch_idx(idx_url),
idx_entries = HrrrClient.parse_idx(idx_text),
ranges = HrrrNativeClient.essential_byte_ranges(idx_entries),
_ = Logger.info("HRRR native downloading #{length(ranges)} ranges for #{valid_time}"),
{:ok, grib_binary} <- HrrrClient.download_grib_ranges(url, ranges),
_ = Logger.info("HRRR native downloaded #{byte_size(grib_binary)} bytes"),
{:ok, grid_data} <- Extractor.extract_grid(grib_binary, points) do
rows =
grid_data
|> Enum.flat_map(fn {{lat, lon}, parsed} ->
profile = HrrrNativeClient.build_native_profile(parsed)
if profile.level_count > 0 do
now = DateTime.truncate(DateTime.utc_now(), :second)
[
profile
|> Map.merge(%{
id: Ecto.UUID.bingenerate(),
valid_time: valid_time,
run_time: valid_time,
lat: lat,
lon: lon,
inserted_at: now,
updated_at: now
})
]
else
[]
end
end)
{inserted, _} =
Repo.insert_all(
HrrrNativeProfile,
rows,
on_conflict: {:replace_all_except, [:id, :inserted_at]},
conflict_target: [:lat, :lon, :valid_time]
)
Logger.info("HrrrNativeGridWorker: upserted #{inserted} profiles for #{valid_time}")
:ok
else
{:error, reason} ->
Logger.warning("HrrrNativeGridWorker failed for #{valid_time}: #{inspect(reason)}")
{:error, reason}
end
end
defp fetch_idx(url) do
case Req.get(url, receive_timeout: 120_000) do
{:ok, %{status: 200, body: body}} -> {:ok, body}
{:ok, %{status: status}} -> {:error, "HRRR native idx HTTP #{status}"}
{:error, reason} -> {:error, reason}
end
end
end

69
lib/mix/tasks/hrrr_native_backfill.ex Normal file
View file

@ -0,0 +1,69 @@
defmodule Mix.Tasks.HrrrNativeBackfill do
@shortdoc "Enqueue HrrrNativeGridWorker jobs for the top-N hours by contact count"
@moduledoc """
Backfills the `hrrr_native_profiles` table by enqueueing one
`HrrrNativeGridWorker` job per distinct `(year, month, day, hour)`
where we have contacts, prioritized by contact count so the most
data-dense hours land first.
Native HRRR files are ~566 MB each, so think about the total
bandwidth before running with a large limit. Typical plan:
mix hrrr_native_backfill --limit 50 # one-time smoke backfill
mix hrrr_native_backfill --limit 500 # full Phase 2 backfill (~280 GB)
Jobs are deduplicated by Oban's unique constraint on
`{year, month, day, hour}` so running the task twice is safe.
"""
use Mix.Task
import Ecto.Query
alias Microwaveprop.Radio.Contact
alias Microwaveprop.Repo
alias Microwaveprop.Workers.HrrrNativeGridWorker
@impl Mix.Task
def run(argv) do
Mix.Task.run("app.start")
{opts, _, _} = OptionParser.parse(argv, switches: [limit: :integer])
limit = Keyword.get(opts, :limit, 50)
hours = top_hours_by_contact_count(limit)
Mix.shell().info("Enqueueing #{length(hours)} HrrrNativeGridWorker jobs")
Enum.each(hours, fn %{year: y, month: m, day: d, hour: h, contacts: n} ->
args = %{"year" => y, "month" => m, "day" => d, "hour" => h}
case Oban.insert(HrrrNativeGridWorker.new(args)) do
{:ok, _job} -> Mix.shell().info(" #{y}-#{pad(m)}-#{pad(d)} #{pad(h)}Z (#{n} contacts)")
{:error, reason} -> Mix.shell().error(" #{y}-#{pad(m)}-#{pad(d)} #{pad(h)}Z failed: #{inspect(reason)}")
end
end)
end
defp pad(n), do: n |> Integer.to_string() |> String.pad_leading(2, "0")
defp top_hours_by_contact_count(limit) do
from(c in Contact,
where: not is_nil(c.pos1),
select: %{
year: fragment("EXTRACT(YEAR FROM ?)::int", c.qso_timestamp),
month: fragment("EXTRACT(MONTH FROM ?)::int", c.qso_timestamp),
day: fragment("EXTRACT(DAY FROM ?)::int", c.qso_timestamp),
hour: fragment("EXTRACT(HOUR FROM ?)::int", c.qso_timestamp),
contacts: count(c.id)
},
group_by: [
fragment("EXTRACT(YEAR FROM ?)::int", c.qso_timestamp),
fragment("EXTRACT(MONTH FROM ?)::int", c.qso_timestamp),
fragment("EXTRACT(DAY FROM ?)::int", c.qso_timestamp),
fragment("EXTRACT(HOUR FROM ?)::int", c.qso_timestamp)
],
order_by: [desc: count(c.id)],
limit: ^limit
)
|> Repo.all()
end
end

45
priv/repo/migrations/20260409211647_create_hrrr_native_profiles.exs Normal file
View file

@ -0,0 +1,45 @@
defmodule Microwaveprop.Repo.Migrations.CreateHrrrNativeProfiles do
use Ecto.Migration
def change do
create table(:hrrr_native_profiles, primary_key: false) do
add :id, :binary_id, primary_key: true, null: false
add :valid_time, :utc_datetime, null: false
add :run_time, :utc_datetime
add :lat, :float, null: false
add :lon, :float, null: false
# Hybrid-sigma level data, sorted ascending by height (level 1 first).
# Each array has the same length (HRRR native has 50 hybrid levels);
# leaving them as arrays avoids a sidecar table and keeps per-profile
# queries cheap.
add :level_count, :integer
add :heights_m, {:array, :float}
add :temp_k, {:array, :float}
add :spfh, {:array, :float}
add :pressure_pa, {:array, :float}
add :u_wind_ms, {:array, :float}
add :v_wind_ms, {:array, :float}
add :tke_m2s2, {:array, :float}
# Cached surface-level scalars for quick lookup without unpacking arrays.
add :surface_temp_k, :float
add :surface_spfh, :float
add :surface_pressure_pa, :float
# Derived fields populated by Phase 2/4. Nullable so the schema
# migration stays additive; the derive task fills them in later.
add :inversion_top_m, :float
add :bulk_richardson, :float
add :theta_e_jump_k, :float
add :shear_at_top_ms, :float
add :ducts, :map
add :best_duct_band_ghz, :float
timestamps(type: :utc_datetime)
end
create unique_index(:hrrr_native_profiles, [:lat, :lon, :valid_time])
create index(:hrrr_native_profiles, [:valid_time])
end
end

167
test/microwaveprop/weather/hrrr_native_client_test.exs Normal file
View file

@ -0,0 +1,167 @@
defmodule Microwaveprop.Weather.HrrrNativeClientTest do
use ExUnit.Case, async: true
alias Microwaveprop.Weather.HrrrClient
alias Microwaveprop.Weather.HrrrNativeClient
describe "hrrr_native_url/3" do
test "builds the wrfnatf00 URL for a given date and hour" do
url = HrrrNativeClient.hrrr_native_url(~D[2026-04-09], 12)
assert url ==
"https://noaa-hrrr-bdp-pds.s3.amazonaws.com/hrrr.20260409/conus/hrrr.t12z.wrfnatf00.grib2"
end
test "pads single-digit hours" do
url = HrrrNativeClient.hrrr_native_url(~D[2026-04-09], 6)
assert url =~ "hrrr.t06z."
end
test "supports non-zero forecast hours" do
url = HrrrNativeClient.hrrr_native_url(~D[2026-04-09], 12, 3)
assert url =~ "wrfnatf03.grib2"
end
end
describe "native_messages/0" do
test "lists 350 messages (7 vars × 50 levels)" do
messages = HrrrNativeClient.native_messages()
assert length(messages) == 350
end
test "every level appears with every essential variable" do
messages = HrrrNativeClient.native_messages()
# Pick level 5 — should have all 7 vars
level5 = Enum.filter(messages, fn %{level: l} -> l == "5 hybrid level" end)
vars = Enum.map(level5, & &1.var) |> Enum.sort()
assert vars == ~w(HGT PRES SPFH TKE TMP UGRD VGRD)
end
test "covers levels 1 through 50" do
levels =
HrrrNativeClient.native_messages()
|> Enum.map(& &1.level)
|> Enum.uniq()
|> Enum.map(fn level_str ->
[digits, _] = String.split(level_str, " ", parts: 2)
String.to_integer(digits)
end)
|> Enum.sort()
assert levels == Enum.to_list(1..50)
end
end
describe "build_native_profile/1" do
test "assembles arrays in level order and caches surface scalars" do
parsed = %{
"HGT:1 hybrid level" => 8.0,
"TMP:1 hybrid level" => 295.0,
"SPFH:1 hybrid level" => 0.010,
"PRES:1 hybrid level" => 101_000.0,
"UGRD:1 hybrid level" => 2.0,
"VGRD:1 hybrid level" => 1.0,
"TKE:1 hybrid level" => 0.5,
"HGT:2 hybrid level" => 25.0,
"TMP:2 hybrid level" => 293.0,
"SPFH:2 hybrid level" => 0.008,
"PRES:2 hybrid level" => 99_800.0,
"UGRD:2 hybrid level" => 3.0,
"VGRD:2 hybrid level" => 1.5,
"TKE:2 hybrid level" => 0.3,
"TMP:surface" => 295.5,
"SPFH:2 m above ground" => 0.011,
"PRES:surface" => 101_325.0
}
profile = HrrrNativeClient.build_native_profile(parsed)
assert profile.level_count == 2
assert profile.heights_m == [8.0, 25.0]
assert profile.temp_k == [295.0, 293.0]
assert profile.spfh == [0.010, 0.008]
assert profile.u_wind_ms == [2.0, 3.0]
assert profile.surface_temp_k == 295.5
assert profile.surface_spfh == 0.011
assert profile.surface_pressure_pa == 101_325.0
end
test "skips levels missing either HGT or TMP" do
parsed = %{
"HGT:1 hybrid level" => 8.0,
"TMP:1 hybrid level" => 295.0,
# level 2 has HGT but no TMP → skipped
"HGT:2 hybrid level" => 25.0,
"SPFH:2 hybrid level" => 0.008
}
profile = HrrrNativeClient.build_native_profile(parsed)
assert profile.level_count == 1
assert profile.heights_m == [8.0]
end
test "orders levels by ascending height, not by hybrid level number" do
# Pathological: level 2 happens to be lower than level 1 (terrain).
parsed = %{
"HGT:1 hybrid level" => 50.0,
"TMP:1 hybrid level" => 290.0,
"HGT:2 hybrid level" => 30.0,
"TMP:2 hybrid level" => 292.0
}
profile = HrrrNativeClient.build_native_profile(parsed)
assert profile.heights_m == [30.0, 50.0]
assert profile.temp_k == [292.0, 290.0]
end
test "falls back to lowest level for surface scalars when parsed has no surface entries" do
parsed = %{
"HGT:1 hybrid level" => 8.0,
"TMP:1 hybrid level" => 295.0,
"SPFH:1 hybrid level" => 0.010,
"PRES:1 hybrid level" => 101_000.0
}
profile = HrrrNativeClient.build_native_profile(parsed)
assert profile.surface_temp_k == 295.0
assert profile.surface_spfh == 0.010
assert profile.surface_pressure_pa == 101_000.0
end
end
describe "essential_byte_ranges/1" do
test "produces one range per essential message present in the idx" do
# Minimal fake idx: level 1 TMP/SPFH/HGT with nothing else.
idx_entries = [
%{msg: 1, offset: 0, var: "TMP", level: "1 hybrid level"},
%{msg: 2, offset: 1000, var: "SPFH", level: "1 hybrid level"},
%{msg: 3, offset: 2000, var: "HGT", level: "1 hybrid level"},
%{msg: 4, offset: 3000, var: "REFC", level: "entire atmosphere"}
]
ranges = HrrrNativeClient.essential_byte_ranges(idx_entries)
# Expect 3 ranges (the three essentials we actually have)
assert length(ranges) == 3
# Each range is {start, end} tuples, all integers
Enum.each(ranges, fn {s, e} ->
assert is_integer(s) and is_integer(e)
assert s < e
end)
end
test "delegates to HrrrClient.byte_ranges_for_messages/2" do
# Just check the two stay in sync by computing the same ranges
# two different ways.
idx_entries = [
%{msg: 1, offset: 0, var: "TMP", level: "1 hybrid level"},
%{msg: 2, offset: 5000, var: "UGRD", level: "1 hybrid level"}
]
assert HrrrNativeClient.essential_byte_ranges(idx_entries) ==
HrrrClient.byte_ranges_for_messages(idx_entries, HrrrNativeClient.native_messages())
end
end
end

59
test/microwaveprop/weather/hrrr_native_profile_test.exs Normal file
View file

@ -0,0 +1,59 @@
defmodule Microwaveprop.Weather.HrrrNativeProfileTest do
use Microwaveprop.DataCase, async: true
alias Microwaveprop.Weather.HrrrNativeProfile
@valid_attrs %{
valid_time: ~U[2026-03-28 18:00:00Z],
run_time: ~U[2026-03-28 18:00:00Z],
lat: 32.9,
lon: -97.0,
level_count: 3,
heights_m: [10.0, 100.0, 500.0],
temp_k: [295.0, 292.0, 285.0],
spfh: [0.010, 0.008, 0.004],
pressure_pa: [101_000.0, 99_800.0, 95_000.0],
u_wind_ms: [2.0, 3.0, 4.0],
v_wind_ms: [1.0, 1.5, 2.0],
tke_m2s2: [0.5, 0.3, 0.1],
surface_temp_k: 295.0,
surface_spfh: 0.010,
surface_pressure_pa: 101_325.0
}
describe "changeset/2" do
test "accepts a full native profile" do
changeset = HrrrNativeProfile.changeset(%HrrrNativeProfile{}, @valid_attrs)
assert changeset.valid?
assert {:ok, profile} = Repo.insert(changeset)
assert profile.level_count == 3
assert profile.heights_m == [10.0, 100.0, 500.0]
end
test "requires valid_time, lat, lon" do
changeset = HrrrNativeProfile.changeset(%HrrrNativeProfile{}, %{})
refute changeset.valid?
assert "can't be blank" in errors_on(changeset).valid_time
assert "can't be blank" in errors_on(changeset).lat
assert "can't be blank" in errors_on(changeset).lon
end
test "rejects mismatched level array lengths" do
attrs = Map.put(@valid_attrs, :temp_k, [295.0, 292.0])
changeset = HrrrNativeProfile.changeset(%HrrrNativeProfile{}, attrs)
refute changeset.valid?
assert errors_on(changeset)[:temp_k] != nil
end
test "enforces (lat, lon, valid_time) uniqueness" do
{:ok, _} = %HrrrNativeProfile{} |> HrrrNativeProfile.changeset(@valid_attrs) |> Repo.insert()
{:error, changeset} =
%HrrrNativeProfile{} |> HrrrNativeProfile.changeset(@valid_attrs) |> Repo.insert()
refute changeset.valid?
assert "has already been taken" in errors_on(changeset).lat
end
end
end

78
test/microwaveprop/workers/hrrr_native_grid_worker_test.exs Normal file
View file

@ -0,0 +1,78 @@
defmodule Microwaveprop.Workers.HrrrNativeGridWorkerTest do
use Microwaveprop.DataCase, async: true
alias Microwaveprop.Radio.Contact
alias Microwaveprop.Workers.HrrrNativeGridWorker
defp create_contact(attrs) do
default = %{
station1: "W5XD",
station2: "K5TR",
qso_timestamp: ~U[2026-03-28 18:00:00Z],
mode: "CW",
band: Decimal.new("10000"),
grid1: "EM12",
grid2: "EM00",
pos1: %{"lat" => 32.9, "lon" => -97.0},
pos2: %{"lat" => 30.3, "lon" => -97.7},
distance_km: Decimal.new("295")
}
{:ok, contact} =
%Contact{}
|> Contact.changeset(Map.merge(default, attrs))
|> Repo.insert()
contact
end
describe "points_of_interest_for_hour/1" do
test "returns snapped distinct (lat, lon) pairs for contacts in the ±30 min window" do
create_contact(%{pos1: %{"lat" => 32.9, "lon" => -97.0}, qso_timestamp: ~U[2026-03-28 17:45:00Z]})
create_contact(%{pos1: %{"lat" => 33.0, "lon" => -97.1}, qso_timestamp: ~U[2026-03-28 18:15:00Z]})
points = HrrrNativeGridWorker.points_of_interest_for_hour(~U[2026-03-28 18:00:00Z])
assert length(points) == 2
assert {32.9, -97.0} in points
assert {33.0, -97.1} in points
end
test "excludes contacts outside the window" do
create_contact(%{qso_timestamp: ~U[2026-03-28 17:00:00Z]})
points = HrrrNativeGridWorker.points_of_interest_for_hour(~U[2026-03-28 19:00:00Z])
assert points == []
end
test "deduplicates coincident positions" do
create_contact(%{station1: "A", pos1: %{"lat" => 32.9, "lon" => -97.0}})
create_contact(%{station1: "B", pos1: %{"lat" => 32.9, "lon" => -97.0}})
points = HrrrNativeGridWorker.points_of_interest_for_hour(~U[2026-03-28 18:00:00Z])
assert points == [{32.9, -97.0}]
end
test "handles contacts with pos1 stored as 'lng'" do
create_contact(%{pos1: %{"lat" => 32.9, "lng" => -97.0}})
points = HrrrNativeGridWorker.points_of_interest_for_hour(~U[2026-03-28 18:00:00Z])
assert points == [{32.9, -97.0}]
end
end
describe "perform/1" do
test "returns :ok without making any HTTP request when no contacts exist for the hour" do
# No contacts at all → no points → no download
job =
HrrrNativeGridWorker.new(%{
"year" => 2026,
"month" => 3,
"day" => 28,
"hour" => 18
})
assert {:ok, %Oban.Job{args: _}} = Oban.insert(job)
end
end
end