Graham McIntire
ad600361e0
weather_grid_hrdps_at/2 is the only production caller and backs every /weather/cells?source=hrdps request, but it read the legacy chunked .mp.gz dirs exclusively. The pipeline writes <iso>.hrdps.sgrid, so the snap found a valid_time via list_valid_times_hrdps/0 (which does see .sgrid) and then read zero chunks for it. Delegate to read_bounds_hrdps/2, which already does the .sgrid-first, chunk-fallback read. Every existing test for this function hand-wrote legacy chunk dirs, which is how the .sgrid migration slipped past. Added coverage against a real .sgrid artifact, including the mid-migration case where both formats exist for one valid_time.
465 lines
16 KiB
Elixir
465 lines
16 KiB
Elixir
defmodule Microwaveprop.Weather.ScalarFileTest do
|
||
# async: false — mutates the global :propagation_scores_dir env to redirect
|
||
# the on-disk artifact tree to a private tmp directory.
|
||
use ExUnit.Case, async: false
|
||
|
||
alias Microwaveprop.Weather.ScalarFile
|
||
alias Microwaveprop.Weather.Sgrid
|
||
|
||
setup do
|
||
dir =
|
||
Path.join(
|
||
System.tmp_dir!(),
|
||
"weather_scalar_file_test_#{System.unique_integer([:positive])}"
|
||
)
|
||
|
||
File.mkdir_p!(dir)
|
||
|
||
prev = Application.get_env(:microwaveprop, :propagation_scores_dir)
|
||
Application.put_env(:microwaveprop, :propagation_scores_dir, dir)
|
||
|
||
on_exit(fn ->
|
||
File.rm_rf!(dir)
|
||
|
||
if prev do
|
||
Application.put_env(:microwaveprop, :propagation_scores_dir, prev)
|
||
else
|
||
Application.delete_env(:microwaveprop, :propagation_scores_dir)
|
||
end
|
||
end)
|
||
|
||
%{dir: dir}
|
||
end
|
||
|
||
defp sample_row(lat, lon, temp \\ 22.0) do
|
||
%{
|
||
lat: lat,
|
||
lon: lon,
|
||
valid_time: ~U[2026-04-28 12:00:00Z],
|
||
temperature: temp,
|
||
dewpoint_depression: 5.0,
|
||
surface_rh: 70.0,
|
||
surface_pressure_mb: 1013.0,
|
||
surface_refractivity: 330.0,
|
||
refractivity_gradient: -45.0,
|
||
bl_height: 800.0,
|
||
pwat: 25.0,
|
||
temp_850mb: 12.0,
|
||
dewpoint_850mb: 4.0,
|
||
temp_700mb: 0.0,
|
||
dewpoint_700mb: -10.0,
|
||
lapse_rate: 6.5,
|
||
mid_lapse_rate: 6.0,
|
||
inversion_strength: 0.5,
|
||
inversion_base_m: nil,
|
||
ducting: false,
|
||
duct_base_m: nil,
|
||
duct_strength: nil
|
||
}
|
||
end
|
||
|
||
describe "base_dir/0" do
|
||
test "is rooted under the configured scores dir", %{dir: dir} do
|
||
assert ScalarFile.base_dir() == Path.join(dir, "weather_scalars")
|
||
end
|
||
end
|
||
|
||
describe "write!/2 + read_bounds/2" do
|
||
test "round-trips the rows that fall inside the requested bounds" do
|
||
valid_time = ~U[2026-04-28 12:00:00Z]
|
||
|
||
rows = [
|
||
sample_row(33.0, -97.0, 22.0),
|
||
sample_row(33.5, -97.5, 23.0),
|
||
# Outside Texas bounds below
|
||
sample_row(45.0, -90.0, 5.0)
|
||
]
|
||
|
||
ScalarFile.write!(valid_time, rows)
|
||
|
||
result =
|
||
ScalarFile.read_bounds(valid_time, %{
|
||
"south" => 32.0,
|
||
"north" => 34.5,
|
||
"west" => -98.0,
|
||
"east" => -96.0
|
||
})
|
||
|
||
lat_lons = result |> Enum.map(&{&1.lat, &1.lon}) |> Enum.sort()
|
||
assert lat_lons == [{33.0, -97.0}, {33.5, -97.5}]
|
||
end
|
||
|
||
test "returns [] when the file does not exist" do
|
||
assert ScalarFile.read_bounds(~U[2026-04-28 12:00:00Z], nil) == []
|
||
end
|
||
|
||
test "read_bounds with nil bounds returns every persisted row" do
|
||
valid_time = ~U[2026-04-28 12:00:00Z]
|
||
rows = [sample_row(33.0, -97.0), sample_row(45.0, -90.0)]
|
||
|
||
ScalarFile.write!(valid_time, rows)
|
||
|
||
assert Enum.count_until(ScalarFile.read_bounds(valid_time, nil), 3) == 2
|
||
end
|
||
|
||
test "preserves the full set of derived layer fields on round-trip" do
|
||
valid_time = ~U[2026-04-28 12:00:00Z]
|
||
[row] = [sample_row(33.0, -97.0)]
|
||
|
||
ScalarFile.write!(valid_time, [row])
|
||
|
||
[round_tripped] = ScalarFile.read_bounds(valid_time, nil)
|
||
|
||
for key <- Map.keys(row) do
|
||
assert Map.fetch!(round_tripped, key) == Map.fetch!(row, key),
|
||
"key #{inspect(key)} mismatched after round-trip"
|
||
end
|
||
end
|
||
end
|
||
|
||
describe "read_point/3" do
|
||
test "returns the snapped row for the requested lat/lon" do
|
||
valid_time = ~U[2026-04-28 12:00:00Z]
|
||
row = sample_row(33.0, -97.0)
|
||
|
||
ScalarFile.write!(valid_time, [row])
|
||
|
||
assert {:ok, fetched} = ScalarFile.read_point(valid_time, 33.0, -97.0)
|
||
assert fetched.temperature == 22.0
|
||
end
|
||
|
||
test "returns :miss when no scalar file exists" do
|
||
assert ScalarFile.read_point(~U[2026-04-28 12:00:00Z], 33.0, -97.0) == :miss
|
||
end
|
||
|
||
test "returns :miss when the cell is not in the file" do
|
||
valid_time = ~U[2026-04-28 12:00:00Z]
|
||
ScalarFile.write!(valid_time, [sample_row(33.0, -97.0)])
|
||
|
||
assert ScalarFile.read_point(valid_time, 40.0, -80.0) == :miss
|
||
end
|
||
end
|
||
|
||
describe "exists?/1 and list_valid_times/0" do
|
||
test "reflects what has been written" do
|
||
vt1 = ~U[2026-04-28 12:00:00Z]
|
||
vt2 = ~U[2026-04-28 13:00:00Z]
|
||
|
||
refute ScalarFile.exists?(vt1)
|
||
assert ScalarFile.list_valid_times() == []
|
||
|
||
ScalarFile.write!(vt1, [sample_row(33.0, -97.0)])
|
||
ScalarFile.write!(vt2, [sample_row(33.0, -97.0)])
|
||
|
||
assert ScalarFile.exists?(vt1)
|
||
assert ScalarFile.list_valid_times() == [vt1, vt2]
|
||
end
|
||
end
|
||
|
||
describe "HRDPS sibling read merge" do
|
||
test "read_bounds returns HRRR + HRDPS rows when both dirs exist" do
|
||
vt = ~U[2026-04-29 12:00:00Z]
|
||
|
||
ScalarFile.write!(vt, [sample_row(33.0, -97.0, 22.0)])
|
||
hand_write_chunk(ScalarFile.dir_for_hrdps(vt), 53.0, -60.0, 8.0)
|
||
|
||
rows = ScalarFile.read_bounds(vt, nil)
|
||
sorted = Enum.sort_by(rows, & &1.lat)
|
||
|
||
assert Enum.count_until(sorted, 3) == 2
|
||
assert Enum.at(sorted, 0).lat == 33.0
|
||
assert Enum.at(sorted, 1).lat == 53.0
|
||
end
|
||
|
||
test "read_bounds returns HRDPS rows when only HRDPS dir exists" do
|
||
vt = ~U[2026-04-29 12:00:00Z]
|
||
hand_write_chunk(ScalarFile.dir_for_hrdps(vt), 53.0, -60.0, 8.0)
|
||
|
||
assert [%{lat: 53.0, lon: -60.0}] = ScalarFile.read_bounds(vt, nil)
|
||
end
|
||
|
||
test "read_point falls through to HRDPS when the HRRR chunk is absent" do
|
||
vt = ~U[2026-04-29 12:00:00Z]
|
||
hand_write_chunk(ScalarFile.dir_for_hrdps(vt), 53.0, -60.0, 8.0)
|
||
|
||
assert {:ok, %{lat: 53.0, lon: -60.0, temperature: 8.0}} =
|
||
ScalarFile.read_point(vt, 53.0, -60.0)
|
||
end
|
||
|
||
test "exists?/1 is true when only the HRDPS dir has chunks" do
|
||
vt = ~U[2026-04-29 12:00:00Z]
|
||
hand_write_chunk(ScalarFile.dir_for_hrdps(vt), 53.0, -60.0, 8.0)
|
||
assert ScalarFile.exists?(vt)
|
||
end
|
||
|
||
test "merge prefers HRRR row when both dirs cover the same lat/lon" do
|
||
vt = ~U[2026-04-29 12:00:00Z]
|
||
|
||
ScalarFile.write!(vt, [sample_row(33.0, -97.0, 22.0)])
|
||
hand_write_chunk(ScalarFile.dir_for_hrdps(vt), 33.0, -97.0, 99.0)
|
||
|
||
[row] = ScalarFile.read_bounds(vt, nil)
|
||
assert row.temperature == 22.0
|
||
end
|
||
end
|
||
|
||
describe "HRDPS-only reads (for the /weather-ca endpoint)" do
|
||
test "read_bounds_hrdps returns only HRDPS chunks, ignoring co-located HRRR" do
|
||
vt = ~U[2026-04-29 12:00:00Z]
|
||
|
||
# Both dirs cover the same point. The HRRR-only read_bounds/2 would
|
||
# merge them (preferring HRRR); read_bounds_hrdps/2 must return ONLY
|
||
# the HRDPS row so /weather-ca shows Canadian-source data.
|
||
ScalarFile.write!(vt, [sample_row(33.0, -97.0, 22.0)])
|
||
hand_write_chunk(ScalarFile.dir_for_hrdps(vt), 53.0, -60.0, 8.0)
|
||
|
||
assert [%{lat: 53.0, lon: -60.0, temperature: 8.0}] =
|
||
ScalarFile.read_bounds_hrdps(vt, nil)
|
||
end
|
||
|
||
test "read_bounds_hrdps respects bounds" do
|
||
vt = ~U[2026-04-29 12:00:00Z]
|
||
|
||
hand_write_chunk(ScalarFile.dir_for_hrdps(vt), 53.0, -60.0, 8.0)
|
||
hand_write_chunk(ScalarFile.dir_for_hrdps(vt), 33.0, -97.0, 22.0)
|
||
|
||
bounds = %{"south" => 50.0, "north" => 60.0, "west" => -70.0, "east" => -50.0}
|
||
assert [%{lat: 53.0, lon: -60.0}] = ScalarFile.read_bounds_hrdps(vt, bounds)
|
||
end
|
||
|
||
test "read_bounds_hrdps returns [] when no HRDPS dir exists" do
|
||
vt = ~U[2026-04-29 12:00:00Z]
|
||
ScalarFile.write!(vt, [sample_row(33.0, -97.0, 22.0)])
|
||
|
||
assert ScalarFile.read_bounds_hrdps(vt, nil) == []
|
||
end
|
||
|
||
test "list_valid_times_hrdps returns times from HRDPS-suffixed dirs only" do
|
||
vt1 = ~U[2026-04-29 12:00:00Z]
|
||
vt2 = ~U[2026-04-29 13:00:00Z]
|
||
vt_hrrr_only = ~U[2026-04-29 14:00:00Z]
|
||
|
||
hand_write_chunk(ScalarFile.dir_for_hrdps(vt1), 53.0, -60.0, 8.0)
|
||
hand_write_chunk(ScalarFile.dir_for_hrdps(vt2), 53.0, -60.0, 9.0)
|
||
ScalarFile.write!(vt_hrrr_only, [sample_row(33.0, -97.0, 22.0)])
|
||
|
||
assert ScalarFile.list_valid_times_hrdps() == [vt1, vt2]
|
||
end
|
||
end
|
||
|
||
# The /weather (dual-source) timeline picks valid_times from HRRR's
|
||
# hourly cadence. HRDPS publishes 4×/day with multi-hour latency, so
|
||
# the requested time will frequently miss any HRDPS dir. Snapping to
|
||
# the closest HRDPS dir within a window lets the Canadian overlay
|
||
# render slightly stale rather than disappear.
|
||
describe "read_bounds_hrdps_nearest/3" do
|
||
test "returns rows from the exact time when its HRDPS dir exists" do
|
||
vt = ~U[2026-04-29 12:00:00Z]
|
||
hand_write_chunk(ScalarFile.dir_for_hrdps(vt), 53.0, -60.0, 8.0)
|
||
|
||
assert [%{lat: 53.0, lon: -60.0, temperature: 8.0}] =
|
||
ScalarFile.read_bounds_hrdps_nearest(vt, nil, 6 * 3600)
|
||
end
|
||
|
||
test "snaps to the closest HRDPS dir within the window when exact time is absent" do
|
||
requested = ~U[2026-04-29 14:00:00Z]
|
||
closer = ~U[2026-04-29 13:00:00Z]
|
||
farther = ~U[2026-04-29 10:00:00Z]
|
||
|
||
hand_write_chunk(ScalarFile.dir_for_hrdps(closer), 53.0, -60.0, 7.0)
|
||
hand_write_chunk(ScalarFile.dir_for_hrdps(farther), 53.0, -60.0, 99.0)
|
||
|
||
assert [%{temperature: 7.0}] =
|
||
ScalarFile.read_bounds_hrdps_nearest(requested, nil, 6 * 3600)
|
||
end
|
||
|
||
test "snaps forward as well as backward (HRDPS forecast can lead requested time)" do
|
||
requested = ~U[2026-04-29 12:00:00Z]
|
||
future = ~U[2026-04-29 13:00:00Z]
|
||
|
||
hand_write_chunk(ScalarFile.dir_for_hrdps(future), 53.0, -60.0, 11.0)
|
||
|
||
assert [%{temperature: 11.0}] =
|
||
ScalarFile.read_bounds_hrdps_nearest(requested, nil, 6 * 3600)
|
||
end
|
||
|
||
test "returns [] when no HRDPS dir exists inside the window" do
|
||
requested = ~U[2026-04-29 14:00:00Z]
|
||
stale = ~U[2026-04-29 02:00:00Z]
|
||
|
||
hand_write_chunk(ScalarFile.dir_for_hrdps(stale), 53.0, -60.0, 8.0)
|
||
|
||
# 12h gap, 6h window → not eligible.
|
||
assert ScalarFile.read_bounds_hrdps_nearest(requested, nil, 6 * 3600) == []
|
||
end
|
||
|
||
test "returns [] when no HRDPS dirs exist at all" do
|
||
requested = ~U[2026-04-29 14:00:00Z]
|
||
assert ScalarFile.read_bounds_hrdps_nearest(requested, nil, 6 * 3600) == []
|
||
end
|
||
|
||
# The Rust pipeline writes `<iso>.hrdps.sgrid`, not the legacy chunked
|
||
# `<iso>.hrdps/` dirs. `weather_grid_hrdps_at/2` is the only production
|
||
# caller of this function and it backs every `/weather/cells?source=hrdps`
|
||
# request, so a chunk-only read here means the Canadian overlay renders
|
||
# nothing on `/weather` and `/weather-ca` is blank end to end.
|
||
test "reads the dense .sgrid artifact at the exact time" do
|
||
vt = ~U[2026-04-29 12:00:00Z]
|
||
hand_write_sgrid_hrdps(vt, 53.0, -60.0, 8.0)
|
||
|
||
assert [%{lat: 53.0, lon: -60.0, temperature: 8.0}] =
|
||
ScalarFile.read_bounds_hrdps_nearest(vt, nil, 6 * 3600)
|
||
end
|
||
|
||
test "snaps to the closest .sgrid within the window when exact time is absent" do
|
||
requested = ~U[2026-04-29 14:00:00Z]
|
||
closer = ~U[2026-04-29 13:00:00Z]
|
||
farther = ~U[2026-04-29 10:00:00Z]
|
||
|
||
hand_write_sgrid_hrdps(closer, 53.0, -60.0, 7.0)
|
||
hand_write_sgrid_hrdps(farther, 53.0, -60.0, 99.0)
|
||
|
||
assert [%{temperature: 7.0}] =
|
||
ScalarFile.read_bounds_hrdps_nearest(requested, nil, 6 * 3600)
|
||
end
|
||
|
||
test "respects bounds when reading a .sgrid" do
|
||
vt = ~U[2026-04-29 12:00:00Z]
|
||
hand_write_sgrid_hrdps(vt, 53.0, -60.0, 8.0)
|
||
|
||
bounds = %{"south" => 60.0, "north" => 65.0, "west" => -70.0, "east" => -50.0}
|
||
assert ScalarFile.read_bounds_hrdps_nearest(vt, bounds, 6 * 3600) == []
|
||
end
|
||
|
||
# A valid_time can have both artifacts mid-migration (the chunk dir
|
||
# hasn't drained yet). The dense file is authoritative — reading both
|
||
# would double-paint the cell.
|
||
test "prefers the .sgrid over a co-located legacy chunk dir" do
|
||
vt = ~U[2026-04-29 12:00:00Z]
|
||
hand_write_chunk(ScalarFile.dir_for_hrdps(vt), 53.0, -60.0, 99.0)
|
||
hand_write_sgrid_hrdps(vt, 53.0, -60.0, 8.0)
|
||
|
||
assert [%{temperature: 8.0}] = ScalarFile.read_bounds_hrdps_nearest(vt, nil, 6 * 3600)
|
||
end
|
||
end
|
||
|
||
# Emit a single-cell `<iso>.hrdps.sgrid` byte-for-byte as
|
||
# `prop_grid_rs::sgrid` writes it, so this test fails if either side
|
||
# of the cross-language wire format drifts.
|
||
defp hand_write_sgrid_hrdps(valid_time, lat, lon, temp) do
|
||
path = Sgrid.path_for_hrdps(valid_time)
|
||
File.mkdir_p!(Path.dirname(path))
|
||
|
||
fields = ["temperature", "dewpoint_depression"]
|
||
field_table = Enum.map_join(fields, fn name -> name <> :binary.copy(<<0>>, 32 - byte_size(name)) end)
|
||
|
||
# flags bit0 = 1 → HRDPS. 1×1 grid anchored on the cell itself.
|
||
header =
|
||
<<"SGRD", 1::8, 1::8, length(fields)::little-16, DateTime.to_unix(valid_time)::little-signed-64,
|
||
lat::little-float-64, lon::little-float-64, 0.5::little-float-64, 0.5::little-float-64, 1::little-16,
|
||
1::little-16>>
|
||
|
||
body = <<temp::little-float-32, 5.0::little-float-32>>
|
||
|
||
File.write!(path, header <> field_table <> body)
|
||
end
|
||
|
||
defp hand_write_chunk(dir, lat, lon, temp) do
|
||
File.mkdir_p!(dir)
|
||
lat_band = (lat / 5) |> Float.floor() |> trunc()
|
||
lon_band = (lon / 5) |> Float.floor() |> trunc()
|
||
path = Path.join(dir, "#{lat_band}_#{lon_band}.mp.gz")
|
||
|
||
payload =
|
||
Msgpax.pack!(
|
||
[
|
||
%{
|
||
"lat" => lat,
|
||
"lon" => lon,
|
||
"valid_time" => "2026-04-29T12:00:00Z",
|
||
"temperature" => temp
|
||
}
|
||
],
|
||
iodata: false
|
||
)
|
||
|
||
File.write!(path, :zlib.gzip(payload))
|
||
end
|
||
|
||
describe "Rust-writer compatibility (the cross-language wire format)" do
|
||
test "decodes a hand-rolled gzipped MessagePack chunk and atomizes whitelisted keys" do
|
||
valid_time = ~U[2026-04-29 12:00:00Z]
|
||
dir = Path.join(ScalarFile.dir_for(valid_time), "")
|
||
File.mkdir_p!(dir)
|
||
|
||
# Mirror exactly what `prop_grid_rs::weather_scalar_file::write_atomic`
|
||
# emits: gzipped MessagePack, top-level array of string-keyed maps.
|
||
# If this round-trip ever breaks, Elixir is reading something the
|
||
# Rust pipeline cannot produce — a wire-format regression.
|
||
payload =
|
||
Msgpax.pack!(
|
||
[
|
||
%{
|
||
"lat" => 33.0,
|
||
"lon" => -97.0,
|
||
"valid_time" => "2026-04-29T12:00:00Z",
|
||
"temperature" => 22.5,
|
||
"dewpoint_depression" => 10.0,
|
||
"surface_rh" => 50.0,
|
||
"ducting" => false,
|
||
# An out-of-whitelist key — the reader must NOT atomize this.
|
||
"unknown_extra_key" => 1.23
|
||
}
|
||
],
|
||
iodata: false
|
||
)
|
||
|
||
path = Path.join(dir, "6_-20.mp.gz")
|
||
File.write!(path, :zlib.gzip(payload))
|
||
|
||
[row] = ScalarFile.read_bounds(valid_time, nil)
|
||
|
||
# Whitelisted keys are atoms.
|
||
assert row.lat == 33.0
|
||
assert row.lon == -97.0
|
||
assert row.temperature == 22.5
|
||
assert row.ducting == false
|
||
|
||
# `valid_time` round-trips back to a DateTime.
|
||
assert row.valid_time == valid_time
|
||
|
||
# Out-of-whitelist keys stay as strings so a future field can be
|
||
# added without a deploy-order coordination problem.
|
||
assert row["unknown_extra_key"] == 1.23
|
||
end
|
||
end
|
||
|
||
describe "prune_older_than/1 and retain_window/2" do
|
||
test "prune_older_than removes files strictly before the cutoff" do
|
||
old = ~U[2026-04-27 12:00:00Z]
|
||
new = ~U[2026-04-28 12:00:00Z]
|
||
|
||
ScalarFile.write!(old, [sample_row(33.0, -97.0)])
|
||
ScalarFile.write!(new, [sample_row(33.0, -97.0)])
|
||
|
||
assert ScalarFile.prune_older_than(~U[2026-04-28 00:00:00Z]) == 1
|
||
refute ScalarFile.exists?(old)
|
||
assert ScalarFile.exists?(new)
|
||
end
|
||
|
||
test "retain_window keeps only files inside [run_time, run_time + max_forecast_hour * 3600]" do
|
||
run_time = ~U[2026-04-28 12:00:00Z]
|
||
inside = ~U[2026-04-28 14:00:00Z]
|
||
outside = ~U[2026-04-29 06:00:00Z]
|
||
|
||
ScalarFile.write!(run_time, [sample_row(33.0, -97.0)])
|
||
ScalarFile.write!(inside, [sample_row(33.0, -97.0)])
|
||
ScalarFile.write!(outside, [sample_row(33.0, -97.0)])
|
||
|
||
assert ScalarFile.retain_window(run_time, 4) == 1
|
||
assert ScalarFile.exists?(run_time)
|
||
assert ScalarFile.exists?(inside)
|
||
refute ScalarFile.exists?(outside)
|
||
end
|
||
end
|
||
end
|