From 444a0acf008bbcbd8de885e672036e4a607b1bbd Mon Sep 17 00:00:00 2001 From: Kilian Ries Date: Fri, 31 Jul 2026 12:56:56 +0200 Subject: [PATCH] feat: run an external program for each reading Closes the README item "call external scripts". --exec /path/to/script runs a program once per reading. The reading is handed over twice so a script can use whichever is easier: as MITEMPR_* environment variables, and as one JSON object on stdin. #!/bin/sh [ -n "$MITEMPR_TEMPERATURE" ] || exit 0 echo "$MITEMPR_NAME is at ${MITEMPR_TEMPERATURE} C" Every variable is always set, empty when the sensor did not report that measurement. Omitting them instead would let a value inherited from mitempr's own environment show through, and an empty string is easy to test for in shell. Rate limiting, because a Pi Zero W has one core and sensors advertise every second or two: - --exec-interval SECS is the shortest gap between two runs for the same sensor, tracked per address. Default 0, i.e. every reading. - At most four hooks run at once. Beyond that, readings are dropped with a warning rather than queued: a backlog of stale temperatures is worse than a gap, and an unbounded queue on this hardware ends badly. - A hook that has not finished within 30 seconds is killed, so a stuck script cannot hold one of those four slots forever. The child is spawned into its own task, so a slow script delays its own next run but never the scan. Its stdout goes to /dev/null on purpose: with --format json, anything the script printed would otherwise interleave with the reading stream and corrupt it. stderr is inherited so a broken script can still complain visibly. Five tests cover the environment mapping, the empty-variable rule, the per-sensor interval, and one end-to-end case that writes a real shell script to a temporary directory and checks what it actually received on stdin and in its environment. Co-Authored-By: Claude Opus 5 (1M context) --- Cargo.toml | 9 +- README.md | 30 +++++- src/exec.rs | 295 ++++++++++++++++++++++++++++++++++++++++++++++++++++ src/main.rs | 15 +++ src/scan.rs | 10 +- 5 files changed, 356 insertions(+), 3 deletions(-) create mode 100644 src/exec.rs diff --git a/Cargo.toml b/Cargo.toml index 498504f..9c6091e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,7 +11,14 @@ categories = ["command-line-utilities", "embedded"] [dependencies] # Only the runtime pieces actually used. "full" additionally pulls in fs, net, # process and io-util, which cost compile time and binary size. -tokio = { version = "1.48", features = ["rt-multi-thread", "macros", "sync", "time", "signal"] } +tokio = { version = "1.48", features = [ + "rt-multi-thread", + "macros", + "sync", + "time", + "signal", + "process", +] } uuid = "1.4" libdbus-sys = { version = "0.2", features = ["vendored"] } # "vendored" is needed for cross compilation! # "bluetoothd" is the D-Bus client API this tool uses; "full" would also build diff --git a/README.md b/README.md index fd9ac1f..ebed580 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,7 @@ Strongly inspired by [Mitemperature2](https://github.com/JsBergbau/MiTemperature ``` mitempr [--config PATH] [--only-known] [--min-rssi DBM] + [--exec PATH] [--exec-interval SECS] [--format text|json] [-v|-vv] [-q] [--watchdog SECS] [--cooldown SECS] ``` @@ -73,11 +74,38 @@ nothing but the event. Unknown keys are rejected rather than ignored, so a typo like `temp_offset` tells you about itself instead of silently doing nothing. +## Calling an external script + +`--exec /path/to/script` runs a program once per reading. The reading arrives +twice, so the script can use whichever is more convenient: as `MITEMPR_*` +environment variables, and as one JSON object on standard input. + +```sh +#!/bin/sh +# Every variable is always set; an empty one means the sensor did not report it. +[ -n "$MITEMPR_TEMPERATURE" ] || exit 0 +echo "$MITEMPR_NAME is at ${MITEMPR_TEMPERATURE} C" +``` + +Available: `MITEMPR_MAC`, `MITEMPR_NAME`, `MITEMPR_FORMAT`, +`MITEMPR_TIMESTAMP`, `MITEMPR_RSSI`, `MITEMPR_TEMPERATURE`, +`MITEMPR_HUMIDITY`, `MITEMPR_BATTERY`, `MITEMPR_VOLTAGE`, `MITEMPR_PRESSURE`, +`MITEMPR_ILLUMINANCE`, `MITEMPR_MOISTURE`. + +Sensors advertise every second or two, which is usually more often than you want +to call out to something: `--exec-interval 60` runs the script at most once a +minute per sensor. At most four hooks run at once and one that has not finished +in 30 seconds is killed, so a slow script cannot pile up processes on a Pi Zero. +When the hook cannot keep up its readings are dropped rather than queued, with a +warning — a backlog of stale temperatures is worse than a gap. + +The script's stdout goes to `/dev/null` so a chatty script cannot corrupt +`--format json`; its stderr is left alone so you can see it complain. + ## TODOs - also decode **encrypted** data - URL callback to Prometheus Push Gateway - - call external scripts - and many more things to fiddle with ;-) ## Cross compiling diff --git a/src/exec.rs b/src/exec.rs new file mode 100644 index 0000000..c801cf7 --- /dev/null +++ b/src/exec.rs @@ -0,0 +1,295 @@ +//! Handing each reading to an external program. + +use crate::output::Reading; +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use std::process::Stdio; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; +use tokio::io::AsyncWriteExt; +use tokio::process::Command; +use tokio::sync::Semaphore; + +/// How many hook processes may be alive at once. A Pi Zero W has one core and +/// 512 MB of RAM, and several sensors advertise every second or two, so an +/// unbounded fan-out of shell scripts is a real way to bring the machine down. +const MAX_PARALLEL: usize = 4; + +/// A hook that has not finished by then is assumed to be stuck and is killed, +/// so it cannot hold one of the slots above forever. +const TIMEOUT: Duration = Duration::from_secs(30); + +/// Runs an external program once per reading. +pub struct Hook { + program: PathBuf, + /// Shortest gap between two runs for the same sensor. + min_interval: Duration, + slots: Arc, + /// When the hook last ran, per sensor address. + last_run: Mutex>, +} + +impl Hook { + pub fn new(program: PathBuf, min_interval: Duration) -> Self { + Self { + program, + min_interval, + slots: Arc::new(Semaphore::new(MAX_PARALLEL)), + last_run: Mutex::new(HashMap::new()), + } + } + + /// Whether enough time has passed since this sensor's last hook run, marking + /// it as run if so. + fn claim(&self, address: &str) -> bool { + let now = Instant::now(); + let mut last_run = self + .last_run + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + + if let Some(previous) = last_run.get(address) + && now.duration_since(*previous) < self.min_interval + { + return false; + } + + last_run.insert(address.to_owned(), now); + true + } + + /// Run the hook for this reading. + /// + /// Returns immediately: the child runs in its own task so that a slow script + /// cannot stall the scan. Readings are dropped rather than queued when the + /// hook cannot keep up, since a queue of stale temperatures is worse than a + /// gap. + pub fn run(&self, reading: &Reading) { + if !self.claim(&reading.address) { + log::trace!( + "exec: skipping {}, less than {:?} since the last run", + reading.address, + self.min_interval + ); + return; + } + + let Ok(permit) = Arc::clone(&self.slots).try_acquire_owned() else { + log::warn!( + "exec: {} hooks already running, dropping the reading from {}", + MAX_PARALLEL, + reading.address + ); + return; + }; + + let program = self.program.clone(); + let environment = environment(reading); + let json = match serde_json::to_string(reading) { + Ok(json) => json, + Err(e) => { + log::error!("exec: could not serialise reading: {e}"); + return; + } + }; + + tokio::spawn(async move { + // Held for as long as the child lives. + let _permit = permit; + + match tokio::time::timeout(TIMEOUT, invoke(&program, &environment, &json)).await { + Ok(Ok(status)) if status.success() => { + log::debug!("exec: {} finished", program.display()); + } + Ok(Ok(status)) => { + log::warn!("exec: {} exited with {status}", program.display()); + } + Ok(Err(e)) => log::warn!("exec: {} failed: {e}", program.display()), + Err(_) => log::warn!( + "exec: {} did not finish within {TIMEOUT:?}, killed", + program.display() + ), + } + }); + } +} + +async fn invoke( + program: &Path, + environment: &[(&'static str, String)], + json: &str, +) -> std::io::Result { + let mut child = Command::new(program) + .envs(environment.iter().map(|(name, value)| (*name, value))) + .stdin(Stdio::piped()) + // Deliberately not inherited: a chatty script would otherwise interleave + // itself with --format json output and corrupt the stream. stderr is + // inherited so a broken script still says so. + .stdout(Stdio::null()) + .kill_on_drop(true) + .spawn()?; + + if let Some(mut stdin) = child.stdin.take() { + stdin.write_all(json.as_bytes()).await?; + stdin.write_all(b"\n").await?; + // Dropping the pipe closes it, so a script reading to EOF is not left + // waiting. + } + + child.wait().await +} + +/// The environment handed to the hook. +/// +/// Every variable is always set, empty when the sensor did not report that +/// measurement. Leaving them out instead would mean a value inherited from +/// mitempr's own environment could show through. +fn environment(reading: &Reading) -> Vec<(&'static str, String)> { + fn number(value: Option) -> String { + value.map(|value| value.to_string()).unwrap_or_default() + } + + vec![ + ("MITEMPR_MAC", reading.address.clone()), + ("MITEMPR_NAME", reading.name.clone().unwrap_or_default()), + ("MITEMPR_FORMAT", reading.format.to_owned()), + ("MITEMPR_TIMESTAMP", reading.timestamp.to_string()), + ("MITEMPR_RSSI", number(reading.rssi_dbm)), + ("MITEMPR_TEMPERATURE", number(reading.temperature_celsius)), + ("MITEMPR_HUMIDITY", number(reading.humidity_percent)), + ("MITEMPR_BATTERY", number(reading.battery_percent)), + ("MITEMPR_VOLTAGE", number(reading.voltage_volts)), + ("MITEMPR_PRESSURE", number(reading.pressure_hpa)), + ("MITEMPR_ILLUMINANCE", number(reading.illuminance_lux)), + ("MITEMPR_MOISTURE", number(reading.moisture_percent)), + ] +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::decoder::{SensorData, SensorFormat}; + use bluer::Address; + + fn reading() -> Reading { + let data = SensorData { + format: SensorFormat::Pvvx, + temperature: Some(22.9), + humidity: Some(64.25), + battery: Some(16), + ..Default::default() + }; + let mut reading = Reading::new( + Address::new([0xA4, 0xC1, 0x38, 0xA0, 0x7B, 0x03]), + Some("Living Room".to_string()), + Some(-67), + &data, + ); + reading.timestamp = 1_770_000_000; + reading + } + + fn variable(environment: &[(&'static str, String)], key: &str) -> String { + environment + .iter() + .find(|(name, _)| *name == key) + .map(|(_, value)| value.clone()) + .unwrap_or_else(|| panic!("{key} should be present")) + } + + #[test] + fn every_measurement_becomes_a_variable() { + let environment = environment(&reading()); + + assert_eq!(variable(&environment, "MITEMPR_MAC"), "A4:C1:38:A0:7B:03"); + assert_eq!(variable(&environment, "MITEMPR_NAME"), "Living Room"); + assert_eq!(variable(&environment, "MITEMPR_FORMAT"), "pvvx"); + assert_eq!(variable(&environment, "MITEMPR_TIMESTAMP"), "1770000000"); + assert_eq!(variable(&environment, "MITEMPR_RSSI"), "-67"); + assert_eq!(variable(&environment, "MITEMPR_TEMPERATURE"), "22.9"); + assert_eq!(variable(&environment, "MITEMPR_HUMIDITY"), "64.25"); + assert_eq!(variable(&environment, "MITEMPR_BATTERY"), "16"); + } + + /// A measurement the sensor did not send must be present but empty, so a + /// script can test for it and cannot pick up a stale inherited value. + #[test] + fn absent_measurements_become_empty_variables() { + let environment = environment(&reading()); + + assert_eq!(variable(&environment, "MITEMPR_PRESSURE"), ""); + assert_eq!(variable(&environment, "MITEMPR_ILLUMINANCE"), ""); + assert_eq!(variable(&environment, "MITEMPR_MOISTURE"), ""); + assert_eq!(variable(&environment, "MITEMPR_VOLTAGE"), ""); + } + + #[test] + fn a_zero_interval_never_holds_a_reading_back() { + let hook = Hook::new(PathBuf::from("/bin/true"), Duration::ZERO); + + assert!(hook.claim("A4:C1:38:A0:7B:03")); + assert!(hook.claim("A4:C1:38:A0:7B:03")); + } + + #[test] + fn the_interval_is_tracked_per_sensor() { + let hook = Hook::new(PathBuf::from("/bin/true"), Duration::from_secs(60)); + + assert!(hook.claim("A4:C1:38:A0:7B:03")); + assert!( + !hook.claim("A4:C1:38:A0:7B:03"), + "the same sensor is held back" + ); + assert!(hook.claim("A4:C1:38:AA:BB:CC"), "a different sensor is not"); + } + + /// The end-to-end path: a real script, the environment and the JSON on + /// stdin. + #[cfg(unix)] + #[tokio::test] + async fn the_hook_receives_the_environment_and_the_json() { + use std::os::unix::fs::PermissionsExt; + + let directory = + std::env::temp_dir().join(format!("mitempr-exec-test-{}", std::process::id())); + std::fs::create_dir_all(&directory).expect("temp directory should be creatable"); + let script = directory.join("hook.sh"); + let output = directory.join("output"); + + std::fs::write( + &script, + format!( + "#!/bin/sh\nread -r line\n{{ echo \"$MITEMPR_NAME|$MITEMPR_TEMPERATURE|$MITEMPR_MOISTURE\"; echo \"$line\"; }} > {}\n", + output.display() + ), + ) + .expect("script should be writable"); + std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)) + .expect("script should be executable"); + + let reading = reading(); + let status = invoke( + &script, + &environment(&reading), + &serde_json::to_string(&reading).expect("reading should serialise"), + ) + .await + .expect("hook should run"); + assert!(status.success(), "hook exited with {status}"); + + let written = + std::fs::read_to_string(&output).expect("hook should have written its output"); + let mut lines = written.lines(); + assert_eq!(lines.next(), Some("Living Room|22.9|")); + assert_eq!( + lines.next(), + Some( + serde_json::to_string(&reading) + .expect("reading should serialise") + .as_str() + ) + ); + + let _ = std::fs::remove_dir_all(&directory); + } +} diff --git a/src/main.rs b/src/main.rs index d2854b0..cee12d6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -5,10 +5,12 @@ use std::time::Duration; mod config; mod decoder; +mod exec; mod output; mod scan; use config::Config; +use exec::Hook; use output::Format; /// Read environmental data from Bluetooth sensors. @@ -27,6 +29,15 @@ struct Args { #[arg(long, value_name = "DBM", allow_negative_numbers = true)] min_rssi: Option, + /// Run this program once per reading, with the reading in MITEMPR_* + /// environment variables and as JSON on its standard input + #[arg(long, value_name = "PATH")] + exec: Option, + + /// Shortest gap in seconds between two --exec runs for the same sensor + #[arg(long, value_name = "SECS", default_value_t = 0, requires = "exec")] + exec_interval: u64, + /// Restart discovery if no reading arrives for this many seconds #[arg(long, default_value_t = 20)] watchdog: u64, @@ -115,6 +126,10 @@ async fn main() -> Result<(), Box> { cooldown: Duration::from_secs(args.cooldown), format: args.format, config, + exec: args + .exec + .clone() + .map(|program| Hook::new(program, Duration::from_secs(args.exec_interval))), }; tokio::select! { diff --git a/src/scan.rs b/src/scan.rs index 4558d96..735c4d9 100644 --- a/src/scan.rs +++ b/src/scan.rs @@ -2,6 +2,7 @@ use crate::config::Config; use crate::decoder; +use crate::exec::Hook; use crate::output::{Format, Reading}; use bluer::{Adapter, AdapterEvent, Address, DeviceProperty, Result}; use futures::StreamExt; @@ -29,6 +30,8 @@ pub struct Settings { pub format: Format, /// Which sensors to report, and how to name and calibrate them. pub config: Config, + /// External program to run per reading, if any. + pub exec: Option, } /// Scan until the adapter disappears or the task is cancelled. @@ -205,7 +208,12 @@ async fn handle_device( } } - Reading::new(addr, name, rssi, &data).emit(settings.format); + let reading = Reading::new(addr, name, rssi, &data); + reading.emit(settings.format); + + if let Some(hook) = &settings.exec { + hook.run(&reading); + } // Only an actual reading counts as proof that the scan is alive. *last_reading_at.lock().await = Instant::now();