Graham McIntire
01b181b1e8
Four changes sized by measured prod telemetry (83m of spans): 1. propagation queue: 2 → 1 slot per pod. Two concurrent forecast-hour steps per pod stacked HRRR grid + native duct grid + scored band map into ~5-6 GiB RSS, OOM-killing every ~15 min. 3-way parallelism cluster-wide still finishes the chain inside the hourly interval. 2. weather queue: 3 → 1 slot per pod. ASOS backfill was 429-thrashing IEM (1,296 retryable jobs; logs were nothing but 429 backoffs). 3. PropagationGridWorker: skip native-level duct fetch on f01..f18. At ~7-11 min/fh and 18 forecast hours, this was the largest single cost per chain. Forecast hours fall back to derived[:min_refractivity_gradient] from the pressure-level profile. f00 still gets full native-level duct analysis. 4. HrrrClient.download_grib_ranges_to_file: parallelize with Task.async_stream (max_concurrency 8). The file-backed variant was sequential, dominating native-duct fetch time on the remaining f00 path. ~20s → ~3s per call.
503 lines
16 KiB
Elixir
503 lines
16 KiB
Elixir
defmodule Microwaveprop.Weather.HrrrClientTest do
|
||
use ExUnit.Case, async: true
|
||
|
||
alias Microwaveprop.Weather.HrrrClient
|
||
|
||
describe "nearest_hrrr_hour/1" do
|
||
test "rounds down when within first 30 minutes" do
|
||
assert HrrrClient.nearest_hrrr_hour(~U[2026-03-28 18:15:00Z]) ==
|
||
~U[2026-03-28 18:00:00Z]
|
||
end
|
||
|
||
test "rounds up when past 30 minutes" do
|
||
assert HrrrClient.nearest_hrrr_hour(~U[2026-03-28 18:45:00Z]) ==
|
||
~U[2026-03-28 19:00:00Z]
|
||
end
|
||
|
||
test "stays same when exactly on the hour" do
|
||
assert HrrrClient.nearest_hrrr_hour(~U[2026-03-28 18:00:00Z]) ==
|
||
~U[2026-03-28 18:00:00Z]
|
||
end
|
||
|
||
test "rounds at exactly 30 minutes" do
|
||
result = HrrrClient.nearest_hrrr_hour(~U[2026-03-28 18:30:00Z])
|
||
assert result in [~U[2026-03-28 18:00:00Z], ~U[2026-03-28 19:00:00Z]]
|
||
end
|
||
end
|
||
|
||
describe "hrrr_url/3" do
|
||
test "builds surface product URL" do
|
||
url = HrrrClient.hrrr_url(~D[2026-03-28], 18, :surface)
|
||
|
||
assert url ==
|
||
"https://noaa-hrrr-bdp-pds.s3.amazonaws.com/hrrr.20260328/conus/hrrr.t18z.wrfsfcf00.grib2"
|
||
end
|
||
|
||
test "builds pressure product URL" do
|
||
url = HrrrClient.hrrr_url(~D[2026-03-28], 6, :pressure)
|
||
|
||
assert url ==
|
||
"https://noaa-hrrr-bdp-pds.s3.amazonaws.com/hrrr.20260328/conus/hrrr.t06z.wrfprsf00.grib2"
|
||
end
|
||
|
||
test "pads single-digit hours" do
|
||
url = HrrrClient.hrrr_url(~D[2026-03-28], 3, :surface)
|
||
assert String.contains?(url, "hrrr.t03z")
|
||
end
|
||
end
|
||
|
||
describe "parse_idx/1" do
|
||
@sample_idx """
|
||
1:0:d=2026032818:TMP:2 m above ground:anl:
|
||
2:1234567:d=2026032818:DPT:2 m above ground:anl:
|
||
3:2345678:d=2026032818:PRES:surface:anl:
|
||
4:3456789:d=2026032818:HPBL:surface:anl:
|
||
5:4567890:d=2026032818:PWAT:entire atmosphere (considered as a single layer):anl:
|
||
6:5678901:d=2026032818:TMP:1000 mb:anl:
|
||
"""
|
||
|
||
test "parses idx into structured list" do
|
||
entries = HrrrClient.parse_idx(@sample_idx)
|
||
|
||
assert length(entries) == 6
|
||
|
||
first = hd(entries)
|
||
assert first.msg == 1
|
||
assert first.offset == 0
|
||
assert first.var == "TMP"
|
||
assert first.level == "2 m above ground"
|
||
end
|
||
|
||
test "correctly extracts offsets" do
|
||
entries = HrrrClient.parse_idx(@sample_idx)
|
||
offsets = Enum.map(entries, & &1.offset)
|
||
assert offsets == [0, 1_234_567, 2_345_678, 3_456_789, 4_567_890, 5_678_901]
|
||
end
|
||
|
||
test "handles empty input" do
|
||
assert HrrrClient.parse_idx("") == []
|
||
end
|
||
|
||
test "skips malformed lines instead of raising" do
|
||
# NOAA S3 occasionally serves an HTML error page as the idx body
|
||
# (e.g. during brief 503 spikes). The old parser would raise
|
||
# ArgumentError on String.to_integer of a non-numeric token,
|
||
# which killed the whole PropagationGridWorker chain step.
|
||
# The tolerant parser returns the well-formed lines and drops
|
||
# the garbage.
|
||
input = """
|
||
<html><body>Service Unavailable</body></html>
|
||
1:0:d=2026032818:TMP:2 m above ground:anl:
|
||
bogus:header:line:with:no:numbers
|
||
2:1234567:d=2026032818:DPT:2 m above ground:anl:
|
||
"""
|
||
|
||
entries = HrrrClient.parse_idx(input)
|
||
|
||
assert length(entries) == 2
|
||
assert Enum.map(entries, & &1.msg) == [1, 2]
|
||
assert Enum.map(entries, & &1.var) == ["TMP", "DPT"]
|
||
end
|
||
end
|
||
|
||
describe "byte_ranges_for_messages/2" do
|
||
test "computes byte ranges from idx entries" do
|
||
entries = [
|
||
%{msg: 1, offset: 0, var: "TMP", level: "2 m above ground"},
|
||
%{msg: 2, offset: 1000, var: "DPT", level: "2 m above ground"},
|
||
%{msg: 3, offset: 2000, var: "PRES", level: "surface"},
|
||
%{msg: 4, offset: 3000, var: "HPBL", level: "surface"}
|
||
]
|
||
|
||
wanted = [
|
||
%{var: "TMP", level: "2 m above ground"},
|
||
%{var: "PRES", level: "surface"}
|
||
]
|
||
|
||
ranges = HrrrClient.byte_ranges_for_messages(entries, wanted)
|
||
|
||
assert length(ranges) == 2
|
||
assert {0, 999} in ranges
|
||
assert {2000, 2999} in ranges
|
||
end
|
||
|
||
test "handles last message in file" do
|
||
entries = [
|
||
%{msg: 1, offset: 0, var: "TMP", level: "2 m above ground"},
|
||
%{msg: 2, offset: 1000, var: "DPT", level: "2 m above ground"}
|
||
]
|
||
|
||
wanted = [%{var: "DPT", level: "2 m above ground"}]
|
||
|
||
ranges = HrrrClient.byte_ranges_for_messages(entries, wanted)
|
||
|
||
# Last message - uses a large end offset
|
||
assert length(ranges) == 1
|
||
[{start, _end}] = ranges
|
||
assert start == 1000
|
||
end
|
||
end
|
||
|
||
describe "fetch_profile/3" do
|
||
test "succeeds when ranges are downloaded individually" do
|
||
grib_data = File.read!("test/fixtures/grib2/hrrr_tmp_2m.grib2")
|
||
|
||
# Idx with multiple surface variables to trigger multi-range download
|
||
grib_size = byte_size(grib_data)
|
||
|
||
idx_text = """
|
||
1:0:d=2026032818:TMP:2 m above ground:anl:
|
||
2:#{grib_size}:d=2026032818:DPT:2 m above ground:anl:
|
||
3:#{grib_size * 2}:d=2026032818:PRES:surface:anl:
|
||
"""
|
||
|
||
Req.Test.stub(HrrrClient, fn conn ->
|
||
if String.ends_with?(conn.request_path, ".idx") do
|
||
Plug.Conn.send_resp(conn, 200, idx_text)
|
||
else
|
||
[range] = Plug.Conn.get_req_header(conn, "range")
|
||
|
||
if String.contains?(range, ",") do
|
||
# Multi-range: S3 ignores Range header, returns full file
|
||
Plug.Conn.send_resp(conn, 200, "full file")
|
||
else
|
||
# Single range: S3 supports this fine
|
||
Plug.Conn.send_resp(conn, 206, grib_data)
|
||
end
|
||
end
|
||
end)
|
||
|
||
assert {:ok, profile} = HrrrClient.fetch_profile(32.90, -97.04, ~U[2026-03-28 18:00:00Z])
|
||
assert is_float(profile.surface_temp_c)
|
||
assert is_struct(profile.run_time, DateTime)
|
||
end
|
||
end
|
||
|
||
describe "build_profile/1" do
|
||
test "assembles profile from parsed data" do
|
||
parsed = %{
|
||
"TMP:1000 mb" => 298.0,
|
||
"DPT:1000 mb" => 291.0,
|
||
"HGT:1000 mb" => 110.0,
|
||
"TMP:975 mb" => 296.0,
|
||
"DPT:975 mb" => 289.0,
|
||
"HGT:975 mb" => 350.0,
|
||
"TMP:950 mb" => 294.0,
|
||
"DPT:950 mb" => 287.0,
|
||
"HGT:950 mb" => 590.0,
|
||
"TMP:2 m above ground" => 299.0,
|
||
"DPT:2 m above ground" => 292.0,
|
||
"PRES:surface" => 101_350.0,
|
||
"HPBL:surface" => 1500.0,
|
||
"PWAT:entire atmosphere (considered as a single layer)" => 25.0
|
||
}
|
||
|
||
result = HrrrClient.build_profile(parsed)
|
||
|
||
assert result.surface_temp_c == 299.0 - 273.15
|
||
assert result.surface_dewpoint_c == 292.0 - 273.15
|
||
assert_in_delta result.surface_pressure_mb, 1013.5, 0.1
|
||
assert result.hpbl_m == 1500.0
|
||
assert result.pwat_mm == 25.0
|
||
assert length(result.profile) == 3
|
||
|
||
first_level = hd(result.profile)
|
||
assert first_level["pres"] == 1000.0
|
||
assert first_level["tmpc"] == 298.0 - 273.15
|
||
assert first_level["dwpc"] == 291.0 - 273.15
|
||
assert first_level["hght"] == 110.0
|
||
end
|
||
|
||
test "includes wind, cloud cover, and precip fields" do
|
||
parsed = %{
|
||
"TMP:2 m above ground" => 299.0,
|
||
"DPT:2 m above ground" => 292.0,
|
||
"PRES:surface" => 101_350.0,
|
||
"HPBL:surface" => 1500.0,
|
||
"PWAT:entire atmosphere (considered as a single layer)" => 25.0,
|
||
"UGRD:10 m above ground" => 3.5,
|
||
"VGRD:10 m above ground" => -2.1,
|
||
"TCDC:entire atmosphere" => 75.0,
|
||
"APCP:surface" => 1.2
|
||
}
|
||
|
||
result = HrrrClient.build_profile(parsed)
|
||
|
||
assert result.wind_u == 3.5
|
||
assert result.wind_v == -2.1
|
||
assert result.cloud_cover_pct == 75.0
|
||
assert result.precip_mm == 1.2
|
||
end
|
||
|
||
test "wind, cloud, and precip fields are nil when missing" do
|
||
parsed = %{
|
||
"TMP:2 m above ground" => 299.0,
|
||
"DPT:2 m above ground" => 292.0,
|
||
"PRES:surface" => 101_350.0,
|
||
"HPBL:surface" => 1500.0,
|
||
"PWAT:entire atmosphere (considered as a single layer)" => 25.0
|
||
}
|
||
|
||
result = HrrrClient.build_profile(parsed)
|
||
|
||
assert result.wind_u == nil
|
||
assert result.wind_v == nil
|
||
assert result.cloud_cover_pct == nil
|
||
assert result.precip_mm == nil
|
||
end
|
||
|
||
test "skips pressure levels with missing data" do
|
||
parsed = %{
|
||
"TMP:1000 mb" => 298.0,
|
||
"DPT:1000 mb" => 291.0,
|
||
"HGT:1000 mb" => 110.0,
|
||
# 975 mb missing HGT
|
||
"TMP:975 mb" => 296.0,
|
||
"DPT:975 mb" => 289.0,
|
||
"TMP:2 m above ground" => 299.0,
|
||
"DPT:2 m above ground" => 292.0,
|
||
"PRES:surface" => 101_350.0,
|
||
"HPBL:surface" => 1500.0,
|
||
"PWAT:entire atmosphere (considered as a single layer)" => 25.0
|
||
}
|
||
|
||
result = HrrrClient.build_profile(parsed)
|
||
assert length(result.profile) == 1
|
||
end
|
||
|
||
test "includes upper-air levels above 700 mb for the skew-T plot" do
|
||
parsed = %{
|
||
"TMP:700 mb" => 275.0,
|
||
"DPT:700 mb" => 265.0,
|
||
"HGT:700 mb" => 3100.0,
|
||
"TMP:500 mb" => 253.0,
|
||
"DPT:500 mb" => 238.0,
|
||
"HGT:500 mb" => 5800.0,
|
||
"TMP:300 mb" => 228.0,
|
||
"DPT:300 mb" => 200.0,
|
||
"HGT:300 mb" => 9400.0,
|
||
"TMP:100 mb" => 205.0,
|
||
"DPT:100 mb" => 180.0,
|
||
"HGT:100 mb" => 16_300.0,
|
||
"TMP:2 m above ground" => 299.0,
|
||
"DPT:2 m above ground" => 292.0,
|
||
"PRES:surface" => 101_350.0,
|
||
"HPBL:surface" => 1500.0,
|
||
"PWAT:entire atmosphere (considered as a single layer)" => 25.0
|
||
}
|
||
|
||
result = HrrrClient.build_profile(parsed)
|
||
pressures = Enum.map(result.profile, & &1["pres"])
|
||
|
||
assert 500.0 in pressures
|
||
assert 300.0 in pressures
|
||
assert 100.0 in pressures
|
||
|
||
level_500 = Enum.find(result.profile, &(&1["pres"] == 500.0))
|
||
assert_in_delta level_500["tmpc"], 253.0 - 273.15, 0.001
|
||
assert_in_delta level_500["dwpc"], 238.0 - 273.15, 0.001
|
||
assert level_500["hght"] == 5800.0
|
||
end
|
||
end
|
||
|
||
describe "pressure_messages/1" do
|
||
test ":grid returns only the near-surface levels needed for scoring" do
|
||
messages = HrrrClient.pressure_messages(:grid)
|
||
|
||
# 13 levels × 3 vars (TMP/DPT/HGT)
|
||
assert length(messages) == 39
|
||
|
||
levels =
|
||
messages
|
||
|> Enum.map(fn m ->
|
||
[num_str, _] = String.split(m.level, " ")
|
||
String.to_integer(num_str)
|
||
end)
|
||
|> Enum.uniq()
|
||
|> Enum.sort()
|
||
|
||
# Narrow list: 1000 → 700 mb, every 25 mb
|
||
assert Enum.max(levels) == 1000
|
||
assert Enum.min(levels) == 700
|
||
assert 700 in levels
|
||
assert 900 in levels
|
||
|
||
# Upper-air levels must be absent — they're the whole reason the grid
|
||
# worker OOMs when this function returns the full profile list.
|
||
refute 500 in levels
|
||
refute 300 in levels
|
||
refute 100 in levels
|
||
end
|
||
|
||
test ":profile returns full troposphere + lower stratosphere for the skew-T" do
|
||
messages = HrrrClient.pressure_messages(:profile)
|
||
|
||
# 25 levels × 3 vars
|
||
assert length(messages) == 75
|
||
|
||
levels =
|
||
messages
|
||
|> Enum.map(fn m ->
|
||
[num_str, _] = String.split(m.level, " ")
|
||
String.to_integer(num_str)
|
||
end)
|
||
|> Enum.uniq()
|
||
|> Enum.sort()
|
||
|
||
assert Enum.max(levels) == 1000
|
||
assert Enum.min(levels) == 100
|
||
assert 500 in levels
|
||
assert 300 in levels
|
||
assert 100 in levels
|
||
end
|
||
|
||
test ":grid and :profile both request TMP, DPT and HGT at every level" do
|
||
for variant <- [:grid, :profile] do
|
||
messages = HrrrClient.pressure_messages(variant)
|
||
vars = messages |> Enum.map(& &1.var) |> Enum.uniq() |> Enum.sort()
|
||
assert vars == ["DPT", "HGT", "TMP"]
|
||
end
|
||
end
|
||
end
|
||
|
||
describe "surface_messages/0" do
|
||
test "returns list of surface message descriptors" do
|
||
messages = HrrrClient.surface_messages()
|
||
|
||
assert is_list(messages)
|
||
assert length(messages) == 9
|
||
|
||
vars = Enum.map(messages, & &1.var)
|
||
assert "TMP" in vars
|
||
assert "DPT" in vars
|
||
assert "PRES" in vars
|
||
assert "HPBL" in vars
|
||
assert "PWAT" in vars
|
||
assert "UGRD" in vars
|
||
assert "VGRD" in vars
|
||
assert "TCDC" in vars
|
||
assert "APCP" in vars
|
||
end
|
||
|
||
test "includes wind messages at 10 m above ground" do
|
||
messages = HrrrClient.surface_messages()
|
||
|
||
ugrd = Enum.find(messages, &(&1.var == "UGRD"))
|
||
assert ugrd.level == "10 m above ground"
|
||
|
||
vgrd = Enum.find(messages, &(&1.var == "VGRD"))
|
||
assert vgrd.level == "10 m above ground"
|
||
end
|
||
|
||
test "includes cloud cover for entire atmosphere" do
|
||
messages = HrrrClient.surface_messages()
|
||
tcdc = Enum.find(messages, &(&1.var == "TCDC"))
|
||
assert tcdc.level == "entire atmosphere"
|
||
end
|
||
|
||
test "includes precipitation at surface" do
|
||
messages = HrrrClient.surface_messages()
|
||
apcp = Enum.find(messages, &(&1.var == "APCP"))
|
||
assert apcp.level == "surface"
|
||
end
|
||
end
|
||
|
||
describe "download_grib_ranges_to_file/3" do
|
||
test "writes bytes to file in offset-ascending order even when fetched out of order" do
|
||
# Two non-adjacent ranges (so they don't get merged). Responses
|
||
# deliberately return with the second range "arriving" first to
|
||
# simulate out-of-order completion under parallel fetches; the
|
||
# final file must still be ordered by offset.
|
||
ranges = [{0, 9}, {100, 119}]
|
||
range_a = String.duplicate("A", 10)
|
||
range_b = String.duplicate("B", 20)
|
||
|
||
Req.Test.stub(HrrrClient, fn conn ->
|
||
[range] = Plug.Conn.get_req_header(conn, "range")
|
||
|
||
case range do
|
||
"bytes=0-9" ->
|
||
Plug.Conn.send_resp(conn, 206, range_a)
|
||
|
||
"bytes=100-119" ->
|
||
Plug.Conn.send_resp(conn, 206, range_b)
|
||
end
|
||
end)
|
||
|
||
tmp = Path.join(System.tmp_dir!(), "hrrr_download_test_#{System.unique_integer([:positive])}.grib2")
|
||
|
||
try do
|
||
assert :ok = HrrrClient.download_grib_ranges_to_file("http://test/grib", ranges, tmp)
|
||
assert File.read!(tmp) == range_a <> range_b
|
||
after
|
||
File.rm(tmp)
|
||
end
|
||
end
|
||
end
|
||
|
||
describe "merge_grid_data/2" do
|
||
test "merges surface and pressure grids by point key" do
|
||
point_a = {32.90, -97.04}
|
||
point_b = {33.10, -96.80}
|
||
|
||
sfc_grid = %{
|
||
point_a => %{
|
||
"TMP:2 m above ground" => 299.0,
|
||
"DPT:2 m above ground" => 292.0,
|
||
"PRES:surface" => 101_350.0,
|
||
"HPBL:surface" => 1500.0,
|
||
"PWAT:entire atmosphere (considered as a single layer)" => 25.0,
|
||
"UGRD:10 m above ground" => 3.5,
|
||
"VGRD:10 m above ground" => -2.1,
|
||
"TCDC:entire atmosphere" => 75.0,
|
||
"APCP:surface" => 1.2
|
||
},
|
||
point_b => %{
|
||
"TMP:2 m above ground" => 297.0,
|
||
"DPT:2 m above ground" => 290.0,
|
||
"PRES:surface" => 101_200.0,
|
||
"HPBL:surface" => 1200.0,
|
||
"PWAT:entire atmosphere (considered as a single layer)" => 22.0,
|
||
"UGRD:10 m above ground" => 2.0,
|
||
"VGRD:10 m above ground" => -1.5,
|
||
"TCDC:entire atmosphere" => 50.0,
|
||
"APCP:surface" => 0.0
|
||
}
|
||
}
|
||
|
||
prs_grid = %{
|
||
point_a => %{
|
||
"TMP:1000 mb" => 298.0,
|
||
"DPT:1000 mb" => 291.0,
|
||
"HGT:1000 mb" => 110.0
|
||
},
|
||
point_b => %{
|
||
"TMP:1000 mb" => 296.0,
|
||
"DPT:1000 mb" => 289.0,
|
||
"HGT:1000 mb" => 105.0
|
||
}
|
||
}
|
||
|
||
merged = HrrrClient.merge_grid_data(sfc_grid, prs_grid)
|
||
|
||
assert map_size(merged) == 2
|
||
assert merged[point_a]["TMP:2 m above ground"] == 299.0
|
||
assert merged[point_a]["TMP:1000 mb"] == 298.0
|
||
assert merged[point_b]["TMP:2 m above ground"] == 297.0
|
||
assert merged[point_b]["TMP:1000 mb"] == 296.0
|
||
end
|
||
|
||
test "handles empty pressure data" do
|
||
point = {32.90, -97.04}
|
||
|
||
sfc_grid = %{
|
||
point => %{"TMP:2 m above ground" => 299.0}
|
||
}
|
||
|
||
merged = HrrrClient.merge_grid_data(sfc_grid, %{})
|
||
|
||
assert map_size(merged) == 1
|
||
assert merged[point]["TMP:2 m above ground"] == 299.0
|
||
end
|
||
end
|
||
end
|