|
|
|
|
@ -4,9 +4,10 @@ use std::task::{Context, Poll};
|
|
|
|
|
use std::time::Duration;
|
|
|
|
|
|
|
|
|
|
use async_trait::async_trait;
|
|
|
|
|
use btleplug::api::{bleuuid::uuid_from_u16, Central, Peripheral as _, ValueNotification, WriteType};
|
|
|
|
|
use btleplug::api::{bleuuid::uuid_from_u16, Central, Characteristic, Peripheral as _, ScanFilter, ValueNotification, WriteType};
|
|
|
|
|
use btleplug::platform::{Adapter, Peripheral};
|
|
|
|
|
use futures::{Stream, StreamExt};
|
|
|
|
|
use std::collections::HashSet;
|
|
|
|
|
use tokio::sync::oneshot;
|
|
|
|
|
use uuid::Uuid;
|
|
|
|
|
|
|
|
|
|
@ -17,7 +18,10 @@ use crate::telemetry::{parse_csc, parse_hr, parse_indoor_bike_data, parse_power,
|
|
|
|
|
|
|
|
|
|
#[async_trait]
|
|
|
|
|
pub(crate) trait SensorSource: Send + Sync {
|
|
|
|
|
async fn scan(&self, seconds: u64) -> Result<Vec<Device>, BleError>;
|
|
|
|
|
/// Начать непрерывный скан. Возвращает поток найденных устройств (по мере
|
|
|
|
|
/// появления, только из рекламы — без подключения и чтения характеристик).
|
|
|
|
|
/// Дроп потока останавливает скан.
|
|
|
|
|
async fn scan_stream(&self) -> Result<Pin<Box<dyn Stream<Item = Device> + Send>>, BleError>;
|
|
|
|
|
|
|
|
|
|
/// Подключиться и вернуть поток распарсенных измерений.
|
|
|
|
|
async fn stream(&self, device: &Device)
|
|
|
|
|
@ -38,6 +42,18 @@ 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);
|
|
|
|
|
const CONTROL_POINT: Uuid = uuid_from_u16(0x2AD9);
|
|
|
|
|
|
|
|
|
|
/// Сервисы, по которым сканируем (только целевые: HR/CSC/FTMS/мощность).
|
|
|
|
|
static TARGET_UUIDS: &[Uuid] = &[
|
|
|
|
|
uuid_from_u16(0x180D), // Heart Rate
|
|
|
|
|
uuid_from_u16(0x1816), // Cycling Speed and Cadence
|
|
|
|
|
uuid_from_u16(0x1826), // Fitness Machine (FTMS)
|
|
|
|
|
uuid_from_u16(0x1818), // Cycling Power
|
|
|
|
|
];
|
|
|
|
|
|
|
|
|
|
/// Интервал опроса перифериалов при непрерывном скане.
|
|
|
|
|
const SCAN_POLL_INTERVAL: Duration = Duration::from_millis(1500);
|
|
|
|
|
|
|
|
|
|
// How long to wait for the trainer to confirm control (Control Point Response 0x80).
|
|
|
|
|
const CONTROL_CONFIRM_TIMEOUT: Duration = Duration::from_secs(3);
|
|
|
|
|
@ -75,20 +91,20 @@ impl BleSource {
|
|
|
|
|
|
|
|
|
|
/// Stream wrapper that cancels the background notification router on drop, so a
|
|
|
|
|
/// reconnect does not leak the previous cycle's routing task.
|
|
|
|
|
struct CancellableStream {
|
|
|
|
|
inner: tokio_stream::wrappers::UnboundedReceiverStream<Telemetry>,
|
|
|
|
|
struct CancellableStream<T> {
|
|
|
|
|
inner: tokio_stream::wrappers::UnboundedReceiverStream<T>,
|
|
|
|
|
cancel: Option<oneshot::Sender<()>>,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
impl Stream for CancellableStream {
|
|
|
|
|
type Item = Telemetry;
|
|
|
|
|
impl<T> Stream for CancellableStream<T> {
|
|
|
|
|
type Item = T;
|
|
|
|
|
|
|
|
|
|
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Telemetry>> {
|
|
|
|
|
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
|
|
|
|
|
Pin::new(&mut self.inner).poll_next(cx)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
impl Drop for CancellableStream {
|
|
|
|
|
impl<T> Drop for CancellableStream<T> {
|
|
|
|
|
fn drop(&mut self) {
|
|
|
|
|
if let Some(cancel) = self.cancel.take() {
|
|
|
|
|
let _ = cancel.send(());
|
|
|
|
|
@ -98,14 +114,85 @@ impl Drop for CancellableStream {
|
|
|
|
|
|
|
|
|
|
#[async_trait]
|
|
|
|
|
impl SensorSource for BleSource {
|
|
|
|
|
async fn scan(&self, seconds: u64) -> Result<Vec<Device>, BleError> {
|
|
|
|
|
crate::ble::scan_devices(&self.adapter, seconds).await
|
|
|
|
|
async fn scan_stream(&self) -> Result<Pin<Box<dyn Stream<Item = Device> + Send>>, BleError> {
|
|
|
|
|
// Сканируем без сервис-фильтра: часть станков не рекламирует сервисы в ADV,
|
|
|
|
|
// и фильтр на уровне адаптера их прячет. Фильтруем уже в коде ниже.
|
|
|
|
|
self.adapter
|
|
|
|
|
.start_scan(ScanFilter { services: Vec::new() })
|
|
|
|
|
.await
|
|
|
|
|
.map_err(|e| {
|
|
|
|
|
crate::logging::log("scan", &format!("start_scan failed: {e}"));
|
|
|
|
|
BleError::ScanFailed(e)
|
|
|
|
|
})?;
|
|
|
|
|
crate::logging::log("scan", "scan started");
|
|
|
|
|
|
|
|
|
|
let adapter = self.adapter.clone();
|
|
|
|
|
let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Device>();
|
|
|
|
|
let (cancel_tx, cancel_rx) = oneshot::channel::<()>();
|
|
|
|
|
|
|
|
|
|
tokio::spawn(async move {
|
|
|
|
|
let mut seen: HashSet<String> = HashSet::new();
|
|
|
|
|
let mut cancel_rx = cancel_rx;
|
|
|
|
|
loop {
|
|
|
|
|
tokio::select! {
|
|
|
|
|
_ = &mut cancel_rx => break,
|
|
|
|
|
_ = tokio::time::sleep(SCAN_POLL_INTERVAL) => {
|
|
|
|
|
let peripherals = match adapter.peripherals().await {
|
|
|
|
|
Ok(p) => p,
|
|
|
|
|
Err(e) => {
|
|
|
|
|
crate::logging::log("scan", &format!("peripherals() failed: {e}"));
|
|
|
|
|
continue;
|
|
|
|
|
}
|
|
|
|
|
};
|
|
|
|
|
for peripheral in peripherals {
|
|
|
|
|
let Some(props) = peripheral.properties().await.unwrap_or(None) else {
|
|
|
|
|
continue;
|
|
|
|
|
};
|
|
|
|
|
let id = peripheral.id().to_string();
|
|
|
|
|
let is_new = seen.insert(id.clone());
|
|
|
|
|
// логируем каждый новый перифериал (с его сервисами), чтобы
|
|
|
|
|
// видеть, что реально видит адаптер — даже до фильтрации.
|
|
|
|
|
if is_new {
|
|
|
|
|
crate::logging::log("scan", &format!("peripheral: name={:?} services={:?}", props.local_name, props.services));
|
|
|
|
|
}
|
|
|
|
|
if !props.services.iter().any(|s| TARGET_UUIDS.contains(s)) {
|
|
|
|
|
continue;
|
|
|
|
|
}
|
|
|
|
|
if !is_new {
|
|
|
|
|
continue;
|
|
|
|
|
}
|
|
|
|
|
let device = Device {
|
|
|
|
|
id,
|
|
|
|
|
address: props.address.to_string(),
|
|
|
|
|
display_name: props.local_name,
|
|
|
|
|
rssi: props.rssi,
|
|
|
|
|
service_uuids: props.services.iter().map(|u| u.to_string()).collect(),
|
|
|
|
|
characteristics: Vec::new(),
|
|
|
|
|
manufacturer_data: props.manufacturer_data.iter().next().map(|(_, v)| hex::encode(v)),
|
|
|
|
|
metadata: None,
|
|
|
|
|
};
|
|
|
|
|
crate::logging::log("scan", &format!("device found: name={:?} id={}", device.display_name, device.id));
|
|
|
|
|
let _ = tx.send(device);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
crate::logging::log("scan", "scan stopped");
|
|
|
|
|
let _ = adapter.stop_scan().await;
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
let stream: Pin<Box<dyn Stream<Item = Device> + Send>> = Box::pin(CancellableStream {
|
|
|
|
|
inner: tokio_stream::wrappers::UnboundedReceiverStream::new(rx),
|
|
|
|
|
cancel: Some(cancel_tx),
|
|
|
|
|
});
|
|
|
|
|
Ok(stream)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async fn stream(&self, device: &Device)
|
|
|
|
|
-> Result<Pin<Box<dyn Stream<Item = Telemetry> + Send>>, BleError>
|
|
|
|
|
{
|
|
|
|
|
let peripheral = self.find_peripheral(device).await?;
|
|
|
|
|
crate::logging::log("ble", &format!("stream for '{}' ({})", device.display_name.as_deref().unwrap_or("?"), device.id));
|
|
|
|
|
ensure_connected(&peripheral).await?;
|
|
|
|
|
|
|
|
|
|
let targets = [
|
|
|
|
|
@ -114,12 +201,14 @@ impl SensorSource for BleSource {
|
|
|
|
|
POWER_MEASUREMENT,
|
|
|
|
|
INDOOR_BIKE_DATA,
|
|
|
|
|
FITNESS_MACHINE_STATUS,
|
|
|
|
|
CONTROL_POINT, // 0x2AD9 — сюда приходит Control Point Response (opcode 0x80)
|
|
|
|
|
];
|
|
|
|
|
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)?;
|
|
|
|
|
crate::logging::log("ble", &format!("subscribed {}", ch.uuid));
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@ -142,12 +231,22 @@ impl SensorSource for BleSource {
|
|
|
|
|
let mut notifications = notifications;
|
|
|
|
|
let mut acc = CscAccumulator::new();
|
|
|
|
|
let mut cancel_rx = cancel_rx;
|
|
|
|
|
// Первые 30 уведомлений логируем с сырыми байтами — чтобы видеть, что
|
|
|
|
|
// реально шлёт устройство (какие флаги/поля в 0x2AD2 и т.п.).
|
|
|
|
|
let mut logged = 0u32;
|
|
|
|
|
loop {
|
|
|
|
|
tokio::select! {
|
|
|
|
|
n = notifications.next() => {
|
|
|
|
|
match n {
|
|
|
|
|
Some(n) => {
|
|
|
|
|
for t in parse_notification(n, &mut acc, &control) {
|
|
|
|
|
let uuid = n.uuid;
|
|
|
|
|
let raw = hex::encode(&n.value);
|
|
|
|
|
let parsed = parse_notification(n, &mut acc, &control);
|
|
|
|
|
if logged < 30 {
|
|
|
|
|
logged += 1;
|
|
|
|
|
crate::logging::log("telemetry", &format!("uuid={uuid} raw={raw} parsed={parsed:?}"));
|
|
|
|
|
}
|
|
|
|
|
for t in parsed {
|
|
|
|
|
let _ = tx.send(t);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
@ -170,7 +269,7 @@ impl SensorSource for BleSource {
|
|
|
|
|
let peripheral = self.find_peripheral(device).await?;
|
|
|
|
|
ensure_connected(&peripheral).await?;
|
|
|
|
|
|
|
|
|
|
let (control_point, payload) = encode_command(cmd);
|
|
|
|
|
let (control_point, payload) = encode_command(cmd.clone());
|
|
|
|
|
let cp_uuid = uuid_from_u16(control_point.short_uuid());
|
|
|
|
|
let ch = peripheral
|
|
|
|
|
.characteristics()
|
|
|
|
|
@ -178,10 +277,8 @@ impl SensorSource for BleSource {
|
|
|
|
|
.find(|c| c.uuid == cp_uuid)
|
|
|
|
|
.ok_or_else(|| BleError::DeviceNotFound(cp_uuid.to_string()))?;
|
|
|
|
|
|
|
|
|
|
peripheral
|
|
|
|
|
.write(&ch, &payload, WriteType::WithoutResponse)
|
|
|
|
|
.await
|
|
|
|
|
.map_err(BleError::WriteFailed)
|
|
|
|
|
crate::logging::log("control", &format!("command {:?} payload={}", cmd, hex::encode(&payload)));
|
|
|
|
|
write_control_point(&peripheral, &ch, &payload).await
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async fn request_control(&self, device: &Device) -> Result<(), BleError> {
|
|
|
|
|
@ -199,10 +296,8 @@ impl SensorSource for BleSource {
|
|
|
|
|
let slot = self.control.lock().unwrap().clone();
|
|
|
|
|
*slot.lock().unwrap() = None;
|
|
|
|
|
let (_, payload) = encode_command(Command::FtmsRequestControl);
|
|
|
|
|
peripheral
|
|
|
|
|
.write(&ch, &payload, WriteType::WithoutResponse)
|
|
|
|
|
.await
|
|
|
|
|
.map_err(BleError::WriteFailed)?;
|
|
|
|
|
crate::logging::log("control", "request_control sent (opcode 0x00)");
|
|
|
|
|
write_control_point(&peripheral, &ch, &payload).await?;
|
|
|
|
|
|
|
|
|
|
// Wait for the Control Point Response (routed by the stream's notification task).
|
|
|
|
|
let result = tokio::time::timeout(CONTROL_CONFIRM_TIMEOUT, async move {
|
|
|
|
|
@ -215,11 +310,17 @@ impl SensorSource for BleSource {
|
|
|
|
|
}
|
|
|
|
|
})
|
|
|
|
|
.await
|
|
|
|
|
.map_err(|_| BleError::ControlTimeout)?;
|
|
|
|
|
.map_err(|_| {
|
|
|
|
|
crate::logging::log("control", "request_control: no confirmation (timeout)");
|
|
|
|
|
BleError::ControlTimeout
|
|
|
|
|
})?;
|
|
|
|
|
|
|
|
|
|
match result {
|
|
|
|
|
ControlResult::Success => Ok(()),
|
|
|
|
|
other => Err(BleError::ControlRejected(other)),
|
|
|
|
|
other => {
|
|
|
|
|
crate::logging::log("control", &format!("control rejected: {:?}", other));
|
|
|
|
|
Err(BleError::ControlRejected(other))
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@ -245,7 +346,9 @@ impl SensorSource for BleSource {
|
|
|
|
|
|
|
|
|
|
async fn ensure_connected(peripheral: &Peripheral) -> Result<(), BleError> {
|
|
|
|
|
if !peripheral.is_connected().await.unwrap_or(false) {
|
|
|
|
|
crate::logging::log("ble", &format!("connecting {}", peripheral.id()));
|
|
|
|
|
peripheral.connect().await.map_err(BleError::ConnectFailed)?;
|
|
|
|
|
crate::logging::log("ble", "connected");
|
|
|
|
|
}
|
|
|
|
|
peripheral
|
|
|
|
|
.discover_services()
|
|
|
|
|
@ -271,12 +374,40 @@ fn parse_notification(
|
|
|
|
|
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 {
|
|
|
|
|
} else if n.uuid == CONTROL_POINT {
|
|
|
|
|
// Control Point Response (opcode 0x80) приходит indication-ом на сам
|
|
|
|
|
// Control Point (0x2AD9), а НЕ на Fitness Machine Status (0x2ADA).
|
|
|
|
|
if let Some(result) = parse_control_result(&n.value) {
|
|
|
|
|
crate::logging::log("control", &format!("control response: {:?}", result));
|
|
|
|
|
*control.lock().unwrap() = Some(result);
|
|
|
|
|
}
|
|
|
|
|
Vec::new()
|
|
|
|
|
} else if n.uuid == FITNESS_MACHINE_STATUS {
|
|
|
|
|
// статус станка (spin down и т.п.) — для ERG пока не нужен
|
|
|
|
|
Vec::new()
|
|
|
|
|
} else {
|
|
|
|
|
Vec::new()
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/// Пишет payload на Control Point: пробуем write-with-response (часть станков
|
|
|
|
|
/// требует именно его), при неудаче — write-without-response.
|
|
|
|
|
async fn write_control_point(
|
|
|
|
|
peripheral: &Peripheral,
|
|
|
|
|
ch: &Characteristic,
|
|
|
|
|
payload: &[u8],
|
|
|
|
|
) -> Result<(), BleError> {
|
|
|
|
|
match peripheral.write(ch, payload, WriteType::WithResponse).await {
|
|
|
|
|
Ok(()) => Ok(()),
|
|
|
|
|
Err(with_resp) => {
|
|
|
|
|
crate::logging::log(
|
|
|
|
|
"control",
|
|
|
|
|
&format!("with-response failed ({with_resp}), falling back to without-response"),
|
|
|
|
|
);
|
|
|
|
|
peripheral
|
|
|
|
|
.write(ch, payload, WriteType::WithoutResponse)
|
|
|
|
|
.await
|
|
|
|
|
.map_err(BleError::WriteFailed)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|