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/Odyssey-Definitions b/Odyssey-Definitions index 5805bd2..a669b65 160000 --- a/Odyssey-Definitions +++ b/Odyssey-Definitions @@ -1 +1 @@ -Subproject commit 5805bd259c11d1913a8b1d711237d2a35cec95b1 +Subproject commit a669b65a64d012bb51b642823764d6fbe0444df6 diff --git a/README.md b/README.md index f72ecf9..3171da0 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 `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 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 manual_sim_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/manual_sim_keymap.example.json b/manual_sim_keymap.example.json new file mode 100644 index 0000000..f7bcd2b --- /dev/null +++ b/manual_sim_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/manual-sim.rs b/src/bin/manual-sim.rs new file mode 100644 index 0000000..703307a --- /dev/null +++ b/src/bin/manual-sim.rs @@ -0,0 +1,284 @@ +use std::collections::HashMap; +use std::io::{self, Write}; +use std::process::exit; +use std::time::{Duration, 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::mqttbytes::QoS; +use rumqttc::v5::{AsyncClient, MqttOptions}; +use tokio_util::sync::CancellationToken; + +#[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, required_unless_present = "list_topics")] + 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, +} + +/// 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(); + } +} + +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}"))?; + 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 { + 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(); + } + *current = current.clamp(*min, *max); + } + SimValue::Discrete { + options, current, .. + } => { + *current = options.choose(&mut rng).unwrap().0; + } + } + } +} + +async fn poll_stub(token: CancellationToken, mut eventloop: rumqttc::v5::EventLoop) { + loop { + tokio::select! { + () = token.cancelled() => break, + result = eventloop.poll() => { + if let Err(e) = result { + print!("MQTT connection error: {e}\r\n"); + io::stdout().flush().ok(); + } + } + } + } +} + +/// 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 Ok(bytes) = payload.write_to_bytes() else { + print!("[{ch}] serialization error for {}\r\n", data.topic); + io::stdout().flush().ok(); + return; + }; + match client + .publish(&data.topic, QoS::AtMostOnce, false, bytes) + .await + { + Ok(()) => { + let values_str: Vec = data.value.iter().map(|v| format!("{v:.2}")).collect(); + 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(); + + 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; + } + + // 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() { + 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); + } + + 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-{}", + UNIX_EPOCH + .elapsed() + .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)); + + println!("Key Mappings:"); + let mut sorted_keys: Vec<_> = components.keys().copied().collect(); + sorted_keys.sort_unstable(); + for key in &sorted_keys { + let component = &components[key]; + println!(" {key} → {} [{}]", component.name, component.unit); + } + println!(); + println!("Press mapped keys to inject. Ctrl+C to exit."); + println!(); + + let guard = RawModeGuard::new().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) { + publish_injection(ch, component, &client).await; + } + } + Some(Err(e)) => { + print!("Terminal event error: {e}\r\n"); + io::stdout().flush().ok(); + break; + } + None => break, + _ => {} + } + } + + drop(guard); + println!("\r\nShutting down..."); + + token.cancel(); + poll_handle.await.ok(); +}