diff --git a/docs/weather.md b/docs/weather.md index 6f5f4c64..47be9d02 100644 --- a/docs/weather.md +++ b/docs/weather.md @@ -14,9 +14,7 @@ Completed so far: - Layer/time changes now refresh overlay tile URLs instead of pushing large `update_weather` payloads (finding #6). - A dedicated `Microwaveprop.Weather.ScalarFile` artifact persists pre-derived weather scalars per `valid_time` on NFS, bucketed into 5°×5° chunks. `weather_grid_at/2` and `weather_point_detail/3` read from it before falling back to raw `ProfilesFile` decoding (findings #1, #2, #3, #4, #7). -Still pending: - -- Have the propagation pipeline write the scalar artifact directly on the Rust side, alongside `.mp.gz`. Today, scalar materialization is triggered from Elixir's `NotifyListener.handle_propagation_ready/1` as soon as the Rust pipeline fires `NOTIFY propagation_ready`, so the gap between profile-write and scalar-availability is just one BEAM derivation pass per new `valid_time`. Closing it fully would require porting `WeatherLayers.derive` (lapse rates, inversion strength, 850/700 mb interpolation, duct summarization) to Rust, plus a compatible writer Elixir's `ScalarFile` reader can decode. Tracked as future work — the runtime cost it would save is small relative to the port effort. +All listed items shipped. The `Microwaveprop.Weather.WeatherLayers.derive/1` port now lives in [`prop_grid_rs::weather_scalar_file`](/Users/graham/dev/ntms/microwaveprop/rust/prop_grid_rs/src/weather_scalar_file.rs:1) and the Rust pipeline writes `ScalarFile` artifacts inline alongside `profiles_file::write_atomic`, using the same gzipped MessagePack chunked-by-5°×5° wire format the Elixir writer produces. The Elixir reader atomizes a whitelist of keys on read so callers see the same shape regardless of which side wrote the bytes. Available infrastructure that should shape the solution: @@ -399,7 +397,7 @@ All listed phases are shipped: - Viewport reads only touch the chunks that overlap the current bounds — for both NFS-backed `ScalarFile` reads and the in-memory `GridCache`. - Point clicks read a single chunk instead of the whole grid file. - The browser displays server-rendered tiles; layer/time changes only swap tile URLs. -- New forecast hours are pre-warmed: `NotifyListener.handle_propagation_ready/1` fires `Weather.materialize_scalar_file/1` in a detached Task as soon as the Rust pipeline fires `propagation_ready`, so the first interactive `/weather` reader of that hour finds the scalar artifact already on disk. +- New forecast hours land with the scalar artifact already on disk: the Rust pipeline writes scalar chunks inline alongside the raw profile, and the Elixir-side `NotifyListener.handle_propagation_ready/1` keeps an idempotent fallback materialization for hours that pre-date the Rust deploy. - Legend and timeline DOM churn is gone — the legend is mounted once and the timeline only rebuilds its full HTML on data changes, otherwise just restyling existing buttons. -Remaining work is a single non-runtime-critical port — move `WeatherLayers.derive` into the Rust pipeline so the BEAM doesn't pay even the one-time derivation pass per new `valid_time`. Documented as future work; the runtime savings are small relative to the port effort. +Both languages produce and consume identical bytes on disk: gzipped MessagePack chunks under `/weather_scalars//_.mp.gz`. There is no Elixir-only or Rust-only branch in the format. diff --git a/lib/microwaveprop/weather/scalar_file.ex b/lib/microwaveprop/weather/scalar_file.ex index b9456653..24baec7a 100644 --- a/lib/microwaveprop/weather/scalar_file.ex +++ b/lib/microwaveprop/weather/scalar_file.ex @@ -17,15 +17,26 @@ defmodule Microwaveprop.Weather.ScalarFile do a tile request can read just the chunks it needs and skip the `SoundingParams.derive/1` + `WeatherLayers.derive/1` work entirely. + ## Wire format + + Each chunk is a gzipped MessagePack `[row_map, ...]`. MessagePack was + picked over ETF so the Rust `prop_grid_rs` worker can produce these + files inline in the propagation pipeline using `rmp-serde` and Elixir + reads them with `Msgpax`. Both producers write identical bytes — there + is no Elixir-only or Rust-only branch. + + Map keys are strings on the wire (MessagePack has no atom type). + Whitelisted keys are atomized on read so callers see the same + Elixir-shaped row regardless of producer. + ## Layout /weather_scalars/ / # e.g. 2026-04-28T12:00:00Z/ - _.etf.gz + _.mp.gz - `lat_band = floor(lat / 5)`, `lon_band = floor(lon / 5)`. Each chunk is - the compressed ETF of `[row_map, ...]`. Writes go through - `rename(2)` → atomic on NFS. + `lat_band = floor(lat / 5)`, `lon_band = floor(lon / 5)`. Writes go + through `rename(2)` → atomic on NFS. ## Chunk size @@ -39,6 +50,34 @@ defmodule Microwaveprop.Weather.ScalarFile do @chunk_step 5 @subdir "weather_scalars" + # Keys produced by the writer that should land as atoms in Elixir. + # Anything outside this set stays a string and is ignored by callers + # (weather.ex / tile_renderer.ex use atom access throughout). + @atom_keys MapSet.new([ + "lat", + "lon", + "valid_time", + "temperature", + "dewpoint_depression", + "surface_rh", + "surface_pressure_mb", + "surface_refractivity", + "refractivity_gradient", + "bl_height", + "pwat", + "temp_850mb", + "dewpoint_850mb", + "temp_700mb", + "dewpoint_700mb", + "lapse_rate", + "mid_lapse_rate", + "inversion_strength", + "inversion_base_m", + "ducting", + "duct_base_m", + "duct_strength" + ]) + @type row :: %{required(:lat) => float(), required(:lon) => float(), optional(atom()) => term()} @type bounds :: %{optional(String.t()) => number()} @@ -90,8 +129,8 @@ defmodule Microwaveprop.Weather.ScalarFile do rows |> Enum.group_by(&chunk_key/1) |> Enum.each(fn {{lat_band, lon_band}, chunk_rows} -> - path = Path.join(dir, "#{lat_band}_#{lon_band}.etf.gz") - binary = :erlang.term_to_binary(chunk_rows, [:compressed]) + path = Path.join(dir, "#{lat_band}_#{lon_band}.mp.gz") + binary = encode_chunk(chunk_rows) tmp = path <> ".tmp." <> unique_suffix() File.write!(tmp, binary, [:binary]) @@ -222,7 +261,7 @@ defmodule Microwaveprop.Weather.ScalarFile do (value / @chunk_step) |> Float.floor() |> trunc() end - defp chunk_filename({lat_band, lon_band}), do: "#{lat_band}_#{lon_band}.etf.gz" + defp chunk_filename({lat_band, lon_band}), do: "#{lat_band}_#{lon_band}.mp.gz" defp chunk_intersects_bounds?(_key, nil), do: true @@ -243,17 +282,73 @@ defmodule Microwaveprop.Weather.ScalarFile do end) end - defp decode_chunk(path) do - case File.read(path) do - {:ok, binary} -> - :erlang.binary_to_term(binary) + # Encode rows as gzipped MessagePack. We strip atoms to strings on the + # wire (Msgpax doesn't speak atom keys natively, and Rust can't read + # them anyway) and re-atomize on the read side. + defp encode_chunk(rows) do + rows + |> Enum.map(&prepare_row_for_encode/1) + |> Msgpax.pack!(iodata: false) + |> :zlib.gzip() + end - {:error, reason} -> - Logger.warning("ScalarFile chunk read failed path=#{path} reason=#{inspect(reason)}") + # Convert atom keys to strings and DateTime values to ISO8601 strings + # so the row is representable in MessagePack. + defp prepare_row_for_encode(row) when is_map(row) do + Map.new(row, fn {k, v} -> + key = if is_atom(k), do: Atom.to_string(k), else: k + {key, encode_value(v)} + end) + end + + defp encode_value(%DateTime{} = dt), do: DateTime.to_iso8601(DateTime.truncate(dt, :second)) + defp encode_value(value), do: value + + defp decode_chunk(path) do + with {:ok, gz} <- File.read(path), + {:ok, binary} <- safe_gunzip(gz), + {:ok, decoded} <- Msgpax.unpack(binary) do + decoded + |> List.wrap() + |> Enum.map(&normalize_row/1) + else + error -> + Logger.warning("ScalarFile chunk read failed path=#{path} reason=#{inspect(error)}") [] end end + defp safe_gunzip(binary) do + {:ok, :zlib.gunzip(binary)} + rescue + e -> {:error, {:gunzip, e}} + end + + # Atomize whitelisted keys and re-parse the `valid_time` ISO8601 back + # into a DateTime so callers see the same shape as the in-memory rows + # produced by `Microwaveprop.Weather.build_grid_cache_rows/3`. + defp normalize_row(row) when is_map(row) do + Map.new(row, fn {k, v} -> + key = atomize_key(k) + {key, normalize_value(key, v)} + end) + end + + defp atomize_key(k) when is_binary(k) do + if MapSet.member?(@atom_keys, k), do: String.to_atom(k), else: k + end + + defp atomize_key(k), do: k + + defp normalize_value(:valid_time, v) when is_binary(v) do + case DateTime.from_iso8601(v) do + {:ok, dt, _offset} -> dt + _ -> v + end + end + + defp normalize_value(_key, v), do: v + defp list_chunk_files(%DateTime{} = valid_time) do dir = dir_for(valid_time) @@ -270,7 +365,7 @@ defmodule Microwaveprop.Weather.ScalarFile do end defp parse_chunk_filename(filename) do - with [_, lat_str, lon_str] <- Regex.run(~r/^(-?\d+)_(-?\d+)\.etf\.gz$/, filename), + with [_, lat_str, lon_str] <- Regex.run(~r/^(-?\d+)_(-?\d+)\.mp\.gz$/, filename), {lat_band, ""} <- Integer.parse(lat_str), {lon_band, ""} <- Integer.parse(lon_str) do {lat_band, lon_band} diff --git a/rust/prop_grid_rs/src/lib.rs b/rust/prop_grid_rs/src/lib.rs index ec69d16c..691cb531 100644 --- a/rust/prop_grid_rs/src/lib.rs +++ b/rust/prop_grid_rs/src/lib.rs @@ -50,3 +50,4 @@ pub mod scorer; pub mod scores_file; pub mod sounding_params; pub mod telemetry; +pub mod weather_scalar_file; diff --git a/rust/prop_grid_rs/src/pipeline.rs b/rust/prop_grid_rs/src/pipeline.rs index ef1154b6..56041f29 100644 --- a/rust/prop_grid_rs/src/pipeline.rs +++ b/rust/prop_grid_rs/src/pipeline.rs @@ -22,6 +22,7 @@ use crate::profiles_file::{self, CellEntry}; use crate::scorer::{self, Conditions}; use crate::scores_file::{self, ScorePoint}; use crate::sounding_params::{self, Level}; +use crate::weather_scalar_file::{self, ScalarRow}; #[derive(Debug, thiserror::Error)] pub enum PipelineError { @@ -33,6 +34,8 @@ pub enum PipelineError { Write(#[from] scores_file::WriteError), #[error("profile write: {0}")] ProfileWrite(#[from] profiles_file::WriteError), + #[error("scalar write: {0}")] + ScalarWrite(#[from] weather_scalar_file::WriteError), #[error("native duct: {0}")] NativeDuct(#[from] native_duct::NativeDuctError), #[error("nexrad: {0}")] @@ -147,12 +150,16 @@ pub async fn run_chain_step( let mut prepared: Vec<(f64, f64, Conditions, scorer::BandInvariants)> = Vec::with_capacity(merged.len()); let mut profile_entries: Vec = Vec::with_capacity(merged.len()); + let mut scalar_rows: Vec = Vec::with_capacity(merged.len()); for (key, cell) in merged.iter() { let (lat, lon) = decoder::key_to_latlon(*key); 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)); profile_entries.push(cell_to_profile_entry(lat, lon, cell)); + if let Some(row) = weather_scalar_file::derive_row(lat, lon, valid_time, cell) { + scalar_rows.push(row); + } } } drop(merged); @@ -170,6 +177,21 @@ pub async fn run_chain_step( }) }; + // Persist the derived scalar artifact alongside the raw profile so + // Elixir's `/weather` reads land on a pre-derived chunk file + // instead of falling back to the cold ProfilesFile decode + per-cell + // SoundingParams + WeatherLayers derivation. Identical wire format + // (gzipped MessagePack chunked by 5°×5°) to what the Elixir writer + // produces — the reader doesn't care which side wrote the bytes. + let scores_dir_for_scalars = scores_dir_owned.clone(); + let scalar_future = { + let rows = scalar_rows; + tokio::task::spawn_blocking(move || { + weather_scalar_file::write_atomic(&scores_dir_for_scalars, valid_time, &rows) + .map(|_| 0u32) + }) + }; + // Step 2: score and write every band in parallel. // // Scoring: rayon's thread pool over the 23 bands. Each band is @@ -219,6 +241,10 @@ pub async fn run_chain_step( .await .expect("blocking join") .map_err(PipelineError::ProfileWrite)?; + scalar_future + .await + .expect("blocking join") + .map_err(PipelineError::ScalarWrite)?; let files_written = bands.len() as u32; Ok(ChainStepStats { @@ -677,12 +703,16 @@ pub async fn run_analysis_step( let mut profile_entries: Vec = Vec::with_capacity(merged.len()); let mut prepared: Vec<(f64, f64, Conditions, scorer::BandInvariants)> = Vec::with_capacity(merged.len()); + let mut scalar_rows: Vec = Vec::with_capacity(merged.len()); for (key, cell) in merged.iter() { let (lat, lon) = decoder::key_to_latlon(*key); 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)); profile_entries.push(cell_to_profile_entry(lat, lon, cell)); + if let Some(row) = weather_scalar_file::derive_row(lat, lon, valid_time, cell) { + scalar_rows.push(row); + } } } @@ -699,6 +729,19 @@ pub async fn run_analysis_step( }) }; + // Same wire format as Elixir's `ScalarFile`. See run_chain_step + // (forecast hours) for the full rationale — both code paths go + // through the same writer so f00 and f01..f18 land identical + // chunked artifacts on NFS. + let scores_dir_for_scalars = scores_dir_owned.clone(); + let scalar_future = { + let rows = scalar_rows; + tokio::task::spawn_blocking(move || { + weather_scalar_file::write_atomic(&scores_dir_for_scalars, valid_time, &rows) + .map(|_| 0u32) + }) + }; + // Bands and score writes run in parallel, same shape as // run_chain_step. Rayon across bands, try_join_all across writes. let bands = band_config::all_bands(); @@ -738,6 +781,10 @@ pub async fn run_analysis_step( .await .expect("blocking join") .map_err(PipelineError::ProfileWrite)?; + scalar_future + .await + .expect("blocking join") + .map_err(PipelineError::ScalarWrite)?; let files_written = bands.len() as u32; Ok(AnalysisStepStats { diff --git a/rust/prop_grid_rs/src/weather_scalar_file.rs b/rust/prop_grid_rs/src/weather_scalar_file.rs new file mode 100644 index 00000000..37242cdb --- /dev/null +++ b/rust/prop_grid_rs/src/weather_scalar_file.rs @@ -0,0 +1,526 @@ +//! Per-cell *derived* weather scalars on disk, the cheap-read sibling +//! of [`profiles_file`][crate::profiles_file]. +//! +//! 1:1 wire-compatible with `Microwaveprop.Weather.ScalarFile` on the +//! Elixir side. Chunked into 5°×5° spatial buckets so a state-sized +//! `/weather` viewport reads only the chunks that overlap. +//! +//! ## Layout +//! +//! ```text +//! /weather_scalars/ +//! / # e.g. 2026-04-29T12:00:00Z/ +//! _.mp.gz +//! ``` +//! +//! `lat_band = floor(lat / 5)`, `lon_band = floor(lon / 5)`. Each chunk +//! is gzipped MessagePack: `[ {row}, {row}, … ]`. Map keys are strings +//! on the wire — the Elixir reader atomizes the whitelist on read, so +//! every key emitted here must match the Elixir whitelist or it gets +//! silently dropped by callers that use atom-style access. +//! +//! Scalar derivation mirrors `Microwaveprop.Weather.WeatherLayers.derive/1` +//! plus the surface fields `Microwaveprop.Weather.build_grid_cache_row/4` +//! adds on top. The Elixir module remains the single source of truth +//! for any cell that hasn't been Rust-materialized yet (the +//! `kickoff_async_scalar_materialize` fallback path). + +use std::collections::HashMap; +use std::fs::File; +use std::io::{BufWriter, Write}; +use std::path::{Path, PathBuf}; + +use chrono::{DateTime, Utc}; +use flate2::write::GzEncoder; +use flate2::Compression; +use serde::Serialize; + +use crate::decoder::CellValues; +use crate::fetcher; +use crate::sounding_params::{ + ducting_detected, min_refractivity_gradient, sat_vap_pres, surface_refractivity, Level, +}; + +const CHUNK_STEP: i32 = 5; +const SUBDIR: &str = "weather_scalars"; + +/// One row in a chunk file. Field names match the Elixir atom-key +/// whitelist exactly — adding a field here without updating the +/// Elixir `@atom_keys` set means the new field round-trips as a +/// string and is invisible to atom-style callers. +#[derive(Debug, Default, Serialize)] +pub struct ScalarRow { + pub lat: f64, + pub lon: f64, + pub valid_time: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub temperature: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub dewpoint_depression: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub surface_rh: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub surface_pressure_mb: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub surface_refractivity: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub refractivity_gradient: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub bl_height: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub pwat: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub temp_850mb: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub dewpoint_850mb: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub temp_700mb: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub dewpoint_700mb: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub lapse_rate: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub mid_lapse_rate: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub inversion_strength: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub inversion_base_m: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub ducting: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub duct_base_m: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub duct_strength: Option, +} + +#[derive(Debug, thiserror::Error)] +pub enum WriteError { + #[error("io: {0}")] + Io(#[from] std::io::Error), + #[error("encode: {0}")] + Encode(#[from] rmp_serde::encode::Error), +} + +/// Build a `ScalarRow` from a single grid cell. Returns `None` for +/// cells whose surface temperature is missing or out of physically +/// plausible range — matches Elixir's `build_grid_cache_row` filter +/// (`temp_c < -80 or temp_c > 60` are dropped). +pub fn derive_row( + lat: f64, + lon: f64, + valid_time: DateTime, + cell: &CellValues, +) -> Option { + let temp_c = cell + .get("TMP:2 m above ground") + .map(|&v| v as f64 - 273.15)?; + if !temp_c.is_finite() || !(-80.0..=60.0).contains(&temp_c) { + return None; + } + + let dewpoint_c = cell.get("DPT:2 m above ground").map(|&v| v as f64 - 273.15); + let dewpoint_depression = dewpoint_c.map(|d| temp_c - d); + let surface_pressure_mb = cell.get("PRES:surface").map(|&v| v as f64 / 100.0); + let bl_height = cell.get("HPBL:surface").map(|&v| v as f64); + let pwat = cell + .get("PWAT:entire atmosphere (considered as a single layer)") + .map(|&v| v as f64); + + let surface_rh = dewpoint_c.map(|d| 100.0 * sat_vap_pres(d) / sat_vap_pres(temp_c)); + + let levels = build_levels(cell); + let derived_min_grad = if levels.len() >= 3 { + min_refractivity_gradient(levels.clone()) + } else { + None + }; + let native_min_grad = cell.get("native_min_gradient").map(|&v| v as f64); + let refractivity_gradient = derived_min_grad.or(native_min_grad); + let ducting = Some(ducting_detected(refractivity_gradient)); + let surface_refractivity_val = surface_refractivity(&levels); + + let mut sorted: Vec<&Level> = levels.iter().collect(); + sorted.sort_by(|a, b| { + b.pres_mb + .partial_cmp(&a.pres_mb) + .unwrap_or(std::cmp::Ordering::Equal) + }); + + let temp_850mb = level_value(&sorted, 850.0, |l| Some(l.tmpc)); + let dewpoint_850mb = level_value(&sorted, 850.0, |l| l.dwpc); + let temp_700mb = level_value(&sorted, 700.0, |l| Some(l.tmpc)); + let dewpoint_700mb = level_value(&sorted, 700.0, |l| l.dwpc); + + let lapse_rate = compute_lapse_rate(&sorted); + let mid_lapse_rate = compute_layer_lapse_rate(&sorted, 850.0, 700.0); + let inversion_strength = compute_inversion_strength(&sorted); + let inversion_base_m = compute_inversion_base_m(&sorted); + + // Ducts: the f01..f18 pipeline only stores the duct-summary scalars + // (`max_duct_thickness_m`, `duct_count`) per cell. Duct base requires + // the full Duct list, which isn't threaded through `CellValues` yet — + // matches the current Elixir behavior on Rust-produced profiles + // (their `:duct_characteristics` is also nil). + let duct_count = cell.get("duct_count").copied().unwrap_or(0.0); + let duct_strength = if duct_count > 0.0 { + cell.get("max_duct_thickness_m").map(|&v| v as f64) + } else { + None + }; + + Some(ScalarRow { + lat, + lon, + valid_time: valid_time.format("%Y-%m-%dT%H:%M:%SZ").to_string(), + temperature: Some(temp_c), + dewpoint_depression, + surface_rh, + surface_pressure_mb, + surface_refractivity: surface_refractivity_val, + refractivity_gradient, + bl_height, + pwat, + temp_850mb, + dewpoint_850mb, + temp_700mb, + dewpoint_700mb, + lapse_rate, + mid_lapse_rate, + inversion_strength, + inversion_base_m, + ducting, + duct_base_m: None, + duct_strength, + }) +} + +/// Absolute path the 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) +} + +/// 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 +/// stale files behind. Each chunk is written `tmp + rename` for an +/// atomic NFS-friendly publish. +pub fn write_atomic( + scores_dir: &Path, + valid_time: DateTime, + rows: &[ScalarRow], +) -> Result { + let dir = dir_for(scores_dir, valid_time); + std::fs::create_dir_all(&dir)?; + + if let Ok(entries) = std::fs::read_dir(&dir) { + for entry in entries.flatten() { + let _ = std::fs::remove_file(entry.path()); + } + } + + let mut chunks: HashMap<(i32, i32), Vec<&ScalarRow>> = HashMap::new(); + for row in rows { + let key = (chunk_band(row.lat), chunk_band(row.lon)); + chunks.entry(key).or_default().push(row); + } + + for ((lat_band, lon_band), chunk_rows) in chunks { + let path = dir.join(format!("{lat_band}_{lon_band}.mp.gz")); + + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + let pid = std::process::id(); + let mut tmp = path.clone().into_os_string(); + tmp.push(format!(".tmp.{nanos}.{pid}")); + let tmp = PathBuf::from(tmp); + + { + let buf = rmp_serde::to_vec_named(&chunk_rows)?; + let file = File::create(&tmp)?; + let mut gz = GzEncoder::new(BufWriter::new(file), Compression::default()); + gz.write_all(&buf)?; + gz.finish()?; + } + + std::fs::rename(&tmp, &path)?; + } + + Ok(dir) +} + +// ── Helpers ────────────────────────────────────────────────────────── + +fn chunk_band(value: f64) -> i32 { + (value / CHUNK_STEP as f64).floor() as i32 +} + +fn build_levels(cell: &CellValues) -> Vec { + fetcher::grid_level_keys() + .iter() + .filter_map(|k| { + let t = cell.get(k.tmp.as_str()).copied()?; + let h = cell.get(k.hgt.as_str()).copied()?; + let d = cell.get(k.dpt.as_str()).copied(); + Some(Level { + pres_mb: k.pres_mb, + hght_m: h as f64, + tmpc: (t as f64) - 273.15, + dwpc: d.map(|v| (v as f64) - 273.15), + }) + }) + .collect() +} + +/// Nearest level to `target_pres` within ±25 mb, mirroring Elixir's +/// `WeatherLayers.level_value/3`. +fn level_value( + sorted: &[&Level], + target_pres: f64, + project: impl Fn(&Level) -> Option, +) -> Option { + let nearest = sorted.iter().min_by(|a, b| { + (a.pres_mb - target_pres) + .abs() + .partial_cmp(&(b.pres_mb - target_pres).abs()) + .unwrap_or(std::cmp::Ordering::Equal) + })?; + if (nearest.pres_mb - target_pres).abs() <= 25.0 { + project(nearest) + } else { + None + } +} + +fn compute_lapse_rate(sorted: &[&Level]) -> Option { + if sorted.len() < 2 { + return None; + } + let surface = sorted.first()?; + let top = sorted.last()?; + let dh_km = (top.hght_m - surface.hght_m) / 1000.0; + if dh_km > 0.0 { + Some((surface.tmpc - top.tmpc) / dh_km) + } else { + None + } +} + +fn compute_layer_lapse_rate(sorted: &[&Level], lower_pres: f64, upper_pres: f64) -> Option { + if !(lower_pres > upper_pres) { + return None; + } + let lower = sorted + .iter() + .find(|l| (l.pres_mb - lower_pres).abs() <= 25.0)?; + let upper = sorted + .iter() + .find(|l| (l.pres_mb - upper_pres).abs() <= 25.0)?; + let dh_km = (upper.hght_m - lower.hght_m) / 1000.0; + if dh_km > 0.0 { + Some((lower.tmpc - upper.tmpc) / dh_km) + } else { + None + } +} + +fn compute_inversion_strength(sorted: &[&Level]) -> Option { + if sorted.is_empty() { + return Some(0.0); + } + let mut max_str = 0.0_f64; + for pair in sorted.windows(2) { + let dt = pair[1].tmpc - pair[0].tmpc; + if dt > max_str { + max_str = dt; + } + } + Some(max_str) +} + +fn compute_inversion_base_m(sorted: &[&Level]) -> Option { + if sorted.is_empty() { + return None; + } + let sfc_hght = sorted.first()?.hght_m; + let mut max_str = 0.0_f64; + let mut base: Option = None; + for pair in sorted.windows(2) { + let dt = pair[1].tmpc - pair[0].tmpc; + if dt > 0.0 && dt > max_str { + max_str = dt; + base = Some(pair[0].hght_m - sfc_hght); + } + } + if max_str > 0.0 { + base + } else { + None + } +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::TimeZone; + use flate2::read::GzDecoder; + use std::io::Read; + use std::sync::Arc; + + fn cell_with(items: &[(&str, f32)]) -> CellValues { + items + .iter() + .map(|&(k, v)| (Arc::::from(k), v)) + .collect() + } + + fn surface_only_cell() -> CellValues { + cell_with(&[ + ("TMP:2 m above ground", 295.65), // 22.5 °C + ("DPT:2 m above ground", 285.65), // 12.5 °C → depression 10 °C + ("PRES:surface", 101_320.0), // 1013.2 mb + ("HPBL:surface", 800.0), + ( + "PWAT:entire atmosphere (considered as a single layer)", + 25.0, + ), + ]) + } + + #[test] + fn drops_cells_with_unphysical_surface_temp() { + let mut cell = surface_only_cell(); + cell.insert("TMP:2 m above ground".into(), 100.0); // way too cold + let vt = Utc.with_ymd_and_hms(2026, 4, 29, 12, 0, 0).unwrap(); + assert!(derive_row(33.0, -97.0, vt, &cell).is_none()); + } + + #[test] + fn surface_only_row_carries_basic_fields() { + let cell = surface_only_cell(); + let vt = Utc.with_ymd_and_hms(2026, 4, 29, 12, 0, 0).unwrap(); + let row = derive_row(33.0, -97.0, vt, &cell).expect("row"); + + assert_eq!(row.lat, 33.0); + assert_eq!(row.lon, -97.0); + assert_eq!(row.valid_time, "2026-04-29T12:00:00Z"); + assert!((row.temperature.unwrap() - 22.5).abs() < 1e-3); + assert!((row.dewpoint_depression.unwrap() - 10.0).abs() < 1e-3); + assert!((row.surface_pressure_mb.unwrap() - 1013.2).abs() < 1e-3); + assert!(row.surface_rh.unwrap() > 30.0 && row.surface_rh.unwrap() < 70.0); + assert_eq!(row.bl_height, Some(800.0)); + assert_eq!(row.pwat, Some(25.0)); + + // No pressure-level data → upper-air fields stay nil. + assert_eq!(row.temp_850mb, None); + assert_eq!(row.lapse_rate, None); + } + + #[test] + fn upper_air_levels_derive_lapse_rate_and_850mb() { + // Need real fetcher::GRID_PRESSURE_LEVELS keys. Build directly + // from grid_level_keys() so the synthetic cell mirrors prod. + let mut cell = surface_only_cell(); + for k in fetcher::grid_level_keys() { + // Plausible synthetic profile: lapse 6.5 °C/km, scale height + // ≈ 8 km. Heights derived from std atmosphere approximation. + let h_m = match k.pres_mb as i32 { + 1000 => 100.0, + 925 => 800.0, + 850 => 1500.0, + 700 => 3000.0, + 500 => 5600.0, + 250 => 10_400.0, + _ => 1500.0 + (1000.0 - k.pres_mb) * 8.0, + }; + let t_c = 22.5 - h_m * 6.5e-3; // °C + let d_c = t_c - 5.0; + cell.insert(k.tmp.clone().into(), (t_c + 273.15) as f32); + cell.insert(k.dpt.clone().into(), (d_c + 273.15) as f32); + cell.insert(k.hgt.clone().into(), h_m as f32); + } + + let vt = Utc.with_ymd_and_hms(2026, 4, 29, 12, 0, 0).unwrap(); + let row = derive_row(33.0, -97.0, vt, &cell).expect("row"); + + let temp_850 = row.temp_850mb.expect("850 mb T"); + // h_m at 850 mb is 1500 → t_c = 22.5 - 1500 * 6.5e-3 = 12.75 °C + assert!( + (temp_850 - 12.75).abs() < 0.5, + "expected ~12.75 °C at 850 mb, got {temp_850}" + ); + + let lapse = row.lapse_rate.expect("lapse_rate"); + assert!( + (lapse - 6.5).abs() < 0.5, + "expected ~6.5 °C/km lapse rate, got {lapse}" + ); + + let mid = row.mid_lapse_rate.expect("mid_lapse_rate"); + assert!( + (mid - 6.5).abs() < 0.5, + "expected ~6.5 mid lapse, got {mid}" + ); + } + + #[test] + fn write_atomic_round_trip_via_msgpack() { + let dir = tempfile::tempdir().unwrap(); + let vt = Utc.with_ymd_and_hms(2026, 4, 29, 12, 0, 0).unwrap(); + + let row = ScalarRow { + lat: 33.0, + lon: -97.0, + valid_time: "2026-04-29T12:00:00Z".to_string(), + temperature: Some(22.5), + dewpoint_depression: Some(10.0), + surface_rh: Some(50.0), + ducting: Some(false), + ..ScalarRow::default() + }; + + write_atomic(dir.path(), vt, &[row]).unwrap(); + + let chunk = dir + .path() + .join(SUBDIR) + .join("2026-04-29T12:00:00Z") + .join("6_-20.mp.gz"); + assert!(chunk.exists()); + + let raw = std::fs::read(&chunk).unwrap(); + let mut gz = GzDecoder::new(&raw[..]); + let mut buf = Vec::new(); + gz.read_to_end(&mut buf).unwrap(); + let decoded: rmpv::Value = rmp_serde::from_slice(&buf).unwrap(); + let arr = decoded.as_array().expect("top-level array"); + assert_eq!(arr.len(), 1); + + let rmpv::Value::Map(pairs) = &arr[0] else { + panic!("row should be a map, got {:?}", arr[0]); + }; + let lat = pairs + .iter() + .find(|(k, _)| k.as_str() == Some("lat")) + .unwrap(); + assert_eq!(lat.1.as_f64(), Some(33.0)); + + let temp = pairs + .iter() + .find(|(k, _)| k.as_str() == Some("temperature")) + .unwrap(); + assert_eq!(temp.1.as_f64(), Some(22.5)); + } + + #[test] + fn chunk_band_floors_match_elixir() { + // floor(33.0 / 5) = 6, floor(-97.0 / 5) = -20 (matches Elixir's + // `chunk_band` Float.floor/trunc). + assert_eq!(chunk_band(33.0), 6); + assert_eq!(chunk_band(-97.0), -20); + assert_eq!(chunk_band(35.0), 7); + assert_eq!(chunk_band(-95.0), -19); + } +} diff --git a/test/microwaveprop/weather/scalar_file_test.exs b/test/microwaveprop/weather/scalar_file_test.exs index e6180634..bb54cd83 100644 --- a/test/microwaveprop/weather/scalar_file_test.exs +++ b/test/microwaveprop/weather/scalar_file_test.exs @@ -155,6 +155,54 @@ defmodule Microwaveprop.Weather.ScalarFileTest do end 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]