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

chore: 初始化测试场景

Browse Source
This commit is contained in:
Del Wang
2026-01-02 13:04:19 +08:00
parent 33da56dead
commit 4ed8d3dca9
22 changed files with 785 additions and 695 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/client-v2/Cargo.lock
Generated
+1
View File
@@ -541,6 +541,7 @@ dependencies = [
"alsa",
"anyhow",
"opus",
"parking_lot",
"postcard",
"serde",
"tokio",
packages/client-v2/Cargo.toml
+1
View File
@@ -21,6 +21,7 @@ anyhow = "1.0"
tokio = { version = "1.48", features = ["full"] }
serde = { version = "1.0", features = ["derive"] }
postcard = { version = "1.0", features = ["alloc", "use-std"] }
parking_lot = "0.12.5"
[target.'cfg(target_os = "linux")'.dependencies]
alsa = "0.11"
packages/client-v2/Makefile
+3 -5
View File
@@ -1,16 +1,14 @@
build-server:
cargo build --release --bin server --features app
clear && cargo build --release --bin server --features app
build-client:
docker run --rm -v $(shell pwd):/app idootop/open-xiaoai-runtime:oh2p \
clear && docker run --rm -v $(shell pwd):/app idootop/open-xiaoai-runtime:oh2p \
cargo build --target armv7-unknown-linux-gnueabihf --release --bin client --features app
run-server:
cargo run --bin server --features app
target/release/server
# 部署到小爱音箱（调试自用）
deploy:
dd if=target/armv7-unknown-linux-gnueabihf/release/client \
| sshpass -p open-xiaoai ssh -o HostKeyAlgorithms=+ssh-rsa root@192.168.31.153 "dd of=/data/client"
dd if=target/armv7-unknown-linux-gnueabihf/release/client \
| sshpass -p open-xiaoai ssh -o HostKeyAlgorithms=+ssh-rsa root@192.168.31.235 "dd of=/data/client"
packages/client-v2/src/app/client/core.rs
+224
View File
@@ -0,0 +1,224 @@
#![cfg(target_os = "linux")]
use crate::audio::codec::OpusCodec;
use crate::audio::config::AudioConfig;
use crate::audio::player::AudioPlayer;
use crate::audio::recorder::AudioRecorder;
use crate::net::discovery::Discovery;
use crate::net::network::{AudioSocket, ClientNetwork, ControlConnection};
use crate::net::protocol::{AudioPacket, ControlPacket, DeviceInfo, RpcResult};
use anyhow::{Context, Result};
use std::net::SocketAddr;
use std::sync::Arc;
use tokio::sync::broadcast;
pub struct Client {
info: DeviceInfo,
}
impl Client {
pub fn new(model: &str, mac: &str, version: u32) -> Self {
Self {
info: DeviceInfo {
model: model.to_string(),
mac: mac.to_string(),
version,
},
}
}
pub async fn run(self) -> Result<()> {
println!("正在寻找服务端...");
let (server_ip, server_port) = Discovery::discover_server().await?;
let server_addr = SocketAddr::new(server_ip, server_port);
println!("发现服务端: {}", server_addr);
let network = ClientNetwork::connect(server_addr).await?;
let mut control = network.into_control();
// 握手
control
.send_packet(&ControlPacket::ClientIdentify {
info: self.info.clone(),
})
.await?;
match control.recv_packet().await? {
ControlPacket::IdentifyOk => println!("认证成功"),
p => return Err(anyhow::anyhow!("认证失败: {:?}", p)),
}
let (stop_tx, _) = broadcast::channel::<()>(1);
let audio_socket = Arc::new(AudioSocket::bind().await?);
loop {
tokio::select! {
packet = control.recv_packet() => {
let packet = packet?;
match packet {
ControlPacket::StartRecording { config } => {
println!("开始录音: {:?}", config);
let socket = audio_socket.clone();
let mut stop_rx = stop_tx.subscribe();
let server_addr = control.peer_addr()?;
tokio::spawn(async move {
if let Err(e) = handle_recording(config, socket, server_addr, stop_rx).await {
eprintln!("录音出错: {}", e);
}
});
}
ControlPacket::StopRecording => {
println!("停止录音");
let _ = stop_tx.send(());
}
ControlPacket::StartPlayback { config } => {
println!("开始播放: {:?}", config);
let socket = audio_socket.clone();
let mut stop_rx = stop_tx.subscribe();
tokio::spawn(async move {
if let Err(e) = handle_playback(config, socket, stop_rx).await {
eprintln!("播放出错: {}", e);
}
});
}
ControlPacket::StopPlayback => {
println!("停止播放");
let _ = stop_tx.send(());
}
ControlPacket::RpcRequest { id, method, args } => {
println!("收到 RPC 请求: {} {:?}", method, args);
let result = handle_rpc(&method, args).await;
control.send_packet(&ControlPacket::RpcResponse { id, result }).await?;
}
ControlPacket::Ping => {
control.send_packet(&ControlPacket::Pong).await?;
}
_ => {}
}
}
}
}
}
}
async fn handle_recording(
config: AudioConfig,
socket: Arc<AudioSocket>,
server_addr: SocketAddr,
mut stop_rx: broadcast::Receiver<()>,
) -> Result<()> {
#[cfg(target_os = "linux")]
{
let recorder = AudioRecorder::new(&config)?;
let mut codec = OpusCodec::new(&config)?;
let mut pcm_buf = vec![0i16; config.frame_size];
let mut opus_buf = vec![0u8; 4096];
loop {
if stop_rx.try_recv().is_ok() {
break;
}
let n = recorder.read(&mut pcm_buf)?;
if n > 0 {
let opus_len = codec.encode(&pcm_buf[..n], &mut opus_buf)?;
let packet = AudioPacket {
data: opus_buf[..opus_len].to_vec(),
};
socket.send_packet(&packet, server_addr).await?;
}
}
}
#[cfg(not(target_os = "linux"))]
{
println!("当前系统不支持 ALSA 录音，模拟发送音频数据...");
loop {
if stop_rx.try_recv().is_ok() {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
let packet = AudioPacket {
data: vec![0u8; 10],
};
socket.send_packet(&packet, server_addr).await?;
}
}
Ok(())
}
async fn handle_playback(
config: AudioConfig,
socket: Arc<AudioSocket>,
mut stop_rx: broadcast::Receiver<()>,
) -> Result<()> {
#[cfg(target_os = "linux")]
{
let player = AudioPlayer::new(&config)?;
let mut codec = OpusCodec::new(&config)?;
let mut pcm_buf = vec![0i16; config.frame_size];
let mut udp_buf = vec![0u8; 4096];
loop {
if stop_rx.try_recv().is_ok() {
break;
}
// 这里简单处理，UDP 接收可能阻塞。实际建议加超时或 select
let (packet, _) = socket.recv_packet(&mut udp_buf).await?;
let pcm_len = codec.decode(&packet.data, &mut pcm_buf)?;
player.write(&pcm_buf[..pcm_len])?;
}
}
#[cfg(not(target_os = "linux"))]
{
println!("当前系统不支持 ALSA 播放，模拟接收音频数据...");
let mut udp_buf = vec![0u8; 4096];
loop {
tokio::select! {
_ = stop_rx.recv() => break,
res = socket.recv_packet(&mut udp_buf) => {
let _ = res?;
}
}
}
}
Ok(())
}
async fn handle_rpc(method: &str, args: Vec<String>) -> RpcResult {
if method == "shell" && !args.is_empty() {
let cmd = &args[0];
println!("执行 Shell: {}", cmd);
// 模拟执行
#[cfg(target_os = "linux")]
{
use std::process::Command;
let output = Command::new("sh").arg("-c").arg(cmd).output();
match output {
Ok(out) => RpcResult {
stdout: String::from_utf8_lossy(&out.stdout).to_string(),
stderr: String::from_utf8_lossy(&out.stderr).to_string(),
code: out.status.code().unwrap_or(0),
},
Err(e) => RpcResult {
stdout: "".to_string(),
stderr: e.to_string(),
code: -1,
},
}
}
#[cfg(not(target_os = "linux"))]
{
RpcResult {
stdout: format!("模拟执行 {} 成功", cmd),
stderr: "".to_string(),
code: 0,
}
}
} else {
RpcResult {
stdout: "".to_string(),
stderr: "未知方法或参数错误".to_string(),
code: -1,
}
}
}
packages/client-v2/src/app/client/entry.rs
+8 -2
View File
@@ -1,8 +1,14 @@
#![cfg(target_os = "linux")]
use crate::app::client::core::Client;
use anyhow::Result;
pub async fn run_client() -> Result<()> {
println!("Hello, Client!");
Ok(())
// 模拟从系统获取信息
let model = "XiaoAi-V2-Simulated";
let mac = "00:11:22:33:44:55";
let version = 1;
let client = Client::new(model, mac, version);
client.run().await
}
packages/client-v2/src/app/client/mod.rs
+1
View File
@@ -1 +1,2 @@
pub mod core;
pub mod entry;
packages/client-v2/src/app/mod.rs
-1
View File
@@ -1,3 +1,2 @@
pub mod client;
pub mod server;
pub mod stereo;
packages/client-v2/src/app/server/core.rs
+305
View File
@@ -0,0 +1,305 @@
use crate::audio::codec::OpusCodec;
use crate::audio::config::AudioConfig;
use crate::audio::wav::{WavReader, WavWriter};
use crate::net::discovery::Discovery;
use crate::net::network::{AudioSocket, ControlConnection, ServerNetwork};
use crate::net::protocol::{AudioPacket, ControlPacket, DeviceInfo, RpcResult};
use anyhow::{Context, Result};
use parking_lot::Mutex;
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use tokio::sync::{broadcast, oneshot};
pub struct RpcManager {
next_id: AtomicU32,
pending: Mutex<HashMap<u32, oneshot::Sender<RpcResult>>>,
}
impl RpcManager {
pub fn new() -> Self {
Self {
next_id: AtomicU32::new(1),
pending: Mutex::new(HashMap::new()),
}
}
pub fn alloc_id(&self) -> (u32, oneshot::Receiver<RpcResult>) {
let id = self.next_id.fetch_add(1, Ordering::SeqCst);
let (tx, rx) = oneshot::channel();
self.pending.lock().insert(id, tx);
(id, rx)
}
pub fn fulfill(&self, id: u32, result: RpcResult) {
if let Some(tx) = self.pending.lock().remove(&id) {
let _ = tx.send(result);
}
}
}
pub struct Session {
pub info: DeviceInfo,
pub control: Arc<tokio::sync::Mutex<ControlConnection>>,
pub addr: SocketAddr,
pub rpc: Arc<RpcManager>,
}
pub struct Server {
sessions: Arc<Mutex<HashMap<SocketAddr, Arc<Session>>>>,
audio_socket: Arc<AudioSocket>,
}
impl Server {
pub async fn new() -> Result<Self> {
Ok(Self {
sessions: Arc::new(Mutex::new(HashMap::new())),
audio_socket: Arc::new(AudioSocket::bind().await?),
})
}
pub async fn run(self: Arc<Self>, port: u16) -> Result<()> {
let network = ServerNetwork::setup(port).await?;
println!("服务端启动在: {}", network.local_addr()?);
// 启动广播
Discovery::start_broadcast(port).await?;
loop {
let (control, addr) = network.accept().await?;
let server = self.clone();
tokio::spawn(async move {
if let Err(e) = server.handle_connection(control, addr).await {
eprintln!("连接 {} 出错: {}", addr, e);
}
server.sessions.lock().remove(&addr);
println!("客户端 {} 断开连接", addr);
});
}
}
async fn handle_connection(
&self,
mut control: ControlConnection,
addr: SocketAddr,
) -> Result<()> {
println!("新客户端连接: {}", addr);
// 握手认证
let info = match control.recv_packet().await? {
ControlPacket::ClientIdentify { info } => info,
p => return Err(anyhow::anyhow!("预期的握手包，收到: {:?}", p)),
};
println!(
"客户端识别: {} ({}) v{}",
info.model, info.mac, info.version
);
control.send_packet(&ControlPacket::IdentifyOk).await?;
let rpc = Arc::new(RpcManager::new());
let control = Arc::new(tokio::sync::Mutex::new(control));
let session = Arc::new(Session {
info,
control: control.clone(),
addr,
rpc: rpc.clone(),
});
self.sessions.lock().insert(addr, session.clone());
// 处理控制消息循环
loop {
let mut ctrl = control.lock().await;
let packet = ctrl.recv_packet().await?;
drop(ctrl); // 释放锁以允许发送
match packet {
ControlPacket::RpcResponse { id, result } => {
rpc.fulfill(id, result);
}
ControlPacket::Ping => {
control
.lock()
.await
.send_packet(&ControlPacket::Pong)
.await?;
}
_ => {}
}
}
}
// 暴露给外部调用的方法
pub async fn call_shell(&self, addr: SocketAddr, cmd: &str) -> Result<RpcResult> {
let session = self
.sessions
.lock()
.get(&addr)
.cloned()
.context("未找到 Session")?;
let (id, rx) = session.rpc.alloc_id();
session
.control
.lock()
.await
.send_packet(&ControlPacket::RpcRequest {
id,
method: "shell".to_string(),
args: vec![cmd.to_string()],
})
.await?;
Ok(rx.await?)
}
pub async fn start_recording(&self, addr: SocketAddr, config: AudioConfig) -> Result<()> {
let session = self
.sessions
.lock()
.get(&addr)
.cloned()
.context("未找到 Session")?;
// 发送开始录音指令
session
.control
.lock()
.await
.send_packet(&ControlPacket::StartRecording {
config: config.clone(),
})
.await?;
let audio_socket = self.audio_socket.clone();
tokio::spawn(async move {
if let Err(e) = save_audio_to_wav(audio_socket, config, "temp/recorded.wav").await {
eprintln!("保存录音失败: {}", e);
}
});
Ok(())
}
pub async fn stop_recording(&self, addr: SocketAddr) -> Result<()> {
let session = self
.sessions
.lock()
.get(&addr)
.cloned()
.context("未找到 Session")?;
session
.control
.lock()
.await
.send_packet(&ControlPacket::StopRecording)
.await?;
Ok(())
}
pub async fn start_playback(&self, addr: SocketAddr, config: AudioConfig) -> Result<()> {
let session = self
.sessions
.lock()
.get(&addr)
.cloned()
.context("未找到 Session")?;
session
.control
.lock()
.await
.send_packet(&ControlPacket::StartPlayback {
config: config.clone(),
})
.await?;
let audio_socket = self.audio_socket.clone();
tokio::spawn(async move {
if let Err(e) = stream_wav_to_client(audio_socket, addr, config, "temp/test.wav").await
{
eprintln!("推流失败: {}", e);
}
});
Ok(())
}
pub async fn stop_playback(&self, addr: SocketAddr) -> Result<()> {
let session = self
.sessions
.lock()
.get(&addr)
.cloned()
.context("未找到 Session")?;
session
.control
.lock()
.await
.send_packet(&ControlPacket::StopPlayback)
.await?;
Ok(())
}
pub fn get_sessions(&self) -> Vec<SocketAddr> {
self.sessions.lock().keys().cloned().collect()
}
}
async fn save_audio_to_wav(
socket: Arc<AudioSocket>,
config: AudioConfig,
path: &str,
) -> Result<()> {
std::fs::create_dir_all("temp")?;
let mut writer = WavWriter::create(path, config.sample_rate, config.channels)?;
let mut codec = OpusCodec::new(&config)?;
let mut pcm_buf = vec![0i16; config.frame_size];
let mut udp_buf = vec![0u8; 4096];
println!("正在录制到 {}...", path);
// 这里需要一个停止机制，目前简单起见，如果一段时间没收到包就停止，或者通过全局状态
// 为简单演示，我们录制 100 个包
for _ in 0..100 {
let (packet, _) = socket.recv_packet(&mut udp_buf).await?;
let pcm_len = codec.decode(&packet.data, &mut pcm_buf)?;
writer.write_samples(&pcm_buf[..pcm_len])?;
}
writer.finalize()?;
println!("录制完成: {}", path);
Ok(())
}
async fn stream_wav_to_client(
socket: Arc<AudioSocket>,
target: SocketAddr,
config: AudioConfig,
path: &str,
) -> Result<()> {
let mut reader = WavReader::open(path)?;
let mut codec = OpusCodec::new(&config)?;
let mut pcm_buf = vec![0i16; config.frame_size];
let mut opus_buf = vec![0u8; 4096];
println!("正在从 {} 推流...", path);
loop {
let n = reader.read_samples(&mut pcm_buf)?;
if n == 0 {
break;
}
let opus_len = codec.encode(&pcm_buf[..n], &mut opus_buf)?;
let packet = AudioPacket {
data: opus_buf[..opus_len].to_vec(),
};
socket.send_packet(&packet, target).await?;
// 控制发送频率，约 20ms 一帧
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
}
println!("推流结束");
Ok(())
}
packages/client-v2/src/app/server/entry.rs
+53 -2
View File
@@ -1,6 +1,57 @@
use crate::app::server::core::Server;
use crate::audio::config::AudioConfig;
use anyhow::Result;
use std::sync::Arc;
pub async fn run_server() -> Result<()> {
println!("Hello, Server!");
Ok(())
let server = Arc::new(Server::new().await?);
let s = server.clone();
// 运行服务器
tokio::spawn(async move {
if let Err(e) = s.run(53531).await {
eprintln!("Server error: {}", e);
}
});
// 等待一个客户端连接并进行演示
println!("等待客户端连接以进行功能演示...");
loop {
let sessions = server.get_sessions();
if !sessions.is_empty() {
let addr = sessions[0];
println!("开始对 {} 进行功能测试...", addr);
// 1. 测试 RPC
println!("测试 RPC: echo hello");
let res = server.call_shell(addr, "echo hello").await?;
println!("RPC 结果: {:?}", res);
// 2. 测试录音
println!("测试录制 2 秒音频...");
server.start_recording(addr, AudioConfig::voice()).await?;
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
server.stop_recording(addr).await?;
println!("录制结束，请检查 temp/recorded.wav");
// 3. 测试播放 (如果 temp/test.wav 存在)
if std::path::Path::new("temp/test.wav").exists() {
println!("测试播放 temp/test.wav...");
server.start_playback(addr, AudioConfig::voice()).await?;
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
server.stop_playback(addr).await?;
println!("播放结束");
} else {
println!("跳过播放测试 (temp/test.wav 不存在)");
}
break;
}
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
}
// 保持运行
loop {
tokio::time::sleep(std::time::Duration::from_secs(3600)).await;
}
}
packages/client-v2/src/app/server/mod.rs
+1
View File
@@ -1 +1,2 @@
pub mod core;
pub mod entry;
packages/client-v2/src/app/stereo/entry.rs
-35
View File
@@ -1,35 +0,0 @@
#![cfg(target_os = "linux")]
use crate::app::master::run_master;
use crate::app::slave::run_slave;
use crate::net::protocol::ChannelRole;
use anyhow::Result;
use std::env;
pub async fn run_stereo() -> Result<()> {
let args: Vec<String> = env::args().collect();
if args.len() < 3 {
eprintln!("用法: {} [master|slave] [left|right]", args[0]);
return Ok(());
}
let mode = if args[1].to_lowercase() == "master" {
"主节点"
} else {
"从节点"
};
let role = if args[2].to_lowercase() == "left" {
ChannelRole::Left
} else {
ChannelRole::Right
};
println!("🚗 当前为: {} {}", mode, role.to_string());
if mode == "主节点" {
run_master(role).await
} else {
run_slave(role).await
}
}
packages/client-v2/src/app/stereo/master.rs
-365
View File
@@ -1,365 +0,0 @@
#![cfg(target_os = "linux")]
use crate::audio::codec::OpusCodec;
use crate::audio::config::AudioConfig;
use crate::audio::player::AudioPlayer;
use crate::net::discovery::Discovery;
use crate::net::network::{ControlConnection, MasterNetwork};
use crate::net::protocol::{AudioPacket, ChannelRole, ControlPacket};
use crate::utils::alsa::AlsaRedirector;
use crate::utils::sync::now_us;
use anyhow::{Result, anyhow};
use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::signal::unix::{SignalKind, signal};
use tokio::sync::Mutex;
pub const SERVER_TCP_PORT: u16 = 53531;
#[derive(Clone)]
struct SlaveSession {
udp_addr: SocketAddr,
role: ChannelRole,
}
pub async fn run_master(master_role: ChannelRole) -> Result<()> {
// 0. 设置 ALSA 重定向
println!("🔥 启动中，请稍等...");
let _alsa_guard = AlsaRedirector::new()?;
// 1. 设置网络 (UDP + TCP)
let network = MasterNetwork::setup(SERVER_TCP_PORT).await?;
let audio_socket = network.audio_socket().clone_inner();
// 2. 启动服务发现广播
Discovery::start_broadcast(SERVER_TCP_PORT).await?;
println!("✅ 服务已启动，等待连接...");
let shutdown_flag = Arc::new(AtomicBool::new(false));
let slaves = Arc::new(Mutex::new(Vec::<SlaveSession>::new()));
// 3. 启动连接监听任务
let slaves_clone = slaves.clone();
let audio_socket_clone = audio_socket.clone();
tokio::spawn(async move {
loop {
match network.accept().await {
Ok((control_conn, client_addr)) => {
let slaves_for_session = slaves_clone.clone();
let audio_socket_for_session = audio_socket_clone.clone();
tokio::spawn(async move {
if let Err(e) = handle_master_session(
control_conn,
audio_socket_for_session,
slaves_for_session,
client_addr.to_string(),
)
.await
{
eprintln!("❌ 会话错误: {:?}", e);
}
});
}
Err(e) => {
eprintln!("❌ Accept 错误: {:?}", e);
}
}
}
});
// 4. 音频处理主循环
let config = AudioConfig::music();
let encode_config = AudioConfig {
channels: 1,
vbr: true,
..AudioConfig::music()
};
let player = AudioPlayer::new(&AudioConfig {
channels: 2,
playback_device: "plug:original_default".into(),
..config.clone()
})?;
let mut raw_buf = vec![0u8; config.frame_size * 2 * 2];
let mut pcm_out = vec![0i16; config.frame_size * 2];
let mut left_pcm = vec![0i16; config.frame_size];
let mut right_pcm = vec![0i16; config.frame_size];
let mut opus_out = vec![0u8; 1500];
let mut seq = 0u32;
// 播放延迟: 只需覆盖网络延迟 + 时钟偏移
let delay_us = 100_000; // 100ms 基础延迟
let frame_duration_us =
(config.frame_size as f64 / config.sample_rate as f64 * 1_000_000.0) as u128;
let mut stream_start_ts = 0;
let mut stream_start_seq = 0;
let shutdown_flag_clone = shutdown_flag.clone();
let audio_loop = async move {
loop {
if shutdown_flag_clone.load(Ordering::Relaxed) {
break;
}
// 打开 FIFO
let mut fifo = match tokio::fs::File::open(AlsaRedirector::fifo_path()).await {
Ok(f) => f,
Err(_) => {
if shutdown_flag_clone.load(Ordering::Relaxed) {
break;
}
tokio::time::sleep(Duration::from_secs(1)).await;
continue;
}
};
// 每个新流开始时，重置编码器状态以避免残留音频导致爆音
let mut left_encoder = OpusCodec::new(&encode_config)?;
let mut right_encoder = OpusCodec::new(&encode_config)?;
loop {
if shutdown_flag_clone.load(Ordering::Relaxed) {
break;
}
// 从 FIFO 读取
if let Err(_) = fifo.read_exact(&mut raw_buf).await {
break; // FIFO 关闭，重新打开
}
let active_slaves = {
let s = slaves.lock().await;
if s.is_empty() { None } else { Some(s.clone()) }
};
let now = now_us();
if stream_start_ts == 0 {
stream_start_ts = now;
stream_start_seq = seq;
}
// 计算该帧应当播放的目标时间
// target_ts = 数据包发送时间 + 播放延迟
let target_ts = stream_start_ts
+ ((seq - stream_start_seq) as u128 * frame_duration_us)
+ delay_us;
// 提取 PCM 数据
for i in 0..config.frame_size {
left_pcm[i] = i16::from_le_bytes([raw_buf[i * 4], raw_buf[i * 4 + 1]]);
right_pcm[i] = i16::from_le_bytes([raw_buf[i * 4 + 2], raw_buf[i * 4 + 3]]);
}
if let Some(slaves_list) = active_slaves {
// 情况 1: 有从节点，进行网络传输，并本地构造静音声道回放
// 1. 检查各声道是否有从节点需要
let needs_left = slaves_list.iter().any(|s| s.role == ChannelRole::Left);
let needs_right = slaves_list.iter().any(|s| s.role == ChannelRole::Right);
// 2. 编码需要的声道
let mut left_bytes = None;
let mut right_bytes = None;
if needs_left {
let len = left_encoder.encode(&left_pcm, &mut opus_out)?;
let packet = AudioPacket {
seq,
timestamp: target_ts,
data: opus_out[..len].to_vec(),
};
left_bytes = Some(postcard::to_allocvec(&packet)?);
}
if needs_right {
let len = right_encoder.encode(&right_pcm, &mut opus_out)?;
let packet = AudioPacket {
seq,
timestamp: target_ts,
data: opus_out[..len].to_vec(),
};
right_bytes = Some(postcard::to_allocvec(&packet)?);
}
// 3. 发送给对应的从节点
for slave in &slaves_list {
let bytes = match slave.role {
ChannelRole::Left => left_bytes.as_ref(),
ChannelRole::Right => right_bytes.as_ref(),
};
if let Some(b) = bytes {
let _ = audio_socket.send_to(b, slave.udp_addr).await;
}
}
// 4. 将非本节点的声道置为静音
for i in 0..config.frame_size {
match master_role {
ChannelRole::Left => {
pcm_out[i * 2] = left_pcm[i];
pcm_out[i * 2 + 1] = 0;
}
ChannelRole::Right => {
pcm_out[i * 2] = 0;
pcm_out[i * 2 + 1] = right_pcm[i];
}
}
}
// 5. 等待播放
let now = now_us();
if now < target_ts {
let wait = target_ts - now;
if wait > 1000 {
tokio::time::sleep(Duration::from_micros(wait as u64)).await;
} else {
// 小于 1ms，直接播放，让播放时机稍微早一点点
}
}
} else {
// 情况 2: 没有从节点，本地立体声播放
for i in 0..config.frame_size {
pcm_out[i * 2] = left_pcm[i];
pcm_out[i * 2 + 1] = right_pcm[i];
}
}
// 统一写入播放器 (始终是立体声)
if let Err(_) = player.write(&pcm_out) {
if shutdown_flag_clone.load(Ordering::Relaxed) {
break;
}
}
seq += 1;
}
// 重置流计时
stream_start_ts = 0;
}
Ok::<(), anyhow::Error>(())
};
tokio::select! {
res = audio_loop => {
if let Err(e) = res {
eprintln!("❌ 音频循环错误: {:?}", e);
}
},
_ = shutdown_signal() => {
// 设置退出标志，通知音频循环停止
shutdown_flag.store(true, Ordering::Relaxed);
},
}
// 显式清理
println!("👋 正在退出...");
AlsaRedirector::cleanup();
// 强制退出
std::process::exit(0);
}
/// 监听系统退出信号 (SIGINT, SIGTERM, SIGQUIT)
async fn shutdown_signal() {
let mut sigint = signal(SignalKind::interrupt()).expect("无法注册 SIGINT 处理器");
let mut sigterm = signal(SignalKind::terminate()).expect("无法注册 SIGTERM 处理器");
let mut sigquit = signal(SignalKind::quit()).expect("无法注册 SIGQUIT 处理器");
tokio::select! {
_ = sigint.recv() => {},
_ = sigterm.recv() => {},
_ = sigquit.recv() => {},
}
}
/// 处理主节点与从节点的会话
async fn handle_master_session(
mut control: ControlConnection,
audio_socket: Arc<tokio::net::UdpSocket>,
slaves: Arc<Mutex<Vec<SlaveSession>>>,
client_tcp_addr: String,
) -> Result<()> {
let mut buf = [0u8; 1024];
// 握手
let pkt = control.recv_packet(&mut buf).await?;
let slave_role = match pkt {
ControlPacket::ClientIdentify { role } => role,
_ => return Err(anyhow!("无效的握手协议")),
};
let xiao = ControlPacket::ServerHello {
udp_port: audio_socket.local_addr()?.port(),
};
control.send_packet(&xiao).await?;
// 等待 UDP 打洞/确认
let mut buf = [0u8; 128];
let (_, client_udp_addr) = audio_socket.recv_from(&mut buf).await?;
println!(
"✅ 从节点已连接: {} {}",
client_tcp_addr,
slave_role.to_string(),
);
// 添加到从节点列表
let session = SlaveSession {
udp_addr: client_udp_addr,
role: slave_role,
};
{
let mut s = slaves.lock().await;
s.push(session.clone());
}
// 分离 TCP 读写，处理控制消息和心跳
let (mut tcp_rx, mut tcp_tx) = control.split();
let mut buf = [0u8; 1024];
loop {
match tcp_rx.read(&mut buf).await {
Ok(0) | Err(_) => {
break;
}
Ok(n) => {
if let Ok(ControlPacket::Ping { client_ts, seq }) = postcard::from_bytes(&buf[..n])
{
let pong = ControlPacket::Pong {
client_ts,
server_ts: now_us(),
seq,
};
if tcp_tx
.write_all(&postcard::to_allocvec(&pong).unwrap())
.await
.is_err()
{
break;
}
}
}
}
}
println!(
"❌ 从节点已断开: {} {}",
client_tcp_addr,
slave_role.to_string(),
);
// 从列表中移除
{
let mut s = slaves.lock().await;
s.retain(|x| x.udp_addr != client_udp_addr);
}
Ok(())
}
packages/client-v2/src/app/stereo/mod.rs
-3
View File
@@ -1,3 +0,0 @@
pub mod entry;
pub mod master;
pub mod slave;
packages/client-v2/src/app/stereo/slave.rs
-199
View File
@@ -1,199 +0,0 @@
#![cfg(target_os = "linux")]
use crate::audio::codec::OpusCodec;
use crate::audio::config::AudioConfig;
use crate::audio::player::AudioPlayer;
use crate::net::discovery::Discovery;
use crate::net::network::SlaveNetwork;
use crate::net::protocol::{AudioPacket, ChannelRole, ControlPacket};
use crate::utils::sync::{ClockSync, now_us};
use anyhow::{Result, anyhow};
use std::sync::Arc;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::sync::{Mutex, mpsc};
/// 运行从节点模式
pub async fn run_slave(role: ChannelRole) -> Result<()> {
loop {
match handle_connection(role.clone()).await {
Err(e) => {
eprintln!("❌ {:?}", e);
tokio::time::sleep(Duration::from_secs(3)).await;
}
Ok(_) => {}
}
}
}
async fn handle_connection(role: ChannelRole) -> Result<()> {
// 1. 发现主节点
println!("🔍 正在扫描主节点...");
let (master_ip, master_tcp_port) = Discovery::discover_master().await?;
let master_tcp_addr = format!("{}:{}", master_ip, master_tcp_port);
// 2. 建立 TCP 连接
println!("🔥 发现主节点: {}", master_tcp_addr);
let network = SlaveNetwork::connect(master_tcp_addr.parse()?).await?;
let (mut control, audio) = network.split();
// 3. 身份认证
control
.send_packet(&ControlPacket::ClientIdentify { role: role.clone() })
.await?;
let mut buf = [0u8; 1024];
let pkt = control.recv_packet(&mut buf).await?;
let server_udp_port = match pkt {
ControlPacket::ServerHello { udp_port } => udp_port,
_ => return Err(anyhow!("身份认证应答异常")),
};
// 4. UDP 打洞
audio
.punch(format!("{}:{}", master_ip, server_udp_port).parse()?)
.await?;
// 5. 初始化音频与同步组件
let config = AudioConfig {
channels: 1,
..AudioConfig::music()
};
let player = AudioPlayer::new(&config)?;
let mut codec = OpusCodec::new(&config)?;
let clock = Arc::new(Mutex::new(ClockSync::new(100)));
// 用于通知主循环 TCP 已断开的消息通道
let (disconnect_tx, mut disconnect_rx) = mpsc::channel::<()>(1);
// 6. 分离 TCP 读写
let (mut tcp_rx, mut tcp_tx) = control.split();
let clock_updater = clock.clone();
let d_tx_ping = disconnect_tx.clone();
let d_tx_pong = disconnect_tx.clone();
// 定时发送 Ping (心跳 & 时间同步)
let _sync_handle = tokio::spawn(async move {
let mut seq = 0;
loop {
let t1 = now_us();
let msg = ControlPacket::Ping { client_ts: t1, seq };
let data = postcard::to_allocvec(&msg).unwrap();
if tcp_tx.write_all(&data).await.is_err() {
let _ = d_tx_ping.send(()).await; // 通知主线程 TCP 失败
break;
}
tokio::time::sleep(Duration::from_millis(200)).await;
seq += 1;
}
});
// 接收 Pong
tokio::spawn(async move {
let mut buf = [0u8; 1024];
loop {
match tcp_rx.read(&mut buf).await {
Ok(n) if n > 0 => {
if let Ok(ControlPacket::Pong {
client_ts,
server_ts,
..
}) = postcard::from_bytes(&buf[..n])
{
let t4 = now_us();
clock_updater.lock().await.update(client_ts, server_ts, t4);
}
}
_ => {
let _ = d_tx_pong.send(()).await; // TCP 断开
break;
}
}
}
});
// 7. 接收音频数据包 (UDP)
let (audio_tx, mut audio_rx) = mpsc::channel(100);
let audio_socket = audio.clone_inner();
tokio::spawn(async move {
let mut buf = [0u8; 2048];
loop {
if let Ok((len, _)) = audio_socket.recv_from(&mut buf).await {
if let Ok(packet) = postcard::from_bytes::<AudioPacket>(&buf[..len]) {
if audio_tx.send(packet).await.is_err() {
break;
}
}
}
}
});
// 8. 播放提示
println!("✅ 主节点已连接，音频串流中...");
let role_str = role.to_string();
tokio::spawn(async move {
let _ = tokio::process::Command::new("sh")
.arg("-c")
.arg(format!(
"/usr/sbin/tts_play.sh \"主节点已连接，{}\" >/dev/null 2>&1",
role_str
))
.status()
.await;
});
// 9. 播放主循环
let mut pcm_buf = vec![0i16; config.frame_size];
let mut last_seq: Option<u32> = None;
loop {
// 检查 TCP 是否已断开
if let Ok(_) = disconnect_rx.try_recv() {
return Err(anyhow!("主节点已断开: {}", master_tcp_addr));
}
// 接收数据包
if let Ok(pkt) = audio_rx.try_recv() {
let now = now_us();
let current_server_time = clock.lock().await.to_server_time(now);
// 检查包是否迟到（目标时间已过）
if current_server_time > pkt.timestamp {
let late_ms = (current_server_time - pkt.timestamp) / 1000;
if late_ms > 50 {
// 迟到超过 50ms，直接丢弃
continue;
}
// 轻微迟到（<50ms），尝试播放
}
last_seq = Some(pkt.seq);
// 精确等待到播放时间
loop {
let now = now_us();
let current_server_time_precise = clock.lock().await.to_server_time(now);
if current_server_time_precise >= pkt.timestamp {
break;
}
let wait_us = (pkt.timestamp - current_server_time_precise) as u64;
if wait_us > 1000 {
tokio::time::sleep(Duration::from_micros(wait_us as u64)).await;
} else {
// 小于 1ms，直接播放，让播放时机稍微早一点点
break;
}
}
// 解码并播放
if let Ok(len) = codec.decode(&pkt.data, &mut pcm_buf) {
let _ = player.write(&pcm_buf[..len]);
}
} else {
tokio::time::sleep(Duration::from_micros(100)).await;
}
}
}
packages/client-v2/src/audio/config.rs
+4 -2
View File
@@ -1,10 +1,12 @@
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum AudioScene {
Music,
Voice,
}
#[derive(Debug, Clone)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AudioConfig {
// ALSA 设备参数，用于录音和播放
pub capture_device: String,
packages/client-v2/src/audio/mod.rs
+1
View File
@@ -2,3 +2,4 @@ pub mod codec;
pub mod config;
pub mod player;
pub mod recorder;
pub mod wav;
packages/client-v2/src/audio/wav.rs
+106
View File
@@ -0,0 +1,106 @@
use anyhow::Result;
use std::fs::File;
use std::io::{BufReader, BufWriter, Read, Seek, SeekFrom, Write};
pub struct WavWriter {
writer: BufWriter<File>,
data_size: u32,
sample_rate: u32,
channels: u16,
}
impl WavWriter {
pub fn create(path: &str, sample_rate: u32, channels: u16) -> Result<Self> {
let file = File::create(path)?;
let mut writer = BufWriter::new(file);
// Header placeholder
writer.write_all(&[0u8; 44])?;
Ok(Self {
writer,
data_size: 0,
sample_rate,
channels,
})
}
pub fn write_samples(&mut self, samples: &[i16]) -> Result<()> {
for &sample in samples {
self.writer.write_all(&sample.to_le_bytes())?;
self.data_size += 2;
}
Ok(())
}
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;
let byte_rate = self.sample_rate * self.channels as u32 * 2;
let block_align = self.channels * 2;
let mut header = [0u8; 44];
header[0..4].copy_from_slice(b"RIFF");
header[4..8].copy_from_slice(&file_size.to_le_bytes());
header[8..12].copy_from_slice(b"WAVE");
header[12..16].copy_from_slice(b"fmt ");
header[16..20].copy_from_slice(&16u32.to_le_bytes());
header[20..22].copy_from_slice(&1u16.to_le_bytes()); // PCM
header[22..24].copy_from_slice(&self.channels.to_le_bytes());
header[24..28].copy_from_slice(&self.sample_rate.to_le_bytes());
header[28..32].copy_from_slice(&byte_rate.to_le_bytes());
header[32..34].copy_from_slice(&block_align.to_le_bytes());
header[34..36].copy_from_slice(&16u16.to_le_bytes()); // bits per sample
header[36..40].copy_from_slice(b"data");
header[40..44].copy_from_slice(&self.data_size.to_le_bytes());
file.write_all(&header)?;
Ok(())
}
}
pub struct WavReader {
reader: BufReader<File>,
pub sample_rate: u32,
pub channels: u16,
pub data_size: u32,
}
impl WavReader {
pub fn open(path: &str) -> Result<Self> {
let file = File::open(path)?;
let mut reader = BufReader::new(file);
let mut header = [0u8; 44];
reader.read_exact(&mut header)?;
if &header[0..4] != b"RIFF" || &header[8..12] != b"WAVE" {
return Err(anyhow::anyhow!("Not a WAV file"));
}
let channels = u16::from_le_bytes([header[22], header[23]]);
let sample_rate = u32::from_le_bytes([header[24], header[25], header[26], header[27]]);
let data_size = u32::from_le_bytes([header[40], header[41], header[42], header[43]]);
Ok(Self {
reader,
sample_rate,
channels,
data_size,
})
}
pub fn read_samples(&mut self, samples: &mut [i16]) -> Result<usize> {
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]]);
}
Ok(sample_count)
}
}
packages/client-v2/src/bin/client.rs
-4
View File
@@ -6,9 +6,5 @@ async fn main() -> Result<()> {
{
xiao::app::client::entry::run_client().await?;
}
#[cfg(not(target_os = "linux"))]
{
eprintln!("Only support Linux");
}
Ok(())
}
packages/client-v2/src/bin/stereo.rs
-14
View File
@@ -1,14 +0,0 @@
use anyhow::Result;
#[tokio::main]
async fn main() -> Result<()> {
#[cfg(feature = "app")]
{
xiao::app::stereo::entry::run_stereo().await?;
}
#[cfg(not(target_os = "linux"))]
{
eprintln!("Only support Linux");
}
Ok(())
}
packages/client-v2/src/net/discovery.rs
+6 -6
View File
@@ -10,13 +10,13 @@ pub const DISCOVERY_PORT: u16 = 53530;
pub struct Discovery;
impl Discovery {
/// 主节点：启动广播，告知从节点自己的 TCP 端口
/// 服务端：启动广播，告知客户端自己的 TCP 端口
pub async fn start_broadcast(tcp_port: u16) -> Result<()> {
let socket = UdpSocket::bind("0.0.0.0:0").await?;
socket.set_broadcast(true)?;
let target: SocketAddr = format!("255.255.255.255:{}", DISCOVERY_PORT).parse()?;
let msg = postcard::to_allocvec(&ControlPacket::ServerHello { udp_port: tcp_port })?;
let msg = postcard::to_allocvec(&ControlPacket::ServerHello { tcp_port })?;
tokio::spawn(async move {
loop {
@@ -28,17 +28,17 @@ impl Discovery {
Ok(())
}
/// 从节点：监听广播，发现主节点的 IP 和 TCP 端口
pub async fn discover_master() -> Result<(IpAddr, u16)> {
/// 客户端：监听广播，发现服务端的 IP 和 TCP 端口
pub async fn discover_server() -> Result<(IpAddr, u16)> {
let socket = UdpSocket::bind(format!("0.0.0.0:{}", DISCOVERY_PORT)).await?;
let mut buf = [0u8; 1024];
loop {
let (len, addr) = socket.recv_from(&mut buf).await?;
if let Ok(ControlPacket::ServerHello { udp_port }) =
if let Ok(ControlPacket::ServerHello { tcp_port }) =
postcard::from_bytes::<ControlPacket>(&buf[..len])
{
return Ok((addr.ip(), udp_port));
return Ok((addr.ip(), tcp_port));
}
}
}
packages/client-v2/src/net/network.rs
+29 -29
View File
@@ -34,11 +34,6 @@ impl AudioSocket {
Ok((packet, addr))
}
pub async fn punch(&self, target: SocketAddr) -> Result<()> {
self.socket.send_to(&[0u8; 1], target).await?;
Ok(())
}
pub fn clone_inner(&self) -> Arc<UdpSocket> {
self.socket.clone()
}
@@ -56,16 +51,20 @@ impl ControlConnection {
pub async fn send_packet(&mut self, packet: &ControlPacket) -> Result<()> {
let bytes = postcard::to_allocvec(packet)?;
let len = bytes.len() as u32;
self.stream.write_u32(len).await?;
self.stream.write_all(&bytes).await?;
Ok(())
}
pub async fn recv_packet(&mut self, buf: &mut [u8]) -> Result<ControlPacket> {
let len = self.stream.read(buf).await?;
if len == 0 {
return Err(anyhow::anyhow!("连接已关闭"));
pub async fn recv_packet(&mut self) -> Result<ControlPacket> {
let len = self.stream.read_u32().await? as usize;
if len > 10 * 1024 * 1024 {
return Err(anyhow::anyhow!("Packet too large: {}", len));
}
let packet = postcard::from_bytes(&buf[..len])?;
let mut buf = vec![0u8; len];
self.stream.read_exact(&mut buf).await?;
let packet = postcard::from_bytes(&buf)?;
Ok(packet)
}
@@ -77,19 +76,21 @@ impl ControlConnection {
) {
self.stream.into_split()
}
pub fn peer_addr(&self) -> Result<SocketAddr> {
self.stream.peer_addr().context("Failed to get peer addr")
}
}
/// 主节点网络管理器
pub struct MasterNetwork {
/// 服务端网络管理器
pub struct ServerNetwork {
listener: TcpListener,
audio: AudioSocket,
}
impl MasterNetwork {
impl ServerNetwork {
pub async fn setup(port: u16) -> Result<Self> {
let listener = TcpListener::bind(format!("0.0.0.0:{}", port)).await?;
let audio = AudioSocket::bind().await?;
Ok(Self { listener, audio })
Ok(Self { listener })
}
pub async fn accept(&self) -> Result<(ControlConnection, SocketAddr)> {
@@ -97,30 +98,29 @@ impl MasterNetwork {
Ok((ControlConnection::new(stream), addr))
}
pub fn audio_socket(&self) -> &AudioSocket {
&self.audio
pub fn local_addr(&self) -> Result<SocketAddr> {
self.listener
.local_addr()
.context("Failed to get local addr")
}
}
/// 从节点网络管理器
pub struct SlaveNetwork {
/// 客户端网络管理器
pub struct ClientNetwork {
control: ControlConnection,
audio: AudioSocket,
}
impl SlaveNetwork {
pub async fn connect(master_addr: SocketAddr) -> Result<Self> {
let stream = TcpStream::connect(master_addr)
impl ClientNetwork {
pub async fn connect(server_addr: SocketAddr) -> Result<Self> {
let stream = TcpStream::connect(server_addr)
.await
.context(format!("无法连接到主节点 TCP 地址: {}", master_addr))?;
let audio = AudioSocket::bind().await?;
.context(format!("无法连接到服务端 TCP 地址: {}", server_addr))?;
Ok(Self {
control: ControlConnection::new(stream),
audio,
})
}
pub fn split(self) -> (ControlConnection, AudioSocket) {
(self.control, self.audio)
pub fn into_control(self) -> ControlConnection {
self.control
}
}
packages/client-v2/src/net/protocol.rs
+42 -28
View File
@@ -1,45 +1,59 @@
use crate::audio::config::AudioConfig;
use serde::{Deserialize, Serialize};
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Copy)]
pub enum ChannelRole {
Left,
Right,
}
impl ChannelRole {
pub fn to_string(&self) -> String {
match self {
ChannelRole::Left => "左声道".to_string(),
ChannelRole::Right => "右声道".to_string(),
}
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
pub struct DeviceInfo {
pub model: String,
pub mac: String,
pub version: u32,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub enum ControlPacket {
// 发现协议
// 服务发现
ServerHello {
udp_port: u16, // UDP 音频流端口
tcp_port: u16,
},
// 握手协议
// 握手与认证
ClientIdentify {
role: ChannelRole,
info: DeviceInfo,
},
// 时间同步 (持续进行)
Ping {
client_ts: u128,
seq: u32,
IdentifyOk,
// 音频控制
StartRecording {
config: AudioConfig,
},
Pong {
client_ts: u128,
server_ts: u128,
seq: u32,
StopRecording,
StartPlayback {
config: AudioConfig,
},
StopPlayback,
// RPC
RpcRequest {
id: u32,
method: String,
args: Vec<String>,
},
RpcResponse {
id: u32,
result: RpcResult,
},
// 心跳
Ping,
Pong,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct RpcResult {
pub stdout: String,
pub stderr: String,
pub code: i32,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct AudioPacket {
pub seq: u32, // 序列号，用于丢包检测
pub timestamp: u128, // 目标播放时间 (主节点时间)
pub data: Vec<u8>, // Opus 编码数据
pub data: Vec<u8>, // Opus 编码数据
}