Hass/open-xiaoai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 完成立体声 demo 的 client / server 构建

Browse Source
This commit is contained in:
Del Wang
2025-12-31 22:53:33 +08:00
committed by Del
parent 06645dc563
commit 392d8bc21b
7 changed files with 171 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/hello/Cargo.toml
+3 -1
View File
@@ -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"
apps/hello/build.sh
+2 -2
View File
@@ -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
apps/hello/src/audio/mod.rs
+4
View File
@@ -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;
apps/hello/src/audio/player.rs
+1
View File
@@ -1,3 +1,4 @@
#![cfg(target_os = "linux")]
use crate::config::AudioConfig;
use alsa::pcm::{Access, Format, HwParams, PCM};
use alsa::Direction;
apps/hello/src/audio/recorder.rs
+1
View File
@@ -1,3 +1,4 @@
#![cfg(target_os = "linux")]
use alsa::pcm::{PCM, HwParams, Format, Access};
use alsa::Direction;
use anyhow::{Result, Context};
apps/hello/src/bin/client.rs
+106 -92
View File
@@ -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<String> = 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::<ControlPacket>(&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<AudioPacket> = 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::<AudioPacket>(&data) {
let _ = tx.send(p).await;
}
env_logger::init();
let args: Vec<String> = 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::<ControlPacket>(&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<AudioPacket> = 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::<AudioPacket>(&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;
}
}
}
}
apps/hello/src/bin/test.rs
+54 -41
View File
@@ -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<u8>)> = 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<u8>)> = 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;
}
}
}