From f0fbb452f5a4ddb05f1ac6c33db8a6bb8c9b8ac0 Mon Sep 17 00:00:00 2001 From: Josh Cunningham Date: Tue, 7 Apr 2026 11:25:52 -0500 Subject: [PATCH 1/7] use ipc for subprocesses instead of cmd args --- Cargo.lock | 229 ++++++++++++++++++++++++++++++++++++++++++++++++++-- Cargo.toml | 1 + src/main.rs | 152 +++++++++++++++++++++++++--------- 3 files changed, 340 insertions(+), 42 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7cdc6fb..5595d65 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -164,6 +164,7 @@ dependencies = [ "clap", "dirs", "indicatif", + "ipc-channel", "netcdf", "pyo3", "rusqlite", @@ -280,6 +281,15 @@ dependencies = [ "cc", ] +[[package]] +name = "cobs" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fa961b519f0b462e3a3b4a34b64d119eeaca1d59af726fe450bbba07a9fc0a1" +dependencies = [ + "thiserror 2.0.18", +] + [[package]] name = "colorchoice" version = "1.0.4" @@ -440,6 +450,18 @@ version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719" +[[package]] +name = "embedded-io" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef1a6892d9eef45c8fa6b9e0086428a2cca8491aca8f787c534a3d6d0bcb3ced" + +[[package]] +name = "embedded-io" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "edd0f118536f44f5ccd48bcb8b111bdc3de888b58c74639dfb034a357d0f206d" + [[package]] name = "encode_unicode" version = "1.0.0" @@ -779,6 +801,25 @@ dependencies = [ "rustversion", ] +[[package]] +name = "ipc-channel" +version = "0.21.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a441490012d80e9aea75fb27503df8e87e9557dcfc6fe4244dde86bfc12e94e3" +dependencies = [ + "crossbeam-channel", + "libc", + "mio", + "postcard", + "rand", + "rustc-hash", + "serde_core", + "tempfile", + "thiserror 2.0.18", + "uuid", + "windows", +] + [[package]] name = "is_terminal_polyfill" version = "1.70.2" @@ -828,9 +869,9 @@ checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2" [[package]] name = "libc" -version = "0.2.177" +version = "0.2.184" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2874a2af47a2325c2001a6e6fad9b16a53b802102b528163885171cf92b15976" +checksum = "48f5d2a454e16a5ea0f4ced81bd44e4cfc7bd3a507b61887c99fd3538b28e4af" [[package]] name = "libloading" @@ -839,7 +880,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d7c4b02199fee7c5d21a5ae7d8cfa79a6ef5bb2fc834d6e9058e89c825efdc55" dependencies = [ "cfg-if", - "windows-link", + "windows-link 0.2.1", ] [[package]] @@ -958,6 +999,17 @@ dependencies = [ "simd-adler32", ] +[[package]] +name = "mio" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "50b7e5b27aa02a74bac8c3f23f448f8d87ff11f92d3aac1a6ed369ee08cc56c1" +dependencies = [ + "libc", + "wasi", + "windows-sys 0.61.2", +] + [[package]] name = "moka" version = "0.12.14" @@ -1205,7 +1257,7 @@ dependencies = [ "libc", "redox_syscall", "smallvec", - "windows-link", + "windows-link 0.2.1", ] [[package]] @@ -1258,6 +1310,27 @@ dependencies = [ "winapi", ] +[[package]] +name = "postcard" +version = "1.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6764c3b5dd454e283a30e6dfe78e9b31096d9e32036b5d1eaac7a6119ccb9a24" +dependencies = [ + "cobs", + "embedded-io 0.4.0", + "embedded-io 0.6.1", + "serde", +] + +[[package]] +name = "ppv-lite86" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" +dependencies = [ + "zerocopy", +] + [[package]] name = "prettyplease" version = "0.2.37" @@ -1373,6 +1446,35 @@ version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "rand" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6db2770f06117d490610c7488547d543617b21bfa07796d7a12f6f1bd53850d1" +dependencies = [ + "rand_chacha", + "rand_core", +] + +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core", +] + +[[package]] +name = "rand_core" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +dependencies = [ + "getrandom 0.3.4", +] + [[package]] name = "rawpointer" version = "0.2.1" @@ -1471,6 +1573,12 @@ dependencies = [ "smallvec", ] +[[package]] +name = "rustc-hash" +version = "2.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94300abf3f1ae2e2b8ffb7b58043de3d399c73fa6f4b73826402a5c457614dbe" + [[package]] name = "rustc_version" version = "0.4.1" @@ -1990,12 +2098,114 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" +[[package]] +name = "windows" +version = "0.61.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9babd3a767a4c1aef6900409f85f5d53ce2544ccdfaa86dad48c91782c6d6893" +dependencies = [ + "windows-collections", + "windows-core", + "windows-future", + "windows-link 0.1.3", + "windows-numerics", +] + +[[package]] +name = "windows-collections" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3beeceb5e5cfd9eb1d76b381630e82c4241ccd0d27f1a39ed41b2760b255c5e8" +dependencies = [ + "windows-core", +] + +[[package]] +name = "windows-core" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0fdd3ddb90610c7638aa2b3a3ab2904fb9e5cdbecc643ddb3647212781c4ae3" +dependencies = [ + "windows-implement", + "windows-interface", + "windows-link 0.1.3", + "windows-result", + "windows-strings", +] + +[[package]] +name = "windows-future" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc6a41e98427b19fe4b73c550f060b59fa592d7d686537eebf9385621bfbad8e" +dependencies = [ + "windows-core", + "windows-link 0.1.3", + "windows-threading", +] + +[[package]] +name = "windows-implement" +version = "0.60.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "windows-interface" +version = "0.59.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "windows-link" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e6ad25900d524eaabdbbb96d20b4311e1e7ae1699af4fb28c17ae66c80d798a" + [[package]] name = "windows-link" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" +[[package]] +name = "windows-numerics" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9150af68066c4c5c07ddc0ce30421554771e528bde427614c61038bc2c92c2b1" +dependencies = [ + "windows-core", + "windows-link 0.1.3", +] + +[[package]] +name = "windows-result" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56f42bd332cc6c8eac5af113fc0c1fd6a8fd2aa08a0119358686e5160d0586c6" +dependencies = [ + "windows-link 0.1.3", +] + +[[package]] +name = "windows-strings" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56e6c93f3a0c3b36176cb1327a4958a0353d5d166c2a35cb268ace15e91d3b57" +dependencies = [ + "windows-link 0.1.3", +] + [[package]] name = "windows-sys" version = "0.48.0" @@ -2020,7 +2230,7 @@ version = "0.61.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" dependencies = [ - "windows-link", + "windows-link 0.2.1", ] [[package]] @@ -2054,6 +2264,15 @@ dependencies = [ "windows_x86_64_msvc 0.52.6", ] +[[package]] +name = "windows-threading" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b66463ad2e0ea3bbf808b7f1d371311c80e115c0b71d60efc142cafbcfb057a6" +dependencies = [ + "windows-link 0.1.3", +] + [[package]] name = "windows_aarch64_gnullvm" version = "0.48.5" diff --git a/Cargo.toml b/Cargo.toml index 7881c62..c2b235d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,6 +25,7 @@ rusqlite = { version = "0.37.0" } clap = { version = "4.5.53", features = ["derive"] } anyhow = "1.0.100" indicatif = "0.17" +ipc-channel = "0.21" toml = "0.8" dirs = "5" zarrs = { version = "0.23", features = ["filesystem"], optional = true } diff --git a/src/main.rs b/src/main.rs index 473942e..14ccfa4 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,9 +1,10 @@ use bmi_driver::{preload_dependencies, BmiError, DivideDataStore, ModelRunner, OutputFormat}; use clap::Parser; use indicatif::{MultiProgress, ProgressBar, ProgressStyle}; +use ipc_channel::ipc::{IpcOneShotServer, IpcSender}; use std::env; use std::fs; -use std::io::{self, BufRead, BufReader, Write}; +use std::io::{self, Write}; use std::path::PathBuf; use std::process::{Command, Stdio}; use std::thread; @@ -25,6 +26,17 @@ struct TomlConfig { progress: Option, } +/// Messages sent from worker processes to the parent over IPC. +#[derive(serde::Serialize, serde::Deserialize, Debug)] +enum WorkerMessage { + /// Worker completed `count` locations so far (cumulative). + Progress(u64), + /// Worker finished successfully. + Done, + /// Worker encountered an error. + Error(String), +} + fn load_toml_config() -> TomlConfig { let config_path = dirs::config_dir() .unwrap_or_else(|| PathBuf::from("~/.config")) @@ -46,6 +58,10 @@ fn load_toml_config() -> TomlConfig { struct Args { data_dir: PathBuf, + /// Path to the realization.json file + #[arg(long)] + config: Option, + /// Number of parallel worker processes (default: number of CPUs) #[arg(short = 'j', long)] jobs: Option, @@ -62,6 +78,10 @@ struct Args { #[arg(long, hide = true)] worker_end: Option, + /// Internal: IPC server name for worker→parent communication + #[arg(long, hide = true)] + ipc_server: Option, + /// Print all unit conversion info and exit without running #[arg(long)] units: bool, @@ -89,7 +109,9 @@ fn main() -> Result<(), BmiError> { let config_dir = data_dir.join("config"); let _ = env::set_current_dir(&data_dir); - let realization = config_dir.join("realization.json"); + let realization = args + .config + .unwrap_or_else(|| config_dir.join("realization.json")); if args.minify { bmi_driver::config::minify_file(&realization)?; @@ -115,8 +137,6 @@ fn main() -> Result<(), BmiError> { let locations: Vec = rows.flatten().collect(); // Apply node-level partitioning for multi-node / SLURM job-array use. - // --node-start and --node-count carve out this node's slice of the location list - // before the internal worker processes further sub-divide it with -j. let node_start = args.node_start.min(locations.len()); let node_end = if args.node_count > 0 { (node_start + args.node_count).min(locations.len()) @@ -137,8 +157,10 @@ fn main() -> Result<(), BmiError> { .or(toml_cfg.progress) .unwrap_or(ProgressMode::Summary); - if let (Some(start), Some(end)) = (args.worker_start, args.worker_end) { - run_worker(&realization, &locations[start..end], &data_dir, start) + if let (Some(start), Some(end), Some(ipc_server)) = + (args.worker_start, args.worker_end, args.ipc_server) + { + run_worker(&realization, &locations[start..end], &data_dir, start, &ipc_server) } else { run_parent(&data_dir, &realization, &locations, jobs, progress) } @@ -319,20 +341,31 @@ fn run_parent( let end = ((i + 1) * chunk_size).min(n_locations); let worker_count = (end - start) as u64; + // Create an IPC one-shot server for this worker + let (server, server_name) = + IpcOneShotServer::::new().map_err(|e| BmiError::FunctionFailed { + model: "runner".into(), + func: format!("Failed to create IPC server: {}", e), + })?; + let mut cmd = Command::new(&exe); cmd.arg(data_dir) + .arg("--config") + .arg(&realization) .arg("--worker-start") .arg(start.to_string()) .arg("--worker-end") - .arg(end.to_string()); + .arg(end.to_string()) + .arg("--ipc-server") + .arg(&server_name); #[cfg(feature = "zarr")] if output_format == OutputFormat::Zarr && !output_vars_csv.is_empty() { cmd.arg("--output-vars").arg(&output_vars_csv); } - let mut child = cmd - .stdout(Stdio::piped()) + let child = cmd + .stdout(Stdio::null()) .stderr(Stdio::inherit()) .spawn() .map_err(|e| BmiError::FunctionFailed { @@ -347,28 +380,40 @@ fn run_parent( pb }); - let stdout = child.stdout.take().unwrap(); let overall_pb = overall_pb.clone(); let handle = thread::spawn(move || { - let reader = BufReader::new(stdout); - let mut prev = 0u64; - for line in reader.lines().flatten() { - if let Ok(count) = line.trim().parse::() { - let delta = count - prev; - if let Some(pb) = &worker_pb { - pb.inc(delta); - } - if let Some(pb) = &overall_pb { - pb.inc(delta); + // Accept the worker's connection — blocks until the worker connects and sends + // its first message. + let (rx, first_msg) = match server.accept() { + Ok(pair) => pair, + Err(e) => { + eprintln!("IPC accept failed for worker: {}", e); + return child; + } + }; + + // Process the first message + handle_worker_message(&first_msg, &worker_pb, &overall_pb); + + // Read remaining messages until the channel closes or worker signals Done/Error + loop { + match rx.recv() { + Ok(msg) => { + let done = matches!(msg, WorkerMessage::Done | WorkerMessage::Error(_)); + handle_worker_message(&msg, &worker_pb, &overall_pb); + if done { + break; + } } - prev = count; + Err(_) => break, // Channel closed } } + if let Some(pb) = worker_pb { pb.finish_and_clear(); } - child.wait() + child }); handles.push(handle); } @@ -376,15 +421,17 @@ fn run_parent( let mut failed = false; for handle in handles { match handle.join() { - Ok(Ok(status)) if status.success() => {} - Ok(Ok(status)) => { - eprintln!("Worker exited with: {}", status); - failed = true; - } - Ok(Err(e)) => { - eprintln!("Worker error: {}", e); - failed = true; - } + Ok(mut child) => match child.wait() { + Ok(status) if status.success() => {} + Ok(status) => { + eprintln!("Worker exited with: {}", status); + failed = true; + } + Err(e) => { + eprintln!("Worker error: {}", e); + failed = true; + } + }, Err(_) => { eprintln!("Worker thread panicked"); failed = true; @@ -428,6 +475,32 @@ fn run_parent( Ok(()) } +/// Process a single WorkerMessage, updating progress bars as needed. +fn handle_worker_message( + msg: &WorkerMessage, + worker_pb: &Option, + overall_pb: &Option, +) { + match msg { + WorkerMessage::Progress(count) => { + // Progress messages carry the cumulative count; we need the delta. + // Since we process messages sequentially per worker, track via pb position. + let current = worker_pb.as_ref().map_or(0, |pb| pb.position()); + let delta = count.saturating_sub(current); + if let Some(pb) = worker_pb { + pb.inc(delta); + } + if let Some(pb) = overall_pb { + pb.inc(delta); + } + } + WorkerMessage::Done => {} + WorkerMessage::Error(msg) => { + eprintln!("Worker error: {}", msg); + } + } +} + fn apply_suggestions( realization: &PathBuf, runner: &ModelRunner, @@ -451,19 +524,15 @@ fn apply_suggestions( .unwrap_or(0); // Build a map: model_name → module index in the JSON array. - // The runner loads modules in dependency order which may differ from config order, - // so match by model_type_name. let config_modules = runner.config.modules(); for s in suggestions { - // Find the module in the config that matches this model let module_json_idx = config_modules .iter() .position(|m| m.params.model_type_name == s.model_name); if let Some(idx) = module_json_idx { if idx < modules { - // Navigate safely with get_mut to avoid creating null entries let params = root .get_mut("global") .and_then(|g| g.get_mut("formulations")) @@ -474,7 +543,6 @@ fn apply_suggestions( .and_then(|m| m.get_mut("params")); if let Some(params) = params { - // Create variables_names_map if it doesn't exist if !params .get("variables_names_map") .is_some_and(|v| v.is_object()) @@ -515,7 +583,16 @@ fn run_worker( locations: &[String], data_dir: &PathBuf, #[cfg_attr(not(feature = "zarr"), allow(unused))] global_start: usize, + ipc_server_name: &str, ) -> Result<(), BmiError> { + // Connect to the parent's IPC server + let tx = IpcSender::::connect(ipc_server_name.to_string()).map_err(|e| { + BmiError::FunctionFailed { + model: "runner".into(), + func: format!("Failed to connect to IPC server: {}", e), + } + })?; + let mut runner = ModelRunner::from_config(realization)?; runner.suppress_warnings = true; let start_epoch = bmi_driver::parse_datetime(&runner.config.time.start_time)?; @@ -552,11 +629,12 @@ fn run_worker( runner.finalize()?; if (i + 1) % report_interval == 0 || i + 1 == locations.len() { - println!("{}", i + 1); + let _ = tx.send(WorkerMessage::Progress((i + 1) as u64)); } } store.finish()?; + let _ = tx.send(WorkerMessage::Done); Ok(()) } From e3ff74b7c0450e94352b27d3cec92fcb29388812 Mon Sep 17 00:00:00 2001 From: Josh Cunningham Date: Tue, 7 Apr 2026 11:26:32 -0500 Subject: [PATCH 2/7] remove unused var_type --- src/variables/forcings.rs | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/variables/forcings.rs b/src/variables/forcings.rs index 67ff0ad..28c5a12 100644 --- a/src/variables/forcings.rs +++ b/src/variables/forcings.rs @@ -30,7 +30,6 @@ pub trait Forcings { #[derive(Debug, Clone)] struct VarInfo { units: String, - var_type: VarType, itemsize: i32, type_str: String, } @@ -205,13 +204,12 @@ impl NetCdfForcings { }) .unwrap_or_default(); - let (type_str, itemsize, var_type) = Self::type_info(&var); + let (type_str, itemsize, _var_type) = Self::type_info(&var); self.var_names.push(name.clone()); self.var_info.insert( name, VarInfo { units, - var_type, itemsize, type_str, }, From dc7ef367a7139b3d1511b5096ec091b3c2a1f466 Mon Sep 17 00:00:00 2001 From: Josh Cunningham Date: Tue, 7 Apr 2026 12:05:37 -0500 Subject: [PATCH 3/7] send all cli args via ipc instead --- src/main.rs | 203 ++++++++++++++++++++++++++++------------------------ 1 file changed, 108 insertions(+), 95 deletions(-) diff --git a/src/main.rs b/src/main.rs index 14ccfa4..d07b079 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,7 +1,7 @@ use bmi_driver::{preload_dependencies, BmiError, DivideDataStore, ModelRunner, OutputFormat}; use clap::Parser; use indicatif::{MultiProgress, ProgressBar, ProgressStyle}; -use ipc_channel::ipc::{IpcOneShotServer, IpcSender}; +use ipc_channel::ipc::{self, IpcOneShotServer, IpcSender}; use std::env; use std::fs; use std::io::{self, Write}; @@ -29,6 +29,8 @@ struct TomlConfig { /// Messages sent from worker processes to the parent over IPC. #[derive(serde::Serialize, serde::Deserialize, Debug)] enum WorkerMessage { + /// First message: worker is ready and provides a channel for receiving config. + Ready(IpcSender), /// Worker completed `count` locations so far (cumulative). Progress(u64), /// Worker finished successfully. @@ -37,6 +39,14 @@ enum WorkerMessage { Error(String), } +/// Configuration sent from parent to worker over IPC. +#[derive(serde::Serialize, serde::Deserialize, Debug)] +struct WorkerConfig { + realization: PathBuf, + locations: Vec, + global_start: usize, +} + fn load_toml_config() -> TomlConfig { let config_path = dirs::config_dir() .unwrap_or_else(|| PathBuf::from("~/.config")) @@ -53,6 +63,29 @@ fn load_toml_config() -> TomlConfig { } } +/// Resolve which output variables to write based on the realization config. +/// +/// - `output_variables: ["all"]` → None (write all available variables) +/// - `output_variables: [...]` → Some(list) (write only those) +/// - `output_variables` empty → Some(vec![main_output_variable]) +fn resolve_output_vars(config: &bmi_driver::config::RealizationConfig) -> Option> { + let formulation = config.global.formulations.first()?; + let output_vars = &formulation.params.output_variables; + + if output_vars.len() == 1 && output_vars[0].eq_ignore_ascii_case("all") { + None + } else if output_vars.is_empty() { + let main = &formulation.params.main_output_variable; + if main.is_empty() { + None + } else { + Some(vec![main.clone()]) + } + } else { + Some(output_vars.clone()) + } +} + #[derive(Parser, Debug)] #[command(version, about, long_about = None)] struct Args { @@ -70,15 +103,7 @@ struct Args { #[arg(long, value_enum)] progress: Option, - /// Internal: start index for worker mode (inclusive) - #[arg(long, hide = true)] - worker_start: Option, - - /// Internal: end index for worker mode (exclusive) - #[arg(long, hide = true)] - worker_end: Option, - - /// Internal: IPC server name for worker→parent communication + /// Internal: IPC server name for worker mode #[arg(long, hide = true)] ipc_server: Option, @@ -97,18 +122,21 @@ struct Args { /// Number of locations to process on this node (0 = all remaining, for SLURM/multi-node use) #[arg(long, default_value_t = 0)] node_count: usize, - - /// Internal: output variable names for zarr workers (comma-separated) - #[arg(long, hide = true)] - output_vars: Option, } fn main() -> Result<(), BmiError> { let args = Args::parse(); let data_dir = fs::canonicalize(&args.data_dir).unwrap(); - let config_dir = data_dir.join("config"); let _ = env::set_current_dir(&data_dir); + preload_dependencies(); + + // Worker mode: receive all config over IPC, skip geopackage/config resolution + if let Some(ipc_server) = args.ipc_server { + return run_worker(&data_dir, &ipc_server); + } + + let config_dir = data_dir.join("config"); let realization = args .config .unwrap_or_else(|| config_dir.join("realization.json")); @@ -119,8 +147,6 @@ fn main() -> Result<(), BmiError> { return Ok(()); } - preload_dependencies(); - let db_path = config_dir .read_dir() .unwrap() @@ -157,13 +183,7 @@ fn main() -> Result<(), BmiError> { .or(toml_cfg.progress) .unwrap_or(ProgressMode::Summary); - if let (Some(start), Some(end), Some(ipc_server)) = - (args.worker_start, args.worker_end, args.ipc_server) - { - run_worker(&realization, &locations[start..end], &data_dir, start, &ipc_server) - } else { - run_parent(&data_dir, &realization, &locations, jobs, progress) - } + run_parent(&data_dir, &realization, &locations, jobs, progress) } fn print_units(realization: &PathBuf, locations: &[String]) -> Result<(), BmiError> { @@ -201,14 +221,10 @@ fn run_parent( // Discover output variable names for zarr (needs to create arrays upfront) #[cfg(feature = "zarr")] { - let config_output_vars: Vec = runner - .config - .global - .formulations - .first() - .map(|f| f.params.output_variables.clone()) - .unwrap_or_default(); - if config_output_vars.is_empty() { + if let Some(vars) = resolve_output_vars(&runner.config) { + discovered_vars = vars; + } else { + // "all" or no main_output — discover from models for model in &runner.models { if let Ok(names) = model.model.get_output_var_names() { for name in names { @@ -218,8 +234,6 @@ fn run_parent( } } } - } else { - discovered_vars = config_output_vars; } } @@ -279,7 +293,7 @@ fn run_parent( // For zarr: create the store before spawning workers so they can write directly #[cfg(feature = "zarr")] - let output_vars_csv: String = if output_format == OutputFormat::Zarr { + if output_format == OutputFormat::Zarr { let runner_tmp = ModelRunner::from_config(realization)?; let start_time = &runner_tmp.config.time.start_time; let interval = runner_tmp.config.time.output_interval; @@ -296,10 +310,7 @@ fn run_parent( locations, &discovered_vars, )?; - discovered_vars.join(",") - } else { - String::new() - }; + } let mp = MultiProgress::new(); @@ -348,23 +359,10 @@ fn run_parent( func: format!("Failed to create IPC server: {}", e), })?; - let mut cmd = Command::new(&exe); - cmd.arg(data_dir) - .arg("--config") - .arg(&realization) - .arg("--worker-start") - .arg(start.to_string()) - .arg("--worker-end") - .arg(end.to_string()) + let child = Command::new(&exe) + .arg(data_dir) .arg("--ipc-server") - .arg(&server_name); - - #[cfg(feature = "zarr")] - if output_format == OutputFormat::Zarr && !output_vars_csv.is_empty() { - cmd.arg("--output-vars").arg(&output_vars_csv); - } - - let child = cmd + .arg(&server_name) .stdout(Stdio::null()) .stderr(Stdio::inherit()) .spawn() @@ -373,6 +371,13 @@ fn run_parent( func: format!("Failed to spawn worker: {}", e), })?; + // Prepare the config to send once the worker is ready + let worker_config = WorkerConfig { + realization: realization.clone(), + locations: locations[start..end].to_vec(), + global_start: start, + }; + let worker_pb = worker_style.as_ref().map(|style| { let pb = mp.add(ProgressBar::new(worker_count)); pb.set_style(style.clone()); @@ -383,8 +388,8 @@ fn run_parent( let overall_pb = overall_pb.clone(); let handle = thread::spawn(move || { - // Accept the worker's connection — blocks until the worker connects and sends - // its first message. + // Accept the worker's connection — blocks until the worker connects. + // The first message is Ready(...) carrying a sender for WorkerConfig. let (rx, first_msg) = match server.accept() { Ok(pair) => pair, Err(e) => { @@ -393,10 +398,15 @@ fn run_parent( } }; - // Process the first message - handle_worker_message(&first_msg, &worker_pb, &overall_pb); + // Send configuration to the worker + if let WorkerMessage::Ready(config_tx) = first_msg { + if let Err(e) = config_tx.send(worker_config) { + eprintln!("Failed to send config to worker: {}", e); + return child; + } + } - // Read remaining messages until the channel closes or worker signals Done/Error + // Read progress messages until Done/Error or channel close loop { match rx.recv() { Ok(msg) => { @@ -406,7 +416,7 @@ fn run_parent( break; } } - Err(_) => break, // Channel closed + Err(_) => break, } } @@ -483,8 +493,6 @@ fn handle_worker_message( ) { match msg { WorkerMessage::Progress(count) => { - // Progress messages carry the cumulative count; we need the delta. - // Since we process messages sequentially per worker, track via pb position. let current = worker_pb.as_ref().map_or(0, |pb| pb.position()); let delta = count.saturating_sub(current); if let Some(pb) = worker_pb { @@ -494,7 +502,7 @@ fn handle_worker_message( pb.inc(delta); } } - WorkerMessage::Done => {} + WorkerMessage::Done | WorkerMessage::Ready(_) => {} WorkerMessage::Error(msg) => { eprintln!("Worker error: {}", msg); } @@ -516,14 +524,12 @@ fn apply_suggestions( func: format!("Failed to parse {}: {}", realization.display(), e), })?; - // Group suggestions by model_idx let modules = root["global"]["formulations"] .as_array() .and_then(|f| f.first()) .and_then(|f| f["params"]["modules"].as_array().map(|a| a.len())) .unwrap_or(0); - // Build a map: model_name → module index in the JSON array. let config_modules = runner.config.modules(); for s in suggestions { @@ -578,14 +584,8 @@ fn apply_suggestions( Ok(()) } -fn run_worker( - realization: &PathBuf, - locations: &[String], - data_dir: &PathBuf, - #[cfg_attr(not(feature = "zarr"), allow(unused))] global_start: usize, - ipc_server_name: &str, -) -> Result<(), BmiError> { - // Connect to the parent's IPC server +fn run_worker(data_dir: &PathBuf, ipc_server_name: &str) -> Result<(), BmiError> { + // Connect to parent and send a Ready message with a config channel let tx = IpcSender::::connect(ipc_server_name.to_string()).map_err(|e| { BmiError::FunctionFailed { model: "runner".into(), @@ -593,19 +593,30 @@ fn run_worker( } })?; - let mut runner = ModelRunner::from_config(realization)?; + let (config_tx, config_rx) = + ipc::channel::().map_err(|e| BmiError::FunctionFailed { + model: "runner".into(), + func: format!("Failed to create config channel: {}", e), + })?; + + tx.send(WorkerMessage::Ready(config_tx)) + .map_err(|e| BmiError::FunctionFailed { + model: "runner".into(), + func: format!("Failed to send Ready: {}", e), + })?; + + let config = config_rx.recv().map_err(|e| BmiError::FunctionFailed { + model: "runner".into(), + func: format!("Failed to receive worker config: {}", e), + })?; + + let mut runner = ModelRunner::from_config(&config.realization)?; runner.suppress_warnings = true; let start_epoch = bmi_driver::parse_datetime(&runner.config.time.start_time)?; let interval = runner.config.time.output_interval; let output_format = runner.config.output_format; let output_path = data_dir.join(&runner.config.output_root); - let output_vars: Vec = runner - .config - .global - .formulations - .first() - .map(|f| f.params.output_variables.clone()) - .unwrap_or_default(); + let output_vars = resolve_output_vars(&runner.config); let mut store = create_output_store( output_format, @@ -613,22 +624,22 @@ fn run_worker( &runner, start_epoch, interval, - global_start, - locations.first().map(|s| s.as_str()), + config.global_start, + config.locations.first().map(|s| s.as_str()), )?; - let report_interval = ((locations.len() as f64) * 0.01).ceil().max(1.0) as usize; + let report_interval = ((config.locations.len() as f64) * 0.01).ceil().max(1.0) as usize; - for (i, location) in locations.iter().enumerate() { + for (i, location) in config.locations.iter().enumerate() { runner.initialize(location)?; runner.run()?; - let columns = collect_columns(&runner, &output_vars); + let columns = collect_columns(&runner, output_vars.as_deref()); store.write_location(location, &columns)?; runner.finalize()?; - if (i + 1) % report_interval == 0 || i + 1 == locations.len() { + if (i + 1) % report_interval == 0 || i + 1 == config.locations.len() { let _ = tx.send(WorkerMessage::Progress((i + 1) as u64)); } } @@ -638,15 +649,17 @@ fn run_worker( Ok(()) } -fn collect_columns(runner: &ModelRunner, output_vars: &[String]) -> Vec<(String, Vec)> { - if output_vars.is_empty() { - runner +fn collect_columns( + runner: &ModelRunner, + output_vars: Option<&[String]>, +) -> Vec<(String, Vec)> { + match output_vars { + None => runner .outputs .iter() .map(|(name, vals)| (name.clone(), vals.clone())) - .collect() - } else { - output_vars + .collect(), + Some(vars) => vars .iter() .filter_map(|name| { runner @@ -654,7 +667,7 @@ fn collect_columns(runner: &ModelRunner, output_vars: &[String]) -> Vec<(String, .ok() .map(|vals| (name.clone(), vals.clone())) }) - .collect() + .collect(), } } From 80a83a70250f73ad5f5e5ca5df24c1855192ad25 Mon Sep 17 00:00:00 2001 From: Josh Cunningham Date: Tue, 7 Apr 2026 18:30:51 -0500 Subject: [PATCH 4/7] assume timeless units to be per timestep --- Cargo.lock | 2 +- Cargo.toml | 2 +- src/main.rs | 28 +++--- src/runner.rs | 6 +- src/variables/units.rs | 200 +++++++++++++++++++++++++++++++++++++---- 5 files changed, 208 insertions(+), 30 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5595d65..353b2a2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -158,7 +158,7 @@ dependencies = [ [[package]] name = "bmi-driver" -version = "0.0.3" +version = "0.0.4" dependencies = [ "anyhow", "clap", diff --git a/Cargo.toml b/Cargo.toml index c2b235d..82aad04 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "bmi-driver" -version = "0.0.3" +version = "0.0.4" edition = "2021" description = "Basic Model Interface (BMI) driver with support for C Fortran, and Python models." license = "MIT" diff --git a/src/main.rs b/src/main.rs index d07b079..0e9a543 100644 --- a/src/main.rs +++ b/src/main.rs @@ -95,6 +95,11 @@ struct Args { #[arg(long)] config: Option, + /// Path to the hydrofabric + /// + #[arg(long)] + hf: Option, + /// Number of parallel worker processes (default: number of CPUs) #[arg(short = 'j', long)] jobs: Option, @@ -127,6 +132,9 @@ struct Args { fn main() -> Result<(), BmiError> { let args = Args::parse(); let data_dir = fs::canonicalize(&args.data_dir).unwrap(); + // Resolve optional paths before changing the working directory + let config_arg = args.config.map(|p| fs::canonicalize(&p).unwrap()); + let hf_arg = args.hf.map(|p| fs::canonicalize(&p).unwrap()); let _ = env::set_current_dir(&data_dir); preload_dependencies(); @@ -137,9 +145,7 @@ fn main() -> Result<(), BmiError> { } let config_dir = data_dir.join("config"); - let realization = args - .config - .unwrap_or_else(|| config_dir.join("realization.json")); + let realization = config_arg.unwrap_or_else(|| config_dir.join("realization.json")); if args.minify { bmi_driver::config::minify_file(&realization)?; @@ -147,13 +153,15 @@ fn main() -> Result<(), BmiError> { return Ok(()); } - let db_path = config_dir - .read_dir() - .unwrap() - .filter_map(Result::ok) - .find(|entry| entry.path().extension().map_or(false, |ext| ext == "gpkg")) - .unwrap() - .path(); + let db_path = hf_arg.unwrap_or_else(|| { + config_dir + .read_dir() + .unwrap() + .filter_map(Result::ok) + .find(|entry| entry.path().extension().map_or(false, |ext| ext == "gpkg")) + .unwrap() + .path() + }); let conn = rusqlite::Connection::open(&db_path).unwrap(); let mut stmt = conn.prepare("SELECT divide_id FROM 'divides'").unwrap(); diff --git a/src/runner.rs b/src/runner.rs index faf4c88..1b176fa 100644 --- a/src/runner.rs +++ b/src/runner.rs @@ -185,7 +185,7 @@ impl ModelRunner { self.vars.insert(output.clone(), VarSource::Model(idx)); } - let input_conversions = self.build_unit_conversions(&*model, module); + let input_conversions = self.build_unit_conversions(&*model, module, timestep_info.dt_seconds); self.models.push(ModelInstance { name: module.params.model_type_name.clone(), @@ -314,10 +314,12 @@ impl ModelRunner { } /// Build unit conversions for each input variable mapping. + /// `model_dt` is the model's timestep in seconds, used for rate↔accumulation conversions. fn build_unit_conversions( &self, model: &dyn Bmi, module: &ModuleConfig, + model_dt: f64, ) -> HashMap { let mut conversions = HashMap::new(); for (model_input, source_var) in &module.params.variables_names_map { @@ -326,7 +328,7 @@ impl ModelRunner { if !source_units.is_empty() && !dest_units.is_empty() { let (conv, warning) = - crate::units::find_conversion_or_identity(&source_units, &dest_units); + crate::units::find_conversion_or_identity(&source_units, &dest_units, Some(model_dt)); if let Some(warn) = warning { if !self.suppress_warnings { eprintln!( diff --git a/src/variables/units.rs b/src/variables/units.rs index ebdf083..70c16b2 100644 --- a/src/variables/units.rs +++ b/src/variables/units.rs @@ -566,27 +566,145 @@ const TIME_UNITS: &[(&str, f64)] = &[ ("days", 86400.0), ]; -/// Convenience: get a conversion or fall back to identity with a warning message. -pub fn find_conversion_or_identity(from: &str, to: &str) -> (UnitConversion, Option) { - if let Some(conv) = find_conversion(from, to) { - (conv, None) - } else { - let warning = format!( - "Cannot convert '{}' → '{}': unrecognized or incompatible units, passing through unchanged", - from, to - ); +/// Try to find a conversion between a rate and a non-rate unit (or vice versa) +/// by treating the non-rate unit as "per timestep". +/// +/// For example, converting `mm/s` → `mm` with `dt_seconds = 3600` yields scale = 3600 +/// (i.e., mm/s × 3600s = mm per hour-long timestep). +fn find_rate_accumulation_conversion( + from: &str, + to: &str, + dt_seconds: f64, +) -> Option { + let from_trimmed = from.trim(); + let to_trimmed = to.trim(); + let from_norm = normalize(from_trimmed); + let to_norm = normalize(to_trimmed); + + match (&from_norm, &to_norm) { + // Rate → Dimensional: multiply by timestep to accumulate + // e.g. mm/s → mm means mm/s × dt_seconds = mm/timestep ( - UnitConversion { - from: from.to_string(), - to: to.to_string(), - scale: 1.0, + NormalizedUnit::Rate { + quantity_to_si: q_from, + quantity_cat: cat_from, + time_to_si: t_from, + }, + NormalizedUnit::Dimensional { + category: cat_to, + to_si: si_to, + }, + ) if cat_from == cat_to => { + // Convert rate to SI (quantity_si / second), then multiply by dt to get accumulation, + // then convert from SI quantity to target unit. + // scale = (q_from / t_from) * dt_seconds / si_to + let scale = (q_from / t_from) * dt_seconds / si_to; + Some(UnitConversion { + from: from_trimmed.into(), + to: to_trimmed.into(), + scale, offset: 0.0, + }) + } + + // Dimensional → Rate: divide by timestep + // e.g. mm → mm/s means mm/timestep ÷ dt_seconds = mm/s + ( + NormalizedUnit::Dimensional { + category: cat_from, + to_si: si_from, + }, + NormalizedUnit::Rate { + quantity_to_si: q_to, + quantity_cat: cat_to, + time_to_si: t_to, }, - Some(warning), - ) + ) if cat_from == cat_to => { + // Convert accumulation to SI, divide by dt to get SI rate, convert to target rate. + // scale = (si_from / dt_seconds) / (q_to / t_to) + let scale = (si_from / dt_seconds) / (q_to / t_to); + Some(UnitConversion { + from: from_trimmed.into(), + to: to_trimmed.into(), + scale, + offset: 0.0, + }) + } + + // MassFlux → Dimensional length: kg m-2 s-1 ≈ mm/s for liquid water + (NormalizedUnit::MassFlux { time_to_si: t_from }, NormalizedUnit::Dimensional { + category: UnitCategory::Length, + to_si: si_to, + }) => { + // kg m-2 per t_from seconds = 1 mm per t_from seconds (water density) + // rate in mm/s = 1/t_from, accumulation = rate × dt = dt/t_from mm + // convert mm to target: 0.001 m / si_to + let scale = (0.001 / t_from) * dt_seconds / si_to; + Some(UnitConversion { + from: from_trimmed.into(), + to: to_trimmed.into(), + scale, + offset: 0.0, + }) + } + + // Dimensional length → MassFlux + (NormalizedUnit::Dimensional { + category: UnitCategory::Length, + to_si: si_from, + }, NormalizedUnit::MassFlux { time_to_si: t_to }) => { + // Reverse: accumulation in source units → kg m-2 per t_to seconds + // source mm = si_from/0.001 mm, rate = si_from/(0.001 * dt) mm/s + // target = rate × t_to kg m-2 s-1 ... actually: + // 1 mm/s = 1 kg m-2 s-1, so mm/s × t_to = kg m-2 per t_to + let scale = (si_from / (0.001 * dt_seconds)) * t_to; + Some(UnitConversion { + from: from_trimmed.into(), + to: to_trimmed.into(), + scale, + offset: 0.0, + }) + } + + _ => None, } } +/// Convenience: get a conversion or fall back to identity with a warning message. +/// `dt_seconds` is the timestep duration used to convert between rates and accumulations. +/// If a rate unit is converted to a non-rate unit (or vice versa), the non-rate unit is +/// treated as "per timestep". Pass `None` to disable rate↔accumulation conversion. +pub fn find_conversion_or_identity( + from: &str, + to: &str, + dt_seconds: Option, +) -> (UnitConversion, Option) { + if let Some(conv) = find_conversion(from, to) { + return (conv, None); + } + + // Try rate ↔ accumulation conversion if we have a timestep + if let Some(dt) = dt_seconds { + if let Some(conv) = find_rate_accumulation_conversion(from, to, dt) { + return (conv, None); + } + } + + let warning = format!( + "Cannot convert '{}' → '{}': unrecognized or incompatible units, passing through unchanged", + from, to + ); + ( + UnitConversion { + from: from.to_string(), + to: to.to_string(), + scale: 1.0, + offset: 0.0, + }, + Some(warning), + ) +} + #[cfg(test)] mod tests { use super::*; @@ -683,7 +801,7 @@ mod tests { #[test] fn test_find_conversion_or_identity_unknown() { - let (conv, warning) = find_conversion_or_identity("furlongs", "mm"); + let (conv, warning) = find_conversion_or_identity("furlongs", "mm", None); assert!(conv.is_identity()); assert!(warning.is_some()); } @@ -723,4 +841,54 @@ mod tests { let conv = find_conversion("mm s^-1", "mm h^-1").unwrap(); assert!(conv.to_string().contains("×3600")); } + + // --- Rate ↔ accumulation conversion tests --- + + #[test] + fn test_rate_to_accumulation_mm_per_s_to_mm_hourly() { + // mm/s → mm with 3600s timestep: 1 mm/s × 3600s = 3600 mm + let (conv, warning) = find_conversion_or_identity("mm/s", "mm", Some(3600.0)); + assert!(warning.is_none()); + assert!((conv.scale - 3600.0).abs() < 1e-6); + } + + #[test] + fn test_rate_to_accumulation_mm_per_h_to_mm_hourly() { + // mm/h → mm with 3600s timestep: 1 mm/h × 1h = 1 mm + let (conv, warning) = find_conversion_or_identity("mm/h", "mm", Some(3600.0)); + assert!(warning.is_none()); + assert!((conv.scale - 1.0).abs() < 1e-6); + } + + #[test] + fn test_rate_to_accumulation_m_per_s_to_mm() { + // m/s → mm with 3600s timestep: 1 m/s × 3600s = 3600 m = 3_600_000 mm + let (conv, warning) = find_conversion_or_identity("m/s", "mm", Some(3600.0)); + assert!(warning.is_none()); + assert!((conv.scale - 3_600_000.0).abs() < 1e-3); + } + + #[test] + fn test_accumulation_to_rate_mm_to_mm_per_s() { + // mm → mm/s with 3600s timestep: 1 mm/3600s = 1/3600 mm/s + let (conv, warning) = find_conversion_or_identity("mm", "mm/s", Some(3600.0)); + assert!(warning.is_none()); + assert!((conv.scale - 1.0 / 3600.0).abs() < 1e-10); + } + + #[test] + fn test_mass_flux_to_accumulation_mm() { + // kg m-2 s-1 → mm with 3600s timestep: 1 kg m-2 s-1 = 1 mm/s, × 3600 = 3600 mm + let (conv, warning) = find_conversion_or_identity("kg m-2 s-1", "mm", Some(3600.0)); + assert!(warning.is_none()); + assert!((conv.scale - 3600.0).abs() < 1e-6); + } + + #[test] + fn test_rate_to_accumulation_without_timestep_returns_warning() { + // Without dt_seconds, rate→accumulation should fail and return identity+warning + let (conv, warning) = find_conversion_or_identity("mm/s", "mm", None); + assert!(warning.is_some()); + assert!(conv.is_identity()); + } } From 2e44e8f755b3ab5c3282a2eb40818648d57bc31e Mon Sep 17 00:00:00 2001 From: Josh Cunningham Date: Wed, 8 Apr 2026 10:49:45 -0500 Subject: [PATCH 5/7] add --inspect to cli --- src/lib.rs | 1 + src/main.rs | 28 +++++----- src/model_info.rs | 131 ++++++++++++++++++++++++++++++++++++++++++++++ src/runner.rs | 40 +------------- 4 files changed, 145 insertions(+), 55 deletions(-) create mode 100644 src/model_info.rs diff --git a/src/lib.rs b/src/lib.rs index 95a8758..543f55b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,6 +3,7 @@ mod types; mod variables; pub mod config; +pub mod model_info; pub mod output; pub mod runner; diff --git a/src/main.rs b/src/main.rs index 0e9a543..8ee73f6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -116,6 +116,10 @@ struct Args { #[arg(long)] units: bool, + /// Print a summary of each model's inputs and outputs, then exit + #[arg(long)] + inspect: bool, + /// Minify realization.json, removing fields bmi-driver doesn't use #[arg(long)] minify: bool, @@ -180,7 +184,11 @@ fn main() -> Result<(), BmiError> { let locations = locations[node_start..node_end].to_vec(); if args.units { - return print_units(&realization, &locations); + return bmi_driver::model_info::print_units(&realization, &locations); + } + + if args.inspect { + return bmi_driver::model_info::inspect_models(&realization, &locations); } // Merge: CLI > TOML > default @@ -194,18 +202,6 @@ fn main() -> Result<(), BmiError> { run_parent(&data_dir, &realization, &locations, jobs, progress) } -fn print_units(realization: &PathBuf, locations: &[String]) -> Result<(), BmiError> { - let mut runner = ModelRunner::from_config(realization)?; - if let Some(loc) = locations.first() { - runner.initialize(loc)?; - runner.print_unit_conversions(false); - runner.finalize()?; - } else { - eprintln!("No locations found."); - } - Ok(()) -} - fn run_parent( data_dir: &PathBuf, realization: &PathBuf, @@ -273,15 +269,15 @@ fn run_parent( // Re-initialize with updated config to show new conversions let mut runner2 = ModelRunner::from_config(realization)?; runner2.initialize(loc)?; - runner2.print_unit_conversions(true); + bmi_driver::model_info::print_unit_conversions(&runner2, true); runner2.finalize()?; } else { eprintln!("Skipping. Running with current config."); - runner.print_unit_conversions(true); + bmi_driver::model_info::print_unit_conversions(&runner, true); runner.finalize()?; } } else { - runner.print_unit_conversions(true); + bmi_driver::model_info::print_unit_conversions(&runner, true); runner.finalize()?; } } diff --git a/src/model_info.rs b/src/model_info.rs new file mode 100644 index 0000000..68fc142 --- /dev/null +++ b/src/model_info.rs @@ -0,0 +1,131 @@ +use std::path::Path; + +use crate::error::BmiResult; +use crate::runner::ModelRunner; + +/// Print all unit conversion info for the first location, then return. +pub fn print_units(realization: &Path, locations: &[String]) -> BmiResult<()> { + let mut runner = ModelRunner::from_config(realization)?; + if let Some(loc) = locations.first() { + runner.initialize(loc)?; + print_unit_conversions(&runner, false); + runner.finalize()?; + } else { + eprintln!("No locations found."); + } + Ok(()) +} + +/// Print a summary of each model's inputs and outputs, then return. +pub fn inspect_models(realization: &Path, locations: &[String]) -> BmiResult<()> { + let mut runner = ModelRunner::from_config(realization)?; + let loc = match locations.first() { + Some(l) => l, + None => { + eprintln!("No locations found."); + return Ok(()); + } + }; + runner.initialize(loc)?; + + let modules: Vec<_> = runner.config.modules().into_iter().cloned().collect(); + + eprintln!("Realization: {} model(s)\n", runner.models.len()); + + for (i, model) in runner.models.iter().enumerate() { + let module = modules.get(i); + eprintln!("━━━ Model {}: {} ━━━", i + 1, model.name); + + if let Some(m) = module { + if !m.params.library_file.is_empty() { + eprintln!(" library: {}", m.params.library_file); + } + if !m.params.init_config.is_empty() { + eprintln!(" init_config: {}", m.params.init_config); + } + } + + eprintln!( + " timestep: {}s ({} steps)", + model.timestep_info.dt_seconds, model.timestep_info.num_steps + ); + + // Inputs + let input_names = model.model.get_input_var_names().unwrap_or_default(); + eprintln!("\n Inputs ({}):", input_names.len()); + for name in &input_names { + let units = model.model.get_var_units(name).unwrap_or_default(); + let units_str = if units.is_empty() { "?" } else { &units }; + + if let Some(source) = model.input_map.get(name) { + let source_label = runner.source_label(source); + let conv_str = model + .input_conversions + .get(name) + .filter(|c| !c.is_identity()) + .map(|c| format!(" [{}]", c)) + .unwrap_or_default(); + eprintln!( + " {} [{}] ← {} ({}){conv_str}", + name, units_str, source, source_label + ); + } else { + eprintln!(" {} [{}] (unmapped)", name, units_str); + } + } + + // Outputs + let output_names = model.model.get_output_var_names().unwrap_or_default(); + eprintln!("\n Outputs ({}):", output_names.len()); + for name in &output_names { + let units = model.model.get_var_units(name).unwrap_or_default(); + let units_str = if units.is_empty() { "?" } else { &units }; + let is_main = name == &model.main_output; + let tag = if is_main { " (main)" } else { "" }; + eprintln!(" {} [{}]{}", name, units_str, tag); + } + + eprintln!(); + } + + runner.finalize()?; + Ok(()) +} + +/// Print unit conversions to stderr. +/// If `active_only` is true, only prints non-identity conversions. +/// If false, prints all variable mappings including those without unit info. +pub fn print_unit_conversions(runner: &ModelRunner, active_only: bool) { + if !active_only { + eprintln!("Unit conversions for this run:"); + } + let mut any = false; + for m in &runner.models { + for (model_input, source_var) in &m.input_map { + let source_label = runner.source_label(source_var); + if let Some(conv) = m.input_conversions.get(model_input) { + if active_only && conv.is_identity() { + continue; + } + eprintln!( + " {}: {} ← {} ({}): {}", + m.name, model_input, source_var, source_label, conv + ); + } else if !active_only { + eprintln!( + " {}: {} ← {} ({}): no unit info available", + m.name, model_input, source_var, source_label + ); + } else { + continue; + } + any = true; + } + } + if active_only && any { + eprintln!(); + } + if !active_only && !any { + eprintln!(" (no variable mappings)"); + } +} diff --git a/src/runner.rs b/src/runner.rs index 1b176fa..5d79892 100644 --- a/src/runner.rs +++ b/src/runner.rs @@ -574,45 +574,7 @@ impl ModelRunner { } } - /// Print unit conversions to stderr. - /// If `active_only` is true, only prints non-identity conversions. - /// If false, prints all variable mappings including those without unit info. - pub fn print_unit_conversions(&self, active_only: bool) { - if !active_only { - eprintln!("Unit conversions for this run:"); - } - let mut any = false; - for m in &self.models { - for (model_input, source_var) in &m.input_map { - let source_label = self.source_label(source_var); - if let Some(conv) = m.input_conversions.get(model_input) { - if active_only && conv.is_identity() { - continue; - } - eprintln!( - " {}: {} ← {} ({}): {}", - m.name, model_input, source_var, source_label, conv - ); - } else if !active_only { - eprintln!( - " {}: {} ← {} ({}): no unit info available", - m.name, model_input, source_var, source_label - ); - } else { - continue; - } - any = true; - } - } - if active_only && any { - eprintln!(); - } - if !active_only && !any { - eprintln!(" (no variable mappings)"); - } - } - - fn source_label(&self, source_var: &str) -> String { + pub fn source_label(&self, source_var: &str) -> String { match self.vars.get(source_var) { Some(VarSource::Forcing) => "forcing".to_string(), Some(VarSource::Model(idx)) => self From 21443f20ea9b0111a09854fc750c699b2ad24179 Mon Sep 17 00:00:00 2001 From: Josh Cunningham Date: Wed, 8 Apr 2026 11:56:02 -0500 Subject: [PATCH 6/7] fix progress bar --- src/main.rs | 37 +++++++++++++------------------------ 1 file changed, 13 insertions(+), 24 deletions(-) diff --git a/src/main.rs b/src/main.rs index 8ee73f6..3590dfb 100644 --- a/src/main.rs +++ b/src/main.rs @@ -411,11 +411,23 @@ fn run_parent( } // Read progress messages until Done/Error or channel close + let mut prev_count: u64 = 0; loop { match rx.recv() { Ok(msg) => { let done = matches!(msg, WorkerMessage::Done | WorkerMessage::Error(_)); - handle_worker_message(&msg, &worker_pb, &overall_pb); + if let WorkerMessage::Progress(count) = &msg { + let delta = count.saturating_sub(prev_count); + prev_count = *count; + if let Some(pb) = &worker_pb { + pb.inc(delta); + } + if let Some(pb) = &overall_pb { + pb.inc(delta); + } + } else if let WorkerMessage::Error(err) = &msg { + eprintln!("Worker error: {}", err); + } if done { break; } @@ -489,29 +501,6 @@ fn run_parent( Ok(()) } -/// Process a single WorkerMessage, updating progress bars as needed. -fn handle_worker_message( - msg: &WorkerMessage, - worker_pb: &Option, - overall_pb: &Option, -) { - match msg { - WorkerMessage::Progress(count) => { - let current = worker_pb.as_ref().map_or(0, |pb| pb.position()); - let delta = count.saturating_sub(current); - if let Some(pb) = worker_pb { - pb.inc(delta); - } - if let Some(pb) = overall_pb { - pb.inc(delta); - } - } - WorkerMessage::Done | WorkerMessage::Ready(_) => {} - WorkerMessage::Error(msg) => { - eprintln!("Worker error: {}", msg); - } - } -} fn apply_suggestions( realization: &PathBuf, From a439b8f62420f83875ef5c648b0ce438718e5ef7 Mon Sep 17 00:00:00 2001 From: Josh Cunningham Date: Wed, 8 Apr 2026 11:56:44 -0500 Subject: [PATCH 7/7] v0.1.0 --- Cargo.lock | 2 +- Cargo.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 353b2a2..bdc737a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -158,7 +158,7 @@ dependencies = [ [[package]] name = "bmi-driver" -version = "0.0.4" +version = "0.1.0" dependencies = [ "anyhow", "clap", diff --git a/Cargo.toml b/Cargo.toml index 82aad04..6698e42 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "bmi-driver" -version = "0.0.4" +version = "0.1.0" edition = "2021" description = "Basic Model Interface (BMI) driver with support for C Fortran, and Python models." license = "MIT"