eric/pmomusic
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Merge branch 'claude/add-pmoplaylist-source-011CUpmZ9YbyUAUshEePVTJi' into claude/add-pmoplaylist-source-011CUq8bHCyjrEqGxCCXuvfh

Browse Source
This commit is contained in:
coissac
2025-11-05 20:22:59 +01:00
committed by GitHub
parent e60535ed3c bf1de53952
commit 9871cfcb70
51 changed files with 3559 additions and 5338 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
pmoaudio/src/lib.rs
View File

@@ -123,6 +123,7 @@ pub use nodes::{
file_source::FileSource,
flac_file_sink::{FlacFileSink, FlacFileSinkStats},
http_source::HttpSource,
resampling_node::ResamplingNode,
AudioError, AudioNode, TypedAudioNode,
};

1
pmoaudio/src/nodes/mod.rs
View File

@@ -24,6 +24,7 @@ pub mod converter_nodes;
pub mod file_source;
pub mod flac_file_sink;
pub mod http_source;
pub mod resampling_node;
// Modules temporairement désactivés
/*

577
pmoaudio/src/nodes/resampling_node.rs Normal file
View File

@@ -0,0 +1,577 @@
//! ResamplingNode - Node de resampling pour normaliser le sample rate
//!
//! Ce node prend en entrée des chunks audio avec des sample rates variables
//! et les resample vers un sample rate cible fixe.
//!
//! # Usage
//!
//! ```rust,no_run
//! use pmoaudio::{ResamplingNode, FileSource};
//!
//! let mut source = FileSource::new("audio.flac");
//! let mut resampler = ResamplingNode::new(48000); // Force 48kHz
//! source.register(Box::new(resampler));
//! ```
//!
//! # Comportement
//!
//! - Détecte automatiquement les changements de sample rate
//! - Recrée le resampler quand nécessaire
//! - Passe les chunks directement si déjà au bon sample rate
//! - Préserve les sync markers (TrackBoundary, etc.)
//!
//! # Performance
//!
//! Le resampling est effectué via libsoxr (très haute qualité).
//! La qualité est adaptée selon la profondeur de bits :
//! - 8-bit : Medium quality
//! - 16-bit : High quality
//! - 24-bit/32-bit : Very high quality
use crate::{
dsp::resampling::{build_resampler, resampling, Resampler},
nodes::{AudioError, TypedAudioNode},
pipeline::{AudioPipelineNode, Node, NodeLogic},
type_constraints::TypeRequirement,
AudioChunk, AudioChunkData, AudioSegment, BitDepth, I24,
};
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tracing;
// ═══════════════════════════════════════════════════════════════════════════
// ResamplingLogic - Logique pure de resampling
// ═══════════════════════════════════════════════════════════════════════════
/// Logique pure de resampling
///
/// Maintient un resampler et le met à jour selon les changements de sample rate.
pub struct ResamplingLogic {
target_sample_rate: u32,
current_resampler: Option<ResamplerState>,
}
struct ResamplerState {
source_hz: u32,
resampler: Resampler,
}
impl ResamplingLogic {
pub fn new(target_sample_rate: u32) -> Self {
Self {
target_sample_rate,
current_resampler: None,
}
}
/// Resample un chunk audio vers le sample rate cible
fn resample_chunk(&mut self, chunk: &AudioChunk) -> Result<AudioChunk, AudioError> {
let source_sr = chunk.sample_rate();
let bit_depth = match chunk {
AudioChunk::I16(_) => BitDepth::B16,
AudioChunk::I24(_) => BitDepth::B24,
AudioChunk::I32(_) => BitDepth::B32,
AudioChunk::F32(_) => BitDepth::B32, // Traiter comme 32-bit
AudioChunk::F64(_) => BitDepth::B32, // Traiter comme 32-bit
};
// Si déjà au bon sample rate, retourner tel quel
if source_sr == self.target_sample_rate {
return Ok(chunk.clone());
}
// Vérifier si on doit recréer le resampler
let need_new_resampler = match &self.current_resampler {
None => true,
Some(state) => state.source_hz != source_sr,
};
if need_new_resampler {
tracing::debug!(
"ResamplingLogic: creating resampler {}Hz → {}Hz (bit_depth={:?})",
source_sr,
self.target_sample_rate,
bit_depth
);
let resampler = build_resampler(source_sr, self.target_sample_rate, bit_depth)
.map_err(|e| AudioError::ProcessingError(format!("Resampler init failed: {}", e)))?;
self.current_resampler = Some(ResamplerState {
source_hz: source_sr,
resampler,
});
}
let state = self.current_resampler.as_mut().unwrap();
// Extraire les canaux L/R en i32
let (left, right) = extract_channels_i32(chunk)?;
// Appliquer le resampling
let (resampled_left, resampled_right) = resampling(&left, &right, &mut state.resampler);
// Recréer le chunk avec le nouveau sample rate
reconstruct_chunk(chunk, resampled_left, resampled_right, self.target_sample_rate)
}
}
#[async_trait::async_trait]
impl NodeLogic for ResamplingLogic {
async fn process(
&mut self,
input: Option<mpsc::Receiver<Arc<AudioSegment>>>,
output: Vec<mpsc::Sender<Arc<AudioSegment>>>,
stop_token: CancellationToken,
) -> Result<(), AudioError> {
let mut rx = input.expect("ResamplingNode must have input");
tracing::debug!(
"ResamplingLogic::process started, target={}Hz, {} children",
self.target_sample_rate,
output.len()
);
loop {
let segment = tokio::select! {
_ = stop_token.cancelled() => {
tracing::debug!("ResamplingLogic cancelled");
break;
}
result = rx.recv() => {
match result {
Some(seg) => seg,
None => {
tracing::debug!("ResamplingLogic received EOF");
break;
}
}
}
};
// Resample si c'est un chunk audio, sinon passer tel quel
let output_segment = if segment.is_audio_chunk() {
if let Some(chunk) = segment.as_chunk() {
let resampled_chunk = self.resample_chunk(chunk)?;
Arc::new(AudioSegment {
order: segment.order,
timestamp_sec: segment.timestamp_sec,
segment: crate::_AudioSegment::Chunk(Arc::new(resampled_chunk)),
})
} else {
segment
}
} else {
segment
};
// Envoyer à tous les enfants
for tx in &output {
tx.send(output_segment.clone())
.await
.map_err(|_| AudioError::ChildDied)?;
}
}
Ok(())
}
}
// ═══════════════════════════════════════════════════════════════════════════
// Helper Functions
// ═══════════════════════════════════════════════════════════════════════════
/// Extrait les canaux L/R d'un AudioChunk en i32
fn extract_channels_i32(chunk: &AudioChunk) -> Result<(Vec<i32>, Vec<i32>), AudioError> {
match chunk {
AudioChunk::I16(data) => {
let frames = data.get_frames();
let left = frames.iter().map(|frame| frame[0] as i32).collect();
let right = frames.iter().map(|frame| frame[1] as i32).collect();
Ok((left, right))
}
AudioChunk::I24(data) => {
let frames = data.get_frames();
let left = frames.iter().map(|frame| frame[0].as_i32()).collect();
let right = frames.iter().map(|frame| frame[1].as_i32()).collect();
Ok((left, right))
}
AudioChunk::I32(data) => {
let frames = data.get_frames();
let left = frames.iter().map(|frame| frame[0]).collect();
let right = frames.iter().map(|frame| frame[1]).collect();
Ok((left, right))
}
AudioChunk::F32(data) => {
let frames = data.get_frames();
// Convertir f32 → i32 (dénormaliser)
let left = frames
.iter()
.map(|frame| (frame[0] * i32::MAX as f32) as i32)
.collect();
let right = frames
.iter()
.map(|frame| (frame[1] * i32::MAX as f32) as i32)
.collect();
Ok((left, right))
}
AudioChunk::F64(data) => {
let frames = data.get_frames();
// Convertir f64 → i32 (dénormaliser)
let left = frames
.iter()
.map(|frame| (frame[0] * i32::MAX as f64) as i32)
.collect();
let right = frames
.iter()
.map(|frame| (frame[1] * i32::MAX as f64) as i32)
.collect();
Ok((left, right))
}
}
}
/// Reconstruit un AudioChunk du même type avec les canaux resamplez
fn reconstruct_chunk(
original: &AudioChunk,
left: Vec<i32>,
right: Vec<i32>,
new_sample_rate: u32,
) -> Result<AudioChunk, AudioError> {
if left.len() != right.len() {
return Err(AudioError::ProcessingError(
"Left and right channel lengths differ after resampling".into(),
));
}
let gain_db = original.gain_db();
match original {
AudioChunk::I16(_) => {
let mut stereo = Vec::with_capacity(left.len());
for i in 0..left.len() {
stereo.push([left[i] as i16, right[i] as i16]);
}
Ok(AudioChunk::I16(AudioChunkData::new(
stereo,
new_sample_rate,
gain_db,
)))
}
AudioChunk::I24(_) => {
let mut stereo = Vec::with_capacity(left.len());
for i in 0..left.len() {
let l = I24::new(left[i])
.ok_or_else(|| AudioError::ProcessingError("Invalid I24 value".into()))?;
let r = I24::new(right[i])
.ok_or_else(|| AudioError::ProcessingError("Invalid I24 value".into()))?;
stereo.push([l, r]);
}
Ok(AudioChunk::I24(AudioChunkData::new(
stereo,
new_sample_rate,
gain_db,
)))
}
AudioChunk::I32(_) => {
let mut stereo = Vec::with_capacity(left.len());
for i in 0..left.len() {
stereo.push([left[i], right[i]]);
}
Ok(AudioChunk::I32(AudioChunkData::new(
stereo,
new_sample_rate,
gain_db,
)))
}
AudioChunk::F32(_) => {
let mut stereo = Vec::with_capacity(left.len());
for i in 0..left.len() {
stereo.push([
left[i] as f32 / i32::MAX as f32,
right[i] as f32 / i32::MAX as f32,
]);
}
Ok(AudioChunk::F32(AudioChunkData::new(
stereo,
new_sample_rate,
gain_db,
)))
}
AudioChunk::F64(_) => {
let mut stereo = Vec::with_capacity(left.len());
for i in 0..left.len() {
stereo.push([
left[i] as f64 / i32::MAX as f64,
right[i] as f64 / i32::MAX as f64,
]);
}
Ok(AudioChunk::F64(AudioChunkData::new(
stereo,
new_sample_rate,
gain_db,
)))
}
}
}
// ═══════════════════════════════════════════════════════════════════════════
// WRAPPER ResamplingNode - Délègue à Node<ResamplingLogic>
// ═══════════════════════════════════════════════════════════════════════════
/// ResamplingNode - Normalise le sample rate vers une valeur cible
///
/// Ce node prend en entrée des chunks audio avec des sample rates variables
/// et les resample vers un sample rate fixe.
pub struct ResamplingNode {
inner: Node<ResamplingLogic>,
}
impl ResamplingNode {
/// Crée un nouveau node de resampling
///
/// * `target_sample_rate` - Sample rate de sortie en Hz (ex: 48000)
pub fn new(target_sample_rate: u32) -> Box<dyn AudioPipelineNode> {
Self::with_channel_size(target_sample_rate, 16)
}
/// Crée un nouveau node de resampling avec taille de canal personnalisée
///
/// * `target_sample_rate` - Sample rate de sortie en Hz
/// * `channel_size` - Taille du canal de communication
pub fn with_channel_size(
target_sample_rate: u32,
channel_size: usize,
) -> Box<dyn AudioPipelineNode> {
let logic = ResamplingLogic::new(target_sample_rate);
Box::new(Self {
inner: Node::new_with_input(logic, channel_size),
})
}
}
#[async_trait::async_trait]
impl AudioPipelineNode for ResamplingNode {
fn get_tx(&self) -> Option<mpsc::Sender<Arc<AudioSegment>>> {
self.inner.get_tx()
}
fn register(&mut self, child: Box<dyn AudioPipelineNode>) {
self.inner.register(child)
}
async fn run(
self: Box<Self>,
stop_token: CancellationToken,
) -> Result<(), AudioError> {
Box::new(self.inner).run(stop_token).await
}
}
impl TypedAudioNode for ResamplingNode {
fn input_type(&self) -> Option<TypeRequirement> {
// Accepte n'importe quel type
Some(TypeRequirement::any())
}
fn output_type(&self) -> Option<TypeRequirement> {
// Produit le même type que l'entrée (mais sample rate changé)
Some(TypeRequirement::any())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{AudioChunk, AudioChunkData, SyncMarker};
#[test]
fn test_extract_channels_i16() {
let chunk = AudioChunk::I16(AudioChunkData::new(
vec![[100, 200], [300, 400]],
48000,
0.0,
));
let (left, right) = extract_channels_i32(&chunk).unwrap();
assert_eq!(left, vec![100i32, 300i32]);
assert_eq!(right, vec![200i32, 400i32]);
}
#[test]
fn test_extract_channels_i24() {
let chunk = AudioChunk::I24(AudioChunkData::new(
vec![
[I24::new(1_000_000).unwrap(), I24::new(2_000_000).unwrap()],
[I24::new(3_000_000).unwrap(), I24::new(4_000_000).unwrap()],
],
48000,
0.0,
));
let (left, right) = extract_channels_i32(&chunk).unwrap();
assert_eq!(left, vec![1_000_000i32, 3_000_000i32]);
assert_eq!(right, vec![2_000_000i32, 4_000_000i32]);
}
#[test]
fn test_reconstruct_chunk_i16() {
let original = AudioChunk::I16(AudioChunkData::new(
vec![[100, 200]],
44100,
0.0,
));
let left = vec![100i32, 300i32];
let right = vec![200i32, 400i32];
let result = reconstruct_chunk(&original, left, right, 48000).unwrap();
if let AudioChunk::I16(data) = result {
assert_eq!(data.get_sample_rate(), 48000);
let frames = data.get_frames();
assert_eq!(frames.len(), 2);
assert_eq!(frames[0], [100i16, 200i16]);
assert_eq!(frames[1], [300i16, 400i16]);
} else {
panic!("Expected I16 chunk");
}
}
#[test]
fn test_reconstruct_chunk_i24() {
let original = AudioChunk::I24(AudioChunkData::new(
vec![[I24::new(1_000_000).unwrap(), I24::new(2_000_000).unwrap()]],
44100,
0.0,
));
let left = vec![1_000_000i32, 3_000_000i32];
let right = vec![2_000_000i32, 4_000_000i32];
let result = reconstruct_chunk(&original, left, right, 48000).unwrap();
if let AudioChunk::I24(data) = result {
assert_eq!(data.get_sample_rate(), 48000);
let frames = data.get_frames();
assert_eq!(frames.len(), 2);
assert_eq!(frames[0][0].as_i32(), 1_000_000);
assert_eq!(frames[0][1].as_i32(), 2_000_000);
} else {
panic!("Expected I24 chunk");
}
}
#[test]
fn test_resample_chunk_no_change_if_same_rate() {
let mut logic = ResamplingLogic::new(48000);
let chunk = AudioChunk::I16(AudioChunkData::new(
vec![[100, 200], [300, 400]],
48000, // Déjà à 48kHz
0.0,
));
let result = logic.resample_chunk(&chunk).unwrap();
// Doit retourner le même chunk sans resampling
if let AudioChunk::I16(data) = result {
assert_eq!(data.get_sample_rate(), 48000);
assert_eq!(data.get_frames().len(), 2);
} else {
panic!("Expected I16 chunk");
}
}
#[tokio::test]
async fn test_resampling_logic_passes_sync_markers() {
let mut logic = ResamplingLogic::new(48000);
let (input_tx, input_rx) = mpsc::channel(10);
let (output_tx, mut output_rx) = mpsc::channel(10);
let stop_token = CancellationToken::new();
// Créer un TrackBoundary
let metadata = Arc::new(tokio::sync::RwLock::new(
pmometadata::MemoryTrackMetadata::new()
));
let boundary = AudioSegment::new_track_boundary(0, 0.0, metadata);
// Envoyer le boundary
input_tx.send(boundary.clone()).await.unwrap();
drop(input_tx);
// Lancer le traitement
tokio::spawn(async move {
logic
.process(Some(input_rx), vec![output_tx], stop_token)
.await
.unwrap();
});
// Vérifier que le boundary passe tel quel
let result = output_rx.recv().await.unwrap();
assert!(result.as_sync_marker().is_some());
if let Some(marker) = result.as_sync_marker() {
assert!(matches!(**marker, SyncMarker::TrackBoundary { .. }));
}
}
#[tokio::test]
async fn test_resampling_node_integration() {
// Test d'intégration complet avec ResamplingNode
let (input_tx, input_rx) = mpsc::channel(10);
let (output_tx, mut output_rx) = mpsc::channel(10);
let stop_token = CancellationToken::new();
let mut logic = ResamplingLogic::new(48000);
// Créer un chunk à 44.1kHz
let chunk_44k = AudioChunk::I16(AudioChunkData::new(
vec![[1000, 2000]; 100], // 100 frames
44100,
0.0,
));
let segment = Arc::new(AudioSegment {
order: 0,
timestamp_sec: 0.0,
segment: crate::_AudioSegment::Chunk(Arc::new(chunk_44k)),
});
input_tx.send(segment).await.unwrap();
drop(input_tx);
// Lancer le traitement
tokio::spawn(async move {
logic
.process(Some(input_rx), vec![output_tx], stop_token)
.await
.unwrap();
});
// Vérifier le résultat
let result = output_rx.recv().await.unwrap();
assert!(result.is_audio_chunk());
if let Some(chunk) = result.as_chunk() {
// Le chunk doit être I16 (même type)
assert!(matches!(chunk.as_ref(), AudioChunk::I16(_)));
// Le sample rate doit être 48000
assert_eq!(chunk.sample_rate(), 48000);
// Le nombre de frames doit avoir changé (ratio ~1.088)
// 100 frames @ 44.1kHz ≈ 109 frames @ 48kHz
if let AudioChunk::I16(data) = chunk.as_ref() {
let frames = data.get_frames().len();
assert!(frames >= 105 && frames <= 115, "Expected ~109 frames, got {}", frames);
}
} else {
panic!("Expected audio chunk");
}
}
}

9
pmoaudio/src/pipeline.rs
View File

@@ -482,6 +482,15 @@ impl<L: NodeLogic> Node<L> {
pub fn logic(&self) -> &L {
&self.logic
}
/// Retourne une référence mutable vers la logique métier du nœud
///
/// Permet de configurer la logique après construction mais avant run().
/// Utile pour définir des options qui ne peuvent pas être connues
/// au moment de la construction du nœud.
pub fn logic_mut(&mut self) -> &mut L {
&mut self.logic
}
}
#[async_trait::async_trait]