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

intégration de pmoflac dans pmoaudiocache. nétoyage du code audio obsolete

Browse Source
This commit is contained in:
Eric Coissac
2025-10-27 23:48:30 +01:00
parent b004c8a8bf
commit b4a8925281
6 changed files with 81 additions and 930 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@@ -14,6 +14,8 @@
**/*.wav **/*.wav
**/*.opus **/*.opus
**/*.mp4 **/*.mp4
**/*.mp3
**/*.ogg
xxx xxx
/dcai/ /dcai/
**/.pmomusic.yml **/.pmomusic.yml

43
Cargo.lock generated
View File

@@ -758,30 +758,15 @@ dependencies = [
"libc", "libc",
] ]
[[package]]
name = "crc"
version = "2.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "49fc9a695bca7f35f5f4c15cddc84415f66a74ea78eef08e90c5024f2b540e23"
dependencies = [
"crc-catalog 1.1.1",
]
[[package]] [[package]]
name = "crc" name = "crc"
version = "3.3.0" version = "3.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9710d3b3739c2e349eb44fe848ad0b7c8cb1e42bd87ee49371df2f7acaf3e675" checksum = "9710d3b3739c2e349eb44fe848ad0b7c8cb1e42bd87ee49371df2f7acaf3e675"
dependencies = [ dependencies = [
"crc-catalog 2.4.0", "crc-catalog",
] ]
[[package]]
name = "crc-catalog"
version = "1.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ccaeedb56da03b09f598226e25e80088cb4cd25f316e6e4df7d695f0feeb1403"
[[package]] [[package]]
name = "crc-catalog" name = "crc-catalog"
version = "2.4.0" version = "2.4.0"
@@ -1162,24 +1147,6 @@ version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1ced73b1dacfc750a6db6c0a0c3a3853c8b41997e2e2c563dc90804ae6867959" checksum = "1ced73b1dacfc750a6db6c0a0c3a3853c8b41997e2e2c563dc90804ae6867959"
[[package]]
name = "flacenc"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fb6da14d3c6605689b5c9ed5187a5218a6d3888e14b747bc18fd4e4bafd452bd"
dependencies = [
"built",
"crc 2.1.0",
"crossbeam-channel",
"heapless",
"log",
"md-5",
"num-traits",
"rustversion",
"seq-macro",
"serde",
]
[[package]] [[package]]
name = "flacenc" name = "flacenc"
version = "0.5.0" version = "0.5.0"
@@ -1187,7 +1154,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "22846e369206f5c3caa4c59071041fe70c1c2863ebdd2341e6ad17490d53772b" checksum = "22846e369206f5c3caa4c59071041fe70c1c2863ebdd2341e6ad17490d53772b"
dependencies = [ dependencies = [
"built", "built",
"crc 3.3.0", "crc",
"crossbeam-channel", "crossbeam-channel",
"heapless", "heapless",
"log", "log",
@@ -2701,8 +2668,6 @@ dependencies = [
"axum 0.8.6", "axum 0.8.6",
"bytes", "bytes",
"chrono", "chrono",
"claxon",
"flacenc 0.4.0",
"futures-util", "futures-util",
"lofty", "lofty",
"paste", "paste",
@@ -2712,11 +2677,9 @@ dependencies = [
"pmoflac", "pmoflac",
"pmoserver", "pmoserver",
"quick-xml 0.37.5", "quick-xml 0.37.5",
"reqwest",
"rusqlite", "rusqlite",
"serde", "serde",
"serde_json", "serde_json",
"symphonia",
"tokio", "tokio",
"tracing", "tracing",
"tracing-subscriber", "tracing-subscriber",
@@ -2861,7 +2824,7 @@ dependencies = [
"chrono", "chrono",
"claxon", "claxon",
"ffmpeg-next", "ffmpeg-next",
"flacenc 0.5.0", "flacenc",
"futures", "futures",
"hex", "hex",
"hound", "hound",

12
pmoaudiocache/Cargo.toml
View File

@@ -20,15 +20,9 @@ chrono = "0.4"
# Métadonnées audio # Métadonnées audio
lofty = "0.22" lofty = "0.22"
# Encodage/décodage audio # Outils de streaming
symphonia = { version = "0.5", features = ["all"] } futures-util = "0.3"
claxon = "0.4" # Décodeur FLAC (legacy) bytes = "1.0"
flacenc = "0.4" # Encodeur FLAC (legacy)
futures-util = "0.3" # Pour le streaming
bytes = "1.0" # Pour la conversion de streams
# HTTP client
reqwest = { version = "0.12", features = ["blocking"] }
# Utilitaires # Utilitaires
anyhow = "1.0" anyhow = "1.0"

289
pmoaudiocache/src/cache.rs
View File

@@ -5,7 +5,7 @@
//! des métadonnées en JSON dans la base de données. //! des métadonnées en JSON dans la base de données.
use anyhow::Result; use anyhow::Result;
use pmocache::{CacheConfig, StreamTransformer}; use pmocache::CacheConfig;
use serde_json::Value; use serde_json::Value;
use std::sync::Arc; use std::sync::Arc;
@@ -33,293 +33,6 @@ impl CacheConfig for AudioConfig {
/// Type alias pour le cache audio avec conversion FLAC /// Type alias pour le cache audio avec conversion FLAC
pub type Cache = pmocache::Cache<AudioConfig>; pub type Cache = pmocache::Cache<AudioConfig>;
/// Créateur de transformer FLAC (legacy)
///
/// ⚠️ DEPRECATED: Cette fonction est conservée pour compatibilité mais utilise
/// une approche avec buffer complet. Utilisez `create_streaming_flac_transformer()`
/// pour de meilleures performances et un vrai streaming.
///
/// Convertit automatiquement tout fichier audio téléchargé en format FLAC
/// en bufferisant d'abord tout le fichier.
///
/// # Workflow
///
/// 1. Buffer tous les bytes du stream
/// 2. Décoder l'audio en PCM via Symphonia
/// 3. Encoder le PCM en FLAC via flacenc
/// 4. Écrire le fichier FLAC complet
///
/// Note: Cette approche nécessite de charger tout le fichier en mémoire.
#[allow(dead_code)]
fn create_flac_transformer_legacy() -> StreamTransformer {
Box::new(|input, mut file, progress| {
Box::pin(async move {
use futures_util::StreamExt;
use tokio::io::AsyncWriteExt;
// 1. Collecter tous les bytes du stream
// Note: Symphonia nécessite un MediaSource avec Read + Seek,
// ce qui n'est pas compatible avec un vrai streaming HTTP.
// Nous devons donc bufferiser les données.
let mut buffer = Vec::new();
let mut stream = input.into_byte_stream();
while let Some(chunk) = stream.next().await {
let chunk = chunk.map_err(|e| format!("Stream error: {}", e))?;
buffer.extend_from_slice(&chunk);
}
tracing::debug!(
"Downloaded {} bytes total, starting FLAC conversion",
buffer.len()
);
// 2. Si c'est déjà du FLAC, on l'écrit directement
if buffer.len() >= 4 && &buffer[0..4] == b"fLaC" {
tracing::debug!("Input is already FLAC, writing directly");
file.write_all(&buffer).await.map_err(|e| e.to_string())?;
file.flush().await.map_err(|e| e.to_string())?;
progress(buffer.len() as u64);
return Ok(());
}
tracing::debug!("Converting to FLAC with Symphonia + flacenc");
// 3. Décoder l'audio avec Symphonia
let (samples, channels, sample_rate, bits_per_sample) = {
use std::io::Cursor;
use symphonia::core::audio::SampleBuffer;
use symphonia::core::codecs::{DecoderOptions, CODEC_TYPE_NULL};
use symphonia::core::errors::Error as SymphoniaError;
use symphonia::core::formats::FormatOptions;
use symphonia::core::io::MediaSourceStream;
use symphonia::core::meta::MetadataOptions;
use symphonia::core::probe::Hint;
let cursor = Cursor::new(buffer);
let mss = MediaSourceStream::new(Box::new(cursor), Default::default());
let hint = Hint::new();
let probed = symphonia::default::get_probe()
.format(
&hint,
mss,
&FormatOptions::default(),
&MetadataOptions::default(),
)
.map_err(|e| {
tracing::error!("Symphonia failed to detect audio format: {}", e);
format!(
"Unable to detect audio format. Error: {}. \
Supported formats: MP3, WAV, OGG, FLAC, AAC, ALAC. \
Please verify the URL points to a valid audio file.",
e
)
})?;
let mut format = probed.format;
let track = format
.tracks()
.iter()
.find(|t| t.codec_params.codec != CODEC_TYPE_NULL)
.ok_or_else(|| {
tracing::error!("No audio track found in the file");
"No audio track found in the file. The file may be corrupted or not a valid audio file.".to_string()
})?;
let codec_name = format!("{:?}", track.codec_params.codec);
tracing::debug!("Detected codec: {}", codec_name);
let mut decoder = symphonia::default::get_codecs()
.make(&track.codec_params, &DecoderOptions::default())
.map_err(|e| {
tracing::error!("Failed to create decoder for codec {}: {}", codec_name, e);
format!(
"Codec '{}' is not supported or failed to initialize. Error: {}",
codec_name, e
)
})?;
let channels = track
.codec_params
.channels
.ok_or_else(|| {
tracing::error!("Audio file missing channel information");
"Audio file is missing channel information. The file may be corrupted."
.to_string()
})?
.count();
let sample_rate = track.codec_params.sample_rate.ok_or_else(|| {
tracing::error!("Audio file missing sample rate information");
"Audio file is missing sample rate information. The file may be corrupted."
.to_string()
})?;
let bits_per_sample = track.codec_params.bits_per_sample.unwrap_or(16);
let mut samples_i32 = Vec::new();
let track_id = track.id;
// Décoder tous les packets
loop {
let packet = match format.next_packet() {
Ok(packet) => packet,
Err(SymphoniaError::ResetRequired) => {
decoder.reset();
continue;
}
Err(SymphoniaError::IoError(e))
if e.kind() == std::io::ErrorKind::UnexpectedEof =>
{
break;
}
Err(e) => {
tracing::error!("Failed to read audio packet: {}", e);
return Err(format!(
"Failed to read audio data: {}. The file may be corrupted.",
e
));
}
};
if packet.track_id() != track_id {
continue;
}
match decoder.decode(&packet) {
Ok(decoded) => {
let spec = *decoded.spec();
let duration = decoded.capacity() as u64;
// Convertir en i32 pour flacenc
// Note: Symphonia retourne des samples i32, nous devons les convertir
// en fonction du bits_per_sample réel
let mut sample_buf = SampleBuffer::<i32>::new(duration, spec);
sample_buf.copy_interleaved_ref(decoded);
samples_i32.extend_from_slice(sample_buf.samples());
}
Err(SymphoniaError::DecodeError(e)) => {
tracing::warn!("Skipping corrupted audio packet: {}", e);
continue;
}
Err(e) => {
tracing::error!("Fatal decode error: {}", e);
return Err(format!("Failed to decode audio: {}. The file may be corrupted or use an unsupported codec variant.", e));
}
}
}
if samples_i32.is_empty() {
tracing::error!("No audio samples could be decoded from the file");
return Err(
"No audio samples could be decoded. The file may be corrupted or empty."
.to_string(),
);
}
tracing::debug!(
"Decoded {} samples (i32), {} channels, {} Hz, {} bits",
samples_i32.len(),
channels,
sample_rate,
bits_per_sample
);
// Normaliser les samples i32 vers la plage appropriée pour flacenc
// Symphonia retourne des samples i32 en pleine échelle (32 bits),
// nous devons les normaliser selon le bits_per_sample réel
let (normalized_samples, target_bits): (Vec<i32>, u32) = match bits_per_sample {
0..=16 => {
// Pour 16 bits ou moins, normaliser vers la plage i16
tracing::debug!("Normalizing to 16-bit");
let samples = samples_i32.iter().map(|&s| (s >> 16) as i32).collect();
(samples, 16)
}
17..=24 => {
// Pour 17-24 bits, normaliser vers la plage 24-bit
tracing::debug!("Normalizing to 24-bit");
let samples = samples_i32.iter().map(|&s| (s >> 8) as i32).collect();
(samples, 24)
}
_ => {
// Pour 25-32 bits, garder la pleine échelle i32
tracing::debug!("Keeping 32-bit");
(samples_i32, 32)
}
};
(normalized_samples, channels, sample_rate, target_bits)
};
tracing::debug!(
"Encoding to FLAC: {} samples, {} channels, {} Hz, {} bits",
samples.len(),
channels,
sample_rate,
bits_per_sample
);
// 4. Encoder en FLAC avec flacenc
// Note: L'encodage FLAC est une opération bloquante/CPU-intensive,
// donc nous l'exécutons dans un thread bloquant pour ne pas bloquer le runtime Tokio
let flac_data = tokio::task::spawn_blocking(move || {
use flacenc::bitsink::ByteSink;
use flacenc::component::BitRepr;
use flacenc::error::Verify;
let config = flacenc::config::Encoder::default()
.into_verified()
.map_err(|e| {
tracing::error!("Failed to create FLAC encoder config: {:?}", e);
format!("Internal error: FLAC encoder configuration failed: {:?}", e)
})?;
let source = flacenc::source::MemSource::from_samples(
&samples,
channels,
bits_per_sample as usize,
sample_rate as usize,
);
let flac_stream =
flacenc::encode_with_fixed_block_size(&config, source, config.block_size)
.map_err(|e| {
tracing::error!("FLAC encoding failed: {:?}", e);
format!("Failed to encode audio to FLAC format: {:?}", e)
})?;
let mut sink = ByteSink::new();
flac_stream.write(&mut sink).map_err(|e| {
tracing::error!("Failed to write FLAC stream: {:?}", e);
format!("Failed to write FLAC data: {:?}", e)
})?;
Ok::<Vec<u8>, String>(sink.into_inner())
})
.await
.map_err(|e| {
tracing::error!("FLAC encoding task panicked: {}", e);
format!("Internal error: FLAC encoding task failed: {}", e)
})??;
tracing::debug!("FLAC encoding complete: {} bytes", flac_data.len());
// 5. Écrire le fichier FLAC
file.write_all(&flac_data)
.await
.map_err(|e| e.to_string())?;
file.flush().await.map_err(|e| e.to_string())?;
// 6. Mettre à jour la progression finale
progress(flac_data.len() as u64);
Ok(())
})
})
}
/// Crée un cache audio avec conversion FLAC automatique /// Crée un cache audio avec conversion FLAC automatique
/// ///
/// # Arguments /// # Arguments

150
pmoaudiocache/src/flac.rs
View File

@@ -1,150 +0,0 @@
// //! Module de conversion audio en FLAC
// //!
// //! Ce module gère la conversion de divers formats audio vers FLAC
// //! pour standardiser le stockage dans le cache.
// use anyhow::{anyhow, Result};
// use std::io::Cursor;
// use symphonia::core::audio::SampleBuffer;
// use symphonia::core::codecs::{DecoderOptions, CODEC_TYPE_NULL};
// use symphonia::core::errors::Error as SymphoniaError;
// use symphonia::core::formats::FormatOptions;
// use symphonia::core::io::MediaSourceStream;
// use symphonia::core::meta::MetadataOptions;
// use symphonia::core::probe::Hint;
// /// Convertit des données audio en FLAC
// ///
// /// Cette fonction accepte n'importe quel format audio supporté par Symphonia
// /// et le convertit en FLAC pour un stockage standardisé.
// ///
// /// # Arguments
// ///
// /// * `data` - Données audio brutes (n'importe quel format)
// /// * `extension` - Extension du fichier source (optionnel, aide à la détection)
// ///
// /// # Returns
// ///
// /// Données audio au format FLAC
// ///
// /// # Exemple
// ///
// /// ```rust,no_run
// /// use pmoaudiocache::flac::convert_to_flac;
// ///
// /// let mp3_data = std::fs::read("track.mp3").unwrap();
// /// let flac_data = convert_to_flac(&mp3_data, Some("mp3")).unwrap();
// /// ```
// pub fn convert_to_flac(data: &[u8], extension: Option<&str>) -> Result<Vec<u8>> {
// // Si c'est déjà du FLAC, on le retourne tel quel
// if is_flac(data) {
// return Ok(data.to_vec());
// }
// // Créer un MediaSource depuis les données (en clonant pour avoir 'static)
// let data_owned = data.to_vec();
// let cursor = Cursor::new(data_owned);
// let mss = MediaSourceStream::new(Box::new(cursor), Default::default());
// // Créer un hint si on a l'extension
// let mut hint = Hint::new();
// if let Some(ext) = extension {
// hint.with_extension(ext);
// }
// // Prober le format
// let probed = symphonia::default::get_probe()
// .format(
// &hint,
// mss,
// &FormatOptions::default(),
// &MetadataOptions::default(),
// )
// .map_err(|e| anyhow!("Impossible de détecter le format audio: {}", e))?;
// let mut format = probed.format;
// // Obtenir le premier track audio
// let track = format
// .tracks()
// .iter()
// .find(|t| t.codec_params.codec != CODEC_TYPE_NULL)
// .ok_or_else(|| anyhow!("Aucune piste audio trouvée"))?;
// // Créer un décodeur
// let mut decoder = symphonia::default::get_codecs()
// .make(&track.codec_params, &DecoderOptions::default())
// .map_err(|e| anyhow!("Impossible de créer le décodeur: {}", e))?;
// // Buffer pour stocker les samples décodés
// let mut samples = Vec::new();
// let track_id = track.id;
// // Décoder tous les packets
// loop {
// let packet = match format.next_packet() {
// Ok(packet) => packet,
// Err(SymphoniaError::ResetRequired) => {
// // Reset du décodeur requis
// decoder.reset();
// continue;
// }
// Err(SymphoniaError::IoError(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
// break;
// }
// Err(e) => return Err(anyhow!("Erreur lors de la lecture: {}", e)),
// };
// // Ignorer les packets qui ne sont pas de notre track
// if packet.track_id() != track_id {
// continue;
// }
// match decoder.decode(&packet) {
// Ok(decoded) => {
// // Convertir les samples en format standard
// let spec = *decoded.spec();
// let duration = decoded.capacity() as u64;
// let mut sample_buf = SampleBuffer::<i16>::new(duration, spec);
// sample_buf.copy_interleaved_ref(decoded);
// samples.extend_from_slice(sample_buf.samples());
// }
// Err(SymphoniaError::DecodeError(_)) => continue,
// Err(e) => return Err(anyhow!("Erreur de décodage: {}", e)),
// }
// }
// if samples.is_empty() {
// return Err(anyhow!("Aucun sample décodé"));
// }
// // Note: Pour l'encodage FLAC, on aurait besoin d'une bibliothèque comme
// // `flacenc` qui n'existe pas encore en Rust. Pour l'instant, on stocke
// // les données telles quelles si c'est déjà du FLAC, sinon on retourne
// // les données originales avec un warning.
// // TODO: Implémenter l'encodage FLAC quand une bibliothèque sera disponible
// tracing::warn!("Encodage FLAC non implémenté, stockage du format original");
// Ok(data.to_vec())
// }
// /// Vérifie si les données sont déjà au format FLAC
// fn is_flac(data: &[u8]) -> bool {
// data.len() >= 4 && &data[0..4] == b"fLaC"
// }
// #[cfg(test)]
// mod tests {
// use super::*;
// #[test]
// fn test_is_flac() {
// let flac_header = b"fLaC\x00\x00\x00\x22";
// assert!(is_flac(flac_header));
// let not_flac = b"RIFF\x00\x00\x00\x00";
// assert!(!is_flac(not_flac));
// }
// }

515
pmoaudiocache/src/streaming.rs
View File

@@ -1,454 +1,91 @@
//! Module de streaming audio avec conversion FLAC progressive //! Audio streaming transformer built on pmoflac.
//! //!
//! Ce module fournit un transformer qui utilise pmoflac pour traiter //! This module wires the generic `pmocache` download pipeline with the new
//! les fichiers audio en vrai streaming, permettant de rendre les fichiers //! streaming transcode helper provided by `pmoflac`. Any supported codec
//! FLAC disponibles progressivement pendant le téléchargement. //! (FLAC, MP3, OGG/Vorbis, Opus, WAV, AIFF) is converted to FLAC on the fly,
//! while native FLAC input is forwarded byte-for-byte without re-encoding.
use futures_util::StreamExt; use bytes::Bytes;
use pmocache::StreamTransformer; use pmocache::StreamTransformer;
use std::sync::Arc; use pmoflac::{transcode_to_flac_stream, AudioCodec, TranscodeOptions};
use tokio::io::AsyncWriteExt; use tokio::io::{AsyncReadExt, AsyncWriteExt};
/// Créateur de transformer FLAC avec streaming progressif /// Creates the transformer consumed by the audio cache.
///
/// Ce transformer améliore significativement la disponibilité des fichiers :
///
/// # Pour les sources FLAC
///
/// Pipeline streaming complet :
/// 1. Télécharger les chunks HTTP au fur et à mesure
/// 2. Décoder FLAC → PCM avec pmoflac (streaming)
/// 3. Re-encoder PCM → FLAC avec pmoflac (streaming)
/// 4. Écrire les frames FLAC immédiatement dans le fichier
///
/// Résultat : Les fichiers sont disponibles pour lecture en ~100ms au lieu
/// d'attendre le téléchargement complet !
///
/// # Pour les autres formats (MP3, OGG, AAC, etc.)
///
/// Pipeline hybride :
/// 1. Buffer complet du fichier (nécessaire pour Symphonia)
/// 2. Décoder avec Symphonia (Read+Seek requis)
/// 3. Encoder avec pmoflac en streaming
/// 4. Écrire progressivement le FLAC
///
/// Résultat : Pas de gain sur la latence initiale, mais meilleures performances
/// d'encodage et usage mémoire optimisé.
pub fn create_streaming_flac_transformer() -> StreamTransformer { pub fn create_streaming_flac_transformer() -> StreamTransformer {
Box::new(|input, file, progress| { Box::new(|input, mut file, progress| {
Box::pin(async move { Box::pin(async move {
// TODO: Implémenter la détection de format pour utiliser le pipeline let byte_stream = input.into_byte_stream();
// streaming complet pour les sources FLAC let reader = StreamToAsyncRead::new(byte_stream);
// Pour l'instant, on utilise toujours le pipeline hybride qui fonctionne
// pour tous les formats
tracing::debug!("Using buffered decode + streaming encode pipeline"); let transcode = transcode_to_flac_stream(reader, TranscodeOptions::default())
buffer_and_convert_to_flac(input, file, progress).await .await
.map_err(|e| format!("Audio transcode error: {}", e))?;
log_stream_info(transcode.input_codec(), transcode.input_stream_info());
let mut flac_stream = transcode.into_stream();
let mut buffer = vec![0u8; 64 * 1024];
let mut total_written = 0u64;
loop {
let read = flac_stream
.read(&mut buffer)
.await
.map_err(|e| format!("Failed to read FLAC data: {}", e))?;
if read == 0 {
break;
}
file.write_all(&buffer[..read])
.await
.map_err(|e| format!("Failed to write FLAC file: {}", e))?;
total_written += read as u64;
progress(total_written);
}
file.flush()
.await
.map_err(|e| format!("Failed to flush FLAC file: {}", e))?;
flac_stream
.wait()
.await
.map_err(|e| format!("FLAC encoder error: {}", e))?;
Ok(())
}) })
}) })
} }
#[derive(Debug, Clone, Copy)] fn log_stream_info(codec: AudioCodec, info: &pmoflac::StreamInfo) {
enum AudioFormat {
Flac,
Other,
}
/// Détecte le format audio en analysant les premiers bytes
async fn detect_format(_input: &pmocache::download::CacheInput) -> Result<AudioFormat, String> {
// On ne peut pas peek CacheInput directement sans le consommer,
// donc on va détecter pendant le traitement du stream
// Pour l'instant, on retourne Other par défaut
Ok(AudioFormat::Other)
}
/// Pipeline streaming complet pour FLAC → FLAC
///
/// Cette fonction implémente le vrai streaming sans buffer :
/// - Lecture progressive du stream HTTP
/// - Décodage FLAC → PCM au fur et à mesure
/// - Re-encodage PCM → FLAC au fur et à mesure
/// - Écriture progressive dans le fichier
async fn stream_flac_to_flac(
input: pmocache::download::CacheInput,
mut file: tokio::fs::File,
progress: Arc<dyn Fn(u64) + Send + Sync>,
) -> Result<(), String> {
use pmoflac::{decode_flac_stream, encode_flac_stream, EncoderOptions, PcmFormat};
use tokio::io::AsyncReadExt;
// Convertir le CacheInput en stream
let stream = input.into_byte_stream();
// Créer un lecteur depuis le stream de bytes
let reader = StreamToAsyncRead::new(stream);
// Décoder le FLAC en PCM (streaming)
let decoded_stream = decode_flac_stream(reader)
.await
.map_err(|e| format!("FLAC decode error: {}", e))?;
let info = decoded_stream.info().clone();
tracing::debug!( tracing::debug!(
"FLAC stream info: {} Hz, {} channels, {} bits/sample", "Detected codec {:?}: {} Hz, {} channels, {} bits/sample (passthrough={})",
codec,
info.sample_rate, info.sample_rate,
info.channels, info.channels,
info.bits_per_sample info.bits_per_sample,
codec == AudioCodec::Flac
); );
// Créer le format PCM depuis les infos du stream
let pcm_format = PcmFormat {
sample_rate: info.sample_rate,
channels: info.channels,
bits_per_sample: info.bits_per_sample,
};
// Options d'encodage
let encoder_options = EncoderOptions {
compression_level: 5,
verify: false,
total_samples: info.total_samples,
block_size: Some(info.max_block_size as u32),
};
// Re-encoder PCM → FLAC (streaming)
// Note: encode_flac_stream consomme decoded_stream
let mut encoded_stream = encode_flac_stream(decoded_stream, pcm_format, encoder_options)
.await
.map_err(|e| format!("FLAC encode error: {}", e))?;
// Écrire le FLAC encodé dans le fichier au fur et à mesure
let mut total_written = 0u64;
let mut buffer = vec![0u8; 64 * 1024]; // Buffer de 64 KB
loop {
let n = encoded_stream
.read(&mut buffer)
.await
.map_err(|e| format!("Failed to read encoded FLAC: {}", e))?;
if n == 0 {
break;
}
file.write_all(&buffer[..n])
.await
.map_err(|e| format!("Failed to write to file: {}", e))?;
total_written += n as u64;
progress(total_written);
}
file.flush()
.await
.map_err(|e| format!("Failed to flush file: {}", e))?;
// Attendre que la tâche d'encodage se termine
// (l'encoder attend automatiquement que le decoder se termine)
encoded_stream
.wait()
.await
.map_err(|e| format!("Encoder/Decoder error: {}", e))?;
tracing::debug!(
"Streaming FLAC conversion complete: {} bytes",
total_written
);
Ok(())
} }
/// Pipeline hybride pour autres formats → FLAC /// Adapter exposing a byte stream as `AsyncRead`.
///
/// Buffer le fichier complet (nécessaire pour Symphonia), puis encode en streaming
async fn buffer_and_convert_to_flac(
input: pmocache::download::CacheInput,
mut file: tokio::fs::File,
progress: Arc<dyn Fn(u64) + Send + Sync>,
) -> Result<(), String> {
use pmoflac::{encode_flac_stream, EncoderOptions, PcmFormat};
// 1. Collecter tous les bytes du stream
let mut buffer = Vec::new();
let mut stream = input.into_byte_stream();
while let Some(chunk) = stream.next().await {
let chunk = chunk.map_err(|e| format!("Stream error: {}", e))?;
buffer.extend_from_slice(&chunk);
}
tracing::debug!(
"Downloaded {} bytes total, starting conversion",
buffer.len()
);
// 2. Si c'est déjà du FLAC, on l'écrit directement
if buffer.len() >= 4 && &buffer[0..4] == b"fLaC" {
tracing::debug!("Input is already FLAC, writing directly");
file.write_all(&buffer).await.map_err(|e| e.to_string())?;
file.flush().await.map_err(|e| e.to_string())?;
progress(buffer.len() as u64);
return Ok(());
}
tracing::debug!("Converting to FLAC with Symphonia + pmoflac");
// 3. Décoder l'audio avec Symphonia (dans un blocking task car c'est CPU-intensive)
let (samples, channels, sample_rate, bits_per_sample) =
tokio::task::spawn_blocking(move || decode_with_symphonia_sync(buffer))
.await
.map_err(|e| format!("Decode task panicked: {}", e))??;
tracing::debug!(
"Decoded {} samples, {} channels, {} Hz, {} bits",
samples.len(),
channels,
sample_rate,
bits_per_sample
);
// 4. Convertir les samples i32 en bytes PCM little-endian
let pcm_bytes = samples_to_pcm_bytes(&samples, bits_per_sample);
// 5. Encoder en FLAC avec pmoflac (streaming)
let pcm_format = PcmFormat {
sample_rate,
channels: channels as u8,
bits_per_sample: bits_per_sample as u8,
};
let encoder_options = EncoderOptions {
compression_level: 5,
verify: false,
total_samples: Some((samples.len() / channels) as u64),
block_size: None,
};
// Utiliser tokio::io::duplex pour éviter le problème de lifetime
use std::io::Cursor;
let cursor = Cursor::new(pcm_bytes);
let mut encoded_stream = encode_flac_stream(cursor, pcm_format, encoder_options)
.await
.map_err(|e| format!("FLAC encode error: {}", e))?;
// 6. Écrire le FLAC encodé progressivement
use tokio::io::AsyncReadExt;
let mut total_written = 0u64;
let mut write_buffer = vec![0u8; 64 * 1024];
loop {
let n = encoded_stream
.read(&mut write_buffer)
.await
.map_err(|e| format!("Failed to read encoded FLAC: {}", e))?;
if n == 0 {
break;
}
file.write_all(&write_buffer[..n])
.await
.map_err(|e| format!("Failed to write to file: {}", e))?;
total_written += n as u64;
progress(total_written);
}
file.flush()
.await
.map_err(|e| format!("Failed to flush file: {}", e))?;
encoded_stream
.wait()
.await
.map_err(|e| format!("Encoder error: {}", e))?;
tracing::debug!("FLAC conversion complete: {} bytes", total_written);
Ok(())
}
/// Décode un fichier audio avec Symphonia
///
/// Retourne (samples, channels, sample_rate, bits_per_sample)
fn decode_with_symphonia_sync(buffer: Vec<u8>) -> Result<(Vec<i32>, usize, u32, u32), String> {
use std::io::Cursor;
use symphonia::core::audio::SampleBuffer;
use symphonia::core::codecs::{DecoderOptions, CODEC_TYPE_NULL};
use symphonia::core::errors::Error as SymphoniaError;
use symphonia::core::formats::FormatOptions;
use symphonia::core::io::MediaSourceStream;
use symphonia::core::meta::MetadataOptions;
use symphonia::core::probe::Hint;
let cursor = Cursor::new(buffer);
let mss = MediaSourceStream::new(Box::new(cursor), Default::default());
let hint = Hint::new();
let probed = symphonia::default::get_probe()
.format(
&hint,
mss,
&FormatOptions::default(),
&MetadataOptions::default(),
)
.map_err(|e| {
format!(
"Unable to detect audio format: {}. \
Supported formats: MP3, WAV, OGG, FLAC, AAC, ALAC.",
e
)
})?;
let mut format = probed.format;
let track = format
.tracks()
.iter()
.find(|t| t.codec_params.codec != CODEC_TYPE_NULL)
.ok_or_else(|| {
"No audio track found in the file. The file may be corrupted.".to_string()
})?;
let codec_name = format!("{:?}", track.codec_params.codec);
tracing::debug!("Detected codec: {}", codec_name);
let mut decoder = symphonia::default::get_codecs()
.make(&track.codec_params, &DecoderOptions::default())
.map_err(|e| format!("Codec '{}' is not supported: {}", codec_name, e))?;
let channels = track
.codec_params
.channels
.ok_or_else(|| "Audio file missing channel information.".to_string())?
.count();
let sample_rate = track
.codec_params
.sample_rate
.ok_or_else(|| "Audio file missing sample rate information.".to_string())?;
let bits_per_sample = track.codec_params.bits_per_sample.unwrap_or(16);
let mut samples_i32 = Vec::new();
let track_id = track.id;
// Décoder tous les packets
loop {
let packet = match format.next_packet() {
Ok(packet) => packet,
Err(SymphoniaError::ResetRequired) => {
decoder.reset();
continue;
}
Err(SymphoniaError::IoError(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
break;
}
Err(e) => {
return Err(format!(
"Failed to read audio data: {}. The file may be corrupted.",
e
));
}
};
if packet.track_id() != track_id {
continue;
}
match decoder.decode(&packet) {
Ok(decoded) => {
let spec = *decoded.spec();
let duration = decoded.capacity() as u64;
let mut sample_buf = SampleBuffer::<i32>::new(duration, spec);
sample_buf.copy_interleaved_ref(decoded);
samples_i32.extend_from_slice(sample_buf.samples());
}
Err(SymphoniaError::DecodeError(e)) => {
tracing::warn!("Skipping corrupted audio packet: {}", e);
continue;
}
Err(e) => {
return Err(format!(
"Failed to decode audio: {}. The file may be corrupted.",
e
));
}
}
}
if samples_i32.is_empty() {
return Err("No audio samples could be decoded. The file may be corrupted.".to_string());
}
// Normaliser les samples selon le bits_per_sample
let (normalized_samples, target_bits): (Vec<i32>, u32) = match bits_per_sample {
0..=16 => {
tracing::debug!("Normalizing to 16-bit");
let samples = samples_i32.iter().map(|&s| (s >> 16) as i32).collect();
(samples, 16)
}
17..=24 => {
tracing::debug!("Normalizing to 24-bit");
let samples = samples_i32.iter().map(|&s| (s >> 8) as i32).collect();
(samples, 24)
}
_ => {
tracing::debug!("Keeping 32-bit");
(samples_i32, 32)
}
};
Ok((normalized_samples, channels, sample_rate, target_bits))
}
/// Convertit des samples i32 en bytes PCM little-endian
fn samples_to_pcm_bytes(samples: &[i32], bits_per_sample: u32) -> Vec<u8> {
let bytes_per_sample = (bits_per_sample / 8) as usize;
let mut bytes = Vec::with_capacity(samples.len() * bytes_per_sample);
for &sample in samples {
match bits_per_sample {
16 => {
let s = sample as i16;
bytes.extend_from_slice(&s.to_le_bytes());
}
24 => {
let s_bytes = sample.to_le_bytes();
bytes.extend_from_slice(&s_bytes[0..3]);
}
32 => {
bytes.extend_from_slice(&sample.to_le_bytes());
}
_ => {
// Fallback pour bits non standard
bytes.extend_from_slice(&sample.to_le_bytes());
}
}
}
bytes
}
/// Adaptateur qui convertit un Stream de Bytes en AsyncRead
struct StreamToAsyncRead { struct StreamToAsyncRead {
stream: futures_util::stream::BoxStream<'static, Result<bytes::Bytes, String>>, stream: futures_util::stream::BoxStream<'static, Result<Bytes, String>>,
current_chunk: Option<bytes::Bytes>, current_chunk: Option<Bytes>,
chunk_offset: usize, offset: usize,
} }
impl StreamToAsyncRead { impl StreamToAsyncRead {
fn new( fn new(
stream: std::pin::Pin< stream: std::pin::Pin<Box<dyn futures_util::Stream<Item = Result<Bytes, String>> + Send>>,
Box<dyn futures_util::Stream<Item = Result<bytes::Bytes, String>> + Send>,
>,
) -> Self { ) -> Self {
use futures_util::StreamExt; use futures_util::StreamExt;
Self { Self {
stream: stream.boxed(), stream: stream.boxed(),
current_chunk: None, current_chunk: None,
chunk_offset: 0, offset: 0,
} }
} }
} }
@@ -463,41 +100,33 @@ impl tokio::io::AsyncRead for StreamToAsyncRead {
use std::task::Poll; use std::task::Poll;
loop { loop {
// Si on a un chunk courant, lire dedans
if let Some(chunk) = &self.current_chunk { if let Some(chunk) = &self.current_chunk {
if self.chunk_offset < chunk.len() { if self.offset < chunk.len() {
let available = chunk.len() - self.chunk_offset; let available = chunk.len() - self.offset;
let to_read = std::cmp::min(available, buf.remaining()); let to_copy = available.min(buf.remaining());
buf.put_slice(&chunk[self.chunk_offset..self.chunk_offset + to_read]); buf.put_slice(&chunk[self.offset..self.offset + to_copy]);
self.chunk_offset += to_read; self.offset += to_copy;
return Poll::Ready(Ok(())); return Poll::Ready(Ok(()));
} else {
// Chunk épuisé, passer au suivant
self.current_chunk = None;
self.chunk_offset = 0;
} }
self.current_chunk = None;
self.offset = 0;
} }
// Pas de chunk courant, en récupérer un nouveau
match self.stream.poll_next_unpin(cx) { match self.stream.poll_next_unpin(cx) {
Poll::Ready(Some(Ok(chunk))) => { Poll::Ready(Some(Ok(chunk))) => {
if chunk.is_empty() { if chunk.is_empty() {
continue; continue;
} }
self.current_chunk = Some(chunk); self.current_chunk = Some(chunk);
self.chunk_offset = 0;
} }
Poll::Ready(Some(Err(e))) => { Poll::Ready(Some(Err(e))) => {
return Poll::Ready(Err(std::io::Error::new(std::io::ErrorKind::Other, e))); return Poll::Ready(Err(std::io::Error::new(std::io::ErrorKind::Other, e)));
} }
Poll::Ready(None) => { Poll::Ready(None) => return Poll::Ready(Ok(())),
// Stream terminé Poll::Pending => return Poll::Pending,
return Poll::Ready(Ok(()));
}
Poll::Pending => {
return Poll::Pending;
}
} }
} }
} }
} }
impl Unpin for StreamToAsyncRead {}