diff --git a/lib/microwaveprop/weather/scalar_file.ex b/lib/microwaveprop/weather/scalar_file.ex index 24baec7a..7b7c3172 100644 --- a/lib/microwaveprop/weather/scalar_file.ex +++ b/lib/microwaveprop/weather/scalar_file.ex @@ -94,9 +94,24 @@ defmodule Microwaveprop.Weather.ScalarFile do Path.join(base_dir(), iso_key(valid_time)) end + @doc """ + Sibling dir for HRDPS-derived scalar chunks. The HRRR `dir_for/1` + writer wipes its destination dir of stale chunks before each write; + routing HRDPS to a parallel directory means the two sources don't + clobber each other. Read paths read both dirs and merge. + """ + @spec dir_for_hrdps(DateTime.t()) :: String.t() + def dir_for_hrdps(%DateTime{} = valid_time) do + Path.join(base_dir(), iso_key(valid_time) <> ".hrdps") + end + @spec exists?(DateTime.t()) :: boolean() def exists?(%DateTime{} = valid_time) do - case File.ls(dir_for(valid_time)) do + has_chunks?(dir_for(valid_time)) or has_chunks?(dir_for_hrdps(valid_time)) + end + + defp has_chunks?(dir) do + case File.ls(dir) do {:ok, [_ | _]} -> true _ -> false end @@ -147,18 +162,42 @@ defmodule Microwaveprop.Weather.ScalarFile do """ @spec read_bounds(DateTime.t(), bounds() | nil) :: [row()] def read_bounds(%DateTime{} = valid_time, bounds) do - case list_chunk_files(valid_time) do - [] -> - [] + # Read HRRR + HRDPS chunks, then concat. HRRR wins where both + # sources cover (shouldn't happen in practice — Rust's HRDPS + # pipeline drops cells in HRRR's CONUS bbox before writing — but + # de-dup defensively in case mask-disagreement ever produces + # overlap). The user-visible intent: show HRRR-derived weather data + # everywhere CONUS, then fill the rest of North America with HRDPS. + hrrr = read_chunks(list_chunk_files(valid_time), bounds) + hrdps = read_chunks(list_chunk_files_hrdps(valid_time), bounds) - files -> - files - |> Enum.filter(fn {key, _path} -> chunk_intersects_bounds?(key, bounds) end) - |> Enum.flat_map(fn {_key, path} -> decode_chunk(path) end) - |> filter_bounds(bounds) + case {hrrr, hrdps} do + {[], []} -> [] + {h, []} -> h + {[], c} -> c + {h, c} -> merge_prefer_hrrr(h, c) end end + defp read_chunks([], _bounds), do: [] + + defp read_chunks(files, bounds) do + files + |> Enum.filter(fn {key, _path} -> chunk_intersects_bounds?(key, bounds) end) + |> Enum.flat_map(fn {_key, path} -> decode_chunk(path) end) + |> filter_bounds(bounds) + end + + defp merge_prefer_hrrr(hrrr_rows, hrdps_rows) do + hrrr_keys = + MapSet.new(hrrr_rows, fn %{lat: lat, lon: lon} -> {lat, lon} end) + + extras = + Enum.reject(hrdps_rows, fn %{lat: lat, lon: lon} -> MapSet.member?(hrrr_keys, {lat, lon}) end) + + hrrr_rows ++ extras + end + @doc """ Look up a single derived row at `(valid_time, lat, lon)`. Returns `{:ok, row}` or `:miss` when either the chunk file is absent or the @@ -167,13 +206,29 @@ defmodule Microwaveprop.Weather.ScalarFile do @spec read_point(DateTime.t(), float(), float()) :: {:ok, row()} | :miss def read_point(%DateTime{} = valid_time, lat, lon) when is_number(lat) and is_number(lon) do key = {chunk_band(lat * 1.0), chunk_band(lon * 1.0)} - path = Path.join(dir_for(valid_time), chunk_filename(key)) + chunk_name = chunk_filename(key) + snapped_lat = snap(lat) + snapped_lon = snap(lon) - if File.exists?(path) do - find_point_in_chunk(path, snap(lat), snap(lon)) - else - :miss - end + # Try HRRR first (where most points live), then HRDPS for Canadian + # cells outside HRRR coverage. + Enum.find_value( + [ + Path.join(dir_for(valid_time), chunk_name), + Path.join(dir_for_hrdps(valid_time), chunk_name) + ], + :miss, + fn path -> + if File.exists?(path) do + case find_point_in_chunk(path, snapped_lat, snapped_lon) do + {:ok, row} -> {:ok, row} + :miss -> false + end + else + false + end + end + ) end defp find_point_in_chunk(path, snapped_lat, snapped_lon) do @@ -350,8 +405,14 @@ defmodule Microwaveprop.Weather.ScalarFile do defp normalize_value(_key, v), do: v defp list_chunk_files(%DateTime{} = valid_time) do - dir = dir_for(valid_time) + list_chunk_files_in(dir_for(valid_time)) + end + defp list_chunk_files_hrdps(%DateTime{} = valid_time) do + list_chunk_files_in(dir_for_hrdps(valid_time)) + end + + defp list_chunk_files_in(dir) do case File.ls(dir) do {:ok, entries} -> for entry <- entries, diff --git a/rust/prop_grid_rs/src/pipeline.rs b/rust/prop_grid_rs/src/pipeline.rs index f53446c7..c4d3df7a 100644 --- a/rust/prop_grid_rs/src/pipeline.rs +++ b/rust/prop_grid_rs/src/pipeline.rs @@ -340,6 +340,7 @@ pub async fn run_chain_step_hrdps( let mut prepared: Vec<(f64, f64, Conditions, scorer::BandInvariants)> = Vec::with_capacity(grid.len()); + let mut scalar_rows: Vec = Vec::with_capacity(grid.len()); for (key, mut cell) in grid.into_iter() { let (lat, lon) = decoder::key_to_latlon(key); // HRDPS publishes DEPR (T-Td depression in K) as a primary @@ -350,6 +351,14 @@ pub async fn run_chain_step_hrdps( if let Some(conditions) = cell_to_conditions(&cell, lat, lon, &valid_time) { let invariants = scorer::precompute_band_invariants(&conditions); prepared.push((lat, lon, conditions, invariants)); + // Derive the per-cell scalar row alongside scoring. The /weather + // map reads these via Weather.weather_grid_at to render + // temperature/dewpoint-depression/refractivity layers — surfacing + // HRDPS here is what gives Canadian cells weather data on the + // map north of the HRRR coverage line. + if let Some(row) = weather_scalar_file::derive_row(lat, lon, valid_time, &cell) { + scalar_rows.push(row); + } } } @@ -375,6 +384,15 @@ pub async fn run_chain_step_hrdps( }; let scores_dir_owned = scores_dir.to_path_buf(); + + // Write the HRDPS scalar artifact alongside the scores. Sibling dir + // (`.hrdps/`) so HRRR's `/` write isn't clobbered. Without + // this the /weather map would only show CONUS data. + let scalar_dir = scores_dir_owned.clone(); + let scalar_future = tokio::task::spawn_blocking(move || { + weather_scalar_file::write_atomic_hrdps(&scalar_dir, valid_time, &scalar_rows).map(|_| ()) + }); + let write_handles: Vec<_> = scored .into_iter() .map(|(band_mhz, scores)| { @@ -387,6 +405,10 @@ pub async fn run_chain_step_hrdps( for h in write_handles { h.await.expect("blocking join")?; } + scalar_future + .await + .expect("blocking join") + .map_err(PipelineError::ScalarWrite)?; let files_written = bands.len() as u32; Ok(ChainStepStats { diff --git a/rust/prop_grid_rs/src/weather_scalar_file.rs b/rust/prop_grid_rs/src/weather_scalar_file.rs index 9025d181..b9122c62 100644 --- a/rust/prop_grid_rs/src/weather_scalar_file.rs +++ b/rust/prop_grid_rs/src/weather_scalar_file.rs @@ -194,12 +194,22 @@ pub fn derive_row( }) } -/// Absolute path the writer lands at for `valid_time`. +/// Absolute path the HRRR writer lands at for `valid_time`. pub fn dir_for(scores_dir: &Path, valid_time: DateTime) -> PathBuf { let iso = valid_time.format("%Y-%m-%dT%H:%M:%SZ").to_string(); scores_dir.join(SUBDIR).join(iso) } +/// Sibling directory for HRDPS-derived scalar chunks. Coexists with the +/// HRRR `dir_for` directory so the two sources can write 5°×5° chunks +/// independently — `write_atomic`'s "wipe dir first" sweep would +/// otherwise have whichever source ran last clobber the other's chunks. +/// Readers stitch the two sets together at request time. +pub fn dir_for_hrdps(scores_dir: &Path, valid_time: DateTime) -> PathBuf { + let iso = valid_time.format("%Y-%m-%dT%H:%M:%SZ").to_string(); + scores_dir.join(SUBDIR).join(format!("{iso}.hrdps")) +} + /// Persist `rows` for `valid_time` as 5°×5° chunked `.mp.gz` files. /// Mirrors the Elixir writer: the destination directory is wiped of /// any prior chunks first so a smaller follow-up write doesn't leave @@ -210,7 +220,20 @@ pub fn write_atomic( valid_time: DateTime, rows: &[ScalarRow], ) -> Result { - let dir = dir_for(scores_dir, valid_time); + write_atomic_into(dir_for(scores_dir, valid_time), rows) +} + +/// HRDPS sibling of `write_atomic` that lands at `dir_for_hrdps`. Wipes +/// only its own dir, leaving the HRRR `/` untouched. +pub fn write_atomic_hrdps( + scores_dir: &Path, + valid_time: DateTime, + rows: &[ScalarRow], +) -> Result { + write_atomic_into(dir_for_hrdps(scores_dir, valid_time), rows) +} + +fn write_atomic_into(dir: PathBuf, rows: &[ScalarRow]) -> Result { std::fs::create_dir_all(&dir)?; if let Ok(entries) = std::fs::read_dir(&dir) { diff --git a/test/microwaveprop/weather/scalar_file_test.exs b/test/microwaveprop/weather/scalar_file_test.exs index bb54cd83..e046e052 100644 --- a/test/microwaveprop/weather/scalar_file_test.exs +++ b/test/microwaveprop/weather/scalar_file_test.exs @@ -155,6 +155,75 @@ defmodule Microwaveprop.Weather.ScalarFileTest do 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 length(sorted) == 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 + + 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]