From acd672d0bcab91ad0bade036f3b63e6b3968e2c0 Mon Sep 17 00:00:00 2001 From: wyattb Date: Sun, 29 Mar 2026 18:44:10 -0400 Subject: [PATCH 01/10] #292 - add interactive MQTT injection tool New standalone binary (src/bin/inject) that maps keyboard keys to MQTT topics and publishes a random value within each topic's defined min/max range on keypress. Reuses SimComponent config from simulate.rs. Features: - Key-to-topic mapping via JSON file (--key-map) - --list-topics to discover available topics - Siren host override via --siren-host-url or CALYPSO_SIREN_HOST_URL - Clean shutdown on Ctrl+C Also cfg-gates imd_poll module to target_os=linux so the library crate compiles on non-Linux platforms (simulate, inject, nerdbc, etc). --- Cargo.lock | 70 +++++++ Cargo.toml | 1 + inject_keymap.example.json | 6 + src/bin/inject.rs | 379 +++++++++++++++++++++++++++++++++++++ src/lib.rs | 1 + 5 files changed, 457 insertions(+) create mode 100644 inject_keymap.example.json create mode 100644 src/bin/inject.rs diff --git a/Cargo.lock b/Cargo.lock index b92a984..0e45683 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -130,6 +130,7 @@ dependencies = [ "calypso-cangen", "can-dbc", "clap", + "crossterm", "daedalus", "futures-util", "num_enum", @@ -276,6 +277,32 @@ dependencies = [ "memchr", ] +[[package]] +name = "crossterm" +version = "0.28.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "829d955a0bb380ef178a640b91779e3987da38c9aea133b20614cfed8cdea9c6" +dependencies = [ + "bitflags", + "crossterm_winapi", + "futures-core", + "mio", + "parking_lot", + "rustix 0.38.44", + "signal-hook", + "signal-hook-mio", + "winapi", +] + +[[package]] +name = "crossterm_winapi" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "acdd7c62a3665c7f6830a51635d9ac9b23ed385797f70a83bb8bafe9c572ab2b" +dependencies = [ + "winapi", +] + [[package]] name = "daedalus" version = "0.1.0" @@ -1332,6 +1359,27 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" +[[package]] +name = "signal-hook" +version = "0.3.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d881a16cf4426aa584979d30bd82cb33429027e42122b169753d6ef1085ed6e2" +dependencies = [ + "libc", + "signal-hook-registry", +] + +[[package]] +name = "signal-hook-mio" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b75a19a7a740b25bc7944bdee6172368f988763b744e3d4dfe753f6b4ece40cc" +dependencies = [ + "libc", + "mio", + "signal-hook", +] + [[package]] name = "signal-hook-registry" version = "1.4.8" @@ -1764,6 +1812,28 @@ dependencies = [ "rustix 0.38.44", ] +[[package]] +name = "winapi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" +dependencies = [ + "winapi-i686-pc-windows-gnu", + "winapi-x86_64-pc-windows-gnu", +] + +[[package]] +name = "winapi-i686-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" + +[[package]] +name = "winapi-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + [[package]] name = "windows-link" version = "0.2.1" diff --git a/Cargo.toml b/Cargo.toml index 7f362f6..e3b4aae 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -41,6 +41,7 @@ futures-util = "0.3.31" tracing = "0.1.41" tracing-subscriber = { version = "0.3.19", features = ["ansi", "env-filter"] } tokio-util = { version = "0.7.16", features = ["full"] } +crossterm = { version = "0.28", features = ["event-stream"] } can-dbc = "6.0.0" serde_json.workspace = true schemars.workspace = true diff --git a/inject_keymap.example.json b/inject_keymap.example.json new file mode 100644 index 0000000..f7bcd2b --- /dev/null +++ b/inject_keymap.example.json @@ -0,0 +1,6 @@ +{ + "v": "BMS/Pack/Voltage", + "c": "BMS/Pack/Current", + "s": "BMS/Pack/SOC", + "h": "BMS/Pack/Health" +} diff --git a/src/bin/inject.rs b/src/bin/inject.rs new file mode 100644 index 0000000..d5c49b4 --- /dev/null +++ b/src/bin/inject.rs @@ -0,0 +1,379 @@ +use std::collections::HashMap; +use std::io::{self, Write}; +use std::process::exit; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use calypso::{ + proto::serverdata, + simulatable_message::{SimComponent, SimValue}, + simulate_data::create_simulated_components, +}; +use clap::Parser; +use crossterm::{ + event::{Event, EventStream, KeyCode, KeyEvent, KeyEventKind, KeyModifiers}, + terminal::{disable_raw_mode, enable_raw_mode}, +}; +use futures_util::StreamExt; +use protobuf::Message; +use rand::prelude::*; +use rumqttc::v5::{AsyncClient, MqttOptions}; +use tokio_util::sync::CancellationToken; +use tracing::debug; + +#[derive(Parser, Debug)] +#[command(version, about = "Interactive MQTT injection tool for manual testing")] +struct InjectArgs { + /// Path to JSON key mapping file (maps single characters to MQTT topics) + #[arg(short = 'k', long)] + key_map: Option, + + /// Siren broker host:port + #[arg( + short = 'u', + long, + env = "CALYPSO_SIREN_HOST_URL", + default_value = "127.0.0.1:1883" + )] + siren_host_url: String, + + /// List all available topics and exit + #[arg(long)] + list_topics: bool, +} + +fn parse_key_map(content: &str) -> Result, String> { + let raw: HashMap = + serde_json::from_str(content).map_err(|e| format!("Invalid key map JSON: {e}"))?; + let mut map = HashMap::new(); + for (key_str, topic) in raw { + if key_str.len() != 1 { + return Err(format!( + "Key mapping keys must be single characters, got: '{key_str}'" + )); + } + let ch = key_str.chars().next().unwrap(); + map.insert(ch, topic); + } + Ok(map) +} + +fn load_key_map(path: &str) -> HashMap { + let content = std::fs::read_to_string(path).unwrap_or_else(|e| { + eprintln!("Failed to read key map file '{path}': {e}"); + exit(1); + }); + parse_key_map(&content).unwrap_or_else(|e| { + eprintln!("{e}"); + exit(1); + }) +} + +fn build_topic_components(key_map: &HashMap) -> HashMap { + let mut components = create_simulated_components(); + let mut result = HashMap::new(); + for (&key, topic) in key_map { + if let Some(pos) = components.iter().position(|c| c.name == *topic) { + result.insert(key, components.swap_remove(pos)); + } else { + eprintln!("Warning: no simulated component for topic '{topic}' (key '{key}')"); + } + } + result +} + +/// Generate a fresh random value within each point's defined bounds. +fn randomize_component(component: &mut SimComponent) { + let mut rng = rand::rng(); + for point in &mut component.points { + match &mut point.value { + SimValue::Range { + min, + max, + inc_min, + round, + current, + .. + } => { + *current = rng.random_range(*min..*max); + if *inc_min != 0.0 { + *current = (*current / *inc_min).round() * *inc_min; + } + if *round { + *current = current.round(); + } + } + SimValue::Discrete { + options, current, .. + } => { + let idx = rng.random_range(0..options.len()); + *current = options[idx].0; + } + } + } +} + +async fn poll_stub(token: CancellationToken, mut eventloop: rumqttc::v5::EventLoop) { + loop { + tokio::select! { + () = token.cancelled() => { + debug!("MQTT poll shutting down"); + break; + }, + _ = eventloop.poll() => {} + } + } +} + +#[tokio::main] +async fn main() { + let cli = InjectArgs::parse(); + + if cli.list_topics { + let components = create_simulated_components(); + println!("Available topics ({} total):", components.len()); + for component in &components { + println!(" {} [{}]", component.name, component.unit); + } + return; + } + + let key_map_path = cli.key_map.unwrap_or_else(|| { + eprintln!("--key-map is required (use --list-topics to see available topics)"); + exit(1); + }); + + let key_map = load_key_map(&key_map_path); + if key_map.is_empty() { + eprintln!("Key map is empty"); + exit(1); + } + + let mut components = build_topic_components(&key_map); + if components.is_empty() { + eprintln!("No matching topics found for any key mapping"); + exit(1); + } + + // Set up MQTT connection + let (host, port_str) = cli.siren_host_url.split_once(':').unwrap_or_else(|| { + eprintln!("Invalid siren URL format, expected host:port"); + exit(1); + }); + let port: u16 = port_str.parse().unwrap_or_else(|_| { + eprintln!("Invalid port: {port_str}"); + exit(1); + }); + + let client_id = format!( + "Calypso-Inject-{}", + SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("Time went backwards") + .as_millis() + ); + let mut mqtt_opts = MqttOptions::new(client_id, host, port); + mqtt_opts + .set_keep_alive(Duration::from_secs(20)) + .set_clean_start(true) + .set_connection_timeout(3) + .set_session_expiry_interval(Some(u32::MAX)) + .set_topic_alias_max(Some(600)); + let (client, eventloop) = AsyncClient::new(mqtt_opts, 600); + + let token = CancellationToken::new(); + let poll_handle = tokio::spawn(poll_stub(token.clone(), eventloop)); + + // Print key mappings + println!("Key Mappings:"); + let mut sorted_keys: Vec<_> = components.keys().copied().collect(); + sorted_keys.sort_unstable(); + for key in &sorted_keys { + if let Some(component) = components.get(key) { + println!(" {key} → {} [{}]", component.name, component.unit); + } + } + println!(); + println!("Press mapped keys to inject. Ctrl+C to exit."); + println!(); + + enable_raw_mode().expect("Failed to enable raw mode"); + + let mut reader = EventStream::new(); + + loop { + match reader.next().await { + Some(Ok(Event::Key(KeyEvent { + code: KeyCode::Char('c'), + modifiers, + kind: KeyEventKind::Press, + .. + }))) if modifiers.contains(KeyModifiers::CONTROL) => { + break; + } + Some(Ok(Event::Key(KeyEvent { + code: KeyCode::Char(ch), + kind: KeyEventKind::Press, + .. + }))) => { + if let Some(component) = components.get_mut(&ch) { + randomize_component(component); + let data = component.get_decode_data(); + + let timestamp = UNIX_EPOCH.elapsed().unwrap().as_micros() as u64; + let mut payload = serverdata::ServerData::new(); + payload.unit.clone_from(&data.unit); + payload.values.clone_from(&data.value); + payload.time_us = timestamp; + + let topic = &data.topic; + let values_str: Vec = + data.value.iter().map(|v| format!("{v:.2}")).collect(); + + if let Ok(bytes) = payload.write_to_bytes() { + match client + .publish( + topic.as_str(), + rumqttc::v5::mqttbytes::QoS::AtMostOnce, + false, + bytes, + ) + .await + { + Ok(()) => { + print!( + "[{ch}] {topic} = [{}] {}\r\n", + values_str.join(", "), + data.unit + ); + io::stdout().flush().ok(); + } + Err(e) => { + print!("[{ch}] publish error: {e}\r\n"); + io::stdout().flush().ok(); + } + } + } + } + } + Some(Err(_)) | None => break, + _ => {} + } + } + + disable_raw_mode().expect("Failed to disable raw mode"); + println!("\r\nShutting down..."); + + token.cancel(); + poll_handle.await.ok(); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_valid_key_map() { + let map = parse_key_map(r#"{"a": "BMS/Pack/Voltage", "b": "BMS/Pack/Current"}"#).unwrap(); + assert_eq!(map.len(), 2); + assert_eq!(map[&'a'], "BMS/Pack/Voltage"); + assert_eq!(map[&'b'], "BMS/Pack/Current"); + } + + #[test] + fn parse_key_map_rejects_multi_char_keys() { + let result = parse_key_map(r#"{"ab": "topic"}"#); + assert!(result.is_err()); + assert!(result.unwrap_err().contains("single characters")); + } + + #[test] + fn parse_key_map_rejects_invalid_json() { + let result = parse_key_map("not json"); + assert!(result.is_err()); + } + + #[test] + fn parse_key_map_empty() { + let map = parse_key_map("{}").unwrap(); + assert!(map.is_empty()); + } + + #[test] + fn build_topic_components_filters_to_mapped_keys() { + let components = create_simulated_components(); + if components.is_empty() { + return; + } + let target_name = components[0].name.clone(); + let mut key_map = HashMap::new(); + key_map.insert('x', target_name.clone()); + + let filtered = build_topic_components(&key_map); + assert_eq!(filtered.len(), 1); + assert_eq!(filtered[&'x'].name, target_name); + } + + #[test] + fn build_topic_components_skips_unknown_topics() { + let mut key_map = HashMap::new(); + key_map.insert('z', "NonExistent/Topic/Name".to_string()); + + let filtered = build_topic_components(&key_map); + assert!(filtered.is_empty()); + } + + #[test] + fn randomize_component_stays_within_bounds() { + let mut components = create_simulated_components(); + for component in &mut components { + for _ in 0..10 { + randomize_component(component); + for point in &component.points { + match &point.value { + SimValue::Range { + min, max, current, .. + } => { + assert!( + *current >= *min, + "value {current} below min {min} for topic {}", + component.name + ); + assert!( + *current <= *max, + "value {current} above max {max} for topic {}", + component.name + ); + } + SimValue::Discrete { + options, current, .. + } => { + assert!( + options.iter().any(|(v, _)| (*v - *current).abs() < 0.001), + "discrete value {current} not in options for topic {}", + component.name + ); + } + } + } + } + } + } + + #[test] + fn key_resolves_to_correct_topic() { + let components = create_simulated_components(); + if components.len() < 2 { + return; + } + let topic_a = components[0].name.clone(); + let topic_b = components[1].name.clone(); + + let mut key_map = HashMap::new(); + key_map.insert('a', topic_a.clone()); + key_map.insert('b', topic_b.clone()); + + let filtered = build_topic_components(&key_map); + assert_eq!(filtered[&'a'].name, topic_a); + assert_eq!(filtered[&'b'].name, topic_b); + } +} diff --git a/src/lib.rs b/src/lib.rs index 0ba8db4..1c9735b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -6,4 +6,5 @@ pub mod proto; pub mod simulatable_message; pub mod simulate_data; +#[cfg(target_os = "linux")] pub mod imd_poll; From 28d7c73b2c6612a7d25c827e6da7bb316e1e366f Mon Sep 17 00:00:00 2001 From: wyattb Date: Sun, 29 Mar 2026 19:07:44 -0400 Subject: [PATCH 02/10] #292 - review fixes: clamp bounds, raw mode guard, simplify - Clamp values after rounding in randomize_component to prevent inc_min/round pushing values past min/max bounds - Add RawModeGuard (RAII) so terminal restores on panic - Extract publish_injection helper to reduce nesting in main loop - Remove dead tracing import (no subscriber configured) - Deduplicate stdout flush calls --- src/bin/inject.rs | 97 +++++++++++++++++++++++++---------------------- 1 file changed, 51 insertions(+), 46 deletions(-) diff --git a/src/bin/inject.rs b/src/bin/inject.rs index d5c49b4..afc5d58 100644 --- a/src/bin/inject.rs +++ b/src/bin/inject.rs @@ -16,9 +16,9 @@ use crossterm::{ use futures_util::StreamExt; use protobuf::Message; use rand::prelude::*; +use rumqttc::v5::mqttbytes::QoS; use rumqttc::v5::{AsyncClient, MqttOptions}; use tokio_util::sync::CancellationToken; -use tracing::debug; #[derive(Parser, Debug)] #[command(version, about = "Interactive MQTT injection tool for manual testing")] @@ -41,6 +41,15 @@ struct InjectArgs { list_topics: bool, } +/// RAII guard that restores the terminal from raw mode on drop. +struct RawModeGuard; + +impl Drop for RawModeGuard { + fn drop(&mut self) { + let _ = disable_raw_mode(); + } +} + fn parse_key_map(content: &str) -> Result, String> { let raw: HashMap = serde_json::from_str(content).map_err(|e| format!("Invalid key map JSON: {e}"))?; @@ -101,6 +110,7 @@ fn randomize_component(component: &mut SimComponent) { if *round { *current = current.round(); } + *current = current.clamp(*min, *max); } SimValue::Discrete { options, current, .. @@ -115,15 +125,47 @@ fn randomize_component(component: &mut SimComponent) { async fn poll_stub(token: CancellationToken, mut eventloop: rumqttc::v5::EventLoop) { loop { tokio::select! { - () = token.cancelled() => { - debug!("MQTT poll shutting down"); - break; - }, + () = token.cancelled() => break, _ = eventloop.poll() => {} } } } +/// Randomize a component's value and publish it to the MQTT broker. +async fn publish_injection(ch: char, component: &mut SimComponent, client: &AsyncClient) { + randomize_component(component); + let data = component.get_decode_data(); + + let timestamp = UNIX_EPOCH.elapsed().unwrap().as_micros() as u64; + let mut payload = serverdata::ServerData::new(); + payload.unit.clone_from(&data.unit); + payload.values.clone_from(&data.value); + payload.time_us = timestamp; + + let values_str: Vec = data.value.iter().map(|v| format!("{v:.2}")).collect(); + + let Ok(bytes) = payload.write_to_bytes() else { + return; + }; + match client + .publish(&data.topic, QoS::AtMostOnce, false, bytes) + .await + { + Ok(()) => { + print!( + "[{ch}] {} = [{}] {}\r\n", + data.topic, + values_str.join(", "), + data.unit + ); + } + Err(e) => { + print!("[{ch}] publish error: {e}\r\n"); + } + } + io::stdout().flush().ok(); +} + #[tokio::main] async fn main() { let cli = InjectArgs::parse(); @@ -197,6 +239,7 @@ async fn main() { println!(); enable_raw_mode().expect("Failed to enable raw mode"); + let guard = RawModeGuard; let mut reader = EventStream::new(); @@ -207,52 +250,14 @@ async fn main() { modifiers, kind: KeyEventKind::Press, .. - }))) if modifiers.contains(KeyModifiers::CONTROL) => { - break; - } + }))) if modifiers.contains(KeyModifiers::CONTROL) => break, Some(Ok(Event::Key(KeyEvent { code: KeyCode::Char(ch), kind: KeyEventKind::Press, .. }))) => { if let Some(component) = components.get_mut(&ch) { - randomize_component(component); - let data = component.get_decode_data(); - - let timestamp = UNIX_EPOCH.elapsed().unwrap().as_micros() as u64; - let mut payload = serverdata::ServerData::new(); - payload.unit.clone_from(&data.unit); - payload.values.clone_from(&data.value); - payload.time_us = timestamp; - - let topic = &data.topic; - let values_str: Vec = - data.value.iter().map(|v| format!("{v:.2}")).collect(); - - if let Ok(bytes) = payload.write_to_bytes() { - match client - .publish( - topic.as_str(), - rumqttc::v5::mqttbytes::QoS::AtMostOnce, - false, - bytes, - ) - .await - { - Ok(()) => { - print!( - "[{ch}] {topic} = [{}] {}\r\n", - values_str.join(", "), - data.unit - ); - io::stdout().flush().ok(); - } - Err(e) => { - print!("[{ch}] publish error: {e}\r\n"); - io::stdout().flush().ok(); - } - } - } + publish_injection(ch, component, &client).await; } } Some(Err(_)) | None => break, @@ -260,7 +265,7 @@ async fn main() { } } - disable_raw_mode().expect("Failed to disable raw mode"); + drop(guard); println!("\r\nShutting down..."); token.cancel(); From 6608b9b393ead941b5fecec19b8d0a1b318f40d9 Mon Sep 17 00:00:00 2001 From: wyattb Date: Mon, 30 Mar 2026 21:02:41 -0400 Subject: [PATCH 03/10] #292 - rename inject to manual-sim, code review fixes Rename bin from inject to manual-sim for clarity. Review fixes: - Fix parse_key_map byte-length check to use char count (Unicode correctness) - Add error feedback on protobuf serialization failure - Log MQTT connection errors in poll_stub instead of silently swallowing - Use clap required_unless_present for key_map argument - Couple RawModeGuard with enable_raw_mode via constructor - Log terminal event errors before exiting - Simplify parse_key_map validation with destructure pattern - Use idiomatic choose() for Discrete random selection - Move values_str allocation into success branch - Replace infallible if-let with direct indexing - Remove SystemTime import, use UNIX_EPOCH.elapsed() consistently --- src/bin/{inject.rs => manual-sim.rs} | 62 ++++++++++++++++------------ 1 file changed, 35 insertions(+), 27 deletions(-) rename src/bin/{inject.rs => manual-sim.rs} (88%) diff --git a/src/bin/inject.rs b/src/bin/manual-sim.rs similarity index 88% rename from src/bin/inject.rs rename to src/bin/manual-sim.rs index afc5d58..1394441 100644 --- a/src/bin/inject.rs +++ b/src/bin/manual-sim.rs @@ -1,7 +1,7 @@ use std::collections::HashMap; use std::io::{self, Write}; use std::process::exit; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use std::time::{Duration, UNIX_EPOCH}; use calypso::{ proto::serverdata, @@ -24,7 +24,7 @@ use tokio_util::sync::CancellationToken; #[command(version, about = "Interactive MQTT injection tool for manual testing")] struct InjectArgs { /// Path to JSON key mapping file (maps single characters to MQTT topics) - #[arg(short = 'k', long)] + #[arg(short = 'k', long, required_unless_present = "list_topics")] key_map: Option, /// Siren broker host:port @@ -41,9 +41,16 @@ struct InjectArgs { list_topics: bool, } -/// RAII guard that restores the terminal from raw mode on drop. +/// RAII guard that enables raw mode on creation and restores on drop. struct RawModeGuard; +impl RawModeGuard { + fn new() -> io::Result { + enable_raw_mode()?; + Ok(Self) + } +} + impl Drop for RawModeGuard { fn drop(&mut self) { let _ = disable_raw_mode(); @@ -55,12 +62,12 @@ fn parse_key_map(content: &str) -> Result, String> { serde_json::from_str(content).map_err(|e| format!("Invalid key map JSON: {e}"))?; let mut map = HashMap::new(); for (key_str, topic) in raw { - if key_str.len() != 1 { + let mut chars = key_str.chars(); + let (Some(ch), None) = (chars.next(), chars.next()) else { return Err(format!( "Key mapping keys must be single characters, got: '{key_str}'" )); - } - let ch = key_str.chars().next().unwrap(); + }; map.insert(ch, topic); } Ok(map) @@ -115,8 +122,7 @@ fn randomize_component(component: &mut SimComponent) { SimValue::Discrete { options, current, .. } => { - let idx = rng.random_range(0..options.len()); - *current = options[idx].0; + *current = options.choose(&mut rng).unwrap().0; } } } @@ -126,7 +132,12 @@ async fn poll_stub(token: CancellationToken, mut eventloop: rumqttc::v5::EventLo loop { tokio::select! { () = token.cancelled() => break, - _ = eventloop.poll() => {} + result = eventloop.poll() => { + if let Err(e) = result { + print!("MQTT connection error: {e}\r\n"); + io::stdout().flush().ok(); + } + } } } } @@ -142,9 +153,9 @@ async fn publish_injection(ch: char, component: &mut SimComponent, client: &Asyn payload.values.clone_from(&data.value); payload.time_us = timestamp; - let values_str: Vec = data.value.iter().map(|v| format!("{v:.2}")).collect(); - let Ok(bytes) = payload.write_to_bytes() else { + print!("[{ch}] serialization error for {}\r\n", data.topic); + io::stdout().flush().ok(); return; }; match client @@ -152,6 +163,7 @@ async fn publish_injection(ch: char, component: &mut SimComponent, client: &Asyn .await { Ok(()) => { + let values_str: Vec = data.value.iter().map(|v| format!("{v:.2}")).collect(); print!( "[{ch}] {} = [{}] {}\r\n", data.topic, @@ -179,10 +191,8 @@ async fn main() { return; } - let key_map_path = cli.key_map.unwrap_or_else(|| { - eprintln!("--key-map is required (use --list-topics to see available topics)"); - exit(1); - }); + // clap enforces key_map is present when list_topics is absent + let key_map_path = cli.key_map.unwrap(); let key_map = load_key_map(&key_map_path); if key_map.is_empty() { @@ -196,7 +206,6 @@ async fn main() { exit(1); } - // Set up MQTT connection let (host, port_str) = cli.siren_host_url.split_once(':').unwrap_or_else(|| { eprintln!("Invalid siren URL format, expected host:port"); exit(1); @@ -208,10 +217,7 @@ async fn main() { let client_id = format!( "Calypso-Inject-{}", - SystemTime::now() - .duration_since(UNIX_EPOCH) - .expect("Time went backwards") - .as_millis() + UNIX_EPOCH.elapsed().expect("Time went backwards").as_millis() ); let mut mqtt_opts = MqttOptions::new(client_id, host, port); mqtt_opts @@ -225,21 +231,18 @@ async fn main() { let token = CancellationToken::new(); let poll_handle = tokio::spawn(poll_stub(token.clone(), eventloop)); - // Print key mappings println!("Key Mappings:"); let mut sorted_keys: Vec<_> = components.keys().copied().collect(); sorted_keys.sort_unstable(); for key in &sorted_keys { - if let Some(component) = components.get(key) { - println!(" {key} → {} [{}]", component.name, component.unit); - } + let component = &components[key]; + println!(" {key} → {} [{}]", component.name, component.unit); } println!(); println!("Press mapped keys to inject. Ctrl+C to exit."); println!(); - enable_raw_mode().expect("Failed to enable raw mode"); - let guard = RawModeGuard; + let guard = RawModeGuard::new().expect("Failed to enable raw mode"); let mut reader = EventStream::new(); @@ -260,7 +263,12 @@ async fn main() { publish_injection(ch, component, &client).await; } } - Some(Err(_)) | None => break, + Some(Err(e)) => { + print!("Terminal event error: {e}\r\n"); + io::stdout().flush().ok(); + break; + } + None => break, _ => {} } } From 912299123335136753ca638c5bd7f7e512707a70 Mon Sep 17 00:00:00 2001 From: wyattb Date: Mon, 30 Mar 2026 21:08:08 -0400 Subject: [PATCH 04/10] #292 - add manual-sim README section, remove tests --- README.md | 26 ++++++++++ src/bin/manual-sim.rs | 113 +----------------------------------------- 2 files changed, 27 insertions(+), 112 deletions(-) diff --git a/README.md b/README.md index f72ecf9..1d5e3d7 100644 --- a/README.md +++ b/README.md @@ -39,6 +39,32 @@ Ex. `cansend vcan0 702#01010101FFFFFFFF` Now view calypso interpret the can message and broadcast it on `mqttui` +### Manual Simulation (Interactive Injection) + +`manual-sim` lets you inject MQTT messages by pressing keys in the terminal. Each key maps to a topic and publishes a randomized value within that topic's defined range. + +1. Create a keymap JSON file mapping single characters to topics: + ```json + { + "v": "BMS/Pack/Voltage", + "c": "BMS/Pack/Current", + "s": "BMS/Pack/SOC" + } + ``` + See `inject_keymap.example.json` for a starting point. Use `--list-topics` to see all available topics: + ``` + cargo run --bin manual-sim -- --list-topics + ``` + +2. Run with your keymap: + ``` + cargo run --bin manual-sim -- --key-map inject_keymap.example.json + ``` + +3. Press the mapped keys to inject values. Press `Ctrl+C` to exit. + +To use a remote broker: `cargo run --bin manual-sim -- --key-map inject_keymap.example.json -u 10.0.0.5:1883` + ### Simulation Mode #### Run from build - Same setup as above, then use the entry point `simulate` instead of `main` diff --git a/src/bin/manual-sim.rs b/src/bin/manual-sim.rs index 1394441..fb72409 100644 --- a/src/bin/manual-sim.rs +++ b/src/bin/manual-sim.rs @@ -278,115 +278,4 @@ async fn main() { token.cancel(); poll_handle.await.ok(); -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn parse_valid_key_map() { - let map = parse_key_map(r#"{"a": "BMS/Pack/Voltage", "b": "BMS/Pack/Current"}"#).unwrap(); - assert_eq!(map.len(), 2); - assert_eq!(map[&'a'], "BMS/Pack/Voltage"); - assert_eq!(map[&'b'], "BMS/Pack/Current"); - } - - #[test] - fn parse_key_map_rejects_multi_char_keys() { - let result = parse_key_map(r#"{"ab": "topic"}"#); - assert!(result.is_err()); - assert!(result.unwrap_err().contains("single characters")); - } - - #[test] - fn parse_key_map_rejects_invalid_json() { - let result = parse_key_map("not json"); - assert!(result.is_err()); - } - - #[test] - fn parse_key_map_empty() { - let map = parse_key_map("{}").unwrap(); - assert!(map.is_empty()); - } - - #[test] - fn build_topic_components_filters_to_mapped_keys() { - let components = create_simulated_components(); - if components.is_empty() { - return; - } - let target_name = components[0].name.clone(); - let mut key_map = HashMap::new(); - key_map.insert('x', target_name.clone()); - - let filtered = build_topic_components(&key_map); - assert_eq!(filtered.len(), 1); - assert_eq!(filtered[&'x'].name, target_name); - } - - #[test] - fn build_topic_components_skips_unknown_topics() { - let mut key_map = HashMap::new(); - key_map.insert('z', "NonExistent/Topic/Name".to_string()); - - let filtered = build_topic_components(&key_map); - assert!(filtered.is_empty()); - } - - #[test] - fn randomize_component_stays_within_bounds() { - let mut components = create_simulated_components(); - for component in &mut components { - for _ in 0..10 { - randomize_component(component); - for point in &component.points { - match &point.value { - SimValue::Range { - min, max, current, .. - } => { - assert!( - *current >= *min, - "value {current} below min {min} for topic {}", - component.name - ); - assert!( - *current <= *max, - "value {current} above max {max} for topic {}", - component.name - ); - } - SimValue::Discrete { - options, current, .. - } => { - assert!( - options.iter().any(|(v, _)| (*v - *current).abs() < 0.001), - "discrete value {current} not in options for topic {}", - component.name - ); - } - } - } - } - } - } - - #[test] - fn key_resolves_to_correct_topic() { - let components = create_simulated_components(); - if components.len() < 2 { - return; - } - let topic_a = components[0].name.clone(); - let topic_b = components[1].name.clone(); - - let mut key_map = HashMap::new(); - key_map.insert('a', topic_a.clone()); - key_map.insert('b', topic_b.clone()); - - let filtered = build_topic_components(&key_map); - assert_eq!(filtered[&'a'].name, topic_a); - assert_eq!(filtered[&'b'].name, topic_b); - } -} +} \ No newline at end of file From f27a05c69e45940e6599d8a47975eaa7a8b61b9d Mon Sep 17 00:00:00 2001 From: wyattb Date: Mon, 30 Mar 2026 21:09:07 -0400 Subject: [PATCH 05/10] #292 - apply rustfmt formatting --- src/bin/manual-sim.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/bin/manual-sim.rs b/src/bin/manual-sim.rs index fb72409..e20e545 100644 --- a/src/bin/manual-sim.rs +++ b/src/bin/manual-sim.rs @@ -217,7 +217,10 @@ async fn main() { let client_id = format!( "Calypso-Inject-{}", - UNIX_EPOCH.elapsed().expect("Time went backwards").as_millis() + UNIX_EPOCH + .elapsed() + .expect("Time went backwards") + .as_millis() ); let mut mqtt_opts = MqttOptions::new(client_id, host, port); mqtt_opts @@ -278,4 +281,4 @@ async fn main() { token.cancel(); poll_handle.await.ok(); -} \ No newline at end of file +} From f950e882824c458bd35b0ca9bd2f393988b759fd Mon Sep 17 00:00:00 2001 From: wyattb Date: Mon, 30 Mar 2026 21:09:51 -0400 Subject: [PATCH 06/10] #292 - revert imd_poll cfg gate (pre-existing, out of scope) --- src/lib.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/src/lib.rs b/src/lib.rs index 1c9735b..0ba8db4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -6,5 +6,4 @@ pub mod proto; pub mod simulatable_message; pub mod simulate_data; -#[cfg(target_os = "linux")] pub mod imd_poll; From 1e5824135b8e49cce251ac302feba57b870db011 Mon Sep 17 00:00:00 2001 From: wyattb Date: Tue, 31 Mar 2026 08:56:20 -0400 Subject: [PATCH 07/10] #292 - rename example keymap to manual_sim_keymap.example.json --- README.md | 6 +++--- ...ct_keymap.example.json => manual_sim_keymap.example.json | 0 2 files changed, 3 insertions(+), 3 deletions(-) rename inject_keymap.example.json => manual_sim_keymap.example.json (100%) diff --git a/README.md b/README.md index 1d5e3d7..3171da0 100644 --- a/README.md +++ b/README.md @@ -51,19 +51,19 @@ Now view calypso interpret the can message and broadcast it on `mqttui` "s": "BMS/Pack/SOC" } ``` - See `inject_keymap.example.json` for a starting point. Use `--list-topics` to see all available topics: + See `manual_sim_keymap.example.json` for a starting point. Use `--list-topics` to see all available topics: ``` cargo run --bin manual-sim -- --list-topics ``` 2. Run with your keymap: ``` - cargo run --bin manual-sim -- --key-map inject_keymap.example.json + cargo run --bin manual-sim -- --key-map manual_sim_keymap.example.json ``` 3. Press the mapped keys to inject values. Press `Ctrl+C` to exit. -To use a remote broker: `cargo run --bin manual-sim -- --key-map inject_keymap.example.json -u 10.0.0.5:1883` +To use a remote broker: `cargo run --bin manual-sim -- --key-map manual_sim_keymap.example.json -u 10.0.0.5:1883` ### Simulation Mode #### Run from build diff --git a/inject_keymap.example.json b/manual_sim_keymap.example.json similarity index 100% rename from inject_keymap.example.json rename to manual_sim_keymap.example.json From b1cf497aa25e062dcb9e0f7f156c9588f3be2d7a Mon Sep 17 00:00:00 2001 From: wyattb Date: Tue, 31 Mar 2026 11:12:37 -0400 Subject: [PATCH 08/10] #292 - simplify parse_key_map to idiomatic iterator collect --- src/bin/manual-sim.rs | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/src/bin/manual-sim.rs b/src/bin/manual-sim.rs index e20e545..703307a 100644 --- a/src/bin/manual-sim.rs +++ b/src/bin/manual-sim.rs @@ -60,17 +60,17 @@ impl Drop for RawModeGuard { fn parse_key_map(content: &str) -> Result, String> { let raw: HashMap = serde_json::from_str(content).map_err(|e| format!("Invalid key map JSON: {e}"))?; - let mut map = HashMap::new(); - for (key_str, topic) in raw { - let mut chars = key_str.chars(); - let (Some(ch), None) = (chars.next(), chars.next()) else { - return Err(format!( - "Key mapping keys must be single characters, got: '{key_str}'" - )); - }; - map.insert(ch, topic); - } - Ok(map) + raw.into_iter() + .map(|(key_str, topic)| { + let mut chars = key_str.chars(); + match (chars.next(), chars.next()) { + (Some(ch), None) => Ok((ch, topic)), + _ => Err(format!( + "Key mapping keys must be single characters, got: '{key_str}'" + )), + } + }) + .collect() } fn load_key_map(path: &str) -> HashMap { From 8b57ebc2c4b81fa9cef35b500b88a07174bfb2c6 Mon Sep 17 00:00:00 2001 From: wyattb Date: Sat, 11 Apr 2026 21:39:24 -0400 Subject: [PATCH 09/10] chore: bump Odyssey-Definitions submodule to latest main --- Odyssey-Definitions | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Odyssey-Definitions b/Odyssey-Definitions index 5805bd2..07e8193 160000 --- a/Odyssey-Definitions +++ b/Odyssey-Definitions @@ -1 +1 @@ -Subproject commit 5805bd259c11d1913a8b1d711237d2a35cec95b1 +Subproject commit 07e81934a3cea9f177854a3c8aa27877c7727a77 From dbfd09c2ef205c8f0bf169945a26c76da02140fe Mon Sep 17 00:00:00 2001 From: wyattb Date: Sat, 11 Apr 2026 23:06:48 -0400 Subject: [PATCH 10/10] chore: bump Odyssey-Definitions submodule to latest main Picks up fix for swapped sim min/max on vdig and vdiv (#59). --- Odyssey-Definitions | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Odyssey-Definitions b/Odyssey-Definitions index 07e8193..a669b65 160000 --- a/Odyssey-Definitions +++ b/Odyssey-Definitions @@ -1 +1 @@ -Subproject commit 07e81934a3cea9f177854a3c8aa27877c7727a77 +Subproject commit a669b65a64d012bb51b642823764d6fbe0444df6