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

Retour sur pmoaudio

Browse Source
This commit is contained in:
Eric Coissac
2025-11-01 21:10:57 +01:00
parent 3402b2b434
commit cd385162ff
26 changed files with 4481 additions and 1048 deletions
Download Patch File Download Diff File Expand all files Collapse all files

244
old_code/pmoaudio/nodes/file_source.rs
View File

@@ -1,244 +0,0 @@
use crate::{
nodes::{AudioError, MultiSubscriberNode},
AudioChunk, BitDepth,
};
use pmoflac::{decode_audio_stream, StreamInfo};
use std::{path::PathBuf, sync::Arc};
use tokio::{fs::File, io::AsyncReadExt, sync::mpsc};
/// FileSource - Lit un fichier audio et publie des `AudioChunk`
///
/// Cette source utilise `pmoflac` pour décoder le fichier (FLAC/MP3/OGG/WAV/AIFF)
/// puis transforme les échantillons PCM en `AudioChunk` stéréo.
pub struct FileSource {
path: PathBuf,
chunk_frames: usize,
subscribers: MultiSubscriberNode,
}
impl FileSource {
/// Crée une nouvelle source de fichier.
///
/// * `path` - chemin du fichier audio à lire
/// * `chunk_frames` - nombre d'échantillons par canal par chunk
pub fn new<P: Into<PathBuf>>(path: P, chunk_frames: usize) -> Self {
Self {
path: path.into(),
chunk_frames: chunk_frames.max(1),
subscribers: MultiSubscriberNode::new(),
}
}
/// Ajoute un abonné qui recevra les chunks décodés.
pub fn add_subscriber(&mut self, tx: mpsc::Sender<Arc<AudioChunk>>) {
self.subscribers.add_subscriber(tx);
}
/// Lance la lecture du fichier et diffuse les chunks.
pub async fn run(self) -> Result<(), AudioError> {
let file = File::open(&self.path).await.map_err(|e| {
AudioError::ProcessingError(format!("Failed to open {:?}: {}", self.path, e))
})?;
let mut stream = decode_audio_stream(file)
.await
.map_err(|e| AudioError::ProcessingError(format!("Decode error: {}", e)))?;
let stream_info = stream.info().clone();
validate_stream(&stream_info)?;
let frame_bytes = stream_info.bytes_per_sample() * stream_info.channels as usize;
let chunk_byte_len = self.chunk_frames * frame_bytes;
let mut pending = Vec::new();
let mut read_buf = vec![0u8; frame_bytes * 512.max(self.chunk_frames)];
let mut chunk_index = 0u64;
loop {
if pending.len() < chunk_byte_len {
let read = stream.read(&mut read_buf).await.map_err(|e| {
AudioError::ProcessingError(format!("I/O error while decoding: {}", e))
})?;
if read == 0 {
break;
}
pending.extend_from_slice(&read_buf[..read]);
}
if pending.is_empty() {
break;
}
let frames_in_pending = pending.len() / frame_bytes;
let frames_to_emit = frames_in_pending.min(self.chunk_frames);
let take_bytes = frames_to_emit * frame_bytes;
let chunk_bytes = pending.drain(..take_bytes).collect::<Vec<u8>>();
let chunk = bytes_to_chunk(&chunk_bytes, &stream_info, frames_to_emit, chunk_index)?;
self.subscribers.push(chunk).await?;
chunk_index += 1;
}
// Reste éventuel (moins qu'un chunk complet)
if !pending.is_empty() {
let frames = pending.len() / frame_bytes;
if frames > 0 {
let chunk = bytes_to_chunk(&pending, &stream_info, frames, chunk_index)?;
self.subscribers.push(chunk).await?;
}
}
stream
.wait()
.await
.map_err(|e| AudioError::ProcessingError(format!("Decode task failed: {}", e)))?;
Ok(())
}
}
fn validate_stream(info: &StreamInfo) -> Result<(), AudioError> {
if !(1..=2).contains(&info.channels) {
return Err(AudioError::ProcessingError(format!(
"Unsupported channel count: {}",
info.channels
)));
}
match info.bits_per_sample {
8 | 16 | 24 | 32 => Ok(()),
other => Err(AudioError::ProcessingError(format!(
"Unsupported bit depth: {}",
other
))),
}
}
fn bytes_to_chunk(
chunk_bytes: &[u8],
info: &StreamInfo,
frames: usize,
order: u64,
) -> Result<Arc<AudioChunk>, AudioError> {
let bytes_per_sample = info.bytes_per_sample();
let channels = info.channels as usize;
let frame_bytes = bytes_per_sample * channels;
let mut left = Vec::with_capacity(frames);
let mut right = Vec::with_capacity(frames);
for frame_idx in 0..frames {
let base = frame_idx * frame_bytes;
let l = sample_to_f32(
&chunk_bytes[base..base + bytes_per_sample],
info.bits_per_sample,
)?;
let r = if channels == 1 {
l
} else {
sample_to_f32(
&chunk_bytes[base + bytes_per_sample..base + 2 * bytes_per_sample],
info.bits_per_sample,
)?
};
left.push(l);
right.push(r);
}
let bit_depth = BitDepth::from_u32_strict(info.bits_per_sample as u32);
Ok(AudioChunk::from_channels_f32(
order,
left,
right,
info.sample_rate,
bit_depth,
))
}
fn sample_to_f32(sample_bytes: &[u8], bits: u8) -> Result<f32, AudioError> {
let sample = match bits {
8 => i8::from_le_bytes([sample_bytes[0]]) as i32,
16 => i16::from_le_bytes(sample_bytes.try_into().unwrap()) as i32,
24 => {
let mut buf = [0u8; 4];
buf[..3].copy_from_slice(sample_bytes);
// Sign extend manually
if sample_bytes[2] & 0x80 != 0 {
buf[3] = 0xFF;
}
i32::from_le_bytes(buf)
}
32 => i32::from_le_bytes(sample_bytes.try_into().unwrap()),
other => {
return Err(AudioError::ProcessingError(format!(
"Unsupported bit depth: {}",
other
)))
}
};
let max = ((1i64 << (bits as i64 - 1)).saturating_sub(1)) as f32;
Ok((sample as f32) / max)
}
#[cfg(test)]
mod tests {
use super::*;
use pmoflac::{encode_flac_stream, EncoderOptions, PcmFormat};
use std::io::Cursor;
use tokio::io::AsyncWriteExt;
use tokio::sync::mpsc;
#[tokio::test]
async fn test_file_source_decodes_flac() {
let temp_dir = tempfile::tempdir().unwrap();
let flac_path = temp_dir.path().join("test.flac");
let sample_rate = 48_000;
let frames = 256;
let mut pcm = Vec::with_capacity(frames * 4);
for i in 0..frames {
let sample = ((i % 32) as f32 / 31.0 * 2.0 - 1.0) * 0.5; // simple ramp
let sample_i16 = (sample * 32767.0) as i16;
pcm.extend_from_slice(&sample_i16.to_le_bytes());
pcm.extend_from_slice(&sample_i16.to_le_bytes());
}
let format = PcmFormat {
sample_rate,
channels: 2,
bits_per_sample: 16,
};
let mut flac_stream =
encode_flac_stream(Cursor::new(pcm.clone()), format, EncoderOptions::default())
.await
.unwrap();
let mut file = File::create(&flac_path).await.expect("create flac file");
tokio::io::copy(&mut flac_stream, &mut file)
.await
.expect("write flac");
file.flush().await.expect("flush file");
flac_stream.wait().await.unwrap();
let mut source = FileSource::new(&flac_path, 64);
let (tx, mut rx) = mpsc::channel(4);
source.add_subscriber(tx);
tokio::spawn(async move {
source.run().await.unwrap();
});
let mut received = 0usize;
while let Some(chunk) = rx.recv().await {
received += chunk.len();
assert_eq!(chunk.sample_rate(), sample_rate);
let scale = 1.0 / chunk.bit_depth().max_value();
if let Some(frame) = chunk.frames().first() {
assert!(((frame[0] as f32) * scale).abs() <= 1.0); // sample range sanity
}
}
assert_eq!(received, frames);
}
}

300
old_code/pmoaudio/nodes/flac_file_sink.rs
View File

@@ -1,300 +0,0 @@
use crate::{nodes::AudioError, AudioChunk};
use pmoflac::{encode_flac_stream, EncoderOptions, PcmFormat};
use std::{
collections::VecDeque,
path::PathBuf,
pin::Pin,
sync::Arc,
task::{Context, Poll},
};
use tokio::{
fs::File,
io::{self, AsyncRead, AsyncWriteExt, ReadBuf},
sync::mpsc,
};
/// Sink qui encode les `AudioChunk` reçus au format FLAC.
pub struct FlacFileSink {
rx: mpsc::Receiver<Arc<AudioChunk>>,
path: PathBuf,
encoder_options: EncoderOptions,
pcm_buffer_capacity: usize,
}
impl FlacFileSink {
/// Crée un sink FLAC avec les options par défaut (compression 5).
pub fn new<P: Into<PathBuf>>(
path: P,
channel_size: usize,
) -> (Self, mpsc::Sender<Arc<AudioChunk>>) {
Self::with_options(path, channel_size, EncoderOptions::default())
}
/// Crée un sink FLAC avec des options explicites.
pub fn with_options<P: Into<PathBuf>>(
path: P,
channel_size: usize,
encoder_options: EncoderOptions,
) -> (Self, mpsc::Sender<Arc<AudioChunk>>) {
let (tx, rx) = mpsc::channel(channel_size);
let sink = Self {
rx,
path: path.into(),
encoder_options,
pcm_buffer_capacity: 8,
};
(sink, tx)
}
/// Lance l'encodage vers le fichier cible.
pub async fn run(self) -> Result<FlacFileSinkStats, AudioError> {
let FlacFileSink {
mut rx,
path,
encoder_options,
pcm_buffer_capacity,
} = self;
let first_chunk = rx.recv().await.ok_or_else(|| {
AudioError::ProcessingError("FlacFileSink: no audio data received".into())
})?;
if first_chunk.len() == 0 {
return Err(AudioError::ProcessingError(
"FlacFileSink: received empty chunk".into(),
));
}
let format = PcmFormat {
sample_rate: first_chunk.sample_rate(),
channels: 2,
bits_per_sample: 16,
};
if let Err(err) = format.validate() {
return Err(AudioError::ProcessingError(format!(
"Invalid PCM format: {}",
err
)));
}
let (pcm_tx, pcm_rx) = mpsc::channel::<Vec<u8>>(pcm_buffer_capacity);
let pump_handle = tokio::spawn(pump_chunks(first_chunk, rx, pcm_tx));
let reader = ByteStreamReader::new(pcm_rx);
let mut flac_stream = encode_flac_stream(reader, format, encoder_options)
.await
.map_err(|e| AudioError::ProcessingError(format!("FLAC encode init failed: {}", e)))?;
let mut output = File::create(&path).await.map_err(|e| {
AudioError::ProcessingError(format!("Failed to create {:?}: {}", path, e))
})?;
tokio::io::copy(&mut flac_stream, &mut output)
.await
.map_err(|e| AudioError::ProcessingError(format!("FLAC write failed: {}", e)))?;
output.flush().await.map_err(|e| {
AudioError::ProcessingError(format!("Failed to flush {:?}: {}", path, e))
})?;
flac_stream
.wait()
.await
.map_err(|e| AudioError::ProcessingError(format!("FLAC encoder task failed: {}", e)))?;
let pump_stats = pump_handle
.await
.map_err(|e| AudioError::ProcessingError(format!("Pump task panicked: {}", e)))??;
Ok(FlacFileSinkStats {
path,
chunks_received: pump_stats.chunks,
total_samples: pump_stats.samples,
total_duration_sec: pump_stats.duration_sec,
})
}
}
struct PumpStats {
chunks: u64,
samples: u64,
duration_sec: f64,
}
async fn pump_chunks(
first_chunk: Arc<AudioChunk>,
mut rx: mpsc::Receiver<Arc<AudioChunk>>,
pcm_tx: mpsc::Sender<Vec<u8>>,
) -> Result<PumpStats, AudioError> {
let mut chunks = 0u64;
let mut samples = 0u64;
let mut duration_sec = 0.0f64;
let expected_rate = first_chunk.sample_rate();
let mut current = Some(first_chunk);
loop {
let chunk_opt = if let Some(ch) = current.take() {
Some(ch)
} else {
rx.recv().await
};
let chunk = match chunk_opt {
Some(ch) => ch,
None => break,
};
if chunk.sample_rate() != expected_rate {
return Err(AudioError::ProcessingError(format!(
"FlacFileSink: inconsistent sample rate ({} vs {})",
chunk.sample_rate(),
expected_rate
)));
}
let pcm_bytes = chunk_to_pcm_bytes(&chunk);
if pcm_bytes.is_empty() {
continue;
}
pcm_tx
.send(pcm_bytes)
.await
.map_err(|_| AudioError::SendError)?;
chunks += 1;
samples += chunk.len() as u64;
duration_sec += chunk.len() as f64 / expected_rate as f64;
}
Ok(PumpStats {
chunks,
samples,
duration_sec,
})
}
fn chunk_to_pcm_bytes(chunk: &AudioChunk) -> Vec<u8> {
let len = chunk.len();
let mut bytes = Vec::with_capacity(len * 4);
let gain = chunk.gain_linear() as f32;
let scale = 1.0f32 / chunk.bit_depth().max_value();
for frame in chunk.frames() {
let left = (frame[0] as f32 * scale * gain).clamp(-1.0, 1.0);
let right = (frame[1] as f32 * scale * gain).clamp(-1.0, 1.0);
let left_i16 = (left * 32767.0) as i16;
let right_i16 = (right * 32767.0) as i16;
bytes.extend_from_slice(&left_i16.to_le_bytes());
bytes.extend_from_slice(&right_i16.to_le_bytes());
}
bytes
}
struct ByteStreamReader {
rx: mpsc::Receiver<Vec<u8>>,
buffer: VecDeque<u8>,
finished: bool,
}
impl ByteStreamReader {
fn new(rx: mpsc::Receiver<Vec<u8>>) -> Self {
Self {
rx,
buffer: VecDeque::new(),
finished: false,
}
}
}
impl AsyncRead for ByteStreamReader {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
loop {
if !self.buffer.is_empty() {
let to_copy = self.buffer.len().min(buf.remaining());
if to_copy == 0 {
return Poll::Ready(Ok(()));
}
// VecDeque::make_contiguous pour copier efficacement
let slice = self.buffer.make_contiguous();
buf.put_slice(&slice[..to_copy]);
self.buffer.drain(..to_copy);
return Poll::Ready(Ok(()));
}
if self.finished {
return Poll::Ready(Ok(()));
}
match Pin::new(&mut self.rx).poll_recv(cx) {
Poll::Ready(Some(bytes)) => {
if bytes.is_empty() {
continue;
}
self.buffer.extend(bytes);
}
Poll::Ready(None) => {
self.finished = true;
return Poll::Ready(Ok(()));
}
Poll::Pending => return Poll::Pending,
}
}
}
}
/// Statistiques produites par le `FlacFileSink`.
#[derive(Debug, Clone)]
pub struct FlacFileSinkStats {
pub path: PathBuf,
pub chunks_received: u64,
pub total_samples: u64,
pub total_duration_sec: f64,
}
#[cfg(test)]
mod tests {
use super::*;
use crate::BitDepth;
use pmoflac::decode_flac_stream;
use tokio::io::AsyncReadExt;
#[tokio::test]
async fn test_flac_file_sink_writes_audio() {
let temp_dir = tempfile::tempdir().unwrap();
let output_path = temp_dir.path().join("output.flac");
let (sink, tx) = FlacFileSink::new(&output_path, 8);
let handle = tokio::spawn(async move { sink.run().await.unwrap() });
let chunk = AudioChunk::from_channels_f32(
0,
vec![0.25; 256],
vec![0.5; 256],
44_100,
BitDepth::B24,
);
tx.send(chunk).await.unwrap();
drop(tx);
let stats = handle.await.unwrap();
assert_eq!(stats.chunks_received, 1);
assert_eq!(stats.total_samples, 256);
let file = File::open(&output_path).await.unwrap();
let mut stream = decode_flac_stream(file).await.unwrap();
let info = stream.info().clone();
assert_eq!(info.channels, 2);
assert_eq!(info.sample_rate, 44_100);
let mut decoded = Vec::new();
stream.read_to_end(&mut decoded).await.unwrap();
stream.wait().await.unwrap();
assert_eq!(decoded.len(), 256 * 4); // 256 frames * 2 channels * 2 bytes
}
}

150
old_code/pmoaudio/nodes/mod.rs
View File

@@ -1,150 +0,0 @@
//! 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 std::sync::Arc;
use tokio::sync::mpsc;
use crate::AudioSegment;
pub mod buffer_node;
pub mod chromecast_sink;
pub mod decoder_node;
pub mod disk_sink;
pub mod dsp_node;
pub mod file_source;
pub mod flac_file_sink;
pub mod mpd_sink;
pub mod sink_node;
pub mod source_node;
pub mod timer_node;
pub mod volume_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<AudioSegment>) -> 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<AudioSegment>>,
}
impl SingleSubscriberNode {
pub fn new(tx: mpsc::Sender<Arc<AudioSegment>>) -> Self {
Self { tx }
}
pub async fn push(&self, chunk: Arc<AudioSegment>) -> 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<AudioSegment>`, 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<AudioSegment>>>,
}
impl MultiSubscriberNode {
pub fn new() -> Self {
Self {
subscribers: Vec::new(),
}
}
pub fn add_subscriber(&mut self, tx: mpsc::Sender<Arc<AudioSegment>>) {
self.subscribers.push(tx);
}
pub async fn push(&self, chunk: Arc<AudioSegment>) -> 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<AudioSegment>) -> 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 {}