diff --git a/rust/prop_grid_rs/Cargo.lock b/rust/prop_grid_rs/Cargo.lock index 1cceaf3f..71fc0918 100644 --- a/rust/prop_grid_rs/Cargo.lock +++ b/rust/prop_grid_rs/Cargo.lock @@ -1355,6 +1355,7 @@ dependencies = [ "byteorder", "bytes", "chrono", + "flate2", "futures", "num_cpus", "png", @@ -1362,6 +1363,8 @@ dependencies = [ "prometheus", "rayon", "reqwest", + "rmp-serde", + "rmpv", "serde", "serde_json", "sqlx", @@ -1618,6 +1621,36 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "rmp" +version = "0.8.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ba8be72d372b2c9b35542551678538b562e7cf86c3315773cae48dfbfe7790c" +dependencies = [ + "num-traits", +] + +[[package]] +name = "rmp-serde" +version = "1.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72f81bee8c8ef9b577d1681a70ebbc962c232461e397b22c208c43c04b67a155" +dependencies = [ + "rmp", + "serde", +] + +[[package]] +name = "rmpv" +version = "1.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7a4e1d4b9b938a26d2996af33229f0ca0956c652c1375067f0b45291c1df8417" +dependencies = [ + "rmp", + "serde", + "serde_bytes", +] + [[package]] name = "rsa" version = "0.9.10" @@ -1726,6 +1759,16 @@ dependencies = [ "serde_derive", ] +[[package]] +name = "serde_bytes" +version = "0.11.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a5d440709e79d88e51ac01c4b72fc6cb7314017bb7da9eeff678aa94c10e3ea8" +dependencies = [ + "serde", + "serde_core", +] + [[package]] name = "serde_core" version = "1.0.228" diff --git a/rust/prop_grid_rs/Cargo.toml b/rust/prop_grid_rs/Cargo.toml index bdcf8974..e026e65d 100644 --- a/rust/prop_grid_rs/Cargo.toml +++ b/rust/prop_grid_rs/Cargo.toml @@ -33,6 +33,9 @@ rayon = "1.10" axum = { version = "0.7", default-features = false, features = ["http1", "tokio"] } prometheus = { version = "0.13", default-features = false } png = "0.17" +rmp-serde = "1.3" +rmpv = { version = "1.3", features = ["with-serde"] } +flate2 = "1" [dev-dependencies] tokio = { version = "1", features = ["full", "test-util"] } diff --git a/rust/prop_grid_rs/src/lib.rs b/rust/prop_grid_rs/src/lib.rs index bebedd27..c0e7f4d9 100644 --- a/rust/prop_grid_rs/src/lib.rs +++ b/rust/prop_grid_rs/src/lib.rs @@ -29,6 +29,7 @@ pub mod metrics; pub mod native_duct; pub mod nexrad; pub mod pipeline; +pub mod profiles_file; pub mod region; pub mod scores_file; pub mod scorer; diff --git a/rust/prop_grid_rs/src/profiles_file.rs b/rust/prop_grid_rs/src/profiles_file.rs new file mode 100644 index 00000000..1dfb1e8d --- /dev/null +++ b/rust/prop_grid_rs/src/profiles_file.rs @@ -0,0 +1,222 @@ +//! On-disk profile store in MessagePack + gzip. +//! +//! Rust-side writer for the f00 ProfilesFile. Replaces the Elixir ETF +//! format (`:erlang.term_to_binary/2`, `.etf.gz`) so Rust can produce +//! the files without porting the Erlang serializer. Elixir's +//! `ProfilesFile` module learns to read `.mp.gz` alongside the legacy +//! `.etf.gz` during the cutover. +//! +//! Wire layout (top-level msgpack map): +//! +//! ```text +//! { +//! "v": 1, +//! "valid_time": "2026-04-19T14:00:00Z", +//! "cells": [ +//! { +//! "lat": 32.9, "lon": -97.0, +//! "profile": { +//! "native_min_gradient": -150.5, +//! "best_duct_freq_ghz": 24.0, +//! "duct_count": 1, +//! "ducts": [ +//! {"base_m": 100, "top_m": 200, "thickness_m": 100, +//! "m_deficit": 12.3, "min_freq_ghz": 15.2} +//! ], +//! "surface_temp_c": 22.5, +//! "surface_dewpoint_c": 18.1, +//! "surface_pressure_mb": 1013.2, +//! "hpbl_m": 320.0, +//! "pwat_mm": 25.3, +//! "profile": [ +//! {"pres_mb":1000,"hght_m":100,"tmpc":20.0,"dwpc":18.0}, +//! ... +//! ] +//! } +//! } +//! ] +//! } +//! ``` +//! +//! All keys are strings — MessagePack has no atom type, and Elixir's +//! reader converts known keys back to atoms via a whitelist. + +use std::fs::File; +use std::io::{BufWriter, Write}; +use std::path::{Path, PathBuf}; + +use chrono::{DateTime, Utc}; +use flate2::Compression; +use flate2::write::GzEncoder; +use serde::Serialize; + +/// Bump whenever the on-disk layout changes in a way that isn't a +/// transparent superset. Elixir reader gates on this. +pub const FORMAT_VERSION: u32 = 1; + +/// One grid cell to persist. `profile` is an arbitrary map of +/// string-keyed values — sized at write time. Kept as a +/// `rmp_serde::Value` so the writer stays agnostic about what the +/// f00 pipeline decides to include. +#[derive(Debug, Serialize)] +pub struct CellEntry { + pub lat: f64, + pub lon: f64, + pub profile: rmpv::Value, +} + +#[derive(Debug, Serialize)] +struct FileBody<'a> { + v: u32, + valid_time: String, + cells: &'a [CellEntry], +} + +#[derive(Debug, thiserror::Error)] +pub enum WriteError { + #[error("io: {0}")] + Io(#[from] std::io::Error), + #[error("encode: {0}")] + Encode(#[from] rmp_serde::encode::Error), +} + +/// Absolute path the writer lands at for `valid_time`. +/// Sibling of the Elixir `.etf.gz` file; the Elixir reader prefers +/// `.mp.gz` when both exist. +pub fn path_for(scores_dir: &Path, valid_time: DateTime) -> PathBuf { + let iso = valid_time.format("%Y-%m-%dT%H:%M:%SZ").to_string(); + scores_dir.join("profiles").join(format!("{iso}.mp.gz")) +} + +/// Write `cells` to `/profiles/.mp.gz` atomically. +/// Pattern mirrors `scores_file::write_atomic`: write to a tmp sibling, +/// rename into place. The NFS rename is atomic, so a concurrent reader +/// sees either the old file, the new file, or nothing — never a torn +/// write. +pub fn write_atomic( + scores_dir: &Path, + valid_time: DateTime, + cells: &[CellEntry], +) -> Result { + let path = path_for(scores_dir, valid_time); + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent)?; + } + 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 file = File::create(&tmp)?; + let mut gz = GzEncoder::new(BufWriter::new(file), Compression::default()); + let body = FileBody { + v: FORMAT_VERSION, + valid_time: valid_time.format("%Y-%m-%dT%H:%M:%SZ").to_string(), + cells, + }; + let buf = rmp_serde::to_vec_named(&body)?; + gz.write_all(&buf)?; + gz.finish()?; + } + + std::fs::rename(&tmp, &path)?; + Ok(path) +} + +/// Round a lat/lon to 3 decimal places — matches Elixir's +/// `Float.round(_, 3)` snap before keying into the profiles map. +pub fn snap_coords(lat: f64, lon: f64) -> (f64, f64) { + ((lat * 1000.0).round() / 1000.0, (lon * 1000.0).round() / 1000.0) +} + +/// Convenience helper: build a `rmpv::Value` map from an +/// iterator of `(name, value)` pairs. +pub fn value_map(entries: impl IntoIterator) -> rmpv::Value { + let pairs: Vec<(rmpv::Value, rmpv::Value)> = entries + .into_iter() + .map(|(k, v)| (rmpv::Value::String(k.into()), v)) + .collect(); + rmpv::Value::Map(pairs) +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::TimeZone; + use flate2::read::GzDecoder; + use std::io::Read; + + #[test] + fn writes_and_reads_round_trip() { + let dir = tempfile::tempdir().unwrap(); + let vt = Utc.with_ymd_and_hms(2026, 4, 19, 14, 0, 0).unwrap(); + let cells = vec![CellEntry { + lat: 32.9, + lon: -97.0, + profile: value_map([ + ("surface_temp_c", rmpv::Value::F64(22.5)), + ("surface_pressure_mb", rmpv::Value::F64(1013.2)), + ("duct_count", rmpv::Value::Integer(1.into())), + ]), + }]; + + let path = write_atomic(dir.path(), vt, &cells).unwrap(); + assert!(path.exists()); + assert!(path.to_string_lossy().ends_with(".mp.gz")); + + // Decode and confirm the top-level shape. + let raw = std::fs::read(&path).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 map = decoded.as_map().unwrap(); + assert!(map.iter().any(|(k, v)| k.as_str() == Some("v") && v.as_u64() == Some(1))); + assert!(map + .iter() + .any(|(k, v)| k.as_str() == Some("valid_time") && v.as_str() == Some("2026-04-19T14:00:00Z"))); + let cells_val = &map.iter().find(|(k, _)| k.as_str() == Some("cells")).unwrap().1; + let cells_arr = cells_val.as_array().unwrap(); + assert_eq!(cells_arr.len(), 1); + } + + #[test] + fn atomic_rename_replaces_existing() { + let dir = tempfile::tempdir().unwrap(); + let vt = Utc.with_ymd_and_hms(2026, 4, 19, 14, 0, 0).unwrap(); + let first = vec![CellEntry { + lat: 32.0, lon: -97.0, + profile: value_map([("v", rmpv::Value::Integer(1.into()))]), + }]; + let second = vec![CellEntry { + lat: 32.0, lon: -97.0, + profile: value_map([("v", rmpv::Value::Integer(2.into()))]), + }]; + let p1 = write_atomic(dir.path(), vt, &first).unwrap(); + let p2 = write_atomic(dir.path(), vt, &second).unwrap(); + assert_eq!(p1, p2); + // Only one file in the profiles dir — the tmp was renamed away. + let entries: Vec<_> = std::fs::read_dir(dir.path().join("profiles")) + .unwrap() + .map(|e| e.unwrap().file_name().into_string().unwrap()) + .collect(); + assert_eq!(entries.len(), 1); + } + + #[test] + fn path_for_layout() { + let vt = Utc.with_ymd_and_hms(2026, 4, 19, 14, 0, 0).unwrap(); + let p = path_for(std::path::Path::new("/data/scores"), vt); + assert_eq!(p.to_str().unwrap(), "/data/scores/profiles/2026-04-19T14:00:00Z.mp.gz"); + } + + #[test] + fn snap_coords_rounds_to_three_decimals() { + assert_eq!(snap_coords(32.8998, -97.04031), (32.900, -97.040)); + } +}