Création du module pmoaudio
This commit is contained in:
241
pmoaudio/src/nodes/buffer_node.rs
Normal file
241
pmoaudio/src/nodes/buffer_node.rs
Normal file
@@ -0,0 +1,241 @@
|
||||
use crate::{AudioChunk, nodes::{AudioError, MultiSubscriberNode}};
|
||||
use std::collections::VecDeque;
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::{mpsc, RwLock};
|
||||
|
||||
/// Subscriber avec son propre offset dans le buffer
|
||||
struct BufferSubscriber {
|
||||
tx: mpsc::Sender<Arc<AudioChunk>>,
|
||||
offset: usize, // Position dans le buffer circulaire
|
||||
}
|
||||
|
||||
/// BufferNode avec buffer circulaire pour support multiroom
|
||||
///
|
||||
/// Ce node maintient un buffer circulaire de chunks et permet à plusieurs
|
||||
/// abonnés de lire avec des offsets différents, ce qui est idéal pour des
|
||||
/// configurations multiroom où différentes pièces peuvent avoir un léger
|
||||
/// délai de synchronisation.
|
||||
///
|
||||
/// # Fonctionnement
|
||||
///
|
||||
/// - Le buffer est implémenté avec un `VecDeque` de taille fixe
|
||||
/// - Chaque abonné peut avoir un offset indépendant (en nombre de chunks)
|
||||
/// - Utilise `try_send` pour éviter de bloquer si un abonné est saturé
|
||||
///
|
||||
/// # Exemples
|
||||
///
|
||||
/// ```no_run
|
||||
/// use pmoaudio::{BufferNode, SinkNode};
|
||||
///
|
||||
/// #[tokio::main]
|
||||
/// async fn main() {
|
||||
/// let (buffer, buffer_tx) = BufferNode::new(50, 10);
|
||||
///
|
||||
/// let (sink1, sink1_tx) = SinkNode::new("Room 1".to_string(), 10);
|
||||
/// let (sink2, sink2_tx) = SinkNode::new("Room 2".to_string(), 10);
|
||||
///
|
||||
/// // Room 1 sans délai
|
||||
/// buffer.add_subscriber_with_offset(sink1_tx, 0).await;
|
||||
///
|
||||
/// // Room 2 avec 5 chunks de retard
|
||||
/// buffer.add_subscriber_with_offset(sink2_tx, 5).await;
|
||||
///
|
||||
/// tokio::spawn(async move { buffer.run().await.unwrap() });
|
||||
/// // ... spawn sinks et source
|
||||
/// }
|
||||
/// ```
|
||||
pub struct BufferNode {
|
||||
buffer: Arc<RwLock<VecDeque<Arc<AudioChunk>>>>,
|
||||
subscribers: Arc<RwLock<Vec<BufferSubscriber>>>,
|
||||
buffer_size: usize,
|
||||
rx: mpsc::Receiver<Arc<AudioChunk>>,
|
||||
next_subscribers: MultiSubscriberNode, // Pour passer au node suivant
|
||||
}
|
||||
|
||||
impl BufferNode {
|
||||
/// Crée un nouveau BufferNode
|
||||
///
|
||||
/// # Arguments
|
||||
/// * `buffer_size` - Taille maximale du buffer circulaire
|
||||
/// * `channel_size` - Taille du channel bounded pour backpressure
|
||||
pub fn new(buffer_size: usize, channel_size: usize) -> (Self, mpsc::Sender<Arc<AudioChunk>>) {
|
||||
let (tx, rx) = mpsc::channel(channel_size);
|
||||
|
||||
let node = Self {
|
||||
buffer: Arc::new(RwLock::new(VecDeque::with_capacity(buffer_size))),
|
||||
subscribers: Arc::new(RwLock::new(Vec::new())),
|
||||
buffer_size,
|
||||
rx,
|
||||
next_subscribers: MultiSubscriberNode::new(),
|
||||
};
|
||||
|
||||
(node, tx)
|
||||
}
|
||||
|
||||
/// Ajoute un abonné avec un offset spécifique (pour multiroom)
|
||||
pub async fn add_subscriber_with_offset(
|
||||
&self,
|
||||
tx: mpsc::Sender<Arc<AudioChunk>>,
|
||||
offset: usize,
|
||||
) {
|
||||
let mut subs = self.subscribers.write().await;
|
||||
subs.push(BufferSubscriber { tx, offset });
|
||||
}
|
||||
|
||||
/// Ajoute un abonné sans offset (commence au chunk courant)
|
||||
pub async fn add_subscriber(&self, tx: mpsc::Sender<Arc<AudioChunk>>) {
|
||||
self.add_subscriber_with_offset(tx, 0).await;
|
||||
}
|
||||
|
||||
/// Ajoute un abonné pour le node suivant (sans buffer)
|
||||
pub fn add_next_subscriber(&mut self, tx: mpsc::Sender<Arc<AudioChunk>>) {
|
||||
self.next_subscribers.add_subscriber(tx);
|
||||
}
|
||||
|
||||
/// Démarre la boucle de traitement du BufferNode
|
||||
pub async fn run(mut self) -> Result<(), AudioError> {
|
||||
let mut chunk_index = 0usize;
|
||||
|
||||
while let Some(chunk) = self.rx.recv().await {
|
||||
// Ajouter au buffer circulaire
|
||||
{
|
||||
let mut buffer = self.buffer.write().await;
|
||||
if buffer.len() >= self.buffer_size {
|
||||
buffer.pop_front();
|
||||
}
|
||||
buffer.push_back(chunk.clone());
|
||||
}
|
||||
|
||||
// Envoyer aux abonnés avec offset
|
||||
{
|
||||
let buffer = self.buffer.read().await;
|
||||
let mut subs = self.subscribers.write().await;
|
||||
|
||||
for sub in subs.iter_mut() {
|
||||
// Calculer l'index dans le buffer en fonction de l'offset
|
||||
let target_index = if chunk_index >= sub.offset {
|
||||
chunk_index - sub.offset
|
||||
} else {
|
||||
continue; // Pas encore assez de données
|
||||
};
|
||||
|
||||
// Vérifier si le chunk est disponible dans le buffer
|
||||
let buffer_age = chunk_index - target_index;
|
||||
if buffer_age < buffer.len() {
|
||||
let chunk_to_send = &buffer[buffer.len() - buffer_age - 1];
|
||||
// try_send non-bloquant pour éviter de bloquer la source
|
||||
let _ = sub.tx.try_send(chunk_to_send.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Push vers les nodes suivants sans buffer
|
||||
self.next_subscribers.try_push(chunk).await?;
|
||||
|
||||
chunk_index += 1;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Version avec push synchrone au lieu de try_push
|
||||
pub async fn run_blocking(mut self) -> Result<(), AudioError> {
|
||||
let mut chunk_index = 0usize;
|
||||
|
||||
while let Some(chunk) = self.rx.recv().await {
|
||||
// Ajouter au buffer circulaire
|
||||
{
|
||||
let mut buffer = self.buffer.write().await;
|
||||
if buffer.len() >= self.buffer_size {
|
||||
buffer.pop_front();
|
||||
}
|
||||
buffer.push_back(chunk.clone());
|
||||
}
|
||||
|
||||
// Envoyer aux abonnés avec offset
|
||||
{
|
||||
let buffer = self.buffer.read().await;
|
||||
let subs = self.subscribers.read().await;
|
||||
|
||||
for sub in subs.iter() {
|
||||
let target_index = if chunk_index >= sub.offset {
|
||||
chunk_index - sub.offset
|
||||
} else {
|
||||
continue;
|
||||
};
|
||||
|
||||
let buffer_age = chunk_index - target_index;
|
||||
if buffer_age < buffer.len() {
|
||||
let chunk_to_send = &buffer[buffer.len() - buffer_age - 1];
|
||||
let _ = sub.tx.send(chunk_to_send.clone()).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Push vers les nodes suivants
|
||||
for _ in 0..self.next_subscribers.subscribers.len() {
|
||||
self.next_subscribers.push(chunk.clone()).await?;
|
||||
}
|
||||
|
||||
chunk_index += 1;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_buffer_node_basic() {
|
||||
let (mut node, tx) = BufferNode::new(10, 5);
|
||||
let (out_tx, mut out_rx) = mpsc::channel(5);
|
||||
|
||||
node.add_next_subscriber(out_tx);
|
||||
|
||||
// Spawn le node
|
||||
tokio::spawn(async move {
|
||||
node.run().await.unwrap();
|
||||
});
|
||||
|
||||
// Envoyer des chunks
|
||||
for i in 0..3 {
|
||||
let chunk = AudioChunk::new(i, vec![0.0; 100], vec![0.0; 100], 48000);
|
||||
tx.send(Arc::new(chunk)).await.unwrap();
|
||||
}
|
||||
|
||||
// Recevoir les chunks
|
||||
for i in 0..3 {
|
||||
let chunk = out_rx.recv().await.unwrap();
|
||||
assert_eq!(chunk.order, i);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_buffer_node_with_offset() {
|
||||
let (node, tx) = BufferNode::new(10, 10);
|
||||
let (out_tx, mut out_rx) = mpsc::channel(10);
|
||||
|
||||
// Ajouter un abonné avec offset de 2 chunks
|
||||
node.add_subscriber_with_offset(out_tx, 2).await;
|
||||
|
||||
// Spawn le node
|
||||
tokio::spawn(async move {
|
||||
node.run().await.unwrap();
|
||||
});
|
||||
|
||||
// Envoyer 5 chunks
|
||||
for i in 0..5 {
|
||||
let chunk = AudioChunk::new(i, vec![0.0; 100], vec![0.0; 100], 48000);
|
||||
tx.send(Arc::new(chunk)).await.unwrap();
|
||||
}
|
||||
|
||||
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
|
||||
|
||||
// L'abonné devrait recevoir les chunks 0, 1, 2 (avec 2 chunks de retard)
|
||||
let chunk = out_rx.try_recv().unwrap();
|
||||
assert_eq!(chunk.order, 0);
|
||||
}
|
||||
}
|
||||
147
pmoaudio/src/nodes/decoder_node.rs
Normal file
147
pmoaudio/src/nodes/decoder_node.rs
Normal file
@@ -0,0 +1,147 @@
|
||||
use crate::{AudioChunk, nodes::{AudioError, MultiSubscriberNode}};
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
/// DecoderNode - Décode des chunks audio
|
||||
///
|
||||
/// Version mock qui passe simplement les chunks (ou simule un décodage simple)
|
||||
pub struct DecoderNode {
|
||||
rx: mpsc::Receiver<Arc<AudioChunk>>,
|
||||
subscribers: MultiSubscriberNode,
|
||||
}
|
||||
|
||||
impl DecoderNode {
|
||||
pub fn new(channel_size: usize) -> (Self, mpsc::Sender<Arc<AudioChunk>>) {
|
||||
let (tx, rx) = mpsc::channel(channel_size);
|
||||
|
||||
let node = Self {
|
||||
rx,
|
||||
subscribers: MultiSubscriberNode::new(),
|
||||
};
|
||||
|
||||
(node, tx)
|
||||
}
|
||||
|
||||
pub fn add_subscriber(&mut self, tx: mpsc::Sender<Arc<AudioChunk>>) {
|
||||
self.subscribers.add_subscriber(tx);
|
||||
}
|
||||
|
||||
/// Mode passthrough - passe les chunks sans modification
|
||||
pub async fn run_passthrough(mut self) -> Result<(), AudioError> {
|
||||
while let Some(chunk) = self.rx.recv().await {
|
||||
self.subscribers.push(chunk).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Mode mock décodage - simule un changement de sample rate
|
||||
pub async fn run_with_resampling(mut self, target_sample_rate: u32) -> Result<(), AudioError> {
|
||||
while let Some(chunk) = self.rx.recv().await {
|
||||
if chunk.sample_rate == target_sample_rate {
|
||||
// Pas besoin de resampling
|
||||
self.subscribers.push(chunk).await?;
|
||||
} else {
|
||||
// Simuler un resampling (mock simple)
|
||||
let ratio = target_sample_rate as f64 / chunk.sample_rate as f64;
|
||||
let new_len = (chunk.len() as f64 * ratio) as usize;
|
||||
|
||||
let (left_data, right_data) = chunk.clone_data();
|
||||
let mut new_left = Vec::with_capacity(new_len);
|
||||
let mut new_right = Vec::with_capacity(new_len);
|
||||
|
||||
// Resampling linéaire simple (mock)
|
||||
for i in 0..new_len {
|
||||
let src_pos = i as f64 / ratio;
|
||||
let src_idx = src_pos as usize;
|
||||
|
||||
if src_idx < left_data.len() - 1 {
|
||||
let frac = src_pos - src_idx as f64;
|
||||
let left_sample =
|
||||
left_data[src_idx] * (1.0 - frac as f32) + left_data[src_idx + 1] * frac as f32;
|
||||
let right_sample =
|
||||
right_data[src_idx] * (1.0 - frac as f32) + right_data[src_idx + 1] * frac as f32;
|
||||
|
||||
new_left.push(left_sample);
|
||||
new_right.push(right_sample);
|
||||
} else if src_idx < left_data.len() {
|
||||
new_left.push(left_data[src_idx]);
|
||||
new_right.push(right_data[src_idx]);
|
||||
}
|
||||
}
|
||||
|
||||
let new_chunk = AudioChunk::new(chunk.order, new_left, new_right, target_sample_rate);
|
||||
self.subscribers.push(Arc::new(new_chunk)).await?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_decoder_passthrough() {
|
||||
let (mut node, tx) = DecoderNode::new(10);
|
||||
let (out_tx, mut out_rx) = mpsc::channel(10);
|
||||
|
||||
node.add_subscriber(out_tx);
|
||||
|
||||
tokio::spawn(async move {
|
||||
node.run_passthrough().await.unwrap();
|
||||
});
|
||||
|
||||
// Envoyer un chunk
|
||||
let chunk = AudioChunk::new(0, vec![1.0, 2.0, 3.0], vec![4.0, 5.0, 6.0], 48000);
|
||||
let chunk_arc = Arc::new(chunk);
|
||||
tx.send(chunk_arc.clone()).await.unwrap();
|
||||
|
||||
// Recevoir le chunk
|
||||
let received = out_rx.recv().await.unwrap();
|
||||
assert!(Arc::ptr_eq(&chunk_arc, &received));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_decoder_resampling() {
|
||||
let (mut node, tx) = DecoderNode::new(10);
|
||||
let (out_tx, mut out_rx) = mpsc::channel(10);
|
||||
|
||||
node.add_subscriber(out_tx);
|
||||
|
||||
tokio::spawn(async move {
|
||||
node.run_with_resampling(96000).await.unwrap();
|
||||
});
|
||||
|
||||
// Envoyer un chunk à 48000 Hz
|
||||
let chunk = AudioChunk::new(0, vec![1.0; 100], vec![1.0; 100], 48000);
|
||||
tx.send(Arc::new(chunk)).await.unwrap();
|
||||
|
||||
// Recevoir le chunk resampleé
|
||||
let received = out_rx.recv().await.unwrap();
|
||||
assert_eq!(received.sample_rate, 96000);
|
||||
// Le chunk devrait être environ 2x plus grand
|
||||
assert!(received.len() > 150 && received.len() < 250);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_decoder_no_resampling_needed() {
|
||||
let (mut node, tx) = DecoderNode::new(10);
|
||||
let (out_tx, mut out_rx) = mpsc::channel(10);
|
||||
|
||||
node.add_subscriber(out_tx);
|
||||
|
||||
tokio::spawn(async move {
|
||||
node.run_with_resampling(48000).await.unwrap();
|
||||
});
|
||||
|
||||
// Envoyer un chunk déjà au bon sample rate
|
||||
let chunk = AudioChunk::new(0, vec![1.0; 100], vec![1.0; 100], 48000);
|
||||
let chunk_arc = Arc::new(chunk);
|
||||
tx.send(chunk_arc.clone()).await.unwrap();
|
||||
|
||||
// Le chunk devrait être passé sans modification
|
||||
let received = out_rx.recv().await.unwrap();
|
||||
assert!(Arc::ptr_eq(&chunk_arc, &received));
|
||||
}
|
||||
}
|
||||
236
pmoaudio/src/nodes/dsp_node.rs
Normal file
236
pmoaudio/src/nodes/dsp_node.rs
Normal file
@@ -0,0 +1,236 @@
|
||||
use crate::{AudioChunk, nodes::{AudioError, MultiSubscriberNode}};
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
/// DspNode - Applique des transformations DSP aux chunks audio
|
||||
///
|
||||
/// Clone les données uniquement si elles doivent être modifiées
|
||||
pub struct DspNode {
|
||||
rx: mpsc::Receiver<Arc<AudioChunk>>,
|
||||
subscribers: MultiSubscriberNode,
|
||||
gain: f32,
|
||||
}
|
||||
|
||||
impl DspNode {
|
||||
pub fn new(channel_size: usize, gain: f32) -> (Self, mpsc::Sender<Arc<AudioChunk>>) {
|
||||
let (tx, rx) = mpsc::channel(channel_size);
|
||||
|
||||
let node = Self {
|
||||
rx,
|
||||
subscribers: MultiSubscriberNode::new(),
|
||||
gain,
|
||||
};
|
||||
|
||||
(node, tx)
|
||||
}
|
||||
|
||||
pub fn add_subscriber(&mut self, tx: mpsc::Sender<Arc<AudioChunk>>) {
|
||||
self.subscribers.add_subscriber(tx);
|
||||
}
|
||||
|
||||
/// Applique le gain aux chunks
|
||||
pub async fn run(mut self) -> Result<(), AudioError> {
|
||||
while let Some(chunk) = self.rx.recv().await {
|
||||
if (self.gain - 1.0).abs() < f32::EPSILON {
|
||||
// Gain = 1.0, pas de transformation nécessaire
|
||||
self.subscribers.push(chunk).await?;
|
||||
} else {
|
||||
// Clone les données pour les modifier
|
||||
let (mut left_data, mut right_data) = chunk.clone_data();
|
||||
|
||||
// Appliquer le gain
|
||||
for sample in &mut left_data {
|
||||
*sample *= self.gain;
|
||||
}
|
||||
for sample in &mut right_data {
|
||||
*sample *= self.gain;
|
||||
}
|
||||
|
||||
let new_chunk = AudioChunk::new(
|
||||
chunk.order,
|
||||
left_data,
|
||||
right_data,
|
||||
chunk.sample_rate,
|
||||
);
|
||||
|
||||
self.subscribers.push(Arc::new(new_chunk)).await?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Met à jour le gain dynamiquement (nécessite un `Arc<RwLock<f32>>` dans une version réelle)
|
||||
pub fn set_gain(&mut self, gain: f32) {
|
||||
self.gain = gain;
|
||||
}
|
||||
}
|
||||
|
||||
/// DspNode avec filtre passe-bas simple (mock)
|
||||
#[allow(dead_code)]
|
||||
pub struct LowPassDspNode {
|
||||
rx: mpsc::Receiver<Arc<AudioChunk>>,
|
||||
subscribers: MultiSubscriberNode,
|
||||
alpha: f32, // Coefficient du filtre
|
||||
prev_left: f32,
|
||||
prev_right: f32,
|
||||
}
|
||||
|
||||
impl LowPassDspNode {
|
||||
#[allow(dead_code)]
|
||||
pub fn new(channel_size: usize, cutoff_ratio: f32) -> (Self, mpsc::Sender<Arc<AudioChunk>>) {
|
||||
let (tx, rx) = mpsc::channel(channel_size);
|
||||
|
||||
// Filtre RC simple: alpha = dt / (RC + dt)
|
||||
// cutoff_ratio entre 0 (tout couper) et 1 (tout passer)
|
||||
let alpha = cutoff_ratio.clamp(0.0, 1.0);
|
||||
|
||||
let node = Self {
|
||||
rx,
|
||||
subscribers: MultiSubscriberNode::new(),
|
||||
alpha,
|
||||
prev_left: 0.0,
|
||||
prev_right: 0.0,
|
||||
};
|
||||
|
||||
(node, tx)
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub fn add_subscriber(&mut self, tx: mpsc::Sender<Arc<AudioChunk>>) {
|
||||
self.subscribers.add_subscriber(tx);
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub async fn run(mut self) -> Result<(), AudioError> {
|
||||
while let Some(chunk) = self.rx.recv().await {
|
||||
let (left_data, right_data) = chunk.clone_data();
|
||||
let mut new_left = Vec::with_capacity(left_data.len());
|
||||
let mut new_right = Vec::with_capacity(right_data.len());
|
||||
|
||||
// Appliquer le filtre
|
||||
for &sample in &left_data {
|
||||
self.prev_left = self.prev_left + self.alpha * (sample - self.prev_left);
|
||||
new_left.push(self.prev_left);
|
||||
}
|
||||
|
||||
for &sample in &right_data {
|
||||
self.prev_right = self.prev_right + self.alpha * (sample - self.prev_right);
|
||||
new_right.push(self.prev_right);
|
||||
}
|
||||
|
||||
let new_chunk = AudioChunk::new(
|
||||
chunk.order,
|
||||
new_left,
|
||||
new_right,
|
||||
chunk.sample_rate,
|
||||
);
|
||||
|
||||
self.subscribers.push(Arc::new(new_chunk)).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_dsp_node_unity_gain() {
|
||||
let (mut node, tx) = DspNode::new(10, 1.0);
|
||||
let (out_tx, mut out_rx) = mpsc::channel(10);
|
||||
|
||||
node.add_subscriber(out_tx);
|
||||
|
||||
tokio::spawn(async move {
|
||||
node.run().await.unwrap();
|
||||
});
|
||||
|
||||
// Envoyer un chunk
|
||||
let chunk = AudioChunk::new(0, vec![1.0, 2.0, 3.0], vec![4.0, 5.0, 6.0], 48000);
|
||||
let chunk_arc = Arc::new(chunk);
|
||||
tx.send(chunk_arc.clone()).await.unwrap();
|
||||
|
||||
// Avec gain = 1.0, le chunk ne devrait pas être cloné
|
||||
let received = out_rx.recv().await.unwrap();
|
||||
assert!(Arc::ptr_eq(&chunk_arc, &received));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_dsp_node_gain() {
|
||||
let (mut node, tx) = DspNode::new(10, 2.0);
|
||||
let (out_tx, mut out_rx) = mpsc::channel(10);
|
||||
|
||||
node.add_subscriber(out_tx);
|
||||
|
||||
tokio::spawn(async move {
|
||||
node.run().await.unwrap();
|
||||
});
|
||||
|
||||
// Envoyer un chunk
|
||||
let chunk = AudioChunk::new(0, vec![1.0, 2.0, 3.0], vec![4.0, 5.0, 6.0], 48000);
|
||||
tx.send(Arc::new(chunk)).await.unwrap();
|
||||
|
||||
// Vérifier que le gain a été appliqué
|
||||
let received = out_rx.recv().await.unwrap();
|
||||
assert_eq!(received.left[0], 2.0);
|
||||
assert_eq!(received.left[1], 4.0);
|
||||
assert_eq!(received.left[2], 6.0);
|
||||
assert_eq!(received.right[0], 8.0);
|
||||
assert_eq!(received.right[1], 10.0);
|
||||
assert_eq!(received.right[2], 12.0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_lowpass_dsp_node() {
|
||||
let (mut node, tx) = LowPassDspNode::new(10, 0.5);
|
||||
let (out_tx, mut out_rx) = mpsc::channel(10);
|
||||
|
||||
node.add_subscriber(out_tx);
|
||||
|
||||
tokio::spawn(async move {
|
||||
node.run().await.unwrap();
|
||||
});
|
||||
|
||||
// Envoyer un chunk avec un signal carré
|
||||
let chunk = AudioChunk::new(
|
||||
0,
|
||||
vec![1.0, 1.0, 1.0, -1.0, -1.0, -1.0],
|
||||
vec![1.0, 1.0, 1.0, -1.0, -1.0, -1.0],
|
||||
48000,
|
||||
);
|
||||
tx.send(Arc::new(chunk)).await.unwrap();
|
||||
|
||||
// Le filtre devrait lisser le signal
|
||||
let received = out_rx.recv().await.unwrap();
|
||||
|
||||
// Vérifier que le signal est lissé (valeurs intermédiaires)
|
||||
assert!(received.left[0].abs() < 1.0); // Premier échantillon lissé
|
||||
assert!(received.left[2].abs() < 1.0); // Signal ne devrait pas atteindre 1.0 immédiatement
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_dsp_node_multiple_subscribers() {
|
||||
let (mut node, tx) = DspNode::new(10, 0.5);
|
||||
let (out_tx1, mut out_rx1) = mpsc::channel(10);
|
||||
let (out_tx2, mut out_rx2) = mpsc::channel(10);
|
||||
|
||||
node.add_subscriber(out_tx1);
|
||||
node.add_subscriber(out_tx2);
|
||||
|
||||
tokio::spawn(async move {
|
||||
node.run().await.unwrap();
|
||||
});
|
||||
|
||||
let chunk = AudioChunk::new(0, vec![2.0, 4.0], vec![2.0, 4.0], 48000);
|
||||
tx.send(Arc::new(chunk)).await.unwrap();
|
||||
|
||||
// Les deux abonnés devraient recevoir le même Arc
|
||||
let received1 = out_rx1.recv().await.unwrap();
|
||||
let received2 = out_rx2.recv().await.unwrap();
|
||||
|
||||
assert!(Arc::ptr_eq(&received1, &received2));
|
||||
assert_eq!(received1.left[0], 1.0); // 2.0 * 0.5
|
||||
assert_eq!(received1.left[1], 2.0); // 4.0 * 0.5
|
||||
}
|
||||
}
|
||||
146
pmoaudio/src/nodes/mod.rs
Normal file
146
pmoaudio/src/nodes/mod.rs
Normal file
@@ -0,0 +1,146 @@
|
||||
//! Nodes du pipeline audio
|
||||
//!
|
||||
//! Ce module contient tous les types de nodes disponibles pour construire
|
||||
//! un pipeline audio, ainsi que les traits et structures de support.
|
||||
|
||||
use crate::AudioChunk;
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
pub mod buffer_node;
|
||||
pub mod decoder_node;
|
||||
pub mod dsp_node;
|
||||
pub mod sink_node;
|
||||
pub mod source_node;
|
||||
pub mod timer_node;
|
||||
|
||||
/// Trait de base pour tous les nodes audio
|
||||
///
|
||||
/// Tous les nodes du pipeline implémentent ce trait pour permettre
|
||||
/// une interface uniforme de traitement des chunks audio.
|
||||
#[async_trait::async_trait]
|
||||
pub trait AudioNode: Send + Sync {
|
||||
/// Push un chunk vers ce node
|
||||
///
|
||||
/// # Erreurs
|
||||
///
|
||||
/// Retourne `AudioError::SendError` si l'envoi échoue
|
||||
async fn push(&mut self, chunk: Arc<AudioChunk>) -> Result<(), AudioError>;
|
||||
|
||||
/// Ferme le node proprement
|
||||
async fn close(&mut self);
|
||||
}
|
||||
|
||||
/// Node avec un seul abonné (pas de clone inutile)
|
||||
///
|
||||
/// Optimisé pour les cas où un node n'a qu'un seul destinataire.
|
||||
/// Le Arc du chunk est simplement transféré sans clonage supplémentaire.
|
||||
///
|
||||
/// # Exemples
|
||||
///
|
||||
/// ```
|
||||
/// use pmoaudio::SingleSubscriberNode;
|
||||
/// use tokio::sync::mpsc;
|
||||
///
|
||||
/// let (tx, rx) = mpsc::channel(10);
|
||||
/// let node = SingleSubscriberNode::new(tx);
|
||||
/// ```
|
||||
pub struct SingleSubscriberNode {
|
||||
tx: mpsc::Sender<Arc<AudioChunk>>,
|
||||
}
|
||||
|
||||
impl SingleSubscriberNode {
|
||||
pub fn new(tx: mpsc::Sender<Arc<AudioChunk>>) -> Self {
|
||||
Self { tx }
|
||||
}
|
||||
|
||||
pub async fn push(&self, chunk: Arc<AudioChunk>) -> Result<(), AudioError> {
|
||||
self.tx
|
||||
.send(chunk)
|
||||
.await
|
||||
.map_err(|_| AudioError::SendError)
|
||||
}
|
||||
}
|
||||
|
||||
/// Node avec plusieurs abonnés (partage le même Arc)
|
||||
///
|
||||
/// Permet de broadcaster un chunk à plusieurs destinations.
|
||||
/// Tous les abonnés reçoivent le même `Arc<AudioChunk>`, donc pas de copie
|
||||
/// des données audio - seul le compteur de référence Arc est incrémenté.
|
||||
///
|
||||
/// # Exemples
|
||||
///
|
||||
/// ```
|
||||
/// use pmoaudio::MultiSubscriberNode;
|
||||
/// use tokio::sync::mpsc;
|
||||
///
|
||||
/// let mut node = MultiSubscriberNode::new();
|
||||
/// let (tx1, rx1) = mpsc::channel(10);
|
||||
/// let (tx2, rx2) = mpsc::channel(10);
|
||||
///
|
||||
/// node.add_subscriber(tx1);
|
||||
/// node.add_subscriber(tx2);
|
||||
/// // Les deux abonnés recevront les mêmes chunks
|
||||
/// ```
|
||||
pub struct MultiSubscriberNode {
|
||||
subscribers: Vec<mpsc::Sender<Arc<AudioChunk>>>,
|
||||
}
|
||||
|
||||
impl MultiSubscriberNode {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
subscribers: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn add_subscriber(&mut self, tx: mpsc::Sender<Arc<AudioChunk>>) {
|
||||
self.subscribers.push(tx);
|
||||
}
|
||||
|
||||
pub async fn push(&self, chunk: Arc<AudioChunk>) -> Result<(), AudioError> {
|
||||
for tx in &self.subscribers {
|
||||
// On partage le même Arc avec tous les abonnés
|
||||
tx.send(chunk.clone())
|
||||
.await
|
||||
.map_err(|_| AudioError::SendError)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn try_push(&self, chunk: Arc<AudioChunk>) -> Result<(), AudioError> {
|
||||
for tx in &self.subscribers {
|
||||
// try_send non-bloquant, ignore si saturé
|
||||
let _ = tx.try_send(chunk.clone());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for MultiSubscriberNode {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
/// Erreurs possibles dans le pipeline audio
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum AudioError {
|
||||
/// Échec d'envoi d'un chunk à travers un channel
|
||||
SendError,
|
||||
/// Échec de réception d'un chunk depuis un channel
|
||||
ReceiveError,
|
||||
/// Erreur de traitement avec message descriptif
|
||||
ProcessingError(String),
|
||||
}
|
||||
|
||||
impl std::fmt::Display for AudioError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
AudioError::SendError => write!(f, "Failed to send audio chunk"),
|
||||
AudioError::ReceiveError => write!(f, "Failed to receive audio chunk"),
|
||||
AudioError::ProcessingError(msg) => write!(f, "Processing error: {}", msg),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for AudioError {}
|
||||
199
pmoaudio/src/nodes/sink_node.rs
Normal file
199
pmoaudio/src/nodes/sink_node.rs
Normal file
@@ -0,0 +1,199 @@
|
||||
use crate::{AudioChunk, nodes::AudioError};
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
/// SinkNode - Node terminal qui consomme les chunks audio
|
||||
///
|
||||
/// Version mock pour tests et logging
|
||||
pub struct SinkNode {
|
||||
rx: mpsc::Receiver<Arc<AudioChunk>>,
|
||||
name: String,
|
||||
}
|
||||
|
||||
impl SinkNode {
|
||||
pub fn new(name: String, channel_size: usize) -> (Self, mpsc::Sender<Arc<AudioChunk>>) {
|
||||
let (tx, rx) = mpsc::channel(channel_size);
|
||||
|
||||
let node = Self { rx, name };
|
||||
|
||||
(node, tx)
|
||||
}
|
||||
|
||||
/// Version silencieuse - consomme les chunks sans action
|
||||
pub async fn run_silent(mut self) -> Result<(), AudioError> {
|
||||
while let Some(_chunk) = self.rx.recv().await {
|
||||
// Ne rien faire, juste consommer
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Version avec logging
|
||||
pub async fn run_with_logging(mut self) -> Result<(), AudioError> {
|
||||
while let Some(chunk) = self.rx.recv().await {
|
||||
println!(
|
||||
"[{}] Received chunk #{} - {} samples @ {} Hz",
|
||||
self.name,
|
||||
chunk.order,
|
||||
chunk.len(),
|
||||
chunk.sample_rate
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Version avec statistiques
|
||||
pub async fn run_with_stats(mut self) -> Result<SinkStats, AudioError> {
|
||||
let mut stats = SinkStats::new(self.name.clone());
|
||||
|
||||
while let Some(chunk) = self.rx.recv().await {
|
||||
stats.process_chunk(&chunk);
|
||||
}
|
||||
|
||||
Ok(stats)
|
||||
}
|
||||
|
||||
/// Version mock pour écriture dans un fichier (simule l'écriture)
|
||||
pub async fn run_mock_file_writer(mut self) -> Result<usize, AudioError> {
|
||||
let mut total_samples = 0;
|
||||
|
||||
while let Some(chunk) = self.rx.recv().await {
|
||||
total_samples += chunk.len();
|
||||
// Simuler l'écriture avec un petit délai
|
||||
tokio::time::sleep(tokio::time::Duration::from_micros(10)).await;
|
||||
}
|
||||
|
||||
Ok(total_samples)
|
||||
}
|
||||
}
|
||||
|
||||
/// Statistiques collectées par un SinkNode
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct SinkStats {
|
||||
pub name: String,
|
||||
pub chunks_received: u64,
|
||||
pub total_samples: u64,
|
||||
pub total_duration_sec: f64,
|
||||
pub peak_left: f32,
|
||||
pub peak_right: f32,
|
||||
pub rms_left: f64,
|
||||
pub rms_right: f64,
|
||||
}
|
||||
|
||||
impl SinkStats {
|
||||
pub fn new(name: String) -> Self {
|
||||
Self {
|
||||
name,
|
||||
chunks_received: 0,
|
||||
total_samples: 0,
|
||||
total_duration_sec: 0.0,
|
||||
peak_left: 0.0,
|
||||
peak_right: 0.0,
|
||||
rms_left: 0.0,
|
||||
rms_right: 0.0,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn process_chunk(&mut self, chunk: &AudioChunk) {
|
||||
self.chunks_received += 1;
|
||||
self.total_samples += chunk.len() as u64;
|
||||
self.total_duration_sec += chunk.len() as f64 / chunk.sample_rate as f64;
|
||||
|
||||
// Calculer les peaks
|
||||
for &sample in chunk.left.iter() {
|
||||
if sample.abs() > self.peak_left {
|
||||
self.peak_left = sample.abs();
|
||||
}
|
||||
}
|
||||
|
||||
for &sample in chunk.right.iter() {
|
||||
if sample.abs() > self.peak_right {
|
||||
self.peak_right = sample.abs();
|
||||
}
|
||||
}
|
||||
|
||||
// Calculer RMS (moyenne des carrés)
|
||||
let sum_squares_left: f64 = chunk.left.iter().map(|&x| (x * x) as f64).sum();
|
||||
let sum_squares_right: f64 = chunk.right.iter().map(|&x| (x * x) as f64).sum();
|
||||
|
||||
self.rms_left = ((self.rms_left.powi(2) * (self.total_samples - chunk.len() as u64) as f64
|
||||
+ sum_squares_left)
|
||||
/ self.total_samples as f64)
|
||||
.sqrt();
|
||||
self.rms_right = ((self.rms_right.powi(2) * (self.total_samples - chunk.len() as u64) as f64
|
||||
+ sum_squares_right)
|
||||
/ self.total_samples as f64)
|
||||
.sqrt();
|
||||
}
|
||||
|
||||
pub fn display(&self) {
|
||||
println!("\n=== Sink Statistics: {} ===", self.name);
|
||||
println!("Chunks received: {}", self.chunks_received);
|
||||
println!("Total samples: {}", self.total_samples);
|
||||
println!("Total duration: {:.3} sec", self.total_duration_sec);
|
||||
println!("Peak L/R: {:.3} / {:.3}", self.peak_left, self.peak_right);
|
||||
println!("RMS L/R: {:.3} / {:.3}", self.rms_left, self.rms_right);
|
||||
println!("========================\n");
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_sink_node_silent() {
|
||||
let (node, tx) = SinkNode::new("test".to_string(), 10);
|
||||
|
||||
let handle = tokio::spawn(async move { node.run_silent().await });
|
||||
|
||||
// Envoyer quelques chunks
|
||||
for i in 0..3 {
|
||||
let chunk = AudioChunk::new(i, vec![0.0; 100], vec![0.0; 100], 48000);
|
||||
tx.send(Arc::new(chunk)).await.unwrap();
|
||||
}
|
||||
|
||||
drop(tx);
|
||||
handle.await.unwrap().unwrap();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_sink_node_stats() {
|
||||
let (node, tx) = SinkNode::new("test".to_string(), 10);
|
||||
|
||||
let handle = tokio::spawn(async move { node.run_with_stats().await });
|
||||
|
||||
// Envoyer des chunks avec signal connu
|
||||
for i in 0..3 {
|
||||
let chunk = AudioChunk::new(i, vec![1.0; 1000], vec![0.5; 1000], 48000);
|
||||
tx.send(Arc::new(chunk)).await.unwrap();
|
||||
}
|
||||
|
||||
drop(tx);
|
||||
let stats = handle.await.unwrap().unwrap();
|
||||
|
||||
assert_eq!(stats.chunks_received, 3);
|
||||
assert_eq!(stats.total_samples, 3000);
|
||||
assert_eq!(stats.peak_left, 1.0);
|
||||
assert_eq!(stats.peak_right, 0.5);
|
||||
assert!((stats.rms_left - 1.0).abs() < 0.001);
|
||||
assert!((stats.rms_right - 0.5).abs() < 0.001);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_sink_node_file_writer() {
|
||||
let (node, tx) = SinkNode::new("writer".to_string(), 10);
|
||||
|
||||
let handle = tokio::spawn(async move { node.run_mock_file_writer().await });
|
||||
|
||||
// Envoyer des chunks
|
||||
for i in 0..5 {
|
||||
let chunk = AudioChunk::new(i, vec![0.0; 100], vec![0.0; 100], 48000);
|
||||
tx.send(Arc::new(chunk)).await.unwrap();
|
||||
}
|
||||
|
||||
drop(tx);
|
||||
let total_samples = handle.await.unwrap().unwrap();
|
||||
|
||||
assert_eq!(total_samples, 500);
|
||||
}
|
||||
}
|
||||
169
pmoaudio/src/nodes/source_node.rs
Normal file
169
pmoaudio/src/nodes/source_node.rs
Normal file
@@ -0,0 +1,169 @@
|
||||
use crate::{AudioChunk, nodes::{AudioError, MultiSubscriberNode}};
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
/// SourceNode - Génère ou lit des chunks audio depuis une source
|
||||
///
|
||||
/// Ce node est la source du pipeline. Version mock pour tests.
|
||||
pub struct SourceNode {
|
||||
subscribers: MultiSubscriberNode,
|
||||
}
|
||||
|
||||
impl SourceNode {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
subscribers: MultiSubscriberNode::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn add_subscriber(&mut self, tx: mpsc::Sender<Arc<AudioChunk>>) {
|
||||
self.subscribers.add_subscriber(tx);
|
||||
}
|
||||
|
||||
/// Génère un chunk de test avec une forme d'onde sinusoïdale
|
||||
pub fn generate_test_chunk(
|
||||
order: u64,
|
||||
size: usize,
|
||||
sample_rate: u32,
|
||||
frequency: f32,
|
||||
) -> AudioChunk {
|
||||
let mut left = Vec::with_capacity(size);
|
||||
let mut right = Vec::with_capacity(size);
|
||||
|
||||
for i in 0..size {
|
||||
let t = (order * size as u64 + i as u64) as f32 / sample_rate as f32;
|
||||
let sample = (2.0 * std::f32::consts::PI * frequency * t).sin();
|
||||
left.push(sample);
|
||||
right.push(sample * 0.8); // Légèrement différent pour la stéréo
|
||||
}
|
||||
|
||||
AudioChunk::new(order, left, right, sample_rate)
|
||||
}
|
||||
|
||||
/// Génère et envoie des chunks de test
|
||||
pub async fn generate_chunks(
|
||||
&self,
|
||||
count: u64,
|
||||
chunk_size: usize,
|
||||
sample_rate: u32,
|
||||
frequency: f32,
|
||||
) -> Result<(), AudioError> {
|
||||
for i in 0..count {
|
||||
let chunk = Self::generate_test_chunk(i, chunk_size, sample_rate, frequency);
|
||||
self.subscribers.push(Arc::new(chunk)).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Génère des chunks silencieux
|
||||
pub async fn generate_silence(
|
||||
&self,
|
||||
count: u64,
|
||||
chunk_size: usize,
|
||||
sample_rate: u32,
|
||||
) -> Result<(), AudioError> {
|
||||
for i in 0..count {
|
||||
let chunk = AudioChunk::new(
|
||||
i,
|
||||
vec![0.0; chunk_size],
|
||||
vec![0.0; chunk_size],
|
||||
sample_rate,
|
||||
);
|
||||
self.subscribers.push(Arc::new(chunk)).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Version streaming : génère des chunks continuellement avec délai
|
||||
pub async fn stream_chunks(
|
||||
&self,
|
||||
chunk_size: usize,
|
||||
sample_rate: u32,
|
||||
frequency: f32,
|
||||
duration_ms: u64,
|
||||
) -> Result<(), AudioError> {
|
||||
let chunk_duration_ms = (chunk_size as f64 / sample_rate as f64 * 1000.0) as u64;
|
||||
let mut order = 0u64;
|
||||
|
||||
let start = tokio::time::Instant::now();
|
||||
let duration = tokio::time::Duration::from_millis(duration_ms);
|
||||
|
||||
while start.elapsed() < duration {
|
||||
let chunk = Self::generate_test_chunk(order, chunk_size, sample_rate, frequency);
|
||||
self.subscribers.push(Arc::new(chunk)).await?;
|
||||
|
||||
order += 1;
|
||||
|
||||
// Attendre pour simuler le timing réel
|
||||
tokio::time::sleep(tokio::time::Duration::from_millis(chunk_duration_ms)).await;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for SourceNode {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_source_node_generation() {
|
||||
let mut source = SourceNode::new();
|
||||
let (tx, mut rx) = mpsc::channel(10);
|
||||
|
||||
source.add_subscriber(tx);
|
||||
|
||||
// Générer 3 chunks
|
||||
source.generate_chunks(3, 100, 48000, 440.0).await.unwrap();
|
||||
|
||||
// Vérifier la réception
|
||||
for i in 0..3 {
|
||||
let chunk = rx.recv().await.unwrap();
|
||||
assert_eq!(chunk.order, i);
|
||||
assert_eq!(chunk.len(), 100);
|
||||
assert_eq!(chunk.sample_rate, 48000);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_sine_wave_generation() {
|
||||
let chunk = SourceNode::generate_test_chunk(0, 48000, 48000, 440.0);
|
||||
|
||||
// Vérifier qu'on a bien une sinusoïde
|
||||
// À 440 Hz avec 48000 samples/s, on devrait avoir 440 cycles
|
||||
let left = &*chunk.left;
|
||||
|
||||
// Trouver les passages par zéro
|
||||
let mut zero_crossings = 0;
|
||||
for i in 1..left.len() {
|
||||
if (left[i - 1] < 0.0 && left[i] >= 0.0) || (left[i - 1] >= 0.0 && left[i] < 0.0) {
|
||||
zero_crossings += 1;
|
||||
}
|
||||
}
|
||||
|
||||
// 440 cycles = 880 passages par zéro (approximativement)
|
||||
assert!(zero_crossings > 850 && zero_crossings < 910);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_source_node_silence() {
|
||||
let mut source = SourceNode::new();
|
||||
let (tx, mut rx) = mpsc::channel(10);
|
||||
|
||||
source.add_subscriber(tx);
|
||||
|
||||
source.generate_silence(2, 100, 48000).await.unwrap();
|
||||
|
||||
for _ in 0..2 {
|
||||
let chunk = rx.recv().await.unwrap();
|
||||
assert!(chunk.left.iter().all(|&x| x == 0.0));
|
||||
assert!(chunk.right.iter().all(|&x| x == 0.0));
|
||||
}
|
||||
}
|
||||
}
|
||||
281
pmoaudio/src/nodes/timer_node.rs
Normal file
281
pmoaudio/src/nodes/timer_node.rs
Normal file
@@ -0,0 +1,281 @@
|
||||
use crate::{AudioChunk, nodes::{AudioError, MultiSubscriberNode}};
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::{mpsc, RwLock};
|
||||
|
||||
/// TimerNode - Node passthrough qui calcule la position temporelle
|
||||
///
|
||||
/// Ce node ne modifie pas les données audio, il les passe directement
|
||||
/// aux abonnés tout en maintenant un compteur de samples pour calculer
|
||||
/// la position en secondes.
|
||||
///
|
||||
/// # Fonctionnement
|
||||
///
|
||||
/// Pour chaque chunk reçu:
|
||||
/// 1. Incrémente `elapsed_samples += chunk.len()`
|
||||
/// 2. Calcule `position_sec = elapsed_samples / sample_rate`
|
||||
/// 3. Push le chunk (sans modification) vers les abonnés
|
||||
///
|
||||
/// # Utilisation
|
||||
///
|
||||
/// Le TimerNode fournit un [`TimerHandle`] qui permet de lire la position
|
||||
/// depuis d'autres threads/tasks sans bloquer le pipeline.
|
||||
///
|
||||
/// # Exemples
|
||||
///
|
||||
/// ```no_run
|
||||
/// use pmoaudio::TimerNode;
|
||||
///
|
||||
/// #[tokio::main]
|
||||
/// async fn main() {
|
||||
/// let (mut timer, timer_tx) = TimerNode::new(10);
|
||||
/// let handle = timer.get_position_handle();
|
||||
///
|
||||
/// tokio::spawn(async move {
|
||||
/// timer.run().await.unwrap();
|
||||
/// });
|
||||
///
|
||||
/// // Lire la position depuis un autre thread
|
||||
/// let position = handle.position_sec().await;
|
||||
/// println!("Position: {:.2} sec", position);
|
||||
/// }
|
||||
/// ```
|
||||
pub struct TimerNode {
|
||||
rx: mpsc::Receiver<Arc<AudioChunk>>,
|
||||
subscribers: MultiSubscriberNode,
|
||||
elapsed_samples: Arc<RwLock<u64>>,
|
||||
current_sample_rate: Arc<RwLock<u32>>,
|
||||
}
|
||||
|
||||
impl TimerNode {
|
||||
/// Crée un nouveau TimerNode
|
||||
pub fn new(channel_size: usize) -> (Self, mpsc::Sender<Arc<AudioChunk>>) {
|
||||
let (tx, rx) = mpsc::channel(channel_size);
|
||||
|
||||
let node = Self {
|
||||
rx,
|
||||
subscribers: MultiSubscriberNode::new(),
|
||||
elapsed_samples: Arc::new(RwLock::new(0)),
|
||||
current_sample_rate: Arc::new(RwLock::new(48000)), // Default
|
||||
};
|
||||
|
||||
(node, tx)
|
||||
}
|
||||
|
||||
/// Ajoute un abonné
|
||||
pub fn add_subscriber(&mut self, tx: mpsc::Sender<Arc<AudioChunk>>) {
|
||||
self.subscribers.add_subscriber(tx);
|
||||
}
|
||||
|
||||
/// Retourne la position actuelle en secondes
|
||||
pub async fn position_sec(&self) -> f64 {
|
||||
let elapsed = *self.elapsed_samples.read().await;
|
||||
let sample_rate = *self.current_sample_rate.read().await;
|
||||
elapsed as f64 / sample_rate as f64
|
||||
}
|
||||
|
||||
/// Retourne le nombre total d'échantillons écoulés
|
||||
pub async fn elapsed_samples(&self) -> u64 {
|
||||
*self.elapsed_samples.read().await
|
||||
}
|
||||
|
||||
/// Reset le compteur
|
||||
pub async fn reset(&self) {
|
||||
let mut elapsed = self.elapsed_samples.write().await;
|
||||
*elapsed = 0;
|
||||
}
|
||||
|
||||
/// Démarre la boucle de traitement du TimerNode
|
||||
pub async fn run(mut self) -> Result<(), AudioError> {
|
||||
while let Some(chunk) = self.rx.recv().await {
|
||||
// Mettre à jour le sample rate si nécessaire
|
||||
{
|
||||
let mut sr = self.current_sample_rate.write().await;
|
||||
if *sr != chunk.sample_rate {
|
||||
*sr = chunk.sample_rate;
|
||||
}
|
||||
}
|
||||
|
||||
// Incrémenter le compteur d'échantillons
|
||||
{
|
||||
let mut elapsed = self.elapsed_samples.write().await;
|
||||
*elapsed += chunk.len() as u64;
|
||||
}
|
||||
|
||||
// Push immédiatement le même chunk vers les abonnés (passthrough)
|
||||
self.subscribers.push(chunk).await?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Version non-bloquante avec try_push
|
||||
pub async fn run_nonblocking(mut self) -> Result<(), AudioError> {
|
||||
while let Some(chunk) = self.rx.recv().await {
|
||||
{
|
||||
let mut sr = self.current_sample_rate.write().await;
|
||||
if *sr != chunk.sample_rate {
|
||||
*sr = chunk.sample_rate;
|
||||
}
|
||||
}
|
||||
|
||||
{
|
||||
let mut elapsed = self.elapsed_samples.write().await;
|
||||
*elapsed += chunk.len() as u64;
|
||||
}
|
||||
|
||||
self.subscribers.try_push(chunk).await?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Retourne un handle pour lire la position depuis d'autres threads
|
||||
pub fn get_position_handle(&self) -> TimerHandle {
|
||||
TimerHandle {
|
||||
elapsed_samples: self.elapsed_samples.clone(),
|
||||
current_sample_rate: self.current_sample_rate.clone(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle pour lire la position du TimerNode depuis d'autres threads
|
||||
///
|
||||
/// Ce handle peut être cloné et utilisé depuis plusieurs threads/tasks
|
||||
/// pour monitorer la position de lecture sans bloquer le pipeline.
|
||||
///
|
||||
/// # Exemples
|
||||
///
|
||||
/// ```no_run
|
||||
/// use pmoaudio::TimerNode;
|
||||
///
|
||||
/// #[tokio::main]
|
||||
/// async fn main() {
|
||||
/// let (mut timer, _tx) = TimerNode::new(10);
|
||||
/// let handle = timer.get_position_handle();
|
||||
/// let handle_clone = handle.clone();
|
||||
///
|
||||
/// // Utiliser depuis plusieurs tasks
|
||||
/// tokio::spawn(async move {
|
||||
/// loop {
|
||||
/// let pos = handle_clone.position_sec().await;
|
||||
/// println!("Position: {:.2}s", pos);
|
||||
/// tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
|
||||
/// }
|
||||
/// });
|
||||
/// }
|
||||
/// ```
|
||||
#[derive(Clone)]
|
||||
pub struct TimerHandle {
|
||||
elapsed_samples: Arc<RwLock<u64>>,
|
||||
current_sample_rate: Arc<RwLock<u32>>,
|
||||
}
|
||||
|
||||
impl TimerHandle {
|
||||
/// Retourne la position actuelle en secondes
|
||||
pub async fn position_sec(&self) -> f64 {
|
||||
let elapsed = *self.elapsed_samples.read().await;
|
||||
let sample_rate = *self.current_sample_rate.read().await;
|
||||
elapsed as f64 / sample_rate as f64
|
||||
}
|
||||
|
||||
/// Retourne le nombre total d'échantillons écoulés
|
||||
pub async fn elapsed_samples(&self) -> u64 {
|
||||
*self.elapsed_samples.read().await
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_timer_node_position_calculation() {
|
||||
let (mut node, tx) = TimerNode::new(10);
|
||||
let (out_tx, mut out_rx) = mpsc::channel(10);
|
||||
|
||||
node.add_subscriber(out_tx);
|
||||
|
||||
let handle = node.get_position_handle();
|
||||
|
||||
// Spawn le node
|
||||
tokio::spawn(async move {
|
||||
node.run().await.unwrap();
|
||||
});
|
||||
|
||||
// Envoyer 3 chunks de 1000 samples à 48000 Hz
|
||||
for i in 0..3 {
|
||||
let chunk = AudioChunk::new(i, vec![0.0; 1000], vec![0.0; 1000], 48000);
|
||||
tx.send(Arc::new(chunk)).await.unwrap();
|
||||
}
|
||||
|
||||
// Attendre que les chunks soient traités
|
||||
for _ in 0..3 {
|
||||
out_rx.recv().await.unwrap();
|
||||
}
|
||||
|
||||
// Vérifier la position
|
||||
let position = handle.position_sec().await;
|
||||
let expected = 3000.0 / 48000.0; // 3 chunks * 1000 samples / 48000 Hz
|
||||
assert!((position - expected).abs() < 0.0001);
|
||||
|
||||
let elapsed = handle.elapsed_samples().await;
|
||||
assert_eq!(elapsed, 3000);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_timer_node_passthrough() {
|
||||
let (mut node, tx) = TimerNode::new(10);
|
||||
let (out_tx, mut out_rx) = mpsc::channel(10);
|
||||
|
||||
node.add_subscriber(out_tx);
|
||||
|
||||
tokio::spawn(async move {
|
||||
node.run().await.unwrap();
|
||||
});
|
||||
|
||||
// Envoyer un chunk
|
||||
let chunk = AudioChunk::new(42, vec![1.0, 2.0, 3.0], vec![4.0, 5.0, 6.0], 48000);
|
||||
let chunk_arc = Arc::new(chunk);
|
||||
tx.send(chunk_arc.clone()).await.unwrap();
|
||||
|
||||
// Recevoir le chunk
|
||||
let received = out_rx.recv().await.unwrap();
|
||||
|
||||
// Vérifier que c'est le même Arc (pas de clone des données)
|
||||
assert!(Arc::ptr_eq(&chunk_arc, &received));
|
||||
assert_eq!(received.order, 42);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_timer_node_sample_rate_change() {
|
||||
let (mut node, tx) = TimerNode::new(10);
|
||||
let (out_tx, mut out_rx) = mpsc::channel(10);
|
||||
|
||||
node.add_subscriber(out_tx);
|
||||
|
||||
let handle = node.get_position_handle();
|
||||
|
||||
tokio::spawn(async move {
|
||||
node.run().await.unwrap();
|
||||
});
|
||||
|
||||
// Chunk à 48000 Hz
|
||||
let chunk1 = AudioChunk::new(0, vec![0.0; 48000], vec![0.0; 48000], 48000);
|
||||
tx.send(Arc::new(chunk1)).await.unwrap();
|
||||
out_rx.recv().await.unwrap();
|
||||
|
||||
// Après 48000 samples à 48000 Hz = 1 seconde
|
||||
let pos1 = handle.position_sec().await;
|
||||
assert!((pos1 - 1.0).abs() < 0.0001);
|
||||
|
||||
// Chunk à 96000 Hz
|
||||
let chunk2 = AudioChunk::new(1, vec![0.0; 96000], vec![0.0; 96000], 96000);
|
||||
tx.send(Arc::new(chunk2)).await.unwrap();
|
||||
out_rx.recv().await.unwrap();
|
||||
|
||||
// Position calculée avec le nouveau sample rate
|
||||
let pos2 = handle.position_sec().await;
|
||||
let expected = (48000.0 + 96000.0) / 96000.0;
|
||||
assert!((pos2 - expected).abs() < 0.0001);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user