From 3274928a5f7202ffa95a1a603cf52bf57a258269 Mon Sep 17 00:00:00 2001 From: Duncan Tourolle Date: Thu, 20 Aug 2026 19:27:04 +0200 Subject: [PATCH] Stream heart rate from a strap or watch into the ride and the FIT file MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A new HRM client in the BLE crate follows the crate's split: the 0x2A37 decoder is a pure function over bytes (u8/u16 formats, the three-state sensor-contact field, straps that append energy/RR data), and only the actor touches the radio. Anything exposing the standard Heart Rate Service works — a chest strap, or a Garmin watch with Broadcast Heart Rate on. A single-slot supervisor in the app owns the link, shaped like the trainer's and the controller's. It publishes bpm on a watch channel the session backend stamps onto each tick's telemetry — never over a heart rate FTMS itself reported, on the same authority rule as the Zwift cadence merge — and it clears the reading after eight silent seconds, so a strap taken off records nothing rather than a flatline of the last real value. From there the existing pipeline does the rest: ride screen tile, FIT records, avg/max in lap and session. The device list routes heart-rate rows to the supervisor, keeps the row alive while connected (a connected monitor stops advertising), and shows the live bpm as proof data is flowing — a connected-but-silent monitor otherwise looks exactly like a working one. Co-Authored-By: Claude Fable 5 --- crates/ble/src/heart_rate.rs | 675 ++++++++++++++++++++++ crates/ble/src/lib.rs | 6 + crates/ble/src/uuids.rs | 12 + src-tauri/src/devices.rs | 97 +++- src-tauri/src/heart_rate.rs | 521 +++++++++++++++++ src-tauri/src/lib.rs | 1 + src-tauri/src/session_backend.rs | 80 ++- src-tauri/src/state.rs | 70 ++- ui/src/components/ConnectionScreen.svelte | 27 +- ui/src/lib/types.ts | 2 + 10 files changed, 1467 insertions(+), 24 deletions(-) create mode 100644 crates/ble/src/heart_rate.rs create mode 100644 src-tauri/src/heart_rate.rs diff --git a/crates/ble/src/heart_rate.rs b/crates/ble/src/heart_rate.rs new file mode 100644 index 0000000..438b060 --- /dev/null +++ b/crates/ble/src/heart_rate.rs @@ -0,0 +1,675 @@ +//! Heart rate monitor client: decoder and connection actor. +//! +//! Structurally a sibling of [`crate::click`]: one task owns the peripheral, +//! callers hold a cheap handle and receive [`HeartRateEvent`]s on a broadcast +//! channel. The wire format lives in this module too, as pure functions over +//! bytes, because it is a single characteristic and does not earn a module of +//! its own. +//! +//! ```text +//! caller ◄──broadcast── HeartRateEvent ◄── actor ◄─notify── HRM / watch +//! ``` +//! +//! Works with anything that exposes the standard Heart Rate Service (`0x180D`) +//! — a chest strap, or a sports watch with *Broadcast Heart Rate* switched on. +//! A watch in broadcast mode accepts a single central, so connecting it here +//! takes it away from any other app; that is a property of the watch, not of +//! this client. +//! +//! Read-only in every sense: nothing is ever written to the device, so — like +//! the Click and unlike the trainer — there is no safety sequence, only the +//! SAF-9 obligation to actually close the link on the way out. + +use std::future::Future; +use std::time::Duration; + +use btleplug::api::{Characteristic, Peripheral as _}; +use btleplug::platform::{Adapter, Peripheral}; +use futures::{Stream, StreamExt}; +use tokio::sync::{broadcast, mpsc, oneshot}; + +use crate::client::{Backoff, InFlight, DISCONNECT_TIMEOUT}; +use crate::error::FtmsError; +use crate::indoor_bike_data::hex; +use crate::scan::{self, DiscoveredDevice, ScanKind}; +use crate::uuids; + +// --------------------------------------------------------------------------- +// Wire format (pure) +// --------------------------------------------------------------------------- + +/// One decoded Heart Rate Measurement (`0x2A37`) notification. +/// +/// Only what the app uses. The characteristic can also carry energy expended +/// and RR intervals; both are skipped, not rejected — a device that sends them +/// still decodes. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct HeartRateMeasurement { + /// Beats per minute. `u16` because the flags allow it, though anything a + /// human produces fits in a `u8`. + pub bpm: u16, + /// Skin contact, where the device reports the feature at all. `None` means + /// "not supported", not "no contact" — the two must not be conflated, or + /// every strap without the feature reads as fallen off. + pub sensor_contact: Option, +} + +/// Why a Heart Rate Measurement frame would not decode. +#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)] +pub enum HrDecodeError { + #[error("empty heart rate measurement")] + Empty, + #[error("heart rate measurement of {len} bytes is too short for its flags (0x{flags:02x})")] + TooShort { flags: u8, len: usize }, +} + +/// Decode a Heart Rate Measurement (`0x2A37`) value. +/// +/// Layout per the Bluetooth SIG GATT specification: a flags byte, then the +/// heart rate as `u8` or `u16` little-endian depending on flags bit 0. Bits +/// 1–2 describe sensor contact: bit 2 says the feature is supported, bit 1 +/// says contact is detected. +pub fn decode_heart_rate(data: &[u8]) -> Result { + let (&flags, rest) = data.split_first().ok_or(HrDecodeError::Empty)?; + + let bpm = if flags & 0x01 == 0 { + *rest.first().ok_or(HrDecodeError::TooShort { + flags, + len: data.len(), + })? as u16 + } else { + match rest { + [lo, hi, ..] => u16::from_le_bytes([*lo, *hi]), + _ => { + return Err(HrDecodeError::TooShort { + flags, + len: data.len(), + }) + } + } + }; + + let sensor_contact = match flags & 0x06 { + 0x06 => Some(true), + 0x04 => Some(false), + _ => None, + }; + + Ok(HeartRateMeasurement { + bpm, + sensor_contact, + }) +} + +// --------------------------------------------------------------------------- +// Selection and configuration +// --------------------------------------------------------------------------- + +/// How to pick a heart rate monitor out of a scan. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum HrSelector { + /// A specific device, by address — what the app uses, since the scan has + /// already identified it. Matched without insisting on the advertisement: + /// a device mid-connection may not be advertising its services. + Address(String), + /// Any peripheral advertising the Heart Rate Service (`0x180D`). + Any, +} + +impl HrSelector { + pub fn matches(&self, d: &DiscoveredDevice) -> bool { + self.matches_parts( + &d.address, + d.services.contains(&uuids::HEART_RATE_SERVICE), + &format!("{:?}", d.id), + ) + } + + /// The matching rule, factored out so it can be unit-tested without a + /// `PeripheralId` (which only the platform backend can construct). + pub(crate) fn matches_parts( + &self, + address: &str, + advertises_hr: bool, + id_debug: &str, + ) -> bool { + match self { + HrSelector::Address(a) => { + address.eq_ignore_ascii_case(a) + || id_debug.to_lowercase().contains(&a.to_lowercase()) + } + HrSelector::Any => advertises_hr, + } + } + + /// Human-readable, for error messages the rider will actually read. + pub fn describe(&self) -> String { + match self { + HrSelector::Address(a) => format!("heart rate monitor at {a}"), + HrSelector::Any => "any heart rate monitor".to_string(), + } + } +} + +/// Tunables for [`HeartRateClient`]. +#[derive(Debug, Clone)] +pub struct HeartRateConfig { + /// How long to look before giving up. A broadcasting device advertises + /// continuously — unlike a Click — so this needs no extra patience. + pub scan_timeout: Duration, + pub backoff: Backoff, + pub channel_capacity: usize, +} + +impl Default for HeartRateConfig { + fn default() -> Self { + Self { + scan_timeout: Duration::from_secs(15), + backoff: Backoff::default(), + channel_capacity: 64, + } + } +} + +// --------------------------------------------------------------------------- +// Events and handle +// --------------------------------------------------------------------------- + +/// Everything the app learns from a heart rate monitor. +#[derive(Debug, Clone, PartialEq)] +pub enum HeartRateEvent { + Connected { + address: String, + name: Option, + /// Read once at connect, where the device exposes `0x2A19` at all. + battery_percent: Option, + }, + /// The link dropped. The actor is retrying — this is not terminal. + Disconnected, + /// Reconnecting ran out of attempts and the actor has stopped (FR-1.11). + /// Terminal: nothing further arrives on this stream. + GaveUp { attempts: u32 }, + /// A decoded measurement. + Measurement { + bpm: u16, + sensor_contact: Option, + }, +} + +enum Cmd { + Shutdown { reply: oneshot::Sender<()> }, +} + +/// Cheap, cloneable handle to a heart rate monitor session. +#[derive(Clone)] +pub struct HeartRateClient { + cmd_tx: mpsc::Sender, + events_tx: broadcast::Sender, + address: String, + name: Option, + battery_percent: Option, +} + +impl HeartRateClient { + /// Carried on the handle rather than left to the event stream, because the + /// first `Connected` is sent before the caller has had a chance to + /// subscribe — same reasoning as [`crate::click::ClickClient::address`]. + pub fn address(&self) -> &str { + &self.address + } + + pub fn name(&self) -> Option<&str> { + self.name.as_deref() + } + + /// Battery level at connect time, where the device reported one. + pub fn battery_percent(&self) -> Option { + self.battery_percent + } + + /// Connect to a heart rate monitor and start streaming measurements. + /// + /// Returns once the measurement characteristic is subscribed, so an `Ok` + /// means data is genuinely on its way — not merely that a link exists. + pub async fn connect( + selector: HrSelector, + config: HeartRateConfig, + ) -> Result { + let adapter = scan::default_adapter().await?; + Self::connect_with_adapter(adapter, selector, config).await + } + + /// As [`HeartRateClient::connect`], but abandoned as soon as `cancel` + /// resolves. Returns `Ok(None)` when it was cancelled; any link the + /// abandoned attempt had opened is closed before this returns (SAF-9, + /// FR-1.10). + pub async fn connect_cancellable( + selector: HrSelector, + config: HeartRateConfig, + cancel: impl Future, + ) -> Result, FtmsError> { + let in_flight = InFlight::default(); + let outcome = { + let attempt = async { + let adapter = scan::default_adapter().await?; + Self::connect_on(adapter, selector, config, &in_flight).await + }; + tokio::pin!(attempt, cancel); + tokio::select! { + result = &mut attempt => Some(result), + () = &mut cancel => None, + } + }; + match outcome { + Some(result) => result.map(Some), + None => { + tracing::info!("hr: connect attempt cancelled"); + in_flight.abandon().await; + Ok(None) + } + } + } + + /// As [`HeartRateClient::connect`], but on a caller-supplied adapter. + pub async fn connect_with_adapter( + adapter: Adapter, + selector: HrSelector, + config: HeartRateConfig, + ) -> Result { + Self::connect_on(adapter, selector, config, &InFlight::default()).await + } + + async fn connect_on( + adapter: Adapter, + selector: HrSelector, + config: HeartRateConfig, + in_flight: &InFlight, + ) -> Result { + let (events_tx, _) = broadcast::channel(config.channel_capacity); + let (cmd_tx, cmd_rx) = mpsc::channel(8); + + let (session, notifications) = + open_session(&adapter, &selector, &config, in_flight).await?; + let (address, name, battery_percent) = ( + session.address.clone(), + session.name.clone(), + session.battery_percent, + ); + // Sent for symmetry with a reconnect; the same facts ride out on the + // handle below because nobody is subscribed yet. + let _ = events_tx.send(HeartRateEvent::Connected { + address: address.clone(), + name: name.clone(), + battery_percent, + }); + + let actor = Actor { + adapter, + // Reconnect to *this* device, not to whatever HRM now answers + // first — two straps in a household is not exotic. + selector: if address.is_empty() { + selector + } else { + HrSelector::Address(address.clone()) + }, + config, + events_tx: events_tx.clone(), + }; + tokio::spawn(actor.run(cmd_rx, session, Box::pin(notifications))); + + Ok(Self { + cmd_tx, + events_tx, + address, + name, + battery_percent, + }) + } + + /// Subscribe to events. Late subscribers see only what arrives after they + /// subscribe. + pub fn events(&self) -> broadcast::Receiver { + self.events_tx.subscribe() + } + + /// Close the link, and wait for it to actually be closed (SAF-9). + /// Idempotent — shutting down a stopped client is a no-op. + pub async fn shutdown(&self) { + let (reply, done) = oneshot::channel(); + if self.cmd_tx.send(Cmd::Shutdown { reply }).await.is_err() { + // The actor is already gone; it tore the link down on its way out. + return; + } + let _ = done.await; + } +} + +// --------------------------------------------------------------------------- +// Session and actor +// --------------------------------------------------------------------------- + +/// Boxed so the initial connection and every reconnect share one type. +type Notifications = std::pin::Pin< + Box + Send>, +>; + +struct Session { + peripheral: Peripheral, + address: String, + name: Option, + battery_percent: Option, + measurement: Characteristic, +} + +/// Find the device, connect and subscribe to Heart Rate Measurement. +async fn open_session( + adapter: &Adapter, + selector: &HrSelector, + config: &HeartRateConfig, + in_flight: &InFlight, +) -> Result<(Session, Notifications), FtmsError> { + // Unfiltered: an address selector must match a device whatever it + // advertises, and some backends ignore service filters anyway. + let peripheral = scan::find_matching( + adapter, + ScanKind::All, + config.scan_timeout, + &selector.describe(), + |d| selector.matches(d), + ) + .await?; + // Cancelling past this point would otherwise strand the link (FR-1.10). + in_flight.hold(peripheral.clone()); + + match setup_session(peripheral.clone()).await { + Ok(session) => { + in_flight.released(); + Ok(session) + } + Err(e) => { + // A failure after `connect()` leaves a live GATT link behind, and a + // watch in broadcast mode accepts one central — held, it will not + // advertise for the retry, so the retry loop would never succeed. + tracing::debug!(error = %e, "hr: session setup failed; disconnecting"); + let _ = peripheral.disconnect().await; + in_flight.released(); + Err(e) + } + } +} + +async fn setup_session(peripheral: Peripheral) -> Result<(Session, Notifications), FtmsError> { + if !peripheral.is_connected().await.unwrap_or(false) { + peripheral.connect().await?; + } + peripheral.discover_services().await?; + + let chars = peripheral.characteristics(); + // Prefer the characteristic inside the Heart Rate Service, but take a + // bare 0x2A37 anywhere — the same leniency the FTMS client extends. + let measurement = chars + .iter() + .find(|c| { + c.uuid == uuids::HEART_RATE_MEASUREMENT + && c.service_uuid == uuids::HEART_RATE_SERVICE + }) + .or_else(|| chars.iter().find(|c| c.uuid == uuids::HEART_RATE_MEASUREMENT)) + .cloned() + .ok_or(FtmsError::MissingCharacteristic( + "Heart Rate Measurement (0x2A37)", + ))?; + + let notifications = peripheral.notifications().await?; + peripheral.subscribe(&measurement).await?; + + // Best effort, once: straps report battery, watches in broadcast mode + // usually do not. A failure here must not cost the session. + let battery_percent = match chars.iter().find(|c| c.uuid == uuids::BATTERY_LEVEL) { + Some(c) => peripheral + .read(c) + .await + .ok() + .and_then(|v| v.first().copied()) + .filter(|p| *p <= 100), + None => None, + }; + + let described = scan::describe(&peripheral).await; + Ok(( + Session { + address: described + .as_ref() + .map(|d| d.address.clone()) + .unwrap_or_default(), + name: described.and_then(|d| d.name), + battery_percent, + peripheral, + measurement, + }, + Box::pin(notifications), + )) +} + +struct Actor { + adapter: Adapter, + selector: HrSelector, + config: HeartRateConfig, + events_tx: broadcast::Sender, +} + +impl Actor { + async fn run( + self, + mut cmd_rx: mpsc::Receiver, + mut current: Session, + mut notifications: Notifications, + ) { + let mut attempt = 0u32; + + loop { + let dropped = tokio::select! { + cmd = cmd_rx.recv() => { + match cmd { + Some(Cmd::Shutdown { reply }) => { + self.teardown(¤t).await; + // Answered only once the link is genuinely closed + // (SAF-9). + let _ = reply.send(()); + return; + } + None => { + self.teardown(¤t).await; + return; + } + } + } + frame = notifications.next() => match frame { + Some(n) => { self.handle(n.uuid, &n.value); false } + // The stream ending is how btleplug reports a dropped link. + None => true, + }, + }; + + if !dropped { + continue; + } + + tracing::warn!( + selector = %self.selector.describe(), + "hr: notification stream ended — link dropped" + ); + let _ = self.events_tx.send(HeartRateEvent::Disconnected); + + // FR-1.6: reconnect with backoff. A watch that left broadcast mode + // or a strap taken off will simply not be found, which is normal. + loop { + if self.config.backoff.exhausted(attempt) { + tracing::warn!("hr: giving up after {attempt} reconnect attempts"); + // Say so before going away (FR-1.11): a silent exit strands + // the supervisor holding a handle whose actor is gone. + let _ = self.events_tx.send(HeartRateEvent::GaveUp { attempts: attempt }); + return; + } + let delay = self.config.backoff.delay(attempt); + attempt += 1; + + // Both the backoff and the attempt stay answerable to shutdown + // (FR-1.10, SAF-9) — an attempt runs for up to `scan_timeout`. + let in_flight = InFlight::default(); + let outcome = { + let attempting = async { + tokio::time::sleep(delay).await; + open_session(&self.adapter, &self.selector, &self.config, &in_flight) + .await + }; + tokio::pin!(attempting); + tokio::select! { + biased; + cmd = cmd_rx.recv() => Err(cmd), + result = &mut attempting => Ok(result), + } + }; + + let result = match outcome { + Ok(result) => result, + Err(cmd) => { + in_flight.abandon().await; + if let Some(Cmd::Shutdown { reply }) = cmd { + let _ = reply.send(()); + } + return; + } + }; + + match result { + Ok((session, stream)) => { + let _ = self.events_tx.send(HeartRateEvent::Connected { + address: session.address.clone(), + name: session.name.clone(), + battery_percent: session.battery_percent, + }); + current = session; + notifications = stream; + attempt = 0; + break; + } + Err(e) => tracing::debug!(error = %e, "hr: reconnect failed"), + } + } + } + } + + /// Decode one notification into an event. + fn handle(&self, uuid: uuid::Uuid, raw: &[u8]) { + if uuid != uuids::HEART_RATE_MEASUREMENT { + tracing::trace!(%uuid, raw = %hex(raw), "hr: notification on an unexpected characteristic"); + return; + } + match decode_heart_rate(raw) { + Ok(m) => { + let _ = self.events_tx.send(HeartRateEvent::Measurement { + bpm: m.bpm, + sensor_contact: m.sensor_contact, + }); + } + // Logged, never fatal (NFR-4). + Err(e) => tracing::debug!(raw = %hex(raw), error = %e, "hr: bad measurement frame"), + } + } + + /// Close the link. Bounded: this runs on the app's exit path (NFR-9). + async fn teardown(&self, session: &Session) { + let unsubscribe = session.peripheral.unsubscribe(&session.measurement); + let _ = tokio::time::timeout(DISCONNECT_TIMEOUT, unsubscribe).await; + match tokio::time::timeout(DISCONNECT_TIMEOUT, session.peripheral.disconnect()).await { + Ok(Ok(())) => tracing::info!("hr: disconnected"), + Ok(Err(e)) => tracing::debug!(error = %e, "hr: disconnect failed"), + Err(_) => tracing::warn!("hr: disconnect timed out"), + } + let _ = self.events_tx.send(HeartRateEvent::Disconnected); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_plain_u8_measurement_decodes() { + // Flags 0x00: u8 heart rate, no contact feature, nothing else. + let m = decode_heart_rate(&[0x00, 142]).unwrap(); + assert_eq!(m.bpm, 142); + assert_eq!(m.sensor_contact, None); + } + + #[test] + fn a_u16_measurement_decodes_little_endian() { + // Flags 0x01: u16 heart rate. 0x0121 = 289 — absurd for a human but + // the format allows it, and the decoder is a decoder, not a referee. + let m = decode_heart_rate(&[0x01, 0x21, 0x01]).unwrap(); + assert_eq!(m.bpm, 289); + } + + #[test] + fn sensor_contact_is_three_states_not_two() { + // Bit 2 = feature supported, bit 1 = contact detected. A strap without + // the feature must read `None`, not "fallen off". + assert_eq!( + decode_heart_rate(&[0x06, 100]).unwrap().sensor_contact, + Some(true) + ); + assert_eq!( + decode_heart_rate(&[0x04, 100]).unwrap().sensor_contact, + Some(false) + ); + assert_eq!(decode_heart_rate(&[0x00, 100]).unwrap().sensor_contact, None); + // Bit 1 without bit 2 is "feature not supported" per the spec. + assert_eq!(decode_heart_rate(&[0x02, 100]).unwrap().sensor_contact, None); + } + + #[test] + fn trailing_fields_are_skipped_not_rejected() { + // Flags 0x18: energy expended + RR intervals present. Real straps + // (Garmin HRM-Dual, Polar H10) send these; the bpm must still decode. + let m = decode_heart_rate(&[0x18, 155, 0x10, 0x00, 0x40, 0x03]).unwrap(); + assert_eq!(m.bpm, 155); + } + + #[test] + fn short_frames_are_errors_not_panics() { + // NFR-4: a malformed packet is a value, never a crash. + assert_eq!(decode_heart_rate(&[]), Err(HrDecodeError::Empty)); + assert!(matches!( + decode_heart_rate(&[0x00]), + Err(HrDecodeError::TooShort { .. }) + )); + // u16 flag with only one byte of payload. + assert!(matches!( + decode_heart_rate(&[0x01, 142]), + Err(HrDecodeError::TooShort { .. }) + )); + } + + #[test] + fn selector_any_requires_the_heart_rate_service() { + let s = HrSelector::Any; + assert!(s.matches_parts("aa:bb:cc:dd:ee:ff", true, "")); + assert!(!s.matches_parts("aa:bb:cc:dd:ee:ff", false, "")); + } + + #[test] + fn selector_address_ignores_the_advertisement() { + // Reconnect runs against a device that may not be advertising its + // services; the address alone identifies it. + let s = HrSelector::Address("AA:BB:CC:DD:EE:FF".into()); + assert!(s.matches_parts("aa:bb:cc:dd:ee:ff", false, "")); + assert!(!s.matches_parts("11:22:33:44:55:66", true, "")); + } + + #[test] + fn selector_address_also_matches_an_opaque_platform_id() { + // macOS gives UUID-shaped PeripheralIds rather than MACs. + let s = HrSelector::Address("1E2F3A4B".into()); + assert!(s.matches_parts("", false, "PeripheralId(1e2f3a4b-....)")); + } +} diff --git a/crates/ble/src/lib.rs b/crates/ble/src/lib.rs index 5f8a256..d11cba3 100644 --- a/crates/ble/src/lib.rs +++ b/crates/ble/src/lib.rs @@ -13,6 +13,7 @@ //! | [`control_point`] | `0x2AD9` encoders and response decoding | no | //! | [`capabilities`] | `0x2ACC`/`0x2AD5`/`0x2AD6`/`0x2AD8` decoding, and the safety gate | no | //! | [`zwift`] | Zwift's proprietary protocol (§2.3.1) | no | +//! | [`heart_rate`] | `0x2A37` decoder (pure) and the HRM connection actor | decoder no, actor yes | //! | [`scan`] | discovery | yes | //! | [`client`] | the connection actor | yes | //! @@ -52,12 +53,17 @@ pub mod click; pub mod client; pub mod control_point; pub mod error; +pub mod heart_rate; pub mod indoor_bike_data; pub mod scan; pub mod uuids; pub mod zwift; pub use click::{ClickClient, ClickConfig, ClickEvent, PodSelector}; +pub use heart_rate::{ + decode_heart_rate, HeartRateClient, HeartRateConfig, HeartRateEvent, HeartRateMeasurement, + HrSelector, +}; pub use capabilities::{ FitnessMachineFeature, InclinationRange, PowerRange, ResistanceLevelRange, TrainerCapabilities, UnsupportedTarget, diff --git a/crates/ble/src/uuids.rs b/crates/ble/src/uuids.rs index 90ed1de..3e6a00c 100644 --- a/crates/ble/src/uuids.rs +++ b/crates/ble/src/uuids.rs @@ -55,6 +55,16 @@ pub const DEVICE_INFORMATION_SERVICE: Uuid = uuid16(0x180A); /// Battery Service — `0x180F`. pub const BATTERY_SERVICE: Uuid = uuid16(0x180F); +/// Battery Level — `0x2A19`, read. +pub const BATTERY_LEVEL: Uuid = uuid16(0x2A19); + +/// Heart Rate Service — `0x180D`. What a strap — or a watch in broadcast +/// mode — advertises. +pub const HEART_RATE_SERVICE: Uuid = uuid16(0x180D); + +/// Heart Rate Measurement — `0x2A37`, notify. +pub const HEART_RATE_MEASUREMENT: Uuid = uuid16(0x2A37); + /// Human-readable name for a well-known UUID, for logging and the probe CLI. /// Returns `None` for anything not recognised. pub fn well_known_name(uuid: Uuid) -> Option<&'static str> { @@ -79,6 +89,8 @@ pub fn well_known_name(uuid: Uuid) -> Option<&'static str> { 0x1816 => "Cycling Speed and Cadence", 0x2A00 => "Device Name", 0x2A19 => "Battery Level", + 0x2A37 => "Heart Rate Measurement", + 0x2A38 => "Body Sensor Location", 0x2A24 => "Model Number String", 0x2A25 => "Serial Number String", 0x2A26 => "Firmware Revision String", diff --git a/src-tauri/src/devices.rs b/src-tauri/src/devices.rs index 87b9bff..237f868 100644 --- a/src-tauri/src/devices.rs +++ b/src-tauri/src/devices.rs @@ -30,6 +30,7 @@ use tokio::sync::watch; use uuid::Uuid; use crate::controller::{ControllerHandle, PodState}; +use crate::heart_rate::{HeartRateHandle, HeartRateStatus}; use crate::trainer::{TrainerHandle, TrainerStatus}; /// One pass of the scanner. Long enough for a trainer to advertise, short @@ -100,6 +101,9 @@ pub struct DeviceInfo { /// Previously paired, so it would auto-connect on launch (FR-1.5). pub remembered: bool, pub battery_pct: Option, + /// Live reading from a connected heart rate monitor — the row's proof that + /// data is actually flowing, the way a pod's battery is. + pub heart_rate_bpm: Option, // No unlock countdown. FR-3.9 assumed the Click v2 needed its Zwift session // refreshing daily; TASK-0 disproved it on this hardware — the pods answer // `RideOn 00 09` unencrypted, with no key exchange and no expiry (§2.3.1). @@ -117,6 +121,9 @@ pub struct PollResult { /// Trainer status, when it changed since the last poll. The caller turns /// this into user-facing notices (FR-1.8, FR-9.4). pub trainer_changed: Option, + /// Heart rate link status, when it changed since the last poll — same + /// contract as `trainer_changed`. + pub hr_changed: Option, } /// What the scan task publishes. @@ -135,6 +142,8 @@ pub struct DeviceRegistry { /// Held so a Click row in the device list connects the same way its card on /// the connection screen does — one path, not two that can disagree. controller: ControllerHandle, + /// The heart rate supervisor, for the same reason. + hr: HeartRateHandle, scan_rx: watch::Receiver, scan_on: watch::Sender, forgotten: HashSet, @@ -142,6 +151,7 @@ pub struct DeviceRegistry { /// The list published last tick, for change detection. published: Vec, last_trainer: TrainerStatus, + last_hr: HeartRateStatus, pub scanning: bool, /// Scanning was switched off by *us*, to get out of the way of a connect — /// not by the rider. Only a suspension is resumed automatically (FR-1.12). @@ -151,14 +161,16 @@ pub struct DeviceRegistry { } impl DeviceRegistry { - pub fn new(trainer: TrainerHandle, controller: ControllerHandle) -> Self { + pub fn new(trainer: TrainerHandle, controller: ControllerHandle, hr: HeartRateHandle) -> Self { let (scan_on, scan_on_rx) = watch::channel(false); let (scan_tx, scan_rx) = watch::channel(ScanSnapshot::default()); tauri::async_runtime::spawn(scan_loop(scan_on_rx, scan_tx)); Self { last_trainer: trainer.status(), + last_hr: hr.status(), trainer, controller, + hr, scan_rx, scan_on, forgotten: HashSet::new(), @@ -193,6 +205,14 @@ impl DeviceRegistry { let trainer_changed = (trainer != self.last_trainer).then(|| trainer.clone()); self.last_trainer = trainer.clone(); + let hr = self.hr.status(); + // Only *link* changes are reported upward: a bpm arriving once a second + // would otherwise raise a notice per heartbeat. The row still updates — + // `build` reads the full status every tick. + let hr_changed = (hr.state != self.last_hr.state || hr.error != self.last_hr.error) + .then(|| hr.clone()); + self.last_hr = hr; + // The scan is switched off for the duration of a connect so that it and // `find_peripheral` do not fight over the one adapter — and nothing else // ever turns it back on. A disconnect, or a connect that failed, would @@ -213,6 +233,7 @@ impl DeviceRegistry { changed, transitions, trainer_changed, + hr_changed, } } @@ -253,6 +274,7 @@ impl DeviceRegistry { services: d.services.iter().map(|u| describe_service(*u)).collect(), remembered: self.remembered.contains(&id), battery_pct: None, + heart_rate_bpm: None, error: None, id, }); @@ -276,6 +298,7 @@ impl DeviceRegistry { services: vec![describe_service(uuids::FITNESS_MACHINE_SERVICE)], remembered: true, battery_pct: None, + heart_rate_bpm: None, error: None, }); out.len() - 1 @@ -327,6 +350,7 @@ impl DeviceRegistry { services: vec![describe_service(ZWIFT_SERVICE)], remembered: true, battery_pct: None, + heart_rate_bpm: None, error: None, }); out.len() - 1 @@ -350,6 +374,53 @@ impl DeviceRegistry { }; } + // And the heart rate monitor, for the same reason again: a connected + // monitor stops advertising, and its row must not vanish the moment it + // starts working. A view of what the supervisor owns, never a second + // link. + let hr = self.hr.status(); + if let Some(address) = hr.address.clone() { + if !self.forgotten.contains(&address) { + let idx = match out.iter().position(|d| d.id == address) { + Some(i) => i, + None => { + out.push(DeviceInfo { + id: address.clone(), + name: hr + .name + .clone() + .unwrap_or_else(|| "Heart rate monitor".into()), + address, + rssi: 0, + kind: DeviceKind::HeartRate, + state: ConnectionState::Idle, + control_acquired: false, + services: vec![describe_service(HEART_RATE_SERVICE)], + remembered: true, + battery_pct: None, + heart_rate_bpm: None, + error: None, + }); + out.len() - 1 + } + }; + let device = &mut out[idx]; + device.kind = DeviceKind::HeartRate; + device.battery_pct = hr.battery_percent; + device.heart_rate_bpm = hr.bpm; + device.error = hr.error.clone(); + device.remembered = true; + if let Some(name) = &hr.name { + device.name = name.clone(); + } + // Idle after a deliberate disconnect keeps whatever the scan + // says about the row, exactly like an idle pod. + if hr.state != ConnectionState::Idle { + device.state = hr.state.clone(); + } + } + } + // Trainers first, then by signal strength: the thing the rider is // looking for should not be below an unnamed peripheral. out.sort_by(|a, b| { @@ -388,9 +459,24 @@ impl DeviceRegistry { return Ok(info); } + // A heart rate row connects the monitor it is, by address — routed to + // the heart rate supervisor so this list and whatever else drives the + // link stay one path. + if device.kind == DeviceKind::HeartRate { + self.remembered.insert(id.to_string()); + self.forgotten.remove(id); + self.hr.connect(Some(device.address.clone())); + let mut info = device; + info.state = ConnectionState::Connecting; + info.error = None; + info.remembered = true; + return Ok(info); + } + if device.kind != DeviceKind::Trainer { return Err(format!( - "{} is neither a trainer nor a Click pod — there is nothing to connect to.", + "{} is not a trainer, Click pod or heart rate monitor — there is nothing to \ + connect to.", device.name )); } @@ -446,6 +532,9 @@ impl DeviceRegistry { if let Some(pod) = device.kind.pod_id() { self.controller.disconnect(Some(pod)); } + if device.kind == DeviceKind::HeartRate { + self.hr.disconnect(); + } device.state = ConnectionState::Idle; device.control_acquired = false; Ok(device) @@ -458,6 +547,9 @@ impl DeviceRegistry { if device.kind == DeviceKind::Trainer && device.control_acquired { self.trainer.disconnect(); } + if device.kind == DeviceKind::HeartRate { + self.hr.disconnect(); + } self.remembered.remove(id); self.forgotten.insert(id.to_string()); self.published.retain(|d| d.id != id); @@ -638,6 +730,7 @@ mod tests { services: Vec::new(), remembered: false, battery_pct: None, + heart_rate_bpm: None, error: None, } } diff --git a/src-tauri/src/heart_rate.rs b/src-tauri/src/heart_rate.rs new file mode 100644 index 0000000..d0f3a33 --- /dev/null +++ b/src-tauri/src/heart_rate.rs @@ -0,0 +1,521 @@ +//! The heart rate supervisor: the app's single owner of a [`HeartRateClient`]. +//! +//! The same shape as [`crate::controller`], with one slot instead of two — the +//! Tauri commands hold a `Mutex` and may never `.await` a radio, so all BLE +//! work happens in one background task they reach over a channel. +//! +//! ```text +//! commands ──connect/disconnect──► [supervisor task] ──► HeartRateClient +//! ride loop ◄──watch>──────────┘ (bpm) +//! device loop ◄──watch───┘ +//! ``` +//! +//! The bpm channel follows the trainer's "never freeze on stale data" rule: +//! a monitor that goes quiet — strap off, watch out of broadcast mode — has +//! its published reading cleared rather than held, so the ride records *no* +//! heart rate rather than a plausible-looking flatline of the last real value. + +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::mpsc::{sync_channel, SyncSender}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use bikecontrol_ble::{ + Backoff, FtmsError, HeartRateClient, HeartRateConfig, HeartRateEvent, HrSelector, +}; +use bikecontrol_core::types::ConnectionState; +use serde::Serialize; +use tokio::sync::{broadcast, mpsc, oneshot, watch}; + +/// How long the monitor may stay silent before the published bpm is cleared. +/// +/// A strap notifies at about 1 Hz, so eight seconds is ~8 missed frames — +/// enough slack for a lossy link, short enough that a strap taken off does not +/// keep writing its last reading into the FIT file. +const STALE_AFTER: Duration = Duration::from_secs(8); +/// Housekeeping tick — staleness only, so it can be lazy. +const HOUSEKEEPING: Duration = Duration::from_secs(2); +/// Upper bound on closing the link at exit. There is no reset sequence, only +/// an unsubscribe and a disconnect (SAF-9, NFR-9). +const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(3); +/// How long auto-reconnect keeps chasing the monitor before giving up +/// (FR-1.11). The trainer's budget, not the Click's: a broadcasting device +/// advertises continuously, so "not found" here means it is genuinely gone. +const RECONNECT_ATTEMPTS: u32 = 20; + +/// The heart rate configuration this app rides with. +fn hr_config() -> HeartRateConfig { + HeartRateConfig { + backoff: Backoff { + max_attempts: Some(RECONNECT_ATTEMPTS), + ..Backoff::default() + }, + ..HeartRateConfig::default() + } +} + +/// What the UI needs to know about the heart rate link (FR-1.7, FR-9.2). +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct HeartRateStatus { + pub state: ConnectionState, + pub address: Option, + pub name: Option, + /// The live reading, also published on the bpm watch channel. Carried here + /// too so the connection screen can show proof the data is flowing — a + /// monitor that is connected but silent looks exactly like a working one + /// otherwise. + pub bpm: Option, + pub battery_percent: Option, + /// Human-readable failure, rendered verbatim (FR-9.2). + pub error: Option, +} + +impl Default for HeartRateStatus { + fn default() -> Self { + Self { + state: ConnectionState::Idle, + address: None, + name: None, + bpm: None, + battery_percent: None, + error: None, + } + } +} + +impl HeartRateStatus { + pub fn is_attached(&self) -> bool { + !matches!( + self.state, + ConnectionState::Idle | ConnectionState::Lost { .. } + ) + } +} + +enum Cmd { + /// Connect to a monitor. `address` when the scanner has already seen it, + /// which is both faster and unambiguous. + Connect { address: Option }, + Disconnect, + /// Close the link and stop, answering only once it is actually closed. + Shutdown { reply: SyncSender<()> }, +} + +/// The outcome of one connect attempt, back from its child task. +struct Attempt { + /// Which attempt this was, so a result superseded by a newer request (or a + /// disconnect) can be told apart from a live one. + generation: u64, + outcome: Result, FtmsError>, +} + +/// Cheap, cloneable handle to the supervisor. +#[derive(Clone)] +pub struct HeartRateHandle { + cmd_tx: mpsc::Sender, + status_rx: watch::Receiver, + bpm_rx: watch::Receiver>, + /// Shared, so every clone of the handle sees the link is already shut. + shut_down: Arc, +} + +impl HeartRateHandle { + /// Start the supervisor task. Needs a Tokio runtime, which + /// `tauri::async_runtime` provides before the app is built. + pub fn spawn() -> Self { + let (cmd_tx, cmd_rx) = mpsc::channel(16); + let (status_tx, status_rx) = watch::channel(HeartRateStatus::default()); + let (bpm_tx, bpm_rx) = watch::channel(None); + + let handle = Self { + cmd_tx, + status_rx, + bpm_rx, + shut_down: Arc::new(AtomicBool::new(false)), + }; + tauri::async_runtime::spawn(run(cmd_rx, status_tx, bpm_tx)); + handle + } + + pub fn status(&self) -> HeartRateStatus { + self.status_rx.borrow().clone() + } + + /// The live reading, for the ride engine. Cleared while disconnected or + /// stale, never a held-over value. + pub fn bpm(&self) -> watch::Receiver> { + self.bpm_rx.clone() + } + + pub fn connect(&self, address: Option) { + let _ = self.cmd_tx.try_send(Cmd::Connect { address }); + } + + pub fn disconnect(&self) { + let _ = self.cmd_tx.try_send(Cmd::Disconnect); + } + + /// Close the link at app exit, waiting for it to be closed (SAF-9). + /// Blocks the calling (non-async) thread until done or [`SHUTDOWN_TIMEOUT`] + /// elapses. Idempotent: one quit delivers several exit events, and the + /// repeats must be silent no-ops. + pub fn shutdown_blocking(&self) { + if self.shut_down.swap(true, Ordering::SeqCst) { + return; + } + let deadline = Instant::now() + SHUTDOWN_TIMEOUT; + let (reply, done) = sync_channel(1); + + let mut cmd = Cmd::Shutdown { reply }; + loop { + match self.cmd_tx.try_send(cmd) { + Ok(()) => break, + Err(mpsc::error::TrySendError::Closed(_)) => return, + Err(mpsc::error::TrySendError::Full(returned)) => { + if Instant::now() >= deadline { + tracing::warn!("hr supervisor unreachable; link left to the OS"); + return; + } + cmd = returned; + std::thread::sleep(Duration::from_millis(20)); + } + } + } + + let remaining = deadline.saturating_duration_since(Instant::now()); + match done.recv_timeout(remaining) { + Ok(()) => tracing::info!("heart rate monitor disconnected"), + Err(e) => tracing::warn!(error = %e, "heart rate monitor did not disconnect in time"), + } + } +} + +// --------------------------------------------------------------------------- +// Supervisor +// --------------------------------------------------------------------------- + +async fn run( + mut cmd_rx: mpsc::Receiver, + status_tx: watch::Sender, + bpm_tx: watch::Sender>, +) { + let mut client: Option = None; + let mut events: Option> = None; + // Held while a connect is in flight; sending on it — or dropping it — + // abandons the attempt (FR-1.10). + let mut cancel: Option> = None; + let mut generation: u64 = 0; + let mut last_measurement: Option = None; + + let (attempt_tx, mut attempt_rx) = mpsc::channel::(2); + let mut housekeeping = tokio::time::interval(HOUSEKEEPING); + + loop { + tokio::select! { + cmd = cmd_rx.recv() => { + let Some(cmd) = cmd else { + // Every handle dropped: nobody can ask for anything again. + shutdown(client.take(), &bpm_tx).await; + return; + }; + match cmd { + Cmd::Connect { address } => { + // A second click while a search is running means + // "connect", which is what is already happening — unless + // it names a different monitor, which takes over. + let current = status_tx.borrow().address.clone(); + let same = address.is_none() + || address.as_deref().is_some_and(|a| { + current.as_deref().is_some_and(|c| c.eq_ignore_ascii_case(a)) + }); + if same && (cancel.is_some() || client.is_some()) { + tracing::debug!("hr: already connected or connecting"); + continue; + } + // Switching monitors: close the old link first (SAF-9). + if let Some(c) = cancel.take() { + let _ = c.send(()); + } + generation += 1; + events = None; + last_measurement = None; + if let Some(c) = client.take() { + c.shutdown().await; + } + let _ = bpm_tx.send(None); + + let selector = match address.clone().or(current) { + Some(a) if !a.trim().is_empty() => HrSelector::Address(a), + _ => HrSelector::Any, + }; + tracing::info!(selector = %selector.describe(), "hr: connecting"); + cancel = Some(start_attempt(generation, selector, &attempt_tx)); + let _ = status_tx.send(HeartRateStatus { + state: ConnectionState::Connecting, + address, + ..HeartRateStatus::default() + }); + } + Cmd::Disconnect => { + if let Some(c) = cancel.take() { + let _ = c.send(()); + } + generation += 1; + events = None; + last_measurement = None; + if let Some(c) = client.take() { + c.shutdown().await; + } + let _ = bpm_tx.send(None); + // Keep who it was: a connected monitor stops appearing + // in scans, so wiping the address would leave the rider + // nothing to click on to reconnect (FR-1.12). + let address = status_tx.borrow().address.clone(); + let name = status_tx.borrow().name.clone(); + let _ = status_tx.send(HeartRateStatus { + address, + name, + ..HeartRateStatus::default() + }); + } + Cmd::Shutdown { reply } => { + if let Some(c) = cancel.take() { + let _ = c.send(()); + } + shutdown(client.take(), &bpm_tx).await; + let _ = status_tx.send(HeartRateStatus::default()); + // Answered only once the link is genuinely closed, so a + // caller blocking the app's exit on this knows what it + // waited for (SAF-9). + let _ = reply.send(()); + return; + } + } + } + + Some(attempt) = attempt_rx.recv() => { + if attempt.generation != generation { + // Superseded — by a disconnect or a newer request. A client + // that arrived anyway owns a live GATT link, and dropping it + // would leave the monitor held (SAF-9). + if let Ok(Some(c)) = attempt.outcome { + tracing::debug!("hr: stale connect; closing"); + tokio::spawn(async move { c.shutdown().await }); + } + continue; + } + cancel = None; + match attempt.outcome { + Ok(Some(c)) => { + tracing::info!( + address = c.address(), + name = c.name().unwrap_or("(no name)"), + "hr: connected" + ); + events = Some(c.events()); + last_measurement = Some(Instant::now()); + let _ = status_tx.send(HeartRateStatus { + state: ConnectionState::Connected, + address: Some(c.address().to_string()), + name: c.name().map(str::to_owned), + battery_percent: c.battery_percent(), + ..HeartRateStatus::default() + }); + client = Some(c); + } + Ok(None) => { + tracing::info!("hr: connect abandoned"); + status_tx.send_modify(|s| s.state = ConnectionState::Idle); + } + Err(e) => { + tracing::warn!(error = %e, "hr: connect failed"); + status_tx.send_modify(|s| { + s.state = ConnectionState::Idle; + s.error = Some(connect_hint(&e)); + }); + } + } + } + + // Only polled while there is a stream; `recv` on a `None` receiver + // would busy-loop, so the branch is disabled instead. + event = async { events.as_mut().unwrap().recv().await }, if events.is_some() => { + match event { + Ok(HeartRateEvent::Measurement { bpm, sensor_contact }) => { + last_measurement = Some(Instant::now()); + // `Some(false)` is the strap saying it has fallen off; + // its reading is noise and recording it would be worse + // than recording nothing. + let reading = (sensor_contact != Some(false)) + .then_some(bpm.min(u8::MAX as u16) as u8); + let _ = bpm_tx.send(reading); + status_tx.send_modify(|s| { + // The first frame after a dropout is the only notice + // that the link is back. + if s.state == ConnectionState::Reconnecting { + s.state = ConnectionState::Connected; + s.error = None; + } + if s.bpm != reading { + s.bpm = reading; + } + }); + } + Ok(HeartRateEvent::Connected { address, name, battery_percent }) => { + // A reconnect inside the client actor. + last_measurement = Some(Instant::now()); + status_tx.send_modify(|s| { + s.state = ConnectionState::Connected; + s.error = None; + if !address.is_empty() { + s.address = Some(address.clone()); + } + s.name = name.clone(); + s.battery_percent = battery_percent; + }); + } + Ok(HeartRateEvent::Disconnected) => { + let _ = bpm_tx.send(None); + status_tx.send_modify(|s| { + // Only a live link "reconnects"; a teardown on the + // way to Idle must not flip the state back. + if s.state == ConnectionState::Connected { + s.state = ConnectionState::Reconnecting; + } + s.bpm = None; + }); + } + Ok(HeartRateEvent::GaveUp { attempts }) => { + // Terminal: the actor has stopped (FR-1.11). + tracing::warn!(attempts, "hr: reconnect gave up"); + events = None; + client = None; + last_measurement = None; + let _ = bpm_tx.send(None); + status_tx.send_modify(|s| { + s.state = ConnectionState::Lost { reason: "stopped answering".into() }; + s.bpm = None; + s.error = Some(format!( + "Gave up reaching the heart rate monitor after {attempts} \ + attempts. Check it is on — a watch must have Broadcast Heart \ + Rate enabled — then connect again." + )); + }); + } + Err(broadcast::error::RecvError::Lagged(n)) => { + tracing::debug!(skipped = n, "hr: event consumer lagged"); + } + Err(broadcast::error::RecvError::Closed) => { + // The actor is gone without a GaveUp — should not + // happen, but a closed stream must not spin the select. + events = None; + } + } + } + + _ = housekeeping.tick() => { + let quiet = client.is_some() + && last_measurement.is_some_and(|t| t.elapsed() >= STALE_AFTER); + if quiet && bpm_tx.borrow().is_some() { + // Clear rather than hold: a strap taken off must not keep + // writing its last reading into the FIT file (FR-1.8). + tracing::info!("hr: no measurement for {STALE_AFTER:?} — clearing the reading"); + let _ = bpm_tx.send(None); + status_tx.send_modify(|s| s.bpm = None); + } + } + } + } +} + +/// Spawn the connect in its own task, so a fifteen-second search cannot hold +/// up a disconnect or a quit (FR-1.10). +fn start_attempt( + generation: u64, + selector: HrSelector, + results: &mpsc::Sender, +) -> oneshot::Sender<()> { + let (cancel_tx, cancel_rx) = oneshot::channel(); + let results = results.clone(); + tauri::async_runtime::spawn(async move { + let outcome = HeartRateClient::connect_cancellable(selector, hr_config(), async { + // Resolves on an explicit cancel *and* on the sender being dropped. + let _ = cancel_rx.await; + }) + .await; + let _ = results.send(Attempt { generation, outcome }).await; + }); + cancel_tx +} + +/// Close the link and clear the published reading. +async fn shutdown(client: Option, bpm_tx: &watch::Sender>) { + let _ = bpm_tx.send(None); + if let Some(c) = client { + c.shutdown().await; + } +} + +/// Turn a connect failure into something the rider can act on (FR-9.2). +fn connect_hint(e: &FtmsError) -> String { + match e { + FtmsError::NotFound(_) => "Could not find the heart rate monitor. A strap needs to be \ + worn to advertise; a watch needs Broadcast Heart Rate \ + switched on. Then try again." + .to_string(), + FtmsError::NoAdapter => { + "No Bluetooth adapter. Check the radio is on and BlueZ is running.".to_string() + } + // A watch in broadcast mode accepts one central; a second connect gets + // torn down mid-handshake and surfaces as a bare BLE error. + FtmsError::Bluetooth(_) | FtmsError::MissingCharacteristic(_) => format!( + "Heart rate monitor is busy: {e}. It accepts one connection at a time — close any \ + other app that is holding it, then try again." + ), + other => other.to_string(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn default_status_is_idle_and_empty() { + let s = HeartRateStatus::default(); + assert_eq!(s.state, ConnectionState::Idle); + assert!(!s.is_attached()); + assert_eq!(s.bpm, None); + } + + #[test] + fn attached_covers_every_live_state() { + for state in [ + ConnectionState::Connecting, + ConnectionState::Connected, + ConnectionState::Reconnecting, + ] { + let s = HeartRateStatus { state: state.clone(), ..HeartRateStatus::default() }; + assert!(s.is_attached(), "{state:?}"); + } + let lost = HeartRateStatus { + state: ConnectionState::Lost { reason: "gone".into() }, + ..HeartRateStatus::default() + }; + assert!(!lost.is_attached()); + } + + #[test] + fn reconnect_is_bounded_so_the_rider_is_eventually_told() { + // FR-1.11 — same reasoning as the trainer's budget. + let cfg = hr_config(); + assert_eq!(cfg.backoff.max_attempts, Some(RECONNECT_ATTEMPTS)); + } + + #[test] + fn a_missing_monitor_is_explained_not_just_reported() { + let hint = connect_hint(&FtmsError::NotFound("any heart rate monitor".into())); + assert!(hint.to_lowercase().contains("broadcast"), "FR-1.8: {hint}"); + } +} diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index dc1e79c..6410eab 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -13,6 +13,7 @@ pub mod controller; pub mod derive; pub mod devices; pub mod events; +pub mod heart_rate; pub mod profile_view; pub mod recording; pub mod samples; diff --git a/src-tauri/src/session_backend.rs b/src-tauri/src/session_backend.rs index a163944..94606e7 100644 --- a/src-tauri/src/session_backend.rs +++ b/src-tauri/src/session_backend.rs @@ -27,6 +27,12 @@ pub struct SessionBackend { session: RideSession, /// Latest decoded Indoor Bike Data, published by [`crate::trainer`]. telemetry: watch::Receiver, + /// Latest reading from a dedicated heart rate monitor, published by + /// [`crate::heart_rate`]. Stamped onto the trainer's telemetry each tick — + /// but never over a heart rate FTMS itself reported, on the same principle + /// as the Zwift cadence merge: a trainer that declares a field is the + /// better authority on it. + heart_rate: watch::Receiver>, last_snapshot: Option, /// Set once a profile has been handed to the session, so a profile swap is /// noticed but the same profile is not reloaded every tick. @@ -36,12 +42,17 @@ pub struct SessionBackend { } impl SessionBackend { - pub fn new(inputs: &RideInputs, telemetry: watch::Receiver) -> Self { + pub fn new( + inputs: &RideInputs, + telemetry: watch::Receiver, + heart_rate: watch::Receiver>, + ) -> Self { let mut session = RideSession::new(inputs.rider, inputs.limits); session.gearing = Gearing::new(inputs.cassette.clone()); Self { session, telemetry, + heart_rate, last_snapshot: None, loaded_profile: None, loaded_cassette: inputs.cassette.clone(), @@ -111,7 +122,10 @@ impl RideBackend for SessionBackend { _ => {} } - let telemetry = *self.telemetry.borrow(); + let mut telemetry = *self.telemetry.borrow(); + if telemetry.heart_rate_bpm.is_none() { + telemetry.heart_rate_bpm = *self.heart_rate.borrow(); + } let mut command = None; let mut snapshot = None; for event in self.session.tick(telemetry, dt_s) { @@ -140,6 +154,12 @@ impl RideBackend for SessionBackend { } } +/// A heart rate channel with nothing on it, for tests about everything else. +#[cfg(test)] +fn no_hr() -> watch::Receiver> { + watch::channel(None).1 +} + #[cfg(test)] mod tests { use super::*; @@ -157,7 +177,7 @@ mod tests { fn real_power_drives_the_ride_forward() { let (tx, rx) = watch::channel(Telemetry::default()); let inputs = running(ControlMode::ManualGrade); - let mut backend = SessionBackend::new(&inputs, rx); + let mut backend = SessionBackend::new(&inputs, rx, no_hr()); // No power: nothing moves. for _ in 0..8 { @@ -190,7 +210,7 @@ mod tests { ..Telemetry::default() }); let inputs = running(ControlMode::ManualGrade); - let mut backend = SessionBackend::new(&inputs, rx); + let mut backend = SessionBackend::new(&inputs, rx, no_hr()); for _ in 0..60 { backend.tick(0.25, &inputs); } @@ -218,7 +238,7 @@ mod tests { // same load in watts and is covered in the engine's own tests. inputs.rider.load_channel = bikecontrol_core::types::LoadChannel::Gradient; inputs.manual_gradient_pct = 5.0; - let mut backend = SessionBackend::new(&inputs, rx); + let mut backend = SessionBackend::new(&inputs, rx, no_hr()); let tick = backend.tick(0.25, &inputs); // The road is the rider's setting exactly; what the trainer is asked // for is the load that road implies through the selected gear, which is @@ -240,7 +260,7 @@ mod tests { let mut inputs = running(ControlMode::ManualGrade); inputs.manual_gradient_pct = 3.0; inputs.gradient_offset_pct = 1.5; - let mut backend = SessionBackend::new(&inputs, rx); + let mut backend = SessionBackend::new(&inputs, rx, no_hr()); for _ in 0..10 { backend.tick(0.25, &inputs); } @@ -253,7 +273,7 @@ mod tests { let (_tx, rx) = watch::channel(Telemetry::default()); let mut inputs = running(ControlMode::Erg); inputs.power_target_w = 275; - let mut backend = SessionBackend::new(&inputs, rx); + let mut backend = SessionBackend::new(&inputs, rx, no_hr()); assert_eq!( backend.tick(0.25, &inputs).command, Some(ControlTarget::Power { watts: 275 }) @@ -274,7 +294,7 @@ mod tests { let mut inputs = running(ControlMode::ManualGrade); inputs.rider.load_channel = bikecontrol_core::types::LoadChannel::Gradient; inputs.manual_gradient_pct = 400.0; - let mut backend = SessionBackend::new(&inputs, rx); + let mut backend = SessionBackend::new(&inputs, rx, no_hr()); let tick = backend.tick(0.25, &inputs); assert_eq!( tick.command, @@ -292,7 +312,7 @@ mod tests { ..Telemetry::default() }); let mut inputs = running(ControlMode::ManualGrade); - let mut backend = SessionBackend::new(&inputs, rx); + let mut backend = SessionBackend::new(&inputs, rx, no_hr()); backend.tick(0.25, &inputs); inputs.status = RideStatus::Paused; inputs.manual_gradient_pct = 9.0; @@ -305,7 +325,7 @@ mod tests { // last value, so a stationary rider was shown 38 km/h indefinitely. let (tx, rx) = watch::channel(Telemetry::default()); let mut inputs = running(ControlMode::ManualGrade); - let mut backend = SessionBackend::new(&inputs, rx); + let mut backend = SessionBackend::new(&inputs, rx, no_hr()); let _ = tx.send(Telemetry { power_w: Some(250), cadence_rpm: Some(85.0), @@ -334,11 +354,43 @@ mod tests { assert!(resumed.virtual_speed_kph > 10.0, "{resumed:?}"); } + #[test] + fn a_monitors_heart_rate_reaches_the_snapshot_but_never_overrides_ftms() { + let (telemetry_tx, telemetry_rx) = watch::channel(Telemetry::default()); + let (hr_tx, hr_rx) = watch::channel(None); + let inputs = running(ControlMode::ManualGrade); + let mut backend = SessionBackend::new(&inputs, telemetry_rx, hr_rx); + + // No monitor, no trainer HR: the field stays absent, so a ride without + // a strap carries no heart rate rather than a zero (see fit::builder). + let snap = backend.tick(0.25, &inputs).snapshot; + assert_eq!(snap.telemetry.heart_rate_bpm, None); + + // The monitor's reading fills the gap FTMS leaves. + let _ = hr_tx.send(Some(147)); + let snap = backend.tick(0.25, &inputs).snapshot; + assert_eq!(snap.telemetry.heart_rate_bpm, Some(147)); + + // A trainer that reports its own heart rate is the better authority. + let _ = telemetry_tx.send(Telemetry { + heart_rate_bpm: Some(151), + ..Telemetry::default() + }); + let snap = backend.tick(0.25, &inputs).snapshot; + assert_eq!(snap.telemetry.heart_rate_bpm, Some(151)); + + // The monitor going quiet clears the reading, not freezes it. + let _ = telemetry_tx.send(Telemetry::default()); + let _ = hr_tx.send(None); + let snap = backend.tick(0.25, &inputs).snapshot; + assert_eq!(snap.telemetry.heart_rate_bpm, None); + } + #[test] fn reset_returns_to_the_start_line() { let (tx, rx) = watch::channel(Telemetry::default()); let inputs = running(ControlMode::ManualGrade); - let mut backend = SessionBackend::new(&inputs, rx); + let mut backend = SessionBackend::new(&inputs, rx, no_hr()); let _ = tx.send(Telemetry { power_w: Some(300), cadence_rpm: Some(90.0), @@ -392,7 +444,7 @@ mod drag_race_tests { ..RideInputs::default() }; inputs.set_gear(3); - let mut backend = SessionBackend::new(&inputs, rx); + let mut backend = SessionBackend::new(&inputs, rx, no_hr()); // Rolling: 180 W at 22 km/h on the flywheel. let _ = tx.send(d100(180, 22.0)); @@ -420,7 +472,7 @@ mod drag_race_tests { mode: ControlMode::Profile, ..RideInputs::default() }; - let mut backend = SessionBackend::new(&inputs, rx); + let mut backend = SessionBackend::new(&inputs, rx, no_hr()); let _ = tx.send(d100(180, 22.0)); for _ in 0..40 { backend.tick(0.25, &inputs); @@ -442,7 +494,7 @@ mod drag_race_tests { ..RideInputs::default() }; inputs.set_gear(gear); - let mut backend = SessionBackend::new(&inputs, rx); + let mut backend = SessionBackend::new(&inputs, rx, no_hr()); let _ = tx.send(d100(180, 22.0)); for _ in 0..60 { backend.tick(0.25, &inputs); diff --git a/src-tauri/src/state.rs b/src-tauri/src/state.rs index 3192c04..4ccf3ca 100644 --- a/src-tauri/src/state.rs +++ b/src-tauri/src/state.rs @@ -19,6 +19,7 @@ use crate::events; use crate::events::{ ConnectionEvent, DeviceList, InputAck, LapSummary, Notice, RideState, RideStatus, }; +use crate::heart_rate::{HeartRateHandle, HeartRateStatus}; use crate::profile_view::{ProfileGeometry, ProfileView}; use crate::recording::{RecorderHandle, Recovered, RideSummary}; use crate::session_backend::SessionBackend; @@ -37,6 +38,7 @@ pub struct Inner { pub devices: DeviceRegistry, pub trainer: TrainerHandle, pub controller: crate::controller::ControllerHandle, + pub heart_rate: HeartRateHandle, pub profile_view: Option, /// Precomputed route geometry, kept Rust-side so the per-tick elevation and /// ascent-remaining lookups are a binary search rather than a scan. @@ -66,20 +68,33 @@ pub struct Inner { /// pedalling, which is the truth. There is deliberately no synthetic rider to /// fall back to — a session a rider could finish and only then discover none of /// it happened is worse than no session at all. -fn build_backend(inputs: &RideInputs, trainer: &TrainerHandle) -> Box { +fn build_backend( + inputs: &RideInputs, + trainer: &TrainerHandle, + heart_rate: &HeartRateHandle, +) -> Box { tracing::info!("ride data source is the trainer; with no trainer attached the ride reads zero"); - Box::new(SessionBackend::new(inputs, trainer.telemetry())) + Box::new(SessionBackend::new( + inputs, + trainer.telemetry(), + heart_rate.bpm(), + )) } impl Inner { - fn new(trainer: TrainerHandle, controller: crate::controller::ControllerHandle) -> Self { + fn new( + trainer: TrainerHandle, + controller: crate::controller::ControllerHandle, + heart_rate: HeartRateHandle, + ) -> Self { let inputs = RideInputs::default(); Self { - backend: build_backend(&inputs, &trainer), - devices: DeviceRegistry::new(trainer.clone(), controller.clone()), + backend: build_backend(&inputs, &trainer, &heart_rate), + devices: DeviceRegistry::new(trainer.clone(), controller.clone(), heart_rate.clone()), inputs, trainer, controller, + heart_rate, profile_view: None, geometry: None, deriver: Deriver::default(), @@ -211,8 +226,9 @@ impl AppState { let limits = RideInputs::default().limits; let trainer = TrainerHandle::spawn(crate::trainer::app_config(limits)); let controller = crate::controller::ControllerHandle::spawn(); + let heart_rate = HeartRateHandle::spawn(); Self { - inner: Arc::new(Mutex::new(Inner::new(trainer, controller))), + inner: Arc::new(Mutex::new(Inner::new(trainer, controller, heart_rate))), recorder: RecorderHandle::default(), } } @@ -233,6 +249,11 @@ impl AppState { self.lock().controller.clone() } + /// The heart rate supervisor handle. + pub fn heart_rate(&self) -> HeartRateHandle { + self.lock().heart_rate.clone() + } + /// Panics are impossible to recover from here, and a poisoned lock means /// the ride loop already died — surface it rather than hide it. pub fn lock(&self) -> std::sync::MutexGuard<'_, Inner> { @@ -499,9 +520,11 @@ pub fn shutdown_devices(app: &AppHandle) { let Some(state) = app.try_state::() else { return; }; - let (trainer, controller) = (state.trainer(), state.controller()); + let (trainer, controller, heart_rate) = + (state.trainer(), state.controller(), state.heart_rate()); trainer.shutdown_blocking(); controller.shutdown_blocking(); + heart_rate.shutdown_blocking(); } /// The scan loop: refreshes the device list from the radio and pushes it when @@ -543,6 +566,11 @@ pub fn spawn_device_loop(app: AppHandle) { } emit_ride_state(&app); } + if let Some(status) = result.hr_changed { + if let Some(notice) = hr_notice(&status) { + notify(&app, notice); + } + } if result.changed { emit_devices(&app); } @@ -590,3 +618,31 @@ fn trainer_notice(status: &TrainerStatus) -> Option { ConnectionState::Idle | ConnectionState::Scanning => None, } } + +/// The same, for the heart rate link (FR-1.8, FR-9.4). Quieter than the +/// trainer's: heart rate is garnish on the ride, not the ride, so only the +/// transitions the rider should act on speak. +fn hr_notice(status: &HeartRateStatus) -> Option { + let name = status + .name + .clone() + .unwrap_or_else(|| "Heart rate monitor".into()); + match &status.state { + ConnectionState::Connected => Some(Notice::info(format!("{name} connected"))), + ConnectionState::Reconnecting => Some(Notice::warn(format!( + "Lost {name} — reconnecting. The ride continues; heart rate pauses until it is back." + ))), + ConnectionState::Lost { reason } => Some(Notice::error( + status + .error + .clone() + .unwrap_or_else(|| format!("{name} unavailable: {reason}")), + )), + // Idle can also carry a failed connect's advice (the supervisor parks + // there rather than in Lost, so the row stays clickable). + ConnectionState::Idle => status.error.clone().map(Notice::warn), + ConnectionState::Connecting + | ConnectionState::Scanning + | ConnectionState::Controlling => None, + } +} diff --git a/ui/src/components/ConnectionScreen.svelte b/ui/src/components/ConnectionScreen.svelte index 01e28e5..f0673ca 100644 --- a/ui/src/components/ConnectionScreen.svelte +++ b/ui/src/components/ConnectionScreen.svelte @@ -140,6 +140,28 @@ : ''} + {:else if device.kind === 'heartRate'} + + + Heart rate + + {device.heartRateBpm != null + ? `${device.heartRateBpm} bpm` + : '—'} + + + {#if device.batteryPct != null} + + Battery + {device.batteryPct}% + + {/if} {:else if device.batteryPct != null} Battery @@ -173,7 +195,10 @@
  • Trainer — turn the pedals for a few seconds.
  • Zwift Click — press any button on the pod.
    • -
  • Heart rate strap — wet the contacts and put it on.
    • +
  • + Heart rate — wet a strap's contacts and put it on; on a sports watch, + switch on Broadcast Heart Rate. +

They will appear here as soon as they advertise. Scanning continues in the background. diff --git a/ui/src/lib/types.ts b/ui/src/lib/types.ts index 345ab45..29d08d4 100644 --- a/ui/src/lib/types.ts +++ b/ui/src/lib/types.ts @@ -252,6 +252,8 @@ export interface DeviceInfo { services: string[]; remembered: boolean; batteryPct: number | null; + /** Live reading from a connected heart rate monitor — proof data is flowing. */ + heartRateBpm: number | null; error: string | null; }