//! wgrib2 subprocess wrapper. 1:1 port of the hot paths in //! `lib/microwaveprop/weather/grib2/wgrib2.ex` used by f01..f48: //! * `extract_grid` (write GRIB2 to tmp → `wgrib2 -lola … bin` → parse) //! * inventory parsing from stdout //! * `parse_lola_binary` (Fortran-unformatted IEEE 754 little-endian f32) //! //! wgrib2 is called as a child process exactly as the Elixir version //! does — same CLI flags, same temp-file lifecycle. Output binary is //! stride-addressed: each message = 4-byte LE record length + N*4 bytes //! of data + 4-byte duplicate record length. use std::collections::HashMap; use std::path::{Path, PathBuf}; use std::process::{Command, Output}; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{SystemTime, UNIX_EPOCH}; use crate::grid::{round3, GridSpec}; const UNDEFINED_VALUE: f32 = 9.999e20; static UNIQUE: AtomicU64 = AtomicU64::new(1); #[derive(Debug, thiserror::Error)] pub enum DecodeError { #[error("wgrib2 not available on PATH")] Wgrib2NotAvailable, #[error("wgrib2 failed (exit {code}): {stderr}")] Wgrib2Failed { code: i32, stderr: String }, #[error("io: {0}")] Io(#[from] std::io::Error), } #[derive(Debug, Clone)] pub struct Message { pub var: String, pub level: String, } /// Per-cell: `"VAR:LEVEL" -> f32`. Key format matches Elixir exactly so /// scorer input can be wired unchanged. /// /// Keys are `Arc` rather than `String`: the hot decode loop /// inserts the same `"VAR:LEVEL"` string into ~92k cell maps per /// GRIB2 message, which was previously a String::clone per cell /// (~5.5M heap allocations + frees per chain step across 60 /// messages). `Arc::clone` is a single atomic increment — keys /// allocate once per message and are ref-counted thereafter. /// /// Lookup is unchanged: `cell.get("TMP:2 m above ground")` works /// because `Arc: Borrow`. pub type CellValues = HashMap, f32>; /// Lat/lon rounded to 3 decimals. f64 keys can't be hashed directly, so /// we key by `(lat_mdeg, lon_mdeg)` where `*_mdeg = round(x * 1000) as i32`. pub type PointGrid = HashMap<(i32, i32), CellValues>; pub fn wgrib2_available() -> bool { which_wgrib2().is_some() } fn which_wgrib2() -> Option { if let Ok(val) = std::env::var("WGRIB2") { return Some(PathBuf::from(val)); } let path_var = std::env::var_os("PATH")?; for dir in std::env::split_paths(&path_var) { let candidate = dir.join("wgrib2"); if candidate.is_file() { return Some(candidate); } } None } pub fn normalize_lon(lon: f64) -> f64 { if lon < 0.0 { lon + 360.0 } else { lon } } pub fn denormalize_lon(lon: f64) -> f64 { if lon > 180.0 { lon - 360.0 } else { lon } } /// Per-cell HRDPS extraction via `wgrib2 -lon LON LAT`. /// /// `extract_grid` (`-lola`) interpolates the entire source grid onto a /// target rectangle, which on HRDPS's rotated lat/lon source runs at /// 10+ minutes per chain step (production observation 2026-04-29). The /// `-lon` per-point path that the probe used is far cheaper because it /// only computes the rotation math at the points we ask for. For /// 57k Canadian-only cells this drops chain-step wall time from "never /// completes" to roughly 30-90 seconds. /// /// Batched at `POINT_BATCH` points per wgrib2 invocation to keep the /// argv list comfortably under Linux's `ARG_MAX` (about 2 MiB) and keep /// stdout parseable in chunks. pub fn extract_points( grib: &[u8], match_pattern: &str, points: &[(f64, f64)], ) -> Result { let wgrib2 = which_wgrib2().ok_or(DecodeError::Wgrib2NotAvailable)?; let nanos = SystemTime::now() .duration_since(UNIX_EPOCH) .map(|d| d.as_nanos()) .unwrap_or(0); let uniq = UNIQUE.fetch_add(1, Ordering::Relaxed); let grib_path: PathBuf = std::env::temp_dir().join(format!("hrdps_{nanos}_{uniq}.grib2")); std::fs::write(&grib_path, grib)?; let result = run_wgrib2_lon_batched(&wgrib2, &grib_path, match_pattern, points); let _ = std::fs::remove_file(&grib_path); result } // 1000 points per wgrib2 invocation. With ~57k Canadian cells that's // 57 invocations per chain step — enough to amortize process-spawn // overhead, small enough that any future wgrib2 regression around // large -lon arg lists trips on a manageable failure mode. // // Note: wgrib2 3.6.0 had a memory-corruption bug surfaced by HRDPS // rotated lat/lon files at *any* batch size (verified at N=200 in // production on 2026-04-30 — `free(): invalid size`). The fix was // bumping the Dockerfile to wgrib2 3.8.0; this batch size is just // a normal performance tuning choice, not a workaround. const POINT_BATCH: usize = 1_000; fn run_wgrib2_lon_batched( wgrib2: &Path, grib_path: &Path, match_pattern: &str, points: &[(f64, f64)], ) -> Result { let mut out: PointGrid = HashMap::with_capacity(points.len()); for batch in points.chunks(POINT_BATCH) { let batch_grid = run_wgrib2_lon_one(wgrib2, grib_path, match_pattern, batch)?; for (key, cell) in batch_grid { out.entry(key).or_default().extend(cell); } } Ok(out) } fn run_wgrib2_lon_one( wgrib2: &Path, grib_path: &Path, match_pattern: &str, batch: &[(f64, f64)], ) -> Result { let mut cmd = Command::new(wgrib2); cmd.arg(grib_path).arg("-s").arg("-match").arg(match_pattern); for &(lat, lon) in batch { cmd.arg("-lon").arg(format!("{lon}")).arg(format!("{lat}")); } let Output { status, stdout, stderr, } = cmd.output()?; if !status.success() { // Surface the FIRST chars of stderr (the actual diagnostic) // rather than stdout (which is mostly per-cell numeric output // that drowned the interesting failure context). status.code() // returns None on signal-kill — flag that with -1 so the // failure mode is distinguishable from a real exit code. let stderr_text = String::from_utf8_lossy(&stderr).to_string(); let stderr_snippet: String = stderr_text.chars().take(400).collect(); let signaled = status.code().is_none(); let suffix = if signaled { format!(" (killed by signal, batch={})", batch.len()) } else { String::new() }; return Err(DecodeError::Wgrib2Failed { code: status.code().unwrap_or(-1), stderr: format!("{stderr_snippet}{suffix}"), }); } let text = String::from_utf8_lossy(&stdout); Ok(parse_lon_output(&text, batch)) } /// Parse `wgrib2 -s -lon` text output. Each record line is a colon- /// separated header followed by per-point `lon=…,lat=…,val=…` segments, /// one per `-lon` arg. /// /// Example line: /// `1:0:d=...:TMP:2 m above ground:anl:lon=280.369,lat=43.670,val=280.97:lon=...` fn parse_lon_output(text: &str, batch: &[(f64, f64)]) -> PointGrid { let mut out: PointGrid = HashMap::new(); for line in text.lines() { if line.is_empty() { continue; } let mut parts = line.split(':'); let _n = parts.next(); let _offset = parts.next(); let _date = parts.next(); let var = match parts.next() { Some(v) => v.to_string(), None => continue, }; let level = match parts.next() { Some(l) => l.to_string(), None => continue, }; // Skip the record-type/forecast-time field (anl, "240 min fcst", …). let _kind = parts.next(); let key: std::sync::Arc = format!("{var}:{level}").into(); for segment in parts { if let Some(value) = parse_lon_segment(segment, batch) { let (lat_key, lon_key, val) = value; out.entry((lat_key, lon_key)) .or_default() .insert(std::sync::Arc::clone(&key), val); } } } out } /// `lon=242.958,lat=32.938,val=306.5` → (lat_key, lon_key, val). /// Snaps to the nearest point in `batch` (handles wgrib2's lon-180/360 /// convention by checking both signs). fn parse_lon_segment(segment: &str, batch: &[(f64, f64)]) -> Option<(i32, i32, f32)> { let mut lon_part = None; let mut lat_part = None; let mut val_part = None; for kv in segment.split(',') { let (k, v) = kv.split_once('=')?; let parsed: f64 = v.trim().parse().ok()?; match k.trim() { "lon" => lon_part = Some(parsed), "lat" => lat_part = Some(parsed), "val" => val_part = Some(parsed), _ => {} } } let raw_lon = lon_part?; let raw_lat = lat_part?; let val = val_part?; if val > UNDEFINED_VALUE as f64 / 2.0 { return None; } let target_lon = denormalize_lon(raw_lon); let snapped = snap_to_batch(raw_lat, target_lon, batch)?; let lat_key = (snapped.0 * 1000.0).round() as i32; let lon_key = (snapped.1 * 1000.0).round() as i32; Some((lat_key, lon_key, val as f32)) } fn snap_to_batch(lat: f64, lon: f64, batch: &[(f64, f64)]) -> Option<(f64, f64)> { // wgrib2 -lon snaps to the nearest source-grid cell, so the returned // lat/lon is close to but not exactly equal to what we asked for. // Match it back to our request (which is on the 0.125° propagation // grid) via the closest batch entry. let mut best: Option<((f64, f64), f64)> = None; for &(p_lat, p_lon) in batch { let dlat = p_lat - lat; let dlon = p_lon - lon; let d2 = dlat * dlat + dlon * dlon; match best { None => best = Some(((p_lat, p_lon), d2)), Some((_, b)) if d2 < b => best = Some(((p_lat, p_lon), d2)), _ => {} } } let (winner, d2) = best?; // Reject anything more than 0.5° away — wgrib2 returned a cell that // doesn't correspond to a point we asked for (probably a rounding // edge case at the bbox boundary). if d2 > 0.25 { return None; } Some(winner) } /// Extract values at the given grid spec from an in-memory GRIB2 blob. /// Writes the blob to a temp file, runs wgrib2, parses the output. pub fn extract_grid( grib: &[u8], match_pattern: &str, grid_spec: GridSpec, ) -> Result { let wgrib2 = which_wgrib2().ok_or(DecodeError::Wgrib2NotAvailable)?; let tmp_dir = std::env::temp_dir(); let uniq = UNIQUE.fetch_add(1, Ordering::Relaxed); let nanos = SystemTime::now() .duration_since(UNIX_EPOCH) .map(|d| d.as_nanos()) .unwrap_or(0); let grib_path = tmp_dir.join(format!("hrrr_{nanos}_{uniq}.grib2")); let bin_path = tmp_dir.join(format!("hrrr_{nanos}_{uniq}.grib2.lola.bin")); let result: Result = (|| { std::fs::write(&grib_path, grib)?; run_wgrib2_lola(&wgrib2, &grib_path, match_pattern, &grid_spec, &bin_path) })(); let _ = std::fs::remove_file(&grib_path); let _ = std::fs::remove_file(&bin_path); result } /// File-path variant — avoids the intermediate `write(tmp, binary)` copy. pub fn extract_grid_from_file( grib_path: &Path, match_pattern: &str, grid_spec: GridSpec, ) -> Result { let wgrib2 = which_wgrib2().ok_or(DecodeError::Wgrib2NotAvailable)?; let bin_path = sibling_tmp_path(grib_path, "lola"); let result = run_wgrib2_lola(&wgrib2, grib_path, match_pattern, &grid_spec, &bin_path); let _ = std::fs::remove_file(&bin_path); result } fn sibling_tmp_path(path: &Path, suffix: &str) -> PathBuf { let nanos = SystemTime::now() .duration_since(UNIX_EPOCH) .map(|d| d.as_nanos()) .unwrap_or(0); let uniq = UNIQUE.fetch_add(1, Ordering::Relaxed); let mut s = path.as_os_str().to_owned(); s.push(format!(".{suffix}.{nanos}.{uniq}.bin")); PathBuf::from(s) } fn run_wgrib2_lola( wgrib2: &Path, grib_path: &Path, match_pattern: &str, grid_spec: &GridSpec, bin_path: &Path, ) -> Result { let lon_spec = format!( "{}:{}:{}", normalize_lon(grid_spec.lon_start), grid_spec.lon_count, grid_spec.lon_step ); let lat_spec = format!( "{}:{}:{}", grid_spec.lat_start, grid_spec.lat_count, grid_spec.lat_step ); let Output { status, stdout, stderr, } = Command::new(wgrib2) .arg(grib_path) .arg("-match") .arg(match_pattern) .arg("-lola") .arg(&lon_spec) .arg(&lat_spec) .arg(bin_path) .arg("bin") .output()?; if !status.success() { let mut combined = stdout; combined.extend_from_slice(&stderr); let text = String::from_utf8_lossy(&combined).to_string(); let snippet: String = text.chars().take(200).collect(); return Err(DecodeError::Wgrib2Failed { code: status.code().unwrap_or(-1), stderr: snippet, }); } let inventory = String::from_utf8_lossy(&stdout); let messages = parse_inventory(&inventory); match std::fs::read(bin_path) { Ok(bin) => Ok(parse_lola_binary(&bin, &messages, grid_spec)), Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(HashMap::new()), Err(e) => Err(e.into()), } } /// Parse `wgrib2`'s stdout inventory (`"n:offset:date:var:level:..."`). /// Malformed lines are dropped. pub fn parse_inventory(out: &str) -> Vec { out.lines() .filter_map(|line| { let mut parts = line.splitn(8, ':'); let _n = parts.next()?; let _offset = parts.next()?; let _date = parts.next()?; let var = parts.next()?; let level = parts.next()?; Some(Message { var: var.to_string(), level: level.to_string(), }) }) .collect() } /// Parse the Fortran-unformatted output file from `wgrib2 -lola`. /// Layout: per message = 4-byte LE u32 record length + N·4 bytes f32 LE /// data + 4-byte LE u32 record length. `N = nx * ny`. Cells whose value /// exceeds `UNDEFINED_VALUE / 2` are treated as missing (the sentinel /// wgrib2 writes for "no data"). pub fn parse_lola_binary(bin: &[u8], messages: &[Message], grid_spec: &GridSpec) -> PointGrid { let nx = grid_spec.lon_count; let ny = grid_spec.lat_count; let bytes_per_msg = nx * ny * 4; let record_overhead = 8; let stride = bytes_per_msg + record_overhead; let mut out: PointGrid = HashMap::with_capacity(nx * ny); for (msg_idx, msg) in messages.iter().enumerate() { let data_offset = msg_idx * stride + 4; if data_offset + bytes_per_msg > bin.len() { continue; } let chunk = &bin[data_offset..data_offset + bytes_per_msg]; // Allocate the "VAR:LEVEL" string once per message. Every // subsequent cell insertion is Arc::clone (one atomic inc), // not a String::clone (heap alloc + copy + free). let key: std::sync::Arc = format!("{}:{}", msg.var, msg.level).into(); for j in 0..ny { let lat = round3(grid_spec.lat_start + j as f64 * grid_spec.lat_step); let lat_key = (lat * 1000.0).round() as i32; for i in 0..nx { let cell_offset = (j * nx + i) * 4; let value = f32::from_le_bytes(chunk[cell_offset..cell_offset + 4].try_into().unwrap()); if value > UNDEFINED_VALUE / 2.0 { continue; } let raw_lon = grid_spec.lon_start + i as f64 * grid_spec.lon_step; let lon = round3(denormalize_lon(raw_lon)); let lon_key = (lon * 1000.0).round() as i32; out.entry((lat_key, lon_key)) .or_default() .insert(std::sync::Arc::clone(&key), value); } } } out } pub fn key_to_latlon(key: (i32, i32)) -> (f64, f64) { (key.0 as f64 / 1000.0, key.1 as f64 / 1000.0) } #[cfg(test)] mod tests { use super::*; #[test] fn lon_round_trip() { for lon in [-125.0, -97.0, -66.0, 0.0, 179.5] { assert!((denormalize_lon(normalize_lon(lon)) - lon).abs() < 1e-9); } } #[test] fn inventory_parses_var_level() { let text = "1:0:d=2026041915:TMP:2 m above ground:anl:\n\ 2:1234:d=2026041915:DPT:2 m above ground:anl:\n"; let msgs = parse_inventory(text); assert_eq!(msgs.len(), 2); assert_eq!(msgs[0].var, "TMP"); assert_eq!(msgs[1].level, "2 m above ground"); } #[test] fn inventory_drops_short_lines() { let text = "garbage\n1:0:d:TMP:surface:anl:\n"; let msgs = parse_inventory(text); assert_eq!(msgs.len(), 1); assert_eq!(msgs[0].var, "TMP"); } #[test] fn parse_lon_output_pulls_var_level_and_snaps_to_batch() { // Match the shape `wgrib2 -s -lon` produces. Two records, two // points each. Note: lon=280.369 is wgrib2's 0-360 form for // -79.631; parse_lon_segment denormalizes it. let text = "1:0:d=2026042912:TMP:2 m above ground:anl:\ lon=280.369,lat=43.670,val=280.97:\ lon=235.000,lat=49.190,val=281.62\n\ 2:3527800:d=2026042912:DPT:2 m above ground:anl:\ lon=280.369,lat=43.670,val=275.00:\ lon=235.000,lat=49.190,val=270.00\n"; let batch = vec![(43.670, -79.631), (49.190, -125.000)]; let grid = parse_lon_output(text, &batch); let toronto_key = ((43.670_f64 * 1000.0).round() as i32, (-79.631_f64 * 1000.0).round() as i32); let toronto = grid.get(&toronto_key).expect("toronto cell"); let tmp = toronto.get("TMP:2 m above ground").copied().expect("tmp present"); assert!((tmp - 280.97).abs() < 0.01); let dpt = toronto.get("DPT:2 m above ground").copied().expect("dpt present"); assert!((dpt - 275.00).abs() < 0.01); } #[test] fn parse_lon_segment_drops_undefined_values() { let batch = vec![(43.670, -79.631)]; let segment = "lon=280.369,lat=43.670,val=9.999e20"; assert!(parse_lon_segment(segment, &batch).is_none()); } #[test] fn parse_lon_segment_rejects_far_off_points() { let batch = vec![(43.670, -79.631)]; // wgrib2 returned a cell for somewhere completely unrelated. let segment = "lon=240.0,lat=10.0,val=290.0"; assert!(parse_lon_segment(segment, &batch).is_none()); } #[test] fn parse_lola_binary_reconstructs_grid() { // Build a synthetic 2-message, 3×2 grid. Layout matches wgrib2: // [4 bytes len][6 f32 values][4 bytes len] per message. let spec = GridSpec { lon_start: -100.0, lon_count: 3, lon_step: 0.5, lat_start: 30.0, lat_count: 2, lat_step: 0.5, }; let msgs = vec![ Message { var: "TMP".into(), level: "surface".into(), }, Message { var: "DPT".into(), level: "surface".into(), }, ]; let mut buf = Vec::new(); for (msg_idx, _msg) in msgs.iter().enumerate() { let len: u32 = 3 * 2 * 4; buf.extend_from_slice(&len.to_le_bytes()); for j in 0..2 { for i in 0..3 { let v = (msg_idx as f32) * 100.0 + j as f32 * 10.0 + i as f32; buf.extend_from_slice(&v.to_le_bytes()); } } buf.extend_from_slice(&len.to_le_bytes()); } let grid = parse_lola_binary(&buf, &msgs, &spec); assert_eq!(grid.len(), 6); // Cell (30.0, -100.0) → i=0, j=0, msg0 value = 0.0, msg1 = 100.0 let k = ((30.0 * 1000.0) as i32, (-100.0 * 1000.0) as i32); let cell = grid.get(&k).unwrap(); assert_eq!(cell["TMP:surface"], 0.0); assert_eq!(cell["DPT:surface"], 100.0); // Cell (30.5, -99.0) → i=2, j=1, msg0 = 12.0 let k2 = ((30.5 * 1000.0) as i32, (-99.0 * 1000.0) as i32); let cell2 = grid.get(&k2).unwrap(); assert_eq!(cell2["TMP:surface"], 12.0); assert_eq!(cell2["DPT:surface"], 112.0); } #[test] fn parse_lola_skips_undefined_sentinel() { let spec = GridSpec { lon_start: -100.0, lon_count: 2, lon_step: 1.0, lat_start: 30.0, lat_count: 1, lat_step: 1.0, }; let msgs = vec![Message { var: "TMP".into(), level: "surface".into(), }]; let mut buf = Vec::new(); let len: u32 = 2 * 4; buf.extend_from_slice(&len.to_le_bytes()); buf.extend_from_slice(&UNDEFINED_VALUE.to_le_bytes()); // i=0 buf.extend_from_slice(&42.0f32.to_le_bytes()); // i=1 buf.extend_from_slice(&len.to_le_bytes()); let grid = parse_lola_binary(&buf, &msgs, &spec); assert_eq!(grid.len(), 1); // only i=1 survived let k = ((30.0 * 1000.0) as i32, (-99.0 * 1000.0) as i32); assert_eq!(grid.get(&k).unwrap()["TMP:surface"], 42.0); } #[test] fn key_to_latlon_roundtrip() { let (lat, lon) = key_to_latlon((32_500, -97_250)); assert_eq!(lat, 32.5); assert_eq!(lon, -97.25); } }