prop/test/microwaveprop/weather/scalar_file_test.exs
Graham McIntire 289cfe5eef
Some checks failed
Build and Push / Build CI test image (push) Successful in 10s
Build base image / Build and push base image (push) Successful in 21s
Build and Push / Build and Push Docker Image (push) Has been cancelled
fix(weather): prune dense .sgrid scalar files
ScalarFile.list_valid_time_dirs/1 stripped only a `.hrdps` suffix
before DateTime.from_iso8601/1, so `<iso>.sgrid` and
`<iso>.hrdps.sgrid` never parsed and were skipped by both
prune_older_than/1 and retain_window/2. Since .sgrid is the format the
pipeline actually writes, nothing on the dense tier was ever deleted —
production held 222 files spanning 8 days against a 48h window.

Strip the known suffixes in order, mirroring ProfilesFile's anchored
parser (which documents this exact failure mode for .hrdps.prop).
Orphaned .tmp.* writes still don't parse, so they stay with the tmp
sweep in Propagation.prune_old_scores/0.

Also corrects the CLAUDE.md entry that recorded this as already fixed —
the call site was present, the suffix parsing was not.
2026-08-05 17:52:06 -05:00

515 lines
18 KiB
Elixir
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

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
# `.sgrid` is the format the pipeline actually writes. Stripping only
# a `.hrdps` suffix before `from_iso8601/1` left every dense file
# unparseable and therefore invisible to both prune paths —
# production accumulated 222 files spanning 8 days against a 48 h
# retention window.
test "prune_older_than removes dense .sgrid files" do
old = ~U[2026-04-27 12:00:00Z]
new = ~U[2026-04-28 12:00:00Z]
File.mkdir_p!(ScalarFile.base_dir())
File.write!(Sgrid.path_for(old), "x")
File.write!(Sgrid.path_for_hrdps(old), "x")
File.write!(Sgrid.path_for(new), "x")
assert ScalarFile.prune_older_than(~U[2026-04-28 00:00:00Z]) == 2
refute File.exists?(Sgrid.path_for(old))
refute File.exists?(Sgrid.path_for_hrdps(old))
assert File.exists?(Sgrid.path_for(new))
end
test "retain_window removes dense .sgrid files outside the window" 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]
File.mkdir_p!(ScalarFile.base_dir())
File.write!(Sgrid.path_for(run_time), "x")
File.write!(Sgrid.path_for_hrdps(inside), "x")
File.write!(Sgrid.path_for(outside), "x")
assert ScalarFile.retain_window(run_time, 4) == 1
assert File.exists?(Sgrid.path_for(run_time))
assert File.exists?(Sgrid.path_for_hrdps(inside))
refute File.exists?(Sgrid.path_for(outside))
end
# Orphaned atomic-write temporaries are swept separately by
# `Propagation.prune_old_scores/0`; they must not parse as a
# valid_time here or a crashed write would be pruned as if it were
# a finished artifact.
test "ignores orphaned .tmp files" do
vt = ~U[2026-04-27 12:00:00Z]
File.mkdir_p!(ScalarFile.base_dir())
tmp = Sgrid.path_for(vt) <> ".tmp.123.4"
File.write!(tmp, "x")
assert ScalarFile.prune_older_than(~U[2026-04-28 00:00:00Z]) == 0
assert File.exists?(tmp)
end
end
end