diff --git a/packages/client-v2/Cargo.lock b/packages/client-v2/Cargo.lock index e6cb3ef..7a007b4 100644 --- a/packages/client-v2/Cargo.lock +++ b/packages/client-v2/Cargo.lock @@ -9,7 +9,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "812947049edcd670a82cd5c73c3661d2e58468577ba8489de58e1a73c04cbd5d" dependencies = [ "alsa-sys", - "bitflags", + "bitflags 2.10.0", "cfg-if", "libc", ] @@ -30,6 +30,12 @@ version = "1.0.100" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a23eb6b1614318a8071c9b2521f36b424b2c83db5eb3a0fead4a6c0809af6e61" +[[package]] +name = "arrayvec" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" + [[package]] name = "atomic-polyfill" version = "1.0.3" @@ -50,12 +56,24 @@ dependencies = [ "pkg-config", ] +[[package]] +name = "bitflags" +version = "1.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a" + [[package]] name = "bitflags" version = "2.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "812e12b5285cc515a9c72a5c1d3b6d46a19dac5acfef5265968c166106e31dd3" +[[package]] +name = "bytemuck" +version = "1.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbdf580320f38b612e485521afda1ee26d10cc9884efaaa750d383e13e3c5f4" + [[package]] name = "byteorder" version = "1.5.0" @@ -140,6 +158,15 @@ version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "edd0f118536f44f5ccd48bcb8b111bdc3de888b58c74639dfb034a357d0f206d" +[[package]] +name = "encoding_rs" +version = "0.8.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75030f3c4f45dafd7586dd6780965a8c7e8e285a5ecb86713e63a79c5b2766f3" +dependencies = [ + "cfg-if", +] + [[package]] name = "errno" version = "0.3.14" @@ -150,6 +177,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "extended" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af9673d8203fcb076b19dfd17e38b3d4ae9f44959416ea532ce72415a6020365" + [[package]] name = "find-msvc-tools" version = "0.1.6" @@ -240,6 +273,12 @@ dependencies = [ "stable_deref_trait", ] +[[package]] +name = "lazy_static" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" + [[package]] name = "libc" version = "0.2.178" @@ -366,7 +405,7 @@ version = "0.5.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" dependencies = [ - "bitflags", + "bitflags 2.10.0", ] [[package]] @@ -473,6 +512,152 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" +[[package]] +name = "symphonia" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5773a4c030a19d9bfaa090f49746ff35c75dfddfa700df7a5939d5e076a57039" +dependencies = [ + "lazy_static", + "symphonia-bundle-flac", + "symphonia-bundle-mp3", + "symphonia-codec-adpcm", + "symphonia-codec-pcm", + "symphonia-codec-vorbis", + "symphonia-core", + "symphonia-format-mkv", + "symphonia-format-ogg", + "symphonia-format-riff", + "symphonia-metadata", +] + +[[package]] +name = "symphonia-bundle-flac" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c91565e180aea25d9b80a910c546802526ffd0072d0b8974e3ebe59b686c9976" +dependencies = [ + "log", + "symphonia-core", + "symphonia-metadata", + "symphonia-utils-xiph", +] + +[[package]] +name = "symphonia-bundle-mp3" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4872dd6bb56bf5eac799e3e957aa1981086c3e613b27e0ac23b176054f7c57ed" +dependencies = [ + "lazy_static", + "log", + "symphonia-core", + "symphonia-metadata", +] + +[[package]] +name = "symphonia-codec-adpcm" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2dddc50e2bbea4cfe027441eece77c46b9f319748605ab8f3443350129ddd07f" +dependencies = [ + "log", + "symphonia-core", +] + +[[package]] +name = "symphonia-codec-pcm" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4e89d716c01541ad3ebe7c91ce4c8d38a7cf266a3f7b2f090b108fb0cb031d95" +dependencies = [ + "log", + "symphonia-core", +] + +[[package]] +name = "symphonia-codec-vorbis" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f025837c309cd69ffef572750b4a2257b59552c5399a5e49707cc5b1b85d1c73" +dependencies = [ + "log", + "symphonia-core", + "symphonia-utils-xiph", +] + +[[package]] +name = "symphonia-core" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ea00cc4f79b7f6bb7ff87eddc065a1066f3a43fe1875979056672c9ef948c2af" +dependencies = [ + "arrayvec", + "bitflags 1.3.2", + "bytemuck", + "lazy_static", + "log", +] + +[[package]] +name = "symphonia-format-mkv" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "122d786d2c43a49beb6f397551b4a050d8229eaa54c7ddf9ee4b98899b8742d0" +dependencies = [ + "lazy_static", + "log", + "symphonia-core", + "symphonia-metadata", + "symphonia-utils-xiph", +] + +[[package]] +name = "symphonia-format-ogg" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b4955c67c1ed3aa8ae8428d04ca8397fbef6a19b2b051e73b5da8b1435639cb" +dependencies = [ + "log", + "symphonia-core", + "symphonia-metadata", + "symphonia-utils-xiph", +] + +[[package]] +name = "symphonia-format-riff" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2d7c3df0e7d94efb68401d81906eae73c02b40d5ec1a141962c592d0f11a96f" +dependencies = [ + "extended", + "log", + "symphonia-core", + "symphonia-metadata", +] + +[[package]] +name = "symphonia-metadata" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "36306ff42b9ffe6e5afc99d49e121e0bd62fe79b9db7b9681d48e29fa19e6b16" +dependencies = [ + "encoding_rs", + "lazy_static", + "log", + "symphonia-core", +] + +[[package]] +name = "symphonia-utils-xiph" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee27c85ab799a338446b68eec77abf42e1a6f1bb490656e121c6e27bfbab9f16" +dependencies = [ + "symphonia-core", + "symphonia-metadata", +] + [[package]] name = "syn" version = "2.0.111" @@ -661,6 +846,7 @@ dependencies = [ "parking_lot", "postcard", "serde", + "symphonia", "tokio", "tokio-util", ] diff --git a/packages/client-v2/Cargo.toml b/packages/client-v2/Cargo.toml index 0b8b87b..8baba73 100644 --- a/packages/client-v2/Cargo.toml +++ b/packages/client-v2/Cargo.toml @@ -11,10 +11,6 @@ panic = "abort" strip = true debug = false -[features] -default = [] -app = [] - [dependencies] opus = "0.3" anyhow = "1.0" @@ -27,3 +23,6 @@ dashmap = "6.1.0" [target.'cfg(target_os = "linux")'.dependencies] alsa = "0.11" + +[target.'cfg(not(target_os = "linux"))'.dependencies] +symphonia = { version = "0.5", features = ["mp3", "wav", "pcm"] } diff --git a/packages/client-v2/Makefile b/packages/client-v2/Makefile index f8bc61e..58a86ee 100644 --- a/packages/client-v2/Makefile +++ b/packages/client-v2/Makefile @@ -1,9 +1,9 @@ build-server: - cargo build --release --bin server --features app + cargo build --release --bin server build-client: docker run --rm -v $(shell pwd):/app idootop/open-xiaoai-runtime:oh2p \ - cargo build --target armv7-unknown-linux-gnueabihf --release --bin client --features app + cargo build --target armv7-unknown-linux-gnueabihf --release --bin client run-server: target/release/server diff --git a/packages/client-v2/src/app/client/mod.rs b/packages/client-v2/src/app/client/mod.rs index 26c2485..20a08a4 100644 --- a/packages/client-v2/src/app/client/mod.rs +++ b/packages/client-v2/src/app/client/mod.rs @@ -25,8 +25,6 @@ //! └─────────────────────────────────────┘ //! ``` -#![cfg(target_os = "linux")] - mod pipeline; mod session; @@ -41,6 +39,7 @@ use crate::net::discovery::Discovery; use crate::net::event::{ClientEvent, NotificationLevel, ServerEvent}; use crate::net::network::{AudioSocket, Connection}; use crate::net::protocol::ControlPacket; +use crate::net::sync::now_us; use anyhow::{Result, anyhow}; use session::handshake; use std::net::SocketAddr; @@ -57,8 +56,8 @@ pub struct ClientConfig { pub client_auth: String, /// 服务端认证 pub server_auth: String, - /// 心跳间隔（秒） - pub heartbeat_interval: u64, + /// 心跳间隔（毫秒） + pub heartbeat_ms: u64, /// 连接超时（秒） pub timeout: u64, /// 客户端型号 @@ -75,7 +74,7 @@ impl Default for ClientConfig { .unwrap_or_else(|_| "xiao-server".to_string()), client_auth: std::env::var("XIAO_CLIENT_AUTH") .unwrap_or_else(|_| "xiao-client".to_string()), - heartbeat_interval: 10, + heartbeat_ms: 200, timeout: 60, // todo 获取设备信息 model: "Open-XiaoAi-V2".to_string(), @@ -192,17 +191,21 @@ impl Client { /// 启动心跳任务 fn spawn_heartbeat(&self, session: Arc) { - let interval = std::time::Duration::from_secs(self.config.heartbeat_interval); + let interval = std::time::Duration::from_millis(self.config.heartbeat_ms); tokio::spawn(async move { + let mut seq = 0; let mut ticker = tokio::time::interval(interval); loop { tokio::select! { _ = session.cancel.cancelled() => break, _ = ticker.tick() => { - if session.send(&ControlPacket::Ping).await.is_err() { + let t1 = now_us(); + let msg = ControlPacket::Ping { client_ts: t1, seq }; + if session.send(&msg).await.is_err() { break; } + seq += 1; } } } @@ -238,10 +241,14 @@ impl Client { /// 处理控制包 async fn handle_packet(&self, session: &Arc, packet: ControlPacket) -> Result<()> { match packet { - ControlPacket::Ping => { - session.send(&ControlPacket::Pong).await?; + ControlPacket::Pong { + client_ts, + server_ts, + .. + } => { + let t4 = now_us(); + session.update_clock(client_ts, server_ts, t4); } - ControlPacket::Pong => {} ControlPacket::RpcResponse { id, result } => { session.resolve_rpc(id, result); } diff --git a/packages/client-v2/src/app/client/pipeline.rs b/packages/client-v2/src/app/client/pipeline.rs index 03eb6fc..7056dd7 100644 --- a/packages/client-v2/src/app/client/pipeline.rs +++ b/packages/client-v2/src/app/client/pipeline.rs @@ -24,9 +24,12 @@ use crate::audio::player::AudioPlayer; use crate::audio::recorder::AudioRecorder; use crate::net::network::AudioSocket; use crate::net::protocol::AudioPacket; +use crate::net::sync::now_us; +use std::collections::VecDeque; use std::net::SocketAddr; -use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; @@ -152,6 +155,8 @@ impl RecordPipeline { Some(samples) => { if let Ok(len) = codec.encode(&samples, &mut opus_buf) { let packet = AudioPacket { + seq: 0, + timestamp: 0, data: opus_buf[..len].to_vec(), }; let _ = socket.send(&packet, target).await; @@ -209,63 +214,86 @@ impl PlaybackPipeline { stop_flag: Arc, ) -> anyhow::Result<()> { // 创建 PCM 数据通道 - let (pcm_tx, pcm_rx) = mpsc::channel::>(32); + let (pcm_tx, pcm_rx) = mpsc::channel::(128); // 启动 ALSA 播放线程（阻塞 I/O） let player_config = config.clone(); let player_stop = stop_flag.clone(); - std::thread::spawn(move || { - let player = match AudioPlayer::new(&player_config) { - Ok(p) => p, - Err(e) => { - eprintln!("[PlaybackPipeline] Failed to create player: {}", e); - return; - } - }; + println!("[PlaybackPipeline] Started"); - // 使用 blocking_recv 在线程中接收 - let mut rx = pcm_rx; - while !player_stop.load(Ordering::SeqCst) { - match rx.blocking_recv() { - Some(samples) => { - if let Err(e) = player.write(&samples) { - eprintln!("[PlaybackPipeline] Write error: {}", e); + // 主循环：从 UDP 接收，解码后发送给播放线程 + let mut udp_buf = vec![0u8; 4096]; + let mut last_time = now_us(); + tokio::spawn(async move { + loop { + match socket.recv(&mut udp_buf).await { + Ok((packet, _src)) => { + let now = now_us(); + let diff = now - last_time; + println!("Received packet now:{} diff:{}ms", now, diff / 1000); + last_time = now; + + if let Err(e) = pcm_tx.send(packet).await { + eprintln!("[PlaybackPipeline] PCM channel send error: {}", e); break; } } - None => { - // 通道关闭 - break; + Err(e) => { + eprintln!("[PlaybackPipeline] Recv error: {}", e); } } } }); + let player = match AudioPlayer::new(&player_config) { + Ok(p) => p, + Err(e) => { + eprintln!("[PlaybackPipeline] Failed to create player: {}", e); + return Err(e); + } + }; + + // 使用 blocking_recv 在线程中接收 + let mut rx = pcm_rx; // 创建 Opus 解码器 - let mut codec = OpusCodec::new(&config)?; - let mut pcm_buf = vec![0i16; config.frame_size]; - let mut udp_buf = vec![0u8; 4096]; + let mut codec = OpusCodec::new(&config).unwrap(); + let mut pcm_buf = vec![0i16; config.frame_size * config.channels as usize]; - println!("[PlaybackPipeline] Started"); + let mut jitter_buffer: VecDeque = VecDeque::new(); - // 主循环：从 UDP 接收，解码后发送给播放线程 - loop { - tokio::select! { - _ = cancel.cancelled() => break, - result = socket.recv(&mut udp_buf) => { - match result { - Ok((packet, _src)) => { - if let Ok(n) = codec.decode(&packet.data, &mut pcm_buf) { - // 使用 try_send 避免阻塞 - let _ = pcm_tx.try_send(pcm_buf[..n].to_vec()); - } - } - Err(e) => { - eprintln!("[PlaybackPipeline] Recv error: {}", e); - } + let mut start_time = 0u128; + let frame_duration_us = + (config.frame_size as f64 / config.sample_rate as f64 * 1_000_000.0) as u128; + + while !player_stop.load(Ordering::SeqCst) { + while let Ok(p) = rx.try_recv() { + jitter_buffer.push_back(p); + } + if let Some(pck) = jitter_buffer.front() { + let now = now_us(); + if start_time == 0 { + start_time = now; + } + + let target_client_time = start_time + (pck.seq as u128) * frame_duration_us; + + if now >= target_client_time { + let packet = jitter_buffer.pop_front().unwrap(); + let samples = codec.decode(&packet.data, &mut pcm_buf)?; + player.write(&pcm_buf[..samples * config.channels as usize])?; + } else if target_client_time - now > 500_000 { + // Too far in the future, maybe clock jumped? + jitter_buffer.pop_front(); + } else { + // Wait until it's time + let wait = (target_client_time - now) as u64; + if wait > 1000 { + tokio::time::sleep(Duration::from_micros(wait)).await; } } + } else { + tokio::time::sleep(Duration::from_millis(5)).await; } } @@ -273,4 +301,3 @@ impl PlaybackPipeline { Ok(()) } } - diff --git a/packages/client-v2/src/app/client/session.rs b/packages/client-v2/src/app/client/session.rs index 7e9276a..2e937c5 100644 --- a/packages/client-v2/src/app/client/session.rs +++ b/packages/client-v2/src/app/client/session.rs @@ -10,6 +10,7 @@ use crate::net::command::{Command, CommandResult}; use crate::net::network::{AudioSocket, Connection}; use crate::net::protocol::{ClientInfo, ControlPacket}; use crate::net::rpc::RpcManager; +use crate::net::sync::ClockSync; use anyhow::{Context, Result, anyhow}; use std::net::SocketAddr; use std::sync::Arc; @@ -94,6 +95,9 @@ pub struct Session { /// 会话取消令牌 pub cancel: CancellationToken, + /// 时间同步 + pub clock: Arc>, + /// 活动管道 pipelines: parking_lot::Mutex, @@ -118,6 +122,7 @@ impl Session { server_audio_addr, rpc: Arc::new(RpcManager::new()), cancel, + clock: Arc::new(parking_lot::Mutex::new(ClockSync::new(100))), pipelines: parking_lot::Mutex::new(ActivePipelines::default()), created_at: std::time::Instant::now(), volume: parking_lot::Mutex::new(100), @@ -144,6 +149,13 @@ impl Session { self.conn.recv().await } + /// 更新时间 + pub fn update_clock(&self, client_send_ts: u128, server_ts: u128, client_recv_ts: u128) { + self.clock + .lock() + .update(client_send_ts, server_ts, client_recv_ts); + } + /// 发起 RPC 调用（新版） pub async fn execute(&self, command: Command) -> Result { let (id, rx) = self.rpc.register(); diff --git a/packages/client-v2/src/app/mod.rs b/packages/client-v2/src/app/mod.rs index 4574649..c07f47e 100644 --- a/packages/client-v2/src/app/mod.rs +++ b/packages/client-v2/src/app/mod.rs @@ -1,3 +1,2 @@ -#[cfg(target_os = "linux")] pub mod client; pub mod server; diff --git a/packages/client-v2/src/app/server/mod.rs b/packages/client-v2/src/app/server/mod.rs index 234379e..f615319 100644 --- a/packages/client-v2/src/app/server/mod.rs +++ b/packages/client-v2/src/app/server/mod.rs @@ -51,6 +51,7 @@ use crate::net::discovery::Discovery; use crate::net::event::{ClientEvent, ServerEvent, ServerEventBus}; use crate::net::network::Connection; use crate::net::protocol::ControlPacket; +use crate::net::sync::now_us; use anyhow::{Context, Result, anyhow}; use std::net::SocketAddr; use std::sync::Arc; @@ -305,10 +306,14 @@ impl Server { /// 处理控制包 async fn handle_packet(&self, session: &Arc, packet: ControlPacket) -> Result<()> { match packet { - ControlPacket::Ping => { - session.send(&ControlPacket::Pong).await?; + ControlPacket::Ping { client_ts, seq } => { + let pong = ControlPacket::Pong { + client_ts, + server_ts: now_us(), + seq, + }; + session.send(&pong).await?; } - ControlPacket::Pong => {} ControlPacket::RpcResponse { id, result } => { session.resolve_rpc(id, result); } diff --git a/packages/client-v2/src/app/server/stream.rs b/packages/client-v2/src/app/server/stream.rs index f3e2040..ca510e4 100644 --- a/packages/client-v2/src/app/server/stream.rs +++ b/packages/client-v2/src/app/server/stream.rs @@ -25,6 +25,7 @@ use crate::audio::config::AudioConfig; use crate::audio::wav::{WavReader, WavWriter}; use crate::net::network::AudioSocket; use crate::net::protocol::AudioPacket; +use crate::net::sync::now_us; use std::net::SocketAddr; use std::sync::Arc; use tokio::sync::broadcast; @@ -98,33 +99,78 @@ impl FilePlaybackStream { target: SocketAddr, cancel: CancellationToken, ) -> anyhow::Result<()> { - let mut codec = OpusCodec::new(&config)?; - let mut pcm = vec![0i16; config.frame_size]; - let mut opus_buf = vec![0u8; 4096]; - let frame_duration = std::time::Duration::from_millis(20); - let mut interval = tokio::time::interval(frame_duration); + #[cfg(not(target_os = "linux"))] + { + use crate::audio::reader::AudioReader; + let mut codec = OpusCodec::new(&config)?; + let mut pcm = vec![0i16; config.frame_size * config.channels as usize]; + let mut opus_buf = vec![0u8; 4096]; - println!("[FilePlayback] Started -> {}", target); + println!("[FilePlayback] Started -> {}", target); - loop { - tokio::select! { - _ = cancel.cancelled() => break, - _ = interval.tick() => { - match reader.read_samples(&mut pcm) { - Ok(0) => { - println!("[FilePlayback] EOF reached"); - break; - } - Ok(n) => { - if let Ok(len) = codec.encode(&pcm[..n], &mut opus_buf) { + let mut seq = 0u32; + let delay_us = 0_000; // 100ms 基础延迟 + + let mut stream_start_ts = 0; + let frame_duration_us = + (config.frame_size as f64 / config.sample_rate as f64 * 1_000_000.0) as u128; + + let mut reader = AudioReader::new("temp/test.wav")?; + + loop { + tokio::select! { + _ = cancel.cancelled() => break, + result = async{ + if let Some((left_pcm, right_pcm)) = reader.read_chunk(config.frame_size)?{ + let actual_len = left_pcm.len(); + + // 1. 先将 pcm 缓冲区清零（处理尾帧时的静音填充） + pcm.fill(0); + + // 2. 只循环实际读取到的长度 + if config.channels == 2 { + for i in 0..actual_len { + pcm[i * 2] = left_pcm[i]; + pcm[i * 2 + 1] = right_pcm[i]; + } + } else { + pcm[..actual_len].copy_from_slice(&left_pcm); + } + + // 3. 编码时依然使用固定的 frame_size + let input_len = config.frame_size * config.channels as usize; + if let Ok(len) = codec.encode(&pcm[..input_len], &mut opus_buf) { + let now = now_us(); + if stream_start_ts == 0 { + // 初始化流开始时间戳 + stream_start_ts = now; + } + let target_ts = stream_start_ts + ((seq) as u128 * frame_duration_us) + delay_us; let packet = AudioPacket { + seq, + timestamp: target_ts, data: opus_buf[..len].to_vec(), }; + let _ = socket.send(&packet, target).await; + seq += 1; + + // 音频发送时长超过音频播放 1s 时进行等待（控制数据超前缓冲 1s） + if target_ts > now + 1_000_000 { + tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; + } + + return Ok(()) } + } - Err(e) => { - eprintln!("[FilePlayback] Read error: {}", e); + Err(anyhow::anyhow!("Unexpected EOF")) + } => { + if let Err(e) = result { + if e.to_string() == "EOF" { + break; + } + eprintln!("[FilePlayback] Error: {}", e); break; } } @@ -299,4 +345,3 @@ impl ForwardStream { Ok(()) } } - diff --git a/packages/client-v2/src/audio/config.rs b/packages/client-v2/src/audio/config.rs index 23f8b1b..effa18b 100644 --- a/packages/client-v2/src/audio/config.rs +++ b/packages/client-v2/src/audio/config.rs @@ -23,7 +23,7 @@ impl AudioConfig { pub fn voice_16k() -> Self { Self { capture_device: "plug:Capture".to_string(), - playback_device: "plug:default".to_string(), + playback_device: "default".to_string(), sample_rate: 16_000, channels: 1, frame_size: 320, // 20ms @@ -37,7 +37,7 @@ impl AudioConfig { pub fn music_48k() -> Self { Self { capture_device: "plug:Capture".to_string(), - playback_device: "plug:default".to_string(), + playback_device: "default".to_string(), sample_rate: 48_000, channels: 2, frame_size: 960, // 20ms diff --git a/packages/client-v2/src/audio/mod.rs b/packages/client-v2/src/audio/mod.rs index 2caf4ee..1a437b4 100644 --- a/packages/client-v2/src/audio/mod.rs +++ b/packages/client-v2/src/audio/mod.rs @@ -3,3 +3,4 @@ pub mod config; pub mod player; pub mod recorder; pub mod wav; +pub mod reader; \ No newline at end of file diff --git a/packages/client-v2/src/audio/player.rs b/packages/client-v2/src/audio/player.rs index 70a8b43..f803cf5 100644 --- a/packages/client-v2/src/audio/player.rs +++ b/packages/client-v2/src/audio/player.rs @@ -1,44 +1,58 @@ -#![cfg(target_os = "linux")] - use crate::audio::config::AudioConfig; -use alsa::Direction; -use alsa::pcm::{Access, Format, HwParams, PCM}; -use anyhow::{Context, Result}; +use anyhow::{Context, Result, anyhow}; pub struct AudioPlayer { - pcm: PCM, + #[cfg(target_os = "linux")] + pcm: alsa::pcm::PCM, } impl AudioPlayer { pub fn new(config: &AudioConfig) -> Result { - let pcm = PCM::new(&config.playback_device, Direction::Playback, false) - .context("Failed to open playback PCM device")?; + #[cfg(target_os = "linux")] { - let hwp = HwParams::any(&pcm).context("Failed to get HwParams")?; - hwp.set_access(Access::RWInterleaved)?; - hwp.set_format(Format::s16())?; - hwp.set_rate_near(config.sample_rate, alsa::ValueOr::Nearest)?; - hwp.set_channels(config.channels as u32)?; - pcm.hw_params(&hwp)?; + use alsa::Direction; + use alsa::pcm::{Access, Format, HwParams, PCM}; + + let pcm = PCM::new(&config.playback_device, Direction::Playback, false) + .context("Failed to open playback PCM device")?; + { + let hwp = HwParams::any(&pcm).context("Failed to get HwParams")?; + hwp.set_access(Access::RWInterleaved)?; + hwp.set_format(Format::s16())?; + hwp.set_rate(config.sample_rate, alsa::ValueOr::Nearest)?; + hwp.set_channels(config.channels as u32)?; + pcm.hw_params(&hwp)?; + pcm.prepare().context("Failed to prepare PCM")?; + } + Ok(Self { pcm }) + } + #[cfg(not(target_os = "linux"))] + { + Err(anyhow!("Linux Only")) } - pcm.prepare().context("Failed to prepare PCM")?; - Ok(Self { pcm }) } pub fn write(&self, buf: &[i16]) -> Result { - let res = self.pcm.io_i16()?.writei(buf); - match res { - Ok(n) => Ok(n), - Err(e) if e.errno() == 32 => { - println!("ALSA write underrun, preparing PCM"); - // Broken pipe (underrun) - self.pcm.prepare()?; - self.pcm - .io_i16()? - .writei(buf) - .context("ALSA write retry failed") + #[cfg(target_os = "linux")] + { + let res = self.pcm.io_i16()?.writei(buf); + match res { + Ok(n) => Ok(n), + Err(e) if e.errno() == 32 => { + println!("ALSA write underrun, preparing PCM"); + // Broken pipe (underrun) + self.pcm.prepare()?; + self.pcm + .io_i16()? + .writei(buf) + .context("ALSA write retry failed") + } + Err(e) => Err(e.into()), } - Err(e) => Err(e.into()), + } + #[cfg(not(target_os = "linux"))] + { + Err(anyhow!("Linux Only")) } } } diff --git a/packages/client-v2/src/audio/reader.rs b/packages/client-v2/src/audio/reader.rs new file mode 100644 index 0000000..f9c9de8 --- /dev/null +++ b/packages/client-v2/src/audio/reader.rs @@ -0,0 +1,140 @@ +#![cfg(not(target_os = "linux"))] +use anyhow::{Context, Result}; +use std::fs::File; +use std::path::Path; +use symphonia::core::audio::SampleBuffer; +use symphonia::core::codecs::{Decoder, DecoderOptions}; +use symphonia::core::formats::{FormatOptions, FormatReader}; +use symphonia::core::io::MediaSourceStream; +use symphonia::core::meta::MetadataOptions; +use symphonia::core::probe::Hint; + +pub struct AudioReader { + format: Box, + decoder: Box, + track_id: u32, + sample_buf: Option>, + channels: usize, + left_buffer: Vec, + right_buffer: Vec, +} + +impl AudioReader { + pub fn new(path: &str) -> Result { + let src = + File::open(Path::new(path)).context(format!("Failed to open audio file: {}", path))?; + let mss = MediaSourceStream::new(Box::new(src), Default::default()); + + let mut hint = Hint::new(); + if path.ends_with(".wav") { + hint.with_extension("wav"); + } else if path.ends_with(".mp3") { + hint.with_extension("mp3"); + } + + let probed = symphonia::default::get_probe() + .format( + &hint, + mss, + &FormatOptions::default(), + &MetadataOptions::default(), + ) + .context("Failed to probe audio format")?; + + let format = probed.format; + let track = format + .tracks() + .iter() + .find(|t| t.codec_params.codec != symphonia::core::codecs::CODEC_TYPE_NULL) + .context("No supported audio track found")?; + + let track_id = track.id; + let decoder = symphonia::default::get_codecs() + .make(&track.codec_params, &DecoderOptions::default()) + .context("Failed to create decoder")?; + + let channels = track.codec_params.channels.map(|c| c.count()).unwrap_or(1); + + Ok(Self { + format, + decoder, + track_id, + sample_buf: None, + channels, + left_buffer: Vec::new(), + right_buffer: Vec::new(), + }) + } + + pub fn read_chunk(&mut self, chunk_size: usize) -> Result, Vec)>> { + while self.left_buffer.len() < chunk_size { + if let Some((l, r)) = self.read_frame_internal()? { + self.left_buffer.extend(l); + self.right_buffer.extend(r); + } else { + break; + } + } + + if self.left_buffer.is_empty() { + return Ok(None); + } + + let actual_size = std::cmp::min(chunk_size, self.left_buffer.len()); + let left = self.left_buffer.drain(0..actual_size).collect(); + let right = self.right_buffer.drain(0..actual_size).collect(); + + Ok(Some((left, right))) + } + + fn read_frame_internal(&mut self) -> Result, Vec)>> { + loop { + let packet = match self.format.next_packet() { + Ok(packet) => packet, + Err(symphonia::core::errors::Error::IoError(ref e)) + if e.kind() == std::io::ErrorKind::UnexpectedEof => + { + return Ok(None); + } + Err(e) => return Err(e.into()), + }; + + if packet.track_id() != self.track_id { + continue; + } + + let decoded = self + .decoder + .decode(&packet) + .context("Failed to decode packet")?; + + if self.sample_buf.is_none() { + self.sample_buf = Some(SampleBuffer::::new( + decoded.capacity() as u64, + *decoded.spec(), + )); + } + + if let Some(buf) = self.sample_buf.as_mut() { + buf.copy_interleaved_ref(decoded); + let samples = buf.samples(); + + let mut left = Vec::with_capacity(samples.len() / self.channels); + let mut right = Vec::with_capacity(samples.len() / self.channels); + + if self.channels == 2 { + for i in (0..samples.len()).step_by(2) { + left.push(samples[i]); + right.push(samples[i + 1]); + } + } else { + for &s in samples { + left.push(s); + right.push(s); + } + } + return Ok(Some((left, right))); + } + } + } +} diff --git a/packages/client-v2/src/audio/recorder.rs b/packages/client-v2/src/audio/recorder.rs index 33c9069..dfac2eb 100644 --- a/packages/client-v2/src/audio/recorder.rs +++ b/packages/client-v2/src/audio/recorder.rs @@ -1,42 +1,56 @@ -#![cfg(target_os = "linux")] - use crate::audio::config::AudioConfig; -use alsa::Direction; -use alsa::pcm::{Access, Format, HwParams, PCM}; -use anyhow::{Context, Result}; +use anyhow::{Context, Result, anyhow}; pub struct AudioRecorder { - pcm: PCM, + #[cfg(target_os = "linux")] + pcm: alsa::pcm::PCM, } impl AudioRecorder { pub fn new(config: &AudioConfig) -> Result { - let pcm = PCM::new(&config.capture_device, Direction::Capture, false)?; + #[cfg(target_os = "linux")] { - let hwp = HwParams::any(&pcm)?; - hwp.set_access(Access::RWInterleaved)?; - hwp.set_format(Format::s16())?; - hwp.set_rate_near(config.sample_rate, alsa::ValueOr::Nearest)?; - hwp.set_channels_near(config.channels as u32)?; - pcm.hw_params(&hwp)?; + use alsa::Direction; + use alsa::pcm::{Access, Format, HwParams, PCM}; + + let pcm = PCM::new(&config.capture_device, Direction::Capture, false)?; + { + let hwp = HwParams::any(&pcm)?; + hwp.set_access(Access::RWInterleaved)?; + hwp.set_format(Format::s16())?; + hwp.set_rate_near(config.sample_rate, alsa::ValueOr::Nearest)?; + hwp.set_channels_near(config.channels as u32)?; + pcm.hw_params(&hwp)?; + } + pcm.prepare()?; + Ok(Self { pcm }) + } + #[cfg(not(target_os = "linux"))] + { + Err(anyhow!("Linux Only")) } - pcm.prepare()?; - Ok(Self { pcm }) } pub fn read(&self, buf: &mut [i16]) -> Result { - match self.pcm.io_i16()?.readi(buf) { - Ok(n) => Ok(n), - Err(e) if e.errno() == 32 => { - // 32 = Broken pipe (Overrun) - println!("ALSA recording overrun, recovering..."); - self.pcm.prepare()?; - self.pcm - .io_i16()? - .readi(buf) - .context("ALSA read retry failed") + #[cfg(target_os = "linux")] + { + match self.pcm.io_i16()?.readi(buf) { + Ok(n) => Ok(n), + Err(e) if e.errno() == 32 => { + // 32 = Broken pipe (Overrun) + println!("ALSA recording overrun, recovering..."); + self.pcm.prepare()?; + self.pcm + .io_i16()? + .readi(buf) + .context("ALSA read retry failed") + } + Err(e) => Err(e.into()), } - Err(e) => Err(e.into()), + } + #[cfg(not(target_os = "linux"))] + { + Err(anyhow!("Linux Only")) } } } diff --git a/packages/client-v2/src/audio/wav.rs b/packages/client-v2/src/audio/wav.rs index 261ab16..24dd61f 100644 --- a/packages/client-v2/src/audio/wav.rs +++ b/packages/client-v2/src/audio/wav.rs @@ -36,7 +36,7 @@ impl WavWriter { pub fn finalize(mut self) -> Result<()> { self.writer.flush()?; let mut file = self.writer.into_inner()?; - + file.seek(SeekFrom::Start(0))?; let file_size = 36 + self.data_size; @@ -93,14 +93,43 @@ impl WavReader { }) } - pub fn read_samples(&mut self, samples: &mut [i16]) -> Result { - let mut bytes = vec![0u8; samples.len() * 2]; - let n = self.reader.read(&mut bytes)?; - let sample_count = n / 2; - for i in 0..sample_count { - samples[i] = i16::from_le_bytes([bytes[i * 2], bytes[i * 2 + 1]]); + /// 读取指定的采样数。如果是立体声，则返回左右声道独立的 Vec。 + /// chunk_size: 每个声道需要读取的采样点数。 + pub fn read_chunk(&mut self, chunk_size: usize) -> Result, Vec)>> { + let channels = self.channels as usize; + let total_samples_to_read = chunk_size * channels; + let mut bytes = vec![0u8; total_samples_to_read * 2]; // 预分配完整大小的字节数组 + + let bytes_read = self.reader.read(&mut bytes)?; + + // 如果连 1 字节都没读到，说明到文件结尾了 + if bytes_read == 0 { + return Ok(None); } - Ok(sample_count) + + let mut left = vec![0i16; chunk_size]; // 预填 0 + let mut right = vec![0i16; chunk_size]; // 预填 0 + + // 实际读到了多少个采样点（总数） + let total_samples_read = bytes_read / 2; + // 计算每个声道实际读到了多少个点 + let samples_per_channel = total_samples_read / channels; + + for i in 0..samples_per_channel { + let base = i * channels * 2; + + // 左声道 (或单声道) + left[i] = i16::from_le_bytes([bytes[base], bytes[base + 1]]); + + if channels == 2 { + // 右声道 + right[i] = i16::from_le_bytes([bytes[base + 2], bytes[base + 3]]); + } else { + // 单声道填充双声道时，复制左声道 + right[i] = left[i]; + } + } + + Ok(Some((left, right))) } } - diff --git a/packages/client-v2/src/bin/client.rs b/packages/client-v2/src/bin/client.rs index 55927dd..3dcb27f 100644 --- a/packages/client-v2/src/bin/client.rs +++ b/packages/client-v2/src/bin/client.rs @@ -6,21 +6,11 @@ //! - 音频录制和播放 //! - 事件处理 -#[cfg(target_os = "linux")] use std::sync::Arc; -#[cfg(target_os = "linux")] use xiao::app::client::{Client, ClientConfig}; -#[cfg(target_os = "linux")] use xiao::net::command::Command; -#[cfg(target_os = "linux")] use xiao::net::event::NotificationLevel; -#[cfg(not(target_os = "linux"))] -fn main() { - println!("This client only works on Linux due to ALSA dependencies."); -} - -#[cfg(target_os = "linux")] #[tokio::main] async fn main() -> anyhow::Result<()> { println!("╔═══════════════════════════════════════════════════════╗"); diff --git a/packages/client-v2/src/bin/server.rs b/packages/client-v2/src/bin/server.rs index 774ed52..e9de390 100644 --- a/packages/client-v2/src/bin/server.rs +++ b/packages/client-v2/src/bin/server.rs @@ -109,7 +109,7 @@ async fn main() -> anyhow::Result<()> { match server.start_play(addr, "temp/test.wav").await { Ok(_) => { println!(" ▶️ Playback started..."); - tokio::time::sleep(std::time::Duration::from_secs(5)).await; + tokio::time::sleep(std::time::Duration::from_secs(60)).await; server.stop_play(addr).await?; println!(" ⏹️ Playback stopped"); } diff --git a/packages/client-v2/src/net/command.rs b/packages/client-v2/src/net/command.rs index 150c717..33e9142 100644 --- a/packages/client-v2/src/net/command.rs +++ b/packages/client-v2/src/net/command.rs @@ -321,4 +321,3 @@ impl CommandResult { } } } - diff --git a/packages/client-v2/src/net/event.rs b/packages/client-v2/src/net/event.rs index 2dbefd6..1cc5cec 100644 --- a/packages/client-v2/src/net/event.rs +++ b/packages/client-v2/src/net/event.rs @@ -45,16 +45,10 @@ pub enum ServerEvent { }, /// 客户端加入（广播给其他客户端） - ClientJoined { - addr: String, - model: String, - }, + ClientJoined { addr: String, model: String }, /// 客户端离开 - ClientLeft { - addr: String, - model: String, - }, + ClientLeft { addr: String, model: String }, /// 服务器消息/通知 Notification { @@ -71,9 +65,7 @@ pub enum ServerEvent { }, /// 播放完成 - PlaybackComplete { - filename: String, - }, + PlaybackComplete { filename: String }, /// 服务器状态更新 ServerStatus { @@ -82,10 +74,7 @@ pub enum ServerEvent { }, /// 自定义事件 - Custom { - name: String, - payload: Vec, - }, + Custom { name: String, payload: Vec }, } /// 客户端事件（Client → Server） @@ -99,16 +88,10 @@ pub enum ClientEvent { }, /// 音频电平 - AudioLevel { - level_db: f32, - is_silent: bool, - }, + AudioLevel { level_db: f32, is_silent: bool }, /// 按键事件 - KeyPress { - key: String, - action: KeyAction, - }, + KeyPress { key: String, action: KeyAction }, /// 警告/错误 Alert { @@ -117,10 +100,7 @@ pub enum ClientEvent { }, /// 自定义事件 - Custom { - name: String, - payload: Vec, - }, + Custom { name: String, payload: Vec }, } /// 通知级别 @@ -283,4 +263,3 @@ where handler.handle(event).await; } } - diff --git a/packages/client-v2/src/net/mod.rs b/packages/client-v2/src/net/mod.rs index dc26e0c..5f4f8d5 100644 --- a/packages/client-v2/src/net/mod.rs +++ b/packages/client-v2/src/net/mod.rs @@ -14,3 +14,4 @@ pub mod event; pub mod network; pub mod protocol; pub mod rpc; +pub mod sync; diff --git a/packages/client-v2/src/net/protocol.rs b/packages/client-v2/src/net/protocol.rs index e4f82df..d1c9268 100644 --- a/packages/client-v2/src/net/protocol.rs +++ b/packages/client-v2/src/net/protocol.rs @@ -19,8 +19,9 @@ pub struct ClientInfo { /// 音频数据包 - UDP 通道传输 #[derive(Serialize, Deserialize, Debug, Clone)] pub struct AudioPacket { - /// Opus 编码的音频数据 - pub data: Vec, + pub seq: u32, // 序列号，用于丢包检测 + pub timestamp: u128, // 目标播放时间 (主节点时间) + pub data: Vec, // Opus 编码数据 } // ==================== 控制包 ==================== @@ -51,8 +52,15 @@ pub enum ControlPacket { }, // ========== 心跳 ========== - Ping, - Pong, + Ping { + client_ts: u128, + seq: u32, + }, + Pong { + client_ts: u128, + server_ts: u128, + seq: u32, + }, // ========== RPC ========== /// RPC 请求（新版，使用 Command 类型） diff --git a/packages/client-v2/src/net/sync.rs b/packages/client-v2/src/net/sync.rs new file mode 100644 index 0000000..4f33cc0 --- /dev/null +++ b/packages/client-v2/src/net/sync.rs @@ -0,0 +1,224 @@ +use std::collections::VecDeque; +use std::time::{SystemTime, UNIX_EPOCH}; + +/// 获取当前微秒级时间戳 +pub fn now_us() -> u128 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("时间倒流") + .as_micros() +} + +/// 时钟同步管理器，用于计算主从节点间的时钟偏移 +/// 采用改进的 NTP 算法 + Kalman 滤波思想 +pub struct ClockSync { + /// 偏移量样本窗口 + offsets: VecDeque, + /// 当前估计的时钟偏移 (server_time - client_time) + pub current_offset: i128, + /// RTT 样本窗口 + rtts: VecDeque, + /// 当前估计的最小 RTT + min_rtt: i128, + /// 窗口大小 + window_size: usize, + /// 时钟漂移率 (ppm: parts per million) + /// 正值表示从节点时钟比主节点快 + drift_rate: f64, + /// 上次更新时间 + last_update_time: u128, + /// 漂移率估计窗口 + drift_samples: VecDeque, +} + +#[derive(Clone, Copy)] +struct OffsetSample { + offset: i128, + rtt: i128, + timestamp: u128, +} + +#[derive(Clone, Copy)] +struct DriftSample { + offset: i128, + timestamp: u128, +} + +impl ClockSync { + pub fn new(window_size: usize) -> Self { + Self { + offsets: VecDeque::with_capacity(window_size), + current_offset: 0, + rtts: VecDeque::with_capacity(window_size), + min_rtt: i128::MAX, + window_size, + drift_rate: 0.0, + last_update_time: now_us(), + drift_samples: VecDeque::with_capacity(60), // 保留 60 秒的样本 + } + } + + /// 更新时钟偏移估计 (NTP 算法) + /// + /// NTP 时间戳标记: + /// t1 = client_send_ts : 客户端发送 Ping 的时间 + /// t2 = server_ts : 服务器接收 Ping 的时间 + /// t3 = server_ts : 服务器发送 Pong 的时间 (假设处理时间忽略不计) + /// t4 = client_recv_ts : 客户端接收 Pong 的时间 + /// + /// RTT = (t4 - t1) - (t3 - t2) = (t4 - t1) (因为 t3 = t2) + /// Offset = ((t2 - t1) + (t3 - t4)) / 2 = ((t2 - t1) + (t2 - t4)) / 2 + /// = t2 - (t1 + t4) / 2 + pub fn update(&mut self, client_send_ts: u128, server_ts: u128, client_recv_ts: u128) { + let t1 = client_send_ts as i128; + let t2 = server_ts as i128; + let t4 = client_recv_ts as i128; + + let rtt = t4 - t1; + + // 过滤异常 RTT (局域网内 > 100ms 视为异常) + if rtt < 0 || rtt > 100_000 { + return; + } + + // 计算时钟偏移: offset = server_time - client_time + // offset = t2 - (t1 + t4) / 2 + let offset = t2 - (t1 + t4) / 2; + + // 更新 RTT 窗口 + self.rtts.push_back(rtt); + if self.rtts.len() > self.window_size { + self.rtts.pop_front(); + } + self.min_rtt = *self.rtts.iter().min().unwrap_or(&rtt); + + // 更新偏移量窗口 + let sample = OffsetSample { + offset, + rtt, + timestamp: client_recv_ts, + }; + self.offsets.push_back(sample); + if self.offsets.len() > self.window_size { + self.offsets.pop_front(); + } + + // 偏移量估计: 使用低 RTT 样本的中位数 + // 原理: RTT 较小的样本受网络抖动影响小，时间测量更准确 + let mut low_rtt_offsets: Vec = self + .offsets + .iter() + .filter(|s| s.rtt <= self.min_rtt + 5000) // 5ms 容差 + .map(|s| s.offset) + .collect(); + + if !low_rtt_offsets.is_empty() { + low_rtt_offsets.sort_unstable(); + let new_offset = low_rtt_offsets[low_rtt_offsets.len() / 2]; + + // 漂移率估计 + self.estimate_drift(new_offset, client_recv_ts); + + // 平滑更新偏移量 (避免突变) + let alpha = 0.3; // 低通滤波系数 + self.current_offset = + (alpha * new_offset as f64 + (1.0 - alpha) * self.current_offset as f64) as i128; + } + + self.last_update_time = client_recv_ts; + } + + /// 估计时钟漂移率 + /// 时钟漂移率 = d(offset) / dt + fn estimate_drift(&mut self, offset: i128, timestamp: u128) { + self.drift_samples + .push_back(DriftSample { offset, timestamp }); + if self.drift_samples.len() > 60 { + self.drift_samples.pop_front(); + } + + // 至少需要 10 秒的数据才能估计漂移 + if self.drift_samples.len() < 10 { + return; + } + + // 使用线性回归估计漂移率 + let first = self.drift_samples.front().unwrap(); + let last = self.drift_samples.back().unwrap(); + + let dt = (last.timestamp - first.timestamp) as f64; + let d_offset = (last.offset - first.offset) as f64; + + if dt > 10_000_000.0 { + // 超过 10 秒 + // drift_rate 单位: 微秒/秒 = ppm + let new_drift = d_offset / (dt / 1_000_000.0); + + // 平滑更新漂移率 + let beta = 0.1; + self.drift_rate = beta * new_drift + (1.0 - beta) * self.drift_rate; + } + } + + /// 将本地时间转换为服务器(主节点)时间 + /// 考虑时钟漂移补偿 + pub fn to_server_time(&self, client_time: u128) -> u128 { + let base_server_time = (client_time as i128 + self.current_offset) as u128; + + // 漂移补偿: 根据距离上次同步的时间，补偿时钟漂移 + let elapsed_since_update = client_time.saturating_sub(self.last_update_time) as f64; + let drift_correction = (self.drift_rate * elapsed_since_update / 1_000_000.0) as i128; + + (base_server_time as i128 + drift_correction) as u128 + } + + /// 将服务器(主节点)时间转换为本地时间 + pub fn to_client_time(&self, server_time: u128) -> u128 { + // 简化版本，不考虑漂移补偿 (播放时主要用 to_server_time) + (server_time as i128 - self.current_offset) as u128 + } + + /// 获取当前估计的 RTT (微秒) + pub fn get_rtt(&self) -> i128 { + self.min_rtt + } + + /// 获取当前时钟漂移率 (ppm) + pub fn get_drift_rate(&self) -> f64 { + self.drift_rate + } + + /// 获取同步质量评估 (0-100, 越高越好) + pub fn get_sync_quality(&self) -> u8 { + if self.offsets.is_empty() { + return 0; + } + + // 基于 RTT 稳定性和偏移量方差评估 + let rtt_variance = self.calculate_variance(&self.rtts.iter().copied().collect::>()); + let offset_variance = + self.calculate_variance(&self.offsets.iter().map(|s| s.offset).collect::>()); + + // RTT 越稳定，方差越小，质量越高 + let rtt_score = ((100_000.0 - rtt_variance.min(100_000.0)) / 100_000.0 * 50.0) as u8; + let offset_score = ((50_000.0 - offset_variance.min(50_000.0)) / 50_000.0 * 50.0) as u8; + + rtt_score + offset_score + } + + fn calculate_variance(&self, samples: &[i128]) -> f64 { + if samples.is_empty() { + return 0.0; + } + let mean = samples.iter().sum::() as f64 / samples.len() as f64; + let variance = samples + .iter() + .map(|&x| { + let diff = x as f64 - mean; + diff * diff + }) + .sum::() + / samples.len() as f64; + variance.sqrt() + } +}