diff --git a/apps/hello/Cargo.toml b/apps/hello/Cargo.toml index be9db18..ed05169 100644 --- a/apps/hello/Cargo.toml +++ b/apps/hello/Cargo.toml @@ -12,7 +12,6 @@ strip = true debug = false [dependencies] -alsa = "0.11" opus = "0.3" anyhow = "1.0" tokio = { version = "1.47", features = ["full"] } @@ -22,3 +21,6 @@ postcard = { version = "1.0", features = ["alloc", "use-std"] } log = "0.4" env_logger = "0.11" byteorder = "1.5" + +[target.'cfg(target_os = "linux")'.dependencies] +alsa = "0.11" diff --git a/apps/hello/build.sh b/apps/hello/build.sh index aa86e55..8ed4e00 100755 --- a/apps/hello/build.sh +++ b/apps/hello/build.sh @@ -7,7 +7,7 @@ set -e # # 1. 编译 demo # echo "Compiling hello demo..." # docker run --rm -v $(pwd):/app/hello open-xiaoai-runtime \ -# cargo build --manifest-path hello/Cargo.toml --target armv7-unknown-linux-gnueabihf --release +# cargo build --manifest-path hello/Cargo.toml --target armv7-unknown-linux-gnueabihf --bin client --release # # 2. 上传二进制文件到小爱音箱 # function upload_to_xiaoai() { @@ -21,5 +21,5 @@ set -e # upload_to_xiaoai client 192.168.31.153 # left # upload_to_xiaoai client 192.168.31.235 # right -cargo build --release +cargo build --bin server --release ./target/release/server diff --git a/apps/hello/src/audio/mod.rs b/apps/hello/src/audio/mod.rs index 23d9a65..04d19e1 100644 --- a/apps/hello/src/audio/mod.rs +++ b/apps/hello/src/audio/mod.rs @@ -1,8 +1,12 @@ +#[cfg(target_os = "linux")] pub mod recorder; +#[cfg(target_os = "linux")] pub mod player; pub mod codec; +#[cfg(target_os = "linux")] pub use recorder::AudioRecorder; +#[cfg(target_os = "linux")] pub use player::AudioPlayer; pub use codec::OpusCodec; diff --git a/apps/hello/src/audio/player.rs b/apps/hello/src/audio/player.rs index 5b04ba7..2611d46 100644 --- a/apps/hello/src/audio/player.rs +++ b/apps/hello/src/audio/player.rs @@ -1,3 +1,4 @@ +#![cfg(target_os = "linux")] use crate::config::AudioConfig; use alsa::pcm::{Access, Format, HwParams, PCM}; use alsa::Direction; diff --git a/apps/hello/src/audio/recorder.rs b/apps/hello/src/audio/recorder.rs index 4156518..d3c4c89 100644 --- a/apps/hello/src/audio/recorder.rs +++ b/apps/hello/src/audio/recorder.rs @@ -1,3 +1,4 @@ +#![cfg(target_os = "linux")] use alsa::pcm::{PCM, HwParams, Format, Access}; use alsa::Direction; use anyhow::{Result, Context}; diff --git a/apps/hello/src/bin/client.rs b/apps/hello/src/bin/client.rs index 0a5cb3b..264c2f9 100644 --- a/apps/hello/src/bin/client.rs +++ b/apps/hello/src/bin/client.rs @@ -1,5 +1,9 @@ use anyhow::Result; +#[cfg(not(target_os = "linux"))] +use hello::audio::OpusCodec; +#[cfg(target_os = "linux")] use hello::audio::{AudioPlayer, OpusCodec}; + use hello::config::AudioConfig; use hello::net::{AudioPacket, ChannelRole, ControlPacket, Discovery}; use hello::sync::{ClockSync, now_us}; @@ -11,110 +15,120 @@ use tokio::net::TcpStream; #[tokio::main] async fn main() -> Result<()> { - env_logger::init(); - let args: Vec = env::args().collect(); - if args.len() < 2 { - eprintln!("Usage: {} [left|right]", args[0]); + #[cfg(not(target_os = "linux"))] + { + println!("Client is only supported on Linux (ALSA required)"); return Ok(()); } - let role = if args[1] == "left" { - ChannelRole::Left - } else { - ChannelRole::Right - }; - let config = AudioConfig { - sample_rate: 48000, - channels: 1, - frame_size: 960, - ..AudioConfig::default() - }; - - println!("Searching for server..."); - let server_addr = Discovery::client_discover_server().await?; - let mut stream = TcpStream::connect(server_addr).await?; - - // 1. Identify - stream - .write_all(&postcard::to_allocvec(&ControlPacket::ClientIdentify { - role: role.clone(), - })?) - .await?; - - // 2. Sync - let mut clock = ClockSync::new(); - let t1 = now_us(); - stream - .write_all(&postcard::to_allocvec(&ControlPacket::Ping { - client_ts: t1, - })?) - .await?; - let mut buf = [0u8; 1024]; - let n = stream.read(&mut buf).await?; - if let Ok(ControlPacket::Pong { - client_ts, - server_ts, - }) = postcard::from_bytes::(&buf[..n]) + #[cfg(target_os = "linux")] { - let t4 = now_us(); - clock.update(client_ts, server_ts, t4); - println!( - "Clock synced. Offset: {}us, RTT: {}us", - clock.offset, - t4 - t1 - ); - } - - let player = AudioPlayer::new(&config)?; - let mut codec = OpusCodec::new(&config)?; - let mut jitter_buffer: VecDeque = VecDeque::new(); - let mut pcm_buf = vec![0i16; config.frame_size]; - - let (tx, mut rx) = tokio::sync::mpsc::channel(100); - tokio::spawn(async move { - loop { - let mut s_buf = [0u8; 4]; - if stream.read_exact(&mut s_buf).await.is_err() { - break; - } - let size = u32::from_be_bytes(s_buf) as usize; - let mut data = vec![0u8; size]; - if stream.read_exact(&mut data).await.is_err() { - break; - } - if let Ok(p) = postcard::from_bytes::(&data) { - let _ = tx.send(p).await; - } + env_logger::init(); + let args: Vec = env::args().collect(); + if args.len() < 2 { + eprintln!("Usage: {} [left|right]", args[0]); + return Ok(()); } - }); + let role = if args[1] == "left" { + ChannelRole::Left + } else { + ChannelRole::Right + }; - println!("Streaming started. Playing channel {:?}", role); + let config = AudioConfig { + sample_rate: 48000, + channels: 1, + frame_size: 960, + ..AudioConfig::default() + }; - loop { - while let Ok(p) = rx.try_recv() { - jitter_buffer.push_back(p); + println!("Searching for server..."); + let server_addr = Discovery::client_discover_server().await?; + let mut stream = TcpStream::connect(server_addr).await?; + + // 1. Identify + stream + .write_all(&postcard::to_allocvec(&ControlPacket::ClientIdentify { + role: role.clone(), + })?) + .await?; + + // 2. Sync + let mut clock = ClockSync::new(); + let t1 = now_us(); + stream + .write_all(&postcard::to_allocvec(&ControlPacket::Ping { + client_ts: t1, + })?) + .await?; + let mut buf = [0u8; 1024]; + let n = stream.read(&mut buf).await?; + if let Ok(ControlPacket::Pong { + client_ts, + server_ts, + .. + }) = postcard::from_bytes::(&buf[..n]) + { + let t4 = now_us(); + clock.update(client_ts, server_ts, t4); + println!( + "Clock synced. Offset: {}us, RTT: {}us", + clock.offset, + t4 - t1 + ); } - if let Some(p) = jitter_buffer.front() { - let target_client_time = clock.to_client_time(p.timestamp); - let now = now_us(); + let player = AudioPlayer::new(&config)?; + let mut codec = OpusCodec::new(&config)?; + let mut jitter_buffer: VecDeque = VecDeque::new(); + let mut pcm_buf = vec![0i16; config.frame_size]; - if now >= target_client_time { - let packet = jitter_buffer.pop_front().unwrap(); - let len = codec.decode(&packet.data, &mut pcm_buf)?; - player.write(&pcm_buf[..len])?; - } 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; + let (tx, mut rx) = tokio::sync::mpsc::channel(100); + tokio::spawn(async move { + loop { + let mut s_buf = [0u8; 4]; + if stream.read_exact(&mut s_buf).await.is_err() { + break; + } + let size = u32::from_be_bytes(s_buf) as usize; + let mut data = vec![0u8; size]; + if stream.read_exact(&mut data).await.is_err() { + break; + } + if let Ok(p) = postcard::from_bytes::(&data) { + let _ = tx.send(p).await; } } - } else { - tokio::time::sleep(Duration::from_millis(5)).await; + }); + + println!("Streaming started. Playing channel {:?}", role); + + loop { + while let Ok(p) = rx.try_recv() { + jitter_buffer.push_back(p); + } + + if let Some(p) = jitter_buffer.front() { + let target_client_time = clock.to_client_time(p.timestamp); + let now = now_us(); + + if now >= target_client_time { + let packet = jitter_buffer.pop_front().unwrap(); + let len = codec.decode(&packet.data, &mut pcm_buf)?; + player.write(&pcm_buf[..len])?; + } 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; + } } } } diff --git a/apps/hello/src/bin/test.rs b/apps/hello/src/bin/test.rs index 6559f36..0c3b7ba 100644 --- a/apps/hello/src/bin/test.rs +++ b/apps/hello/src/bin/test.rs @@ -1,59 +1,72 @@ +#[cfg(not(target_os = "linux"))] +use hello::audio::OpusCodec; +#[cfg(target_os = "linux")] use hello::audio::{AudioPlayer, AudioRecorder, OpusCodec}; -use hello::config::AudioConfig; + use anyhow::Result; +use hello::config::AudioConfig; use std::collections::VecDeque; use std::time::{Duration, Instant}; fn main() -> Result<()> { - let config = AudioConfig::default(); - println!("Starting Audio Modular Demo with 1s delay"); - println!( - "Config: capture={}, playback={}, rate={}, channels={}, frame_size={}", - config.capture_device, - config.playback_device, - config.sample_rate, - config.channels, - config.frame_size - ); + #[cfg(not(target_os = "linux"))] + { + println!("Test utility is only supported on Linux (ALSA required)"); + return Ok(()); + } - let recorder = AudioRecorder::new(&config)?; - let player = AudioPlayer::new(&config)?; - let mut codec = OpusCodec::new(&config)?; + #[cfg(target_os = "linux")] + { + let config = AudioConfig::default(); + println!("Starting Audio Modular Demo with 1s delay"); + println!( + "Config: capture={}, playback={}, rate={}, channels={}, frame_size={}", + config.capture_device, + config.playback_device, + config.sample_rate, + config.channels, + config.frame_size + ); - // 使用队列存储编码后的 Opus 数据块及录制时间 - let mut delay_queue: VecDeque<(Instant, Vec)> = VecDeque::new(); - let delay_duration = Duration::from_secs(1); + let recorder = AudioRecorder::new(&config)?; + let player = AudioPlayer::new(&config)?; + let mut codec = OpusCodec::new(&config)?; - let mut pcm_in = vec![0i16; config.frame_size * config.channels as usize]; - let mut opus_buf = vec![0u8; 1024]; + // 使用队列存储编码后的 Opus 数据块及录制时间 + let mut delay_queue: VecDeque<(Instant, Vec)> = VecDeque::new(); + let delay_duration = Duration::from_secs(1); - println!("Recording... Delaying playback by 1s. Press Ctrl+C to stop."); + let mut pcm_in = vec![0i16; config.frame_size * config.channels as usize]; + let mut opus_buf = vec![0u8; 1024]; - loop { - // 1. 录制原始 PCM - let read = recorder.read(&mut pcm_in)?; - if read == config.frame_size { - // 2. 编码并存入队列 - let opus_len = codec.encode(&pcm_in, &mut opus_buf)?; - let opus_data = opus_buf[..opus_len].to_vec(); - delay_queue.push_back((Instant::now(), opus_data)); - } + println!("Recording... Delaying playback by 1s. Press Ctrl+C to stop."); - // 3. 处理延迟播放：检查队列头部数据是否已等待超过 1s - while let Some((timestamp, _)) = delay_queue.front() { - if timestamp.elapsed() >= delay_duration { - let (_, opus_data) = delay_queue.pop_front().unwrap(); + loop { + // 1. 录制原始 PCM + let read = recorder.read(&mut pcm_in)?; + if read == config.frame_size { + // 2. 编码并存入队列 + let opus_len = codec.encode(&pcm_in, &mut opus_buf)?; + let opus_data = opus_buf[..opus_len].to_vec(); + delay_queue.push_back((Instant::now(), opus_data)); + } - let mut pcm_out = vec![0i16; config.frame_size * config.channels as usize]; - let decoded_len = codec.decode(&opus_data, &mut pcm_out)?; + // 3. 处理延迟播放：检查队列头部数据是否已等待超过 1s + while let Some((timestamp, _)) = delay_queue.front() { + if timestamp.elapsed() >= delay_duration { + let (_, opus_data) = delay_queue.pop_front().unwrap(); - if decoded_len == config.frame_size { - // 4. 播放还原后的 PCM - player.write(&pcm_out)?; + let mut pcm_out = vec![0i16; config.frame_size * config.channels as usize]; + let decoded_len = codec.decode(&opus_data, &mut pcm_out)?; + + if decoded_len == config.frame_size { + // 4. 播放还原后的 PCM + player.write(&pcm_out)?; + } + } else { + // 头部数据还未到 1s，由于队列是按时间排序的，后面的肯定也没到 + break; } - } else { - // 头部数据还未到 1s，由于队列是按时间排序的，后面的肯定也没到 - break; } } }