|
|
|
|
@ -5,53 +5,19 @@ mod state;
|
|
|
|
|
mod telemetry;
|
|
|
|
|
mod commands;
|
|
|
|
|
|
|
|
|
|
use std::pin::Pin;
|
|
|
|
|
use std::sync::{Arc, Mutex, RwLock};
|
|
|
|
|
use std::time::Duration;
|
|
|
|
|
use std::sync::Arc;
|
|
|
|
|
|
|
|
|
|
use futures::{Stream, StreamExt};
|
|
|
|
|
use serde::Serialize;
|
|
|
|
|
use tauri::{AppHandle, Emitter, Manager, async_runtime};
|
|
|
|
|
use tokio::time;
|
|
|
|
|
|
|
|
|
|
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::telemetry::Telemetry;
|
|
|
|
|
use crate::state::State;
|
|
|
|
|
use btleplug::api::Manager as ApiManager;
|
|
|
|
|
use btleplug::platform::Manager as BleManager;
|
|
|
|
|
|
|
|
|
|
/// Сколько секунд сканируем BLE на старте.
|
|
|
|
|
const SCAN_SECONDS: u64 = 5;
|
|
|
|
|
|
|
|
|
|
#[derive(Clone, Serialize)]
|
|
|
|
|
struct TelemetryEvent {
|
|
|
|
|
device_id: String,
|
|
|
|
|
telemetry: Telemetry,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
#[derive(Clone, Serialize)]
|
|
|
|
|
struct BatteryEvent {
|
|
|
|
|
device_id: String,
|
|
|
|
|
percent: u8,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// 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.
|
|
|
|
|
const RECONNECT_MAX_ATTEMPTS: u32 = 5;
|
|
|
|
|
// Base backoff between reconnect attempts.
|
|
|
|
|
const RECONNECT_BACKOFF: Duration = Duration::from_secs(2);
|
|
|
|
|
|
|
|
|
|
#[derive(Clone, Serialize)]
|
|
|
|
|
struct ConnectionEvent {
|
|
|
|
|
device_id: String,
|
|
|
|
|
category: String,
|
|
|
|
|
status: &'static str, // "reconnecting" | "reconnected" | "gave_up"
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
#[tauri::command]
|
|
|
|
|
async fn init(app: AppHandle) -> Result<(), String> {
|
|
|
|
|
let repo = scan_and_classify(&app).await.map_err(|e| e.to_string())?;
|
|
|
|
|
@ -64,7 +30,6 @@ async fn scan_and_classify(app: &AppHandle) -> Result<Repo, ble::BleError> {
|
|
|
|
|
let state = app.state::<State>();
|
|
|
|
|
let source = state.source.clone();
|
|
|
|
|
let devices = source.scan(SCAN_SECONDS).await?;
|
|
|
|
|
*state.devices.write().unwrap() = devices.clone();
|
|
|
|
|
Ok(repo_from_devices(devices))
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@ -85,231 +50,6 @@ fn present_repo(app: &AppHandle, repo: Repo) -> Result<(), String> {
|
|
|
|
|
Ok(())
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
#[tauri::command]
|
|
|
|
|
async fn connect(app: AppHandle, device_id: String, category: String) -> Result<(), String> {
|
|
|
|
|
let state = app.state::<State>();
|
|
|
|
|
|
|
|
|
|
let device = {
|
|
|
|
|
let devices = state.devices.read().unwrap();
|
|
|
|
|
devices
|
|
|
|
|
.iter()
|
|
|
|
|
.find(|d| d.id == device_id)
|
|
|
|
|
.cloned()
|
|
|
|
|
.ok_or_else(|| "устройство не найдено".to_string())?
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
if let Some(old) = state.streams.lock().unwrap().remove(&category) {
|
|
|
|
|
old.abort();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
let source = state.source.clone();
|
|
|
|
|
let source_for_spawn = source.clone();
|
|
|
|
|
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;
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
state
|
|
|
|
|
.streams
|
|
|
|
|
.lock()
|
|
|
|
|
.unwrap()
|
|
|
|
|
.insert(category.clone(), handle);
|
|
|
|
|
|
|
|
|
|
if let Ok(percent) = source.battery(&device).await {
|
|
|
|
|
let _ = app.emit(
|
|
|
|
|
"battery",
|
|
|
|
|
BatteryEvent {
|
|
|
|
|
device_id: device.id.clone(),
|
|
|
|
|
percent,
|
|
|
|
|
},
|
|
|
|
|
);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// станок: запрашиваем контроль и запоминаем
|
|
|
|
|
if category == "trainer" {
|
|
|
|
|
let _ = source.command(&device, Command::FtmsRequestControl).await;
|
|
|
|
|
*state.trainer.lock().unwrap() = Some(device_id);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
Ok(())
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/// Задать целевую мощность станку (ERG).
|
|
|
|
|
#[tauri::command]
|
|
|
|
|
async fn set_target_power(app: AppHandle, watts: i16) -> Result<(), String> {
|
|
|
|
|
let state = app.state::<State>();
|
|
|
|
|
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);
|
|
|
|
|
Ok(())
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/// Задать целевой каденс станку (rpm; внутри переводится в 0.5-rpm единицы FTMS).
|
|
|
|
|
#[tauri::command]
|
|
|
|
|
async fn set_target_cadence(app: AppHandle, rpm: u16) -> Result<(), String> {
|
|
|
|
|
let state = app.state::<State>();
|
|
|
|
|
let device = trainer_device(&state)?;
|
|
|
|
|
state
|
|
|
|
|
.source
|
|
|
|
|
.command(&device, Command::FtmsSetTargetCadence(rpm * 2))
|
|
|
|
|
.await
|
|
|
|
|
.map_err(|e| e.to_string())?;
|
|
|
|
|
state.trainer_targets.lock().unwrap().cadence = Some(rpm);
|
|
|
|
|
Ok(())
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/// Запустить станок (FTMS Start or Resume, opcode 0x07).
|
|
|
|
|
#[tauri::command]
|
|
|
|
|
async fn trainer_start(app: AppHandle) -> Result<(), String> {
|
|
|
|
|
let state = app.state::<State>();
|
|
|
|
|
let device = trainer_device(&state)?;
|
|
|
|
|
state
|
|
|
|
|
.source
|
|
|
|
|
.command(&device, Command::FtmsStart)
|
|
|
|
|
.await
|
|
|
|
|
.map_err(|e| e.to_string())
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/// Остановить станок (FTMS Stop or Pause, opcode 0x08).
|
|
|
|
|
#[tauri::command]
|
|
|
|
|
async fn trainer_stop(app: AppHandle) -> Result<(), String> {
|
|
|
|
|
let state = app.state::<State>();
|
|
|
|
|
let device = trainer_device(&state)?;
|
|
|
|
|
state
|
|
|
|
|
.source
|
|
|
|
|
.command(&device, Command::FtmsStop)
|
|
|
|
|
.await
|
|
|
|
|
.map_err(|e| e.to_string())
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/// Возвращает Device подключённого станка.
|
|
|
|
|
fn trainer_device(state: &State) -> Result<Device, String> {
|
|
|
|
|
let trainer_id = state
|
|
|
|
|
.trainer
|
|
|
|
|
.lock()
|
|
|
|
|
.unwrap()
|
|
|
|
|
.clone()
|
|
|
|
|
.ok_or("станок не подключен")?;
|
|
|
|
|
let devices = state.devices.read().unwrap();
|
|
|
|
|
devices
|
|
|
|
|
.iter()
|
|
|
|
|
.find(|d| d.id == trainer_id)
|
|
|
|
|
.cloned()
|
|
|
|
|
.ok_or_else(|| "станок не найден".to_string())
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/// Why the watchdog loop stopped consuming a stream.
|
|
|
|
|
enum StreamOutcome {
|
|
|
|
|
/// The stream ended (peripheral disconnected / notifications closed).
|
|
|
|
|
Ended,
|
|
|
|
|
/// No item arrived within `timeout` (stale connection).
|
|
|
|
|
Stale,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async fn consume_with_watchdog<F>(
|
|
|
|
|
stream: &mut Pin<Box<dyn Stream<Item = Telemetry> + Send>>,
|
|
|
|
|
timeout: Duration,
|
|
|
|
|
mut on_item: F,
|
|
|
|
|
) -> StreamOutcome
|
|
|
|
|
where
|
|
|
|
|
F: FnMut(Telemetry),
|
|
|
|
|
{
|
|
|
|
|
loop {
|
|
|
|
|
match time::timeout(timeout, stream.next()).await {
|
|
|
|
|
Ok(Some(t)) => on_item(t),
|
|
|
|
|
Ok(None) => return StreamOutcome::Ended,
|
|
|
|
|
Err(_) => return StreamOutcome::Stale,
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async fn run_watchdog(
|
|
|
|
|
app: AppHandle,
|
|
|
|
|
source: Arc<dyn SensorSource>,
|
|
|
|
|
device: Device,
|
|
|
|
|
category: String,
|
|
|
|
|
) {
|
|
|
|
|
let device_id = device.id.clone();
|
|
|
|
|
let mut attempts: 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;
|
|
|
|
|
emit_connection(&app, &device_id, &category, "reconnecting");
|
|
|
|
|
if attempts >= RECONNECT_MAX_ATTEMPTS {
|
|
|
|
|
emit_connection(&app, &device_id, &category, "gave_up");
|
|
|
|
|
return;
|
|
|
|
|
}
|
|
|
|
|
time::sleep(RECONNECT_BACKOFF * attempts).await;
|
|
|
|
|
continue;
|
|
|
|
|
}
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
// Trainer: re-request control and re-apply last targets after reconnect.
|
|
|
|
|
if category == "trainer" {
|
|
|
|
|
let _ = source.command(&device, Command::FtmsRequestControl).await;
|
|
|
|
|
reapply_trainer_targets(&app, &source, &device).await;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
let app_emit = app.clone();
|
|
|
|
|
let dev_id = device_id.clone();
|
|
|
|
|
let outcome = consume_with_watchdog(
|
|
|
|
|
&mut stream,
|
|
|
|
|
WATCHDOG_STALE_TIMEOUT,
|
|
|
|
|
move |t| {
|
|
|
|
|
let _ = app_emit.emit("telemetry", TelemetryEvent {
|
|
|
|
|
device_id: dev_id.clone(),
|
|
|
|
|
telemetry: t,
|
|
|
|
|
});
|
|
|
|
|
},
|
|
|
|
|
).await;
|
|
|
|
|
|
|
|
|
|
// Stream broke (Ended or Stale). Loop back and reconnect.
|
|
|
|
|
let _ = outcome;
|
|
|
|
|
emit_connection(&app, &device_id, &category, "reconnecting");
|
|
|
|
|
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,
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async fn reapply_trainer_targets(
|
|
|
|
|
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)
|
|
|
|
|
};
|
|
|
|
|
if let Some(watts) = power {
|
|
|
|
|
let _ = source.command(device, Command::FtmsSetTargetPower(watts)).await;
|
|
|
|
|
}
|
|
|
|
|
if let Some(rpm) = cadence {
|
|
|
|
|
let _ = source.command(device, Command::FtmsSetTargetCadence(rpm * 2)).await;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
#[cfg_attr(mobile, tauri::mobile_entry_point)]
|
|
|
|
|
pub fn run() {
|
|
|
|
|
tauri::Builder::default()
|
|
|
|
|
@ -325,66 +65,11 @@ pub fn run() {
|
|
|
|
|
Arc::new(BleSource::new(adapter))
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
app.manage(State {
|
|
|
|
|
source,
|
|
|
|
|
devices: RwLock::new(Vec::new()),
|
|
|
|
|
trainer: Mutex::new(None),
|
|
|
|
|
streams: Mutex::new(std::collections::HashMap::new()),
|
|
|
|
|
trainer_targets: Mutex::new(TrainerTargets::default()),
|
|
|
|
|
});
|
|
|
|
|
app.manage(State { source });
|
|
|
|
|
Ok::<(), Box<dyn std::error::Error>>(())
|
|
|
|
|
})
|
|
|
|
|
})
|
|
|
|
|
.invoke_handler(tauri::generate_handler![
|
|
|
|
|
init,
|
|
|
|
|
connect,
|
|
|
|
|
set_target_power,
|
|
|
|
|
set_target_cadence,
|
|
|
|
|
trainer_start,
|
|
|
|
|
trainer_stop
|
|
|
|
|
])
|
|
|
|
|
.invoke_handler(tauri::generate_handler![init])
|
|
|
|
|
.run(tauri::generate_context!())
|
|
|
|
|
.expect("error while running tauri application");
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
#[cfg(test)]
|
|
|
|
|
mod tests {
|
|
|
|
|
use super::*;
|
|
|
|
|
use futures::stream;
|
|
|
|
|
use std::time::Duration;
|
|
|
|
|
|
|
|
|
|
fn pin(s: impl Stream<Item = Telemetry> + Send + 'static) -> Pin<Box<dyn Stream<Item = Telemetry> + Send>> {
|
|
|
|
|
Box::pin(s)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
#[tokio::test]
|
|
|
|
|
async fn ends_when_stream_is_done() {
|
|
|
|
|
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;
|
|
|
|
|
assert!(matches!(outcome, StreamOutcome::Ended));
|
|
|
|
|
assert_eq!(seen, vec![Telemetry::HeartRate { bpm: 120 }]);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
#[tokio::test]
|
|
|
|
|
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;
|
|
|
|
|
assert!(matches!(outcome, StreamOutcome::Stale));
|
|
|
|
|
assert!(seen.is_empty());
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
#[tokio::test]
|
|
|
|
|
async fn forwards_every_item_via_callback() {
|
|
|
|
|
let items = vec![
|
|
|
|
|
Telemetry::Power { watts: 200 },
|
|
|
|
|
Telemetry::Cadence { rpm: 90.0 },
|
|
|
|
|
];
|
|
|
|
|
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;
|
|
|
|
|
assert_eq!(seen.len(), 2);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|