From 7cd6674bdd118615944df0d623d43e9385de40b9 Mon Sep 17 00:00:00 2001 From: lambda Date: Mon, 26 Oct 2026 13:42:23 +0300 Subject: [PATCH] fix: FTMS indoor bike data, control confirmation, watchdog reconnects --- src-tauri/Cargo.lock | 1 + src-tauri/Cargo.toml | 3 +- src-tauri/src/ble/error.rs | 10 +++ src-tauri/src/ble/mock.rs | 4 + src-tauri/src/ble/source.rs | 115 +++++++++++++++++++++++++--- src-tauri/src/commands.rs | 46 +++++++++++ src-tauri/src/lib.rs | 147 +++++++++++++++++++++++------------- src-tauri/src/state.rs | 9 ++- src-tauri/src/telemetry.rs | 53 +++++++++++++ 9 files changed, 317 insertions(+), 71 deletions(-) diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index 6e65e62..f5b0806 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -3200,6 +3200,7 @@ dependencies = [ "tauri-plugin-opener", "thiserror 1.0.69", "tokio", + "tokio-stream", "uuid", ] diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index 1769f0d..3de69f7 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -23,7 +23,8 @@ tauri-plugin-opener = "2" serde = { version = "1", features = ["derive"] } serde_json = "1" btleplug = "0.11.8" -tokio = { version = "1", features = ["rt-multi-thread", "macros", "time"] } +tokio = { version = "1", features = ["rt-multi-thread", "macros", "time", "sync"] } +tokio-stream = "0.1" hex = "0.4" uuid = "1.16.0" thiserror = "1" diff --git a/src-tauri/src/ble/error.rs b/src-tauri/src/ble/error.rs index 76210bd..44227a3 100644 --- a/src-tauri/src/ble/error.rs +++ b/src-tauri/src/ble/error.rs @@ -1,5 +1,7 @@ use thiserror::Error; +use crate::commands::ControlResult; + /// Ошибки BLE-слоя. В отличие от прежней заглушки сохраняют источник /// (`btleplug::Error`), чтобы при отладке было видно, что именно пошло не так. #[derive(Debug, Error)] @@ -39,4 +41,12 @@ pub enum BleError { /// Не удалось прочитать характеристику. #[error("не удалось прочитать характеристику: {0}")] ReadFailed(btleplug::Error), + + /// Control Point Response вернулся с ошибкой (не Success). + #[error("станок отклонил управление: {0:?}")] + ControlRejected(ControlResult), + + /// Не дождались подтверждения управления за таймаут. + #[error("таймаут подтверждения управления станком")] + ControlTimeout, } diff --git a/src-tauri/src/ble/mock.rs b/src-tauri/src/ble/mock.rs index f13803b..685b087 100644 --- a/src-tauri/src/ble/mock.rs +++ b/src-tauri/src/ble/mock.rs @@ -102,6 +102,10 @@ impl SensorSource for MockSource { Ok(()) } + async fn request_control(&self, _device: &Device) -> Result<(), BleError> { + Ok(()) + } + async fn battery(&self, device: &Device) -> Result { let percent = match device.id.as_str() { "mock-hr" => 15, diff --git a/src-tauri/src/ble/source.rs b/src-tauri/src/ble/source.rs index 53da46f..06a1704 100644 --- a/src-tauri/src/ble/source.rs +++ b/src-tauri/src/ble/source.rs @@ -1,15 +1,17 @@ use std::pin::Pin; +use std::sync::{Arc, Mutex}; +use std::time::Duration; use async_trait::async_trait; use btleplug::api::{bleuuid::uuid_from_u16, Central, Peripheral as _, ValueNotification, WriteType}; use btleplug::platform::{Adapter, Peripheral}; -use futures::{future, Stream, StreamExt}; +use futures::{Stream, StreamExt}; use uuid::Uuid; use crate::ble::BleError; -use crate::commands::{encode_command, Command}; +use crate::commands::{encode_command, parse_control_result, Command, ControlPoint, ControlResult}; use crate::device::Device; -use crate::telemetry::{parse_csc, parse_hr, parse_power, CscAccumulator, Telemetry}; +use crate::telemetry::{parse_csc, parse_hr, parse_indoor_bike_data, parse_power, CscAccumulator, Telemetry}; #[async_trait] pub(crate) trait SensorSource: Send + Sync { @@ -22,6 +24,9 @@ pub(crate) trait SensorSource: Send + Sync { /// Отправить команду на управляющую точку устройства (ERG и т.п.). async fn command(&self, device: &Device, cmd: Command) -> Result<(), BleError>; + /// Запросить контроль у FTMS-станка и дождаться подтверждения (0x80). + async fn request_control(&self, device: &Device) -> Result<(), BleError>; + /// Прочитать уровень заряда батареи (0..=100). Best-effort. async fn battery(&self, device: &Device) -> Result; } @@ -29,15 +34,26 @@ pub(crate) trait SensorSource: Send + Sync { const HR_MEASUREMENT: Uuid = uuid_from_u16(0x2A37); const CSC_MEASUREMENT: Uuid = uuid_from_u16(0x2A5B); const POWER_MEASUREMENT: Uuid = uuid_from_u16(0x2A63); +const INDOOR_BIKE_DATA: Uuid = uuid_from_u16(0x2AD2); +const FITNESS_MACHINE_STATUS: Uuid = uuid_from_u16(0x2ADA); + +// How long to wait for the trainer to confirm control (Control Point Response 0x80). +const CONTROL_CONFIRM_TIMEOUT: Duration = Duration::from_secs(3); +const CONTROL_POLL_INTERVAL: Duration = Duration::from_millis(50); /// Реальный источник данных через btleplug. pub(crate) struct BleSource { adapter: Adapter, + /// Last control result (set by the notification router, awaited by `request_control`). + control: Arc>>, } impl BleSource { pub(crate) fn new(adapter: Adapter) -> Self { - Self { adapter } + Self { + adapter, + control: Arc::new(Mutex::new(None)), + } } async fn find_peripheral(&self, device: &Device) -> Result { @@ -65,9 +81,17 @@ impl SensorSource for BleSource { let peripheral = self.find_peripheral(device).await?; ensure_connected(&peripheral).await?; - let targets = [HR_MEASUREMENT, CSC_MEASUREMENT, POWER_MEASUREMENT]; + let targets = [ + HR_MEASUREMENT, + CSC_MEASUREMENT, + POWER_MEASUREMENT, + INDOOR_BIKE_DATA, + FITNESS_MACHINE_STATUS, + ]; for ch in peripheral.characteristics().iter() { if targets.contains(&ch.uuid) { + // Unsubscribe first (best-effort) so a stale reconnect does not double-subscribe. + let _ = peripheral.unsubscribe(ch).await; peripheral.subscribe(ch).await.map_err(BleError::SubscribeFailed)?; } } @@ -77,9 +101,25 @@ impl SensorSource for BleSource { .await .map_err(BleError::NotificationsFailed)?; - let mut acc = CscAccumulator::new(); - let stream = notifications.filter_map(move |n| future::ready(parse_notification(n, &mut acc))); - Ok(Box::pin(stream)) + // Route notifications on a background task: telemetry -> mpsc, control result -> shared state. + // This keeps the control-result flow concurrent with the telemetry consumer so that + // `request_control` can await confirmation while telemetry is still being consumed. + let (tx, rx) = tokio::sync::mpsc::unbounded_channel::(); + let control = self.control.clone(); + + tokio::spawn(async move { + let mut notifications = notifications; + let mut acc = CscAccumulator::new(); + while let Some(n) = notifications.next().await { + for t in parse_notification(n, &mut acc, &control) { + let _ = tx.send(t); + } + } + }); + + let stream: Pin + Send>> = + Box::pin(tokio_stream::wrappers::UnboundedReceiverStream::new(rx)); + Ok(stream) } async fn command(&self, device: &Device, cmd: Command) -> Result<(), BleError> { @@ -100,6 +140,44 @@ impl SensorSource for BleSource { .map_err(BleError::WriteFailed) } + async fn request_control(&self, device: &Device) -> Result<(), BleError> { + let peripheral = self.find_peripheral(device).await?; + ensure_connected(&peripheral).await?; + + let cp_uuid = uuid_from_u16(ControlPoint::Ftms.short_uuid()); + let ch = peripheral + .characteristics() + .into_iter() + .find(|c| c.uuid == cp_uuid) + .ok_or_else(|| BleError::DeviceNotFound(cp_uuid.to_string()))?; + + // Reset and send Request Control (opcode 0x00). + *self.control.lock().unwrap() = None; + let (_, payload) = encode_command(Command::FtmsRequestControl); + peripheral + .write(&ch, &payload, WriteType::WithoutResponse) + .await + .map_err(BleError::WriteFailed)?; + + // Wait for the Control Point Response (routed by the stream's notification task). + let result = tokio::time::timeout(CONTROL_CONFIRM_TIMEOUT, async { + loop { + let current = *self.control.lock().unwrap(); + if let Some(r) = current { + return r; + } + tokio::time::sleep(CONTROL_POLL_INTERVAL).await; + } + }) + .await + .map_err(|_| BleError::ControlTimeout)?; + + match result { + ControlResult::Success => Ok(()), + other => Err(BleError::ControlRejected(other)), + } + } + async fn battery(&self, device: &Device) -> Result { let peripheral = self.find_peripheral(device).await?; ensure_connected(&peripheral).await?; @@ -131,16 +209,29 @@ async fn ensure_connected(peripheral: &Peripheral) -> Result<(), BleError> { Ok(()) } -fn parse_notification(n: ValueNotification, acc: &mut CscAccumulator) -> Option { +fn parse_notification( + n: ValueNotification, + acc: &mut CscAccumulator, + control: &Arc>>, +) -> Vec { if n.uuid == HR_MEASUREMENT { - parse_hr(&n.value) + parse_hr(&n.value).into_iter().collect() } else if n.uuid == CSC_MEASUREMENT { parse_csc(&n.value) .and_then(|raw| acc.push(raw)) .map(|rpm| Telemetry::Cadence { rpm }) + .into_iter() + .collect() } else if n.uuid == POWER_MEASUREMENT { - parse_power(&n.value) + parse_power(&n.value).into_iter().collect() + } else if n.uuid == INDOOR_BIKE_DATA { + parse_indoor_bike_data(&n.value) + } else if n.uuid == FITNESS_MACHINE_STATUS { + if let Some(result) = parse_control_result(&n.value) { + *control.lock().unwrap() = Some(result); + } + Vec::new() } else { - None + Vec::new() } } diff --git a/src-tauri/src/commands.rs b/src-tauri/src/commands.rs index 5787a15..af2bf08 100644 --- a/src-tauri/src/commands.rs +++ b/src-tauri/src/commands.rs @@ -62,10 +62,56 @@ pub(crate) fn encode_command(cmd: Command) -> (ControlPoint, Vec) { } } +/// Result of a Control Point Response (Fitness Machine Status opcode 0x80). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum ControlResult { + Success, + OpCodeNotSupported, + InvalidParameter, + OperationFailed, + ControlNotPermitted, +} + +/// Parses a Fitness Machine Status (0x2ADA) notification for a Control Point +/// Response. Layout: [opcode=0x80][request opcode][result code]. +pub(crate) fn parse_control_result(bytes: &[u8]) -> Option { + if bytes.first() != Some(&0x80) { + return None; + } + match bytes.get(2) { + Some(0x01) => Some(ControlResult::Success), + Some(0x02) => Some(ControlResult::OpCodeNotSupported), + Some(0x03) => Some(ControlResult::InvalidParameter), + Some(0x04) => Some(ControlResult::OperationFailed), + Some(0x05) => Some(ControlResult::ControlNotPermitted), + _ => None, + } +} + #[cfg(test)] mod tests { use super::*; + #[test] + fn test_parse_control_result_success() { + assert_eq!(parse_control_result(&[0x80, 0x00, 0x01]), Some(ControlResult::Success)); + } + + #[test] + fn test_parse_control_result_rejected() { + assert_eq!( + parse_control_result(&[0x80, 0x00, 0x05]), + Some(ControlResult::ControlNotPermitted) + ); + } + + #[test] + fn test_parse_control_result_not_a_response() { + // opcode != 0x80 (e.g. Fitness Machine Stopped) + assert_eq!(parse_control_result(&[0x02]), None); + assert_eq!(parse_control_result(&[]), None); + } + #[test] fn test_encode_request_control() { assert_eq!( diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 33f7a7d..1da84fd 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -18,7 +18,7 @@ use crate::ble::{BleSource, MockSource, SensorSource}; use crate::commands::Command; use crate::device::Device; use crate::device_repo::{repo_from_devices, Repo}; -use crate::state::{State, TrainerTargets}; +use crate::state::{State, TrainerState}; use crate::telemetry::Telemetry; use btleplug::api::Manager as ApiManager; use btleplug::platform::Manager as BleManager; @@ -40,7 +40,7 @@ struct BatteryEvent { // Watchdog: how long without telemetry before the connection is considered dead. const WATCHDOG_STALE_TIMEOUT: Duration = Duration::from_secs(5); -// How many times to retry reconnect before giving up. +// How many consecutive failed/silent cycles before giving up on reconnect. const RECONNECT_MAX_ATTEMPTS: u32 = 5; // Base backoff between reconnect attempts. const RECONNECT_BACKOFF: Duration = Duration::from_secs(2); @@ -103,12 +103,24 @@ async fn connect(app: AppHandle, device_id: String, category: String) -> Result< } let source = state.source.clone(); - let source_for_spawn = source.clone(); + + // Battery first, before the watchdog task touches the peripheral (avoids a + // connect/discover race on the same peripheral). + if let Ok(percent) = source.battery(&device).await { + let _ = app.emit( + "battery", + BatteryEvent { + device_id: device.id.clone(), + percent, + }, + ); + } + let app_t = app.clone(); let device_t = device.clone(); let category_t = category.clone(); let handle = async_runtime::spawn(async move { - run_watchdog(app_t, source_for_spawn, device_t, category_t).await; + run_watchdog(app_t, source, device_t, category_t).await; }); state @@ -117,20 +129,10 @@ async fn connect(app: AppHandle, device_id: String, category: String) -> Result< .unwrap() .insert(category.clone(), handle); - if let Ok(percent) = source.battery(&device).await { - let _ = app.emit( - "battery", - BatteryEvent { - device_id: device.id.clone(), - percent, - }, - ); - } - - // станок: запрашиваем контроль и запоминаем + // Trainer: remember the device id. Control is requested (with confirmation) + // inside run_watchdog, after the notification subscription is live. if category == "trainer" { - let _ = source.command(&device, Command::FtmsRequestControl).await; - *state.trainer.lock().unwrap() = Some(device_id); + *state.trainer.lock().unwrap() = Some(device.id.clone()); } Ok(()) @@ -140,13 +142,16 @@ async fn connect(app: AppHandle, device_id: String, category: String) -> Result< #[tauri::command] async fn set_target_power(app: AppHandle, watts: i16) -> Result<(), String> { let state = app.state::(); + if watts < 0 { + return Err("мощность не может быть отрицательной".to_string()); + } let device = trainer_device(&state)?; state .source .command(&device, Command::FtmsSetTargetPower(watts)) .await .map_err(|e| e.to_string())?; - state.trainer_targets.lock().unwrap().power = Some(watts); + state.trainer_state.lock().unwrap().power = Some(watts); Ok(()) } @@ -154,13 +159,14 @@ async fn set_target_power(app: AppHandle, watts: i16) -> Result<(), String> { #[tauri::command] async fn set_target_cadence(app: AppHandle, rpm: u16) -> Result<(), String> { let state = app.state::(); + let half_rpm = rpm.checked_mul(2).ok_or("каденс слишком большой")?; let device = trainer_device(&state)?; state .source - .command(&device, Command::FtmsSetTargetCadence(rpm * 2)) + .command(&device, Command::FtmsSetTargetCadence(half_rpm)) .await .map_err(|e| e.to_string())?; - state.trainer_targets.lock().unwrap().cadence = Some(rpm); + state.trainer_state.lock().unwrap().cadence = Some(rpm); Ok(()) } @@ -173,7 +179,9 @@ async fn trainer_start(app: AppHandle) -> Result<(), String> { .source .command(&device, Command::FtmsStart) .await - .map_err(|e| e.to_string()) + .map_err(|e| e.to_string())?; + state.trainer_state.lock().unwrap().running = true; + Ok(()) } /// Остановить станок (FTMS Stop or Pause, opcode 0x08). @@ -185,7 +193,9 @@ async fn trainer_stop(app: AppHandle) -> Result<(), String> { .source .command(&device, Command::FtmsStop) .await - .map_err(|e| e.to_string()) + .map_err(|e| e.to_string())?; + state.trainer_state.lock().unwrap().running = false; + Ok(()) } /// Возвращает Device подключённого станка. @@ -212,19 +222,25 @@ enum StreamOutcome { Stale, } +/// Consumes a telemetry stream, invoking `on_item` per item, until it ends or goes +/// silent for `timeout`. Returns the outcome and whether at least one item was seen. async fn consume_with_watchdog( stream: &mut Pin + Send>>, timeout: Duration, mut on_item: F, -) -> StreamOutcome +) -> (StreamOutcome, bool) where F: FnMut(Telemetry), { + let mut saw = false; loop { match time::timeout(timeout, stream.next()).await { - Ok(Some(t)) => on_item(t), - Ok(None) => return StreamOutcome::Ended, - Err(_) => return StreamOutcome::Stale, + Ok(Some(t)) => { + saw = true; + on_item(t); + } + Ok(None) => return (StreamOutcome::Ended, saw), + Err(_) => return (StreamOutcome::Stale, saw), } } } @@ -236,71 +252,88 @@ async fn run_watchdog( category: String, ) { let device_id = device.id.clone(); - let mut attempts: u32 = 0; + let mut silent_cycles: u32 = 0; loop { // stream() does ensure_connected() internally, so a fresh call = reconnect. let mut stream = match source.stream(&device).await { Ok(s) => { - attempts = 0; emit_connection(&app, &device_id, &category, "reconnected"); s } Err(_) => { - attempts += 1; + silent_cycles += 1; emit_connection(&app, &device_id, &category, "reconnecting"); - if attempts >= RECONNECT_MAX_ATTEMPTS { + if silent_cycles >= RECONNECT_MAX_ATTEMPTS { emit_connection(&app, &device_id, &category, "gave_up"); + app.state::().streams.lock().unwrap().remove(&category); return; } - time::sleep(RECONNECT_BACKOFF * attempts).await; + time::sleep(RECONNECT_BACKOFF * silent_cycles).await; continue; } }; - // Trainer: re-request control and re-apply last targets after reconnect. + // Trainer: request control (with confirmation) and restore targets/start. if category == "trainer" { - let _ = source.command(&device, Command::FtmsRequestControl).await; - reapply_trainer_targets(&app, &source, &device).await; + let _ = source.request_control(&device).await; + reapply_trainer_state(&app, &source, &device).await; } let app_emit = app.clone(); let dev_id = device_id.clone(); - let outcome = consume_with_watchdog( + let (outcome, saw_data) = consume_with_watchdog( &mut stream, WATCHDOG_STALE_TIMEOUT, move |t| { - let _ = app_emit.emit("telemetry", TelemetryEvent { - device_id: dev_id.clone(), - telemetry: t, - }); + let _ = app_emit.emit( + "telemetry", + TelemetryEvent { + device_id: dev_id.clone(), + telemetry: t, + }, + ); }, - ).await; + ) + .await; - // Stream broke (Ended or Stale). Loop back and reconnect. let _ = outcome; + // A healthy cycle (received data) resets the counter; a silent one does not. + if saw_data { + silent_cycles = 0; + } else { + silent_cycles += 1; + } emit_connection(&app, &device_id, &category, "reconnecting"); + if silent_cycles >= RECONNECT_MAX_ATTEMPTS { + emit_connection(&app, &device_id, &category, "gave_up"); + app.state::().streams.lock().unwrap().remove(&category); + return; + } time::sleep(RECONNECT_BACKOFF).await; } } fn emit_connection(app: &AppHandle, device_id: &str, category: &str, status: &'static str) { - let _ = app.emit("connection", ConnectionEvent { - device_id: device_id.to_string(), - category: category.to_string(), - status, - }); + let _ = app.emit( + "connection", + ConnectionEvent { + device_id: device_id.to_string(), + category: category.to_string(), + status, + }, + ); } -async fn reapply_trainer_targets( +async fn reapply_trainer_state( app: &AppHandle, source: &Arc, device: &Device, ) { let state = app.state::(); - let (power, cadence) = { - let targets = state.trainer_targets.lock().unwrap(); - (targets.power, targets.cadence) + let (power, cadence, running) = { + let t = state.trainer_state.lock().unwrap(); + (t.power, t.cadence, t.running) }; if let Some(watts) = power { let _ = source.command(device, Command::FtmsSetTargetPower(watts)).await; @@ -308,6 +341,9 @@ async fn reapply_trainer_targets( if let Some(rpm) = cadence { let _ = source.command(device, Command::FtmsSetTargetCadence(rpm * 2)).await; } + if running { + let _ = source.command(device, Command::FtmsStart).await; + } } #[cfg_attr(mobile, tauri::mobile_entry_point)] @@ -330,7 +366,7 @@ pub fn run() { devices: RwLock::new(Vec::new()), trainer: Mutex::new(None), streams: Mutex::new(std::collections::HashMap::new()), - trainer_targets: Mutex::new(TrainerTargets::default()), + trainer_state: Mutex::new(TrainerState::default()), }); Ok::<(), Box>(()) }) @@ -362,8 +398,9 @@ mod tests { let items = vec![Telemetry::HeartRate { bpm: 120 }]; let mut s = pin(stream::iter(items)); let mut seen = Vec::new(); - let outcome = consume_with_watchdog(&mut s, Duration::from_secs(5), |t| seen.push(t)).await; + let (outcome, saw) = consume_with_watchdog(&mut s, Duration::from_secs(5), |t| seen.push(t)).await; assert!(matches!(outcome, StreamOutcome::Ended)); + assert!(saw); assert_eq!(seen, vec![Telemetry::HeartRate { bpm: 120 }]); } @@ -371,8 +408,9 @@ mod tests { async fn stale_when_silent_past_timeout() { let mut s = pin(stream::pending::()); let mut seen = Vec::new(); - let outcome = consume_with_watchdog(&mut s, Duration::from_millis(50), |t| seen.push(t)).await; + let (outcome, saw) = consume_with_watchdog(&mut s, Duration::from_millis(50), |t| seen.push(t)).await; assert!(matches!(outcome, StreamOutcome::Stale)); + assert!(!saw); assert!(seen.is_empty()); } @@ -384,7 +422,8 @@ mod tests { ]; let mut s = pin(stream::iter(items)); let mut seen = Vec::new(); - let _ = consume_with_watchdog(&mut s, Duration::from_secs(5), |t| seen.push(t)).await; + let (_outcome, saw) = consume_with_watchdog(&mut s, Duration::from_secs(5), |t| seen.push(t)).await; + assert!(saw); assert_eq!(seen.len(), 2); } } diff --git a/src-tauri/src/state.rs b/src-tauri/src/state.rs index acece23..7871228 100644 --- a/src-tauri/src/state.rs +++ b/src-tauri/src/state.rs @@ -6,11 +6,12 @@ use tauri::async_runtime::JoinHandle; use crate::ble::SensorSource; use crate::device::Device; -/// Last trainer targets, reapplied after reconnect. +/// Trainer control state, restored after a reconnect. #[derive(Default)] -pub struct TrainerTargets { +pub struct TrainerState { pub power: Option, // watts pub cadence: Option, // rpm (raw, as entered by user) + pub running: bool, } pub struct State { @@ -21,6 +22,6 @@ pub struct State { pub trainer: Mutex>, /// Потоки телеметрии по категориям ("hr"/"cadence"/"trainer") pub streams: Mutex>>, - /// Last trainer targets, reapplied after reconnect. - pub trainer_targets: Mutex, + /// Trainer control state, restored after a reconnect. + pub trainer_state: Mutex, } diff --git a/src-tauri/src/telemetry.rs b/src-tauri/src/telemetry.rs index faefd09..bcab3e4 100644 --- a/src-tauri/src/telemetry.rs +++ b/src-tauri/src/telemetry.rs @@ -35,6 +35,39 @@ pub(crate) fn parse_power(bytes: &[u8]) -> Option { Some(Telemetry::Power { watts }) } +/// Indoor Bike Data (0x2AD2): variable-length, fields driven by flags. +/// Extracts instantaneous cadence (bit 3) and instantaneous power (bit 5). +pub(crate) fn parse_indoor_bike_data(bytes: &[u8]) -> Vec { + if bytes.len() < 2 { + return Vec::new(); + } + let flags = u16::from_le_bytes([bytes[0], bytes[1]]); + let mut out = Vec::new(); + + // Byte offset of instantaneous cadence (after the 2-byte flags). + let cadence_off = 2 + + if flags & 0x0100 != 0 { 2 } else { 0 } // instantaneous speed + + if flags & 0x0002 != 0 { 2 } else { 0 }; // average speed + + // Byte offset of instantaneous power. + let power_off = cadence_off + + if flags & 0x0008 != 0 { 2 } else { 0 } // instantaneous cadence + + if flags & 0x0010 != 0 { 2 } else { 0 } // average cadence + + if flags & 0x0004 != 0 { 3 } else { 0 } // total distance (uint24) + + if flags & 0x0080 != 0 { 2 } else { 0 }; // resistance level + + if flags & 0x0008 != 0 && cadence_off + 2 <= bytes.len() { + let raw = u16::from_le_bytes([bytes[cadence_off], bytes[cadence_off + 1]]); + out.push(Telemetry::Cadence { rpm: raw as f32 / 2.0 }); + } + if flags & 0x0020 != 0 && power_off + 2 <= bytes.len() { + let raw = u16::from_le_bytes([bytes[power_off], bytes[power_off + 1]]); + out.push(Telemetry::Power { watts: raw as i16 }); + } + + out +} + /// Battery Level (0x2A19): один байт 0..=100. pub(crate) fn parse_battery(bytes: &[u8]) -> Option { let percent = *bytes.first()?; @@ -132,6 +165,26 @@ mod tests { assert_eq!(parse_power(&[0x00, 0x00, 0xC8]), None); } + #[test] + fn test_indoor_bike_cadence_and_power() { + // flags: bit 3 (cadence) + bit 5 (power) = 0x0028 + // cadence uint16 = 180 (0.5 rpm) => 90 rpm; power sint16 = 250 W + let bytes = [0x28, 0x00, 0xB4, 0x00, 0xFA, 0x00]; + assert_eq!( + parse_indoor_bike_data(&bytes), + vec![ + Telemetry::Cadence { rpm: 90.0 }, + Telemetry::Power { watts: 250 }, + ] + ); + } + + #[test] + fn test_indoor_bike_too_short() { + assert!(parse_indoor_bike_data(&[0x28]).is_empty()); + assert!(parse_indoor_bike_data(&[]).is_empty()); + } + #[test] fn test_parse_battery() { assert_eq!(parse_battery(&[100]), Some(Telemetry::Battery { percent: 100 }));