fix: telemetry live updates, input validation, connection race in UI

src-tauri/Cargo.lock

@@ -3200,6 +3200,7 @@ dependencies = [
  "tauri-plugin-opener",
  "thiserror 1.0.69",
  "tokio",
+ "tokio-stream",
  "uuid",
 ]
 

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"

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,
 }

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<u8, BleError> {
         let percent = match device.id.as_str() {
             "mock-hr" => 15,

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<u8, BleError>;
 }
@@ -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<Mutex<Option<ControlResult>>>,
 }
 
 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<Peripheral, BleError> {
@@ -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)?;
 
+        // 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::<Telemetry>();
+        let control = self.control.clone();
+
+        tokio::spawn(async move {
+            let mut notifications = notifications;
             let mut acc = CscAccumulator::new();
-        let stream = notifications.filter_map(move |n| future::ready(parse_notification(n, &mut acc)));
-        Ok(Box::pin(stream))
+            while let Some(n) = notifications.next().await {
+                for t in parse_notification(n, &mut acc, &control) {
+                    let _ = tx.send(t);
+                }
+            }
+        });
+
+        let stream: Pin<Box<dyn Stream<Item = Telemetry> + 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<u8, BleError> {
         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<Telemetry> {
+fn parse_notification(
+    n: ValueNotification,
+    acc: &mut CscAccumulator,
+    control: &Arc<Mutex<Option<ControlResult>>>,
+) -> Vec<Telemetry> {
     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()
     }
 }

src-tauri/src/commands.rs

@@ -62,10 +62,56 @@ pub(crate) fn encode_command(cmd: Command) -> (ControlPoint, Vec<u8>) {
     }
 }
 
+/// 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<ControlResult> {
+    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!(

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::<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::<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<F>(
     stream: &mut Pin<Box<dyn Stream<Item = Telemetry> + 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::<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 {
+                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::<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 {
+    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<dyn SensorSource>,
     device: &Device,
 ) {
     let state = app.state::<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<dyn std::error::Error>>(())
             })
@@ -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::<Telemetry>());
         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);
     }
 }

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<i16>, // watts
     pub cadence: Option<u16>, // rpm (raw, as entered by user)
+    pub running: bool,
 }
 
 pub struct State {
@@ -21,6 +22,6 @@ pub struct State {
     pub trainer: Mutex<Option<String>>,
     /// Потоки телеметрии по категориям ("hr"/"cadence"/"trainer")
     pub streams: Mutex<HashMap<String, JoinHandle<()>>>,
-    /// Last trainer targets, reapplied after reconnect.
-    pub trainer_targets: Mutex<TrainerTargets>,
+    /// Trainer control state, restored after a reconnect.
+    pub trainer_state: Mutex<TrainerState>,
 }

src-tauri/src/telemetry.rs

@@ -35,6 +35,39 @@ pub(crate) fn parse_power(bytes: &[u8]) -> Option<Telemetry> {
     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<Telemetry> {
+    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<Telemetry> {
     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 }));

src/main.js

@@ -13,6 +13,7 @@ const groups = {
 
 const deviceCards = new Map(); // device_id -> li
 const selected = {}; // category -> { btn, card } подключённого датчика
+const pendingStatus = {}; // category -> status (пришёл до установки selected)
 
 function el(tag, className, text) {
   const node = document.createElement(tag);
@@ -108,6 +109,10 @@ async function connect(deviceId, btn, category, card) {
     btn.textContent = "✓";
     btn.classList.add("device__connect--done");
     selected[category] = { btn, card };
+    if (pendingStatus[category]) {
+      applyStatus(selected[category], category, pendingStatus[category]);
+      delete pendingStatus[category];
+    }
     if (category === "trainer") startBtn.disabled = false;
   } catch (e) {
     console.error(e);
@@ -140,17 +145,14 @@ once("battery", (event) => {
   if (card) card.querySelector(".device__head").append(batteryIcon(percent));
 });
 
-once("telemetry", (event) => {
+listen("telemetry", (event) => {
   const t = event.payload.telemetry;
   if (t.Power) setMetric("power-value", t.Power.watts);
   else if (t.Cadence) setMetric("cadence-value", Math.round(t.Cadence.rpm));
   else if (t.HeartRate) setMetric("hr-value", t.HeartRate.bpm);
 });
 
-listen("connection", (event) => {
-  const { category, status } = event.payload;
-  const sel = selected[category];
-  if (!sel) return;
+function applyStatus(sel, category, status) {
   const { btn } = sel;
   if (status === "reconnecting") {
     btn.textContent = "переподключение…";
@@ -165,6 +167,16 @@ listen("connection", (event) => {
     btn.disabled = false;
     selected[category] = null;
   }
+}
+
+listen("connection", (event) => {
+  const { category, status } = event.payload;
+  const sel = selected[category];
+  if (!sel) {
+    pendingStatus[category] = status;
+    return;
+  }
+  applyStatus(sel, category, status);
 });
 
 // --- Таймер ---
@@ -212,8 +224,12 @@ timerBtn.addEventListener("click", async () => {
 
 // --- Целевая мощность/каденс ---
 document.querySelector("#apply-btn").addEventListener("click", async () => {
-  const watts = Number(document.querySelector("#power-input").value);
-  const rpm = Number(document.querySelector("#cadence-input").value);
+  const powerRaw = document.querySelector("#power-input").value.trim();
+  const cadenceRaw = document.querySelector("#cadence-input").value.trim();
+  if (powerRaw === "" || cadenceRaw === "") return;
+  const watts = Number(powerRaw);
+  const rpm = Number(cadenceRaw);
+  if (!Number.isFinite(watts) || !Number.isFinite(rpm)) return;
   try {
     await invoke("set_target_power", { watts });
     await invoke("set_target_cadence", { rpm });