diff --git a/Cargo.lock b/Cargo.lock index df6869eb..146318c0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2616,13 +2616,17 @@ version = "0.1.0" dependencies = [ "async-trait", "bytemuck", + "futures-util", "paste", "pmoflac", "pmometadata", + "reqwest", "soxr", "tempfile", "tokio", "tokio-test", + "tokio-util", + "wiremock", ] [[package]] @@ -2637,6 +2641,7 @@ dependencies = [ "futures-util", "lofty", "paste", + "pmoaudio", "pmocache", "pmoconfig", "pmodidl", diff --git a/pmoaudio/src/audio_segment.rs b/pmoaudio/src/audio_segment.rs index fba2c824..5d22acc4 100644 --- a/pmoaudio/src/audio_segment.rs +++ b/pmoaudio/src/audio_segment.rs @@ -1,4 +1,5 @@ use std::sync::Arc; +use tokio::sync::RwLock; use pmometadata::TrackMetadata; @@ -158,7 +159,7 @@ impl AudioSegment { pub fn new_track_boundary( order: u64, timestamp_sec: f64, - metadata: Arc, + metadata: Arc>, ) -> Arc { let marker = Arc::new(SyncMarker::TrackBoundary { metadata: Arc::clone(&metadata), @@ -299,7 +300,7 @@ impl AudioSegment { } /// Récupère les métadatas du track si c'est un TrackBoundary - pub fn as_track_metadata(&self) -> Option<&Arc> { + pub fn as_track_metadata(&self) -> Option<&Arc>> { match &self.segment { _AudioSegment::Sync(marker) => match &**marker { SyncMarker::TrackBoundary { metadata } => Some(metadata), diff --git a/pmoaudio/src/nodes/file_source.rs b/pmoaudio/src/nodes/file_source.rs index e23b8c7b..510c5d0a 100644 --- a/pmoaudio/src/nodes/file_source.rs +++ b/pmoaudio/src/nodes/file_source.rs @@ -107,7 +107,7 @@ impl FileSource { } // Émettre TrackBoundary - let track_boundary = AudioSegment::new_track_boundary(0, 0.0, Arc::new(metadata)); + let track_boundary = AudioSegment::new_track_boundary(0, 0.0, Arc::new(tokio::sync::RwLock::new(metadata))); self.subscribers.push(track_boundary).await?; } Err(e) => { diff --git a/pmoaudio/src/nodes/flac_file_sink.rs b/pmoaudio/src/nodes/flac_file_sink.rs index fab4de08..5068911f 100644 --- a/pmoaudio/src/nodes/flac_file_sink.rs +++ b/pmoaudio/src/nodes/flac_file_sink.rs @@ -220,7 +220,7 @@ fn generate_track_path(base_path: &Path, track_number: usize) -> PathBuf { /// Signal retourné par pump_segments indiquant pourquoi l'encodage s'est arrêté. enum StopReason { - TrackBoundary(Arc), + TrackBoundary(Arc>), EndOfStream, ChannelClosed, } @@ -229,8 +229,8 @@ enum StopReason { /// Retourne une erreur si EndOfStream est reçu avant tout audio. async fn wait_for_first_audio_chunk_with_metadata( rx: &mut mpsc::Receiver>, -) -> Result<(Arc, Option>), AudioError> { - let mut track_metadata: Option> = None; +) -> Result<(Arc, Option>>), AudioError> { + let mut track_metadata: Option>> = None; loop { let segment = rx @@ -583,7 +583,7 @@ mod tests { metadata.set_year(Some(2024)).await.unwrap(); let track_boundary = - crate::AudioSegment::new_track_boundary(0, 0.0, std::sync::Arc::new(metadata)); + crate::AudioSegment::new_track_boundary(0, 0.0, std::sync::Arc::new(tokio::sync::RwLock::new(metadata))); tx.send(track_boundary).await.unwrap(); // Générer et envoyer des chunks audio diff --git a/pmoaudio/src/nodes/http_source.rs b/pmoaudio/src/nodes/http_source.rs index 3c84dd98..38bb8c4c 100644 --- a/pmoaudio/src/nodes/http_source.rs +++ b/pmoaudio/src/nodes/http_source.rs @@ -257,7 +257,7 @@ impl HttpSource { self.subscribers.push(top_zero).await?; // Émettre TrackBoundary avec les métadonnées HTTP - let track_boundary = AudioSegment::new_track_boundary(0, 0.0, Arc::new(metadata)); + let track_boundary = AudioSegment::new_track_boundary(0, 0.0, Arc::new(tokio::sync::RwLock::new(metadata))); self.subscribers.push(track_boundary).await?; // Préparer la lecture des chunks audio diff --git a/pmoaudio/src/sync_marker.rs b/pmoaudio/src/sync_marker.rs index 679901d1..d187cc92 100644 --- a/pmoaudio/src/sync_marker.rs +++ b/pmoaudio/src/sync_marker.rs @@ -1,9 +1,10 @@ use std::sync::Arc; +use tokio::sync::RwLock; use pmometadata::TrackMetadata; pub enum SyncMarker { - TrackBoundary { metadata: Arc }, + TrackBoundary { metadata: Arc> }, StreamMetadata { key: String, value: String }, TopZeroSync, Heartbeat, diff --git a/pmoaudiocache/Cargo.toml b/pmoaudiocache/Cargo.toml index 1a7b6a22..f61dda24 100644 --- a/pmoaudiocache/Cargo.toml +++ b/pmoaudiocache/Cargo.toml @@ -12,6 +12,8 @@ pmodidl = { path = "../pmodidl" } # Streaming FLAC asynchrone pmoflac = { path = "../pmoflac" } +pmometadata = { path = "../pmometadata" } +pmoaudio = { path = "../pmoaudio" } # Base de données rusqlite = { version = "0.37", features = ["bundled"] } @@ -30,6 +32,7 @@ serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" quick-xml = { version = "0.37", features = ["serialize"] } paste = "1.0" +async-trait = "0.1" # Async tokio = { version = "1.0", features = ["full"] } @@ -44,6 +47,7 @@ tracing = "0.1.41" [dev-dependencies] tracing-subscriber = "0.3" +tempfile = "3" [features] default = ["pmoserver"] diff --git a/pmoaudiocache/src/lib.rs b/pmoaudiocache/src/lib.rs index 5ae8b138..dffb8602 100644 --- a/pmoaudiocache/src/lib.rs +++ b/pmoaudiocache/src/lib.rs @@ -79,7 +79,9 @@ pub mod cache; pub mod metadata; pub mod metadata_ext; +pub mod nodes; pub mod streaming; +pub mod track_metadata; #[cfg(feature = "pmoserver")] pub mod openapi; @@ -90,7 +92,9 @@ pub mod config_ext; // Re-exports principaux pub use cache::{add_with_metadata_extraction, get_metadata, new_cache, AudioConfig, Cache}; pub use metadata::AudioMetadata; -pub use metadata_ext::AudioMetadataExt; +pub use metadata_ext::{AudioMetadataExt, AudioTrackMetadataExt}; +pub use nodes::{FlacCacheSink, FlacCacheSinkStats}; +pub use track_metadata::AudioCacheTrackMetadata; #[cfg(feature = "pmoconfig")] pub use config_ext::AudioCacheConfigExt; diff --git a/pmoaudiocache/src/metadata_ext.rs b/pmoaudiocache/src/metadata_ext.rs index 3ddd25a2..cd920a6d 100644 --- a/pmoaudiocache/src/metadata_ext.rs +++ b/pmoaudiocache/src/metadata_ext.rs @@ -3,8 +3,11 @@ //! Ce module utilise la macro `define_metadata_properties!` de pmocache //! pour générer automatiquement des méthodes d'accès typées aux métadonnées audio. -use crate::AudioConfig; +use crate::{AudioCacheTrackMetadata, AudioConfig}; use pmocache::define_metadata_properties; +use pmometadata::TrackMetadata; +use std::sync::Arc; +use tokio::sync::RwLock; // Génération automatique du trait AudioMetadataExt avec toutes les propriétés audio define_metadata_properties! { @@ -33,3 +36,15 @@ define_metadata_properties! { bit_depth: i64 as i64, } } + +/// Fournit un accès direct à une implémentation `TrackMetadata` basée sur le cache. +pub trait AudioTrackMetadataExt { + fn track_metadata(&self, pk: impl Into) -> Arc>; +} + +impl AudioTrackMetadataExt for Arc> { + fn track_metadata(&self, pk: impl Into) -> Arc> { + let metadata = AudioCacheTrackMetadata::new(self.clone(), pk); + Arc::new(RwLock::new(metadata)) + } +} diff --git a/pmoaudiocache/src/nodes/flac_cache_sink.rs b/pmoaudiocache/src/nodes/flac_cache_sink.rs new file mode 100644 index 00000000..f4b50fb0 --- /dev/null +++ b/pmoaudiocache/src/nodes/flac_cache_sink.rs @@ -0,0 +1,560 @@ +//! Sink qui encode les AudioSegment au format FLAC et les stocke dans le cache audio + +use crate::metadata_ext::AudioTrackMetadataExt; +use pmoaudio::{ + nodes::{AudioError, TypedAudioNode, DEFAULT_CHANNEL_SIZE}, + type_constraints::TypeRequirement, + AudioChunk, AudioSegment, SyncMarker, _AudioSegment, +}; +use pmoflac::{encode_flac_stream, EncoderOptions, PcmFormat}; +use std::{ + collections::VecDeque, + io::Cursor, + pin::Pin, + sync::Arc, + task::{Context, Poll}, +}; +use tokio::{ + io::{self, AsyncRead, ReadBuf}, + sync::{mpsc, RwLock}, +}; + +/// Sink qui encode les `AudioSegment` reçus au format FLAC et les stocke dans le cache audio. +/// +/// Ce sink : +/// - Filtre les chunks audio et ignore les autres syncmarkers (sauf TrackBoundary et EndOfStream) +/// - Crée une nouvelle entrée de cache pour chaque TrackBoundary rencontré +/// - Adapte automatiquement l'encodage FLAC selon la profondeur de bit du chunk (8/16/24/32-bit) +/// - Copie les métadonnées du TrackBoundary dans le cache après ingestion +/// - Termine l'encodage proprement quand il reçoit EndOfStream +pub struct FlacCacheSink { + rx: mpsc::Receiver>, + cache: Arc, + collection: Option, + encoder_options: EncoderOptions, + pcm_buffer_capacity: usize, +} + +impl FlacCacheSink { + /// Crée un sink FLAC cache avec les options par défaut (compression 5, buffer de 16 segments). + /// + /// # Arguments + /// + /// * `cache` - Arc vers le cache audio où stocker les fichiers FLAC encodés + pub fn new(cache: Arc) -> (Self, mpsc::Sender>) { + Self::with_channel_size(cache, DEFAULT_CHANNEL_SIZE) + } + + /// Crée un sink FLAC cache avec une taille de buffer MPSC personnalisée. + /// + /// # Arguments + /// + /// * `cache` - Arc vers le cache audio + /// * `channel_size` - Taille du buffer MPSC (nombre de segments en attente avant backpressure) + pub fn with_channel_size( + cache: Arc, + channel_size: usize, + ) -> (Self, mpsc::Sender>) { + Self::with_config(cache, channel_size, EncoderOptions::default(), None) + } + + /// Crée un sink FLAC cache avec une configuration complète. + /// + /// # Arguments + /// + /// * `cache` - Arc vers le cache audio + /// * `channel_size` - Taille du buffer MPSC + /// * `encoder_options` - Options d'encodage FLAC (compression, etc.) + /// * `collection` - Collection optionnelle à laquelle appartiennent les fichiers + pub fn with_config( + cache: Arc, + channel_size: usize, + encoder_options: EncoderOptions, + collection: Option, + ) -> (Self, mpsc::Sender>) { + let (tx, rx) = mpsc::channel(channel_size); + let sink = Self { + rx, + cache, + collection, + encoder_options, + pcm_buffer_capacity: 8, + }; + (sink, tx) + } + + /// Lance l'encodage et l'ingestion dans le cache. + /// + /// Cette méthode crée une nouvelle entrée de cache pour chaque TrackBoundary rencontré. + /// Les métadonnées du TrackBoundary sont copiées dans le cache après l'ingestion. + pub async fn run(self) -> Result { + let FlacCacheSink { + mut rx, + cache, + collection, + encoder_options, + pcm_buffer_capacity, + } = self; + + let mut all_tracks = Vec::new(); + let mut track_number = 0; + + loop { + // Attendre le premier chunk audio pour cette track, en capturant les métadonnées du TrackBoundary + let (first_segment, track_metadata) = + match wait_for_first_audio_chunk_with_metadata(&mut rx).await { + Ok(result) => result, + Err(_) => { + // Plus d'audio disponible + if all_tracks.is_empty() { + return Err(AudioError::ProcessingError( + "No audio data received".into(), + )); + } + break; + } + }; + + // Extraire les informations du premier chunk + let first_chunk = first_segment.as_chunk().unwrap(); + let sample_rate = first_chunk.sample_rate(); + let bits_per_sample = get_chunk_bit_depth(first_chunk); + + let format = PcmFormat { + sample_rate, + channels: 2, + bits_per_sample, + }; + if let Err(err) = format.validate() { + return Err(AudioError::ProcessingError(format!( + "Invalid PCM format: {}", + err + ))); + } + + // Créer le pipeline d'encodage pour cette track + let (pcm_tx, pcm_rx) = mpsc::channel::>(pcm_buffer_capacity); + + // Préparer les options d'encodage avec les métadonnées du TrackBoundary + let mut options_with_metadata = encoder_options.clone(); + options_with_metadata.metadata = track_metadata.clone(); + + // Créer l'encoder + let reader = ByteStreamReader::new(pcm_rx); + let mut flac_stream = encode_flac_stream(reader, format, options_with_metadata) + .await + .map_err(|e| { + AudioError::ProcessingError(format!("FLAC encode init failed: {}", e)) + })?; + + // Créer un buffer pour collecter le FLAC encodé + let mut flac_buffer = Vec::new(); + + // Exécuter pump et copy en parallèle + let pump_future = pump_track_segments( + first_segment, + &mut rx, + pcm_tx, + bits_per_sample, + sample_rate, + ); + let copy_future = async { + tokio::io::copy(&mut flac_stream, &mut flac_buffer) + .await + .map_err(|e| AudioError::ProcessingError(format!("FLAC write failed: {}", e)))?; + flac_stream + .wait() + .await + .map_err(|e| AudioError::ProcessingError(format!("Encoder failed: {}", e)))?; + Ok::<_, AudioError>(()) + }; + + // Attendre les deux tâches en parallèle + let (copy_result, pump_result): (Result<(), AudioError>, Result<(u64, u64, f64, StopReason), AudioError>) = + tokio::join!(copy_future, pump_future); + copy_result?; + let (chunks, samples, duration_sec, stop_reason) = pump_result?; + + // Ingérer le FLAC dans le cache + let flac_reader = Cursor::new(flac_buffer.clone()); + let collection_ref = collection.as_deref(); + let pk = cache + .add_from_reader(None, flac_reader, Some(flac_buffer.len() as u64), collection_ref) + .await + .map_err(|e| { + AudioError::ProcessingError(format!("Failed to add to cache: {}", e)) + })?; + + // Copier les métadonnées du TrackBoundary dans le cache + if let Some(src_metadata) = track_metadata { + let dest_metadata = cache.track_metadata(&pk); + + // Utiliser copy_metadata_into pour copier toutes les métadonnées + pmometadata::copy_metadata_into(&src_metadata, &dest_metadata) + .await + .map_err(|e| { + AudioError::ProcessingError(format!( + "Failed to copy metadata to cache: {}", + e + )) + })?; + } + + // Ajouter les stats de cette track + all_tracks.push(TrackStats { + pk, + track_number, + chunks_received: chunks, + total_samples: samples, + total_duration_sec: duration_sec, + }); + + // Vérifier le stop_reason pour savoir si on continue + match stop_reason { + StopReason::TrackBoundary(_metadata) => { + // Continuer avec la prochaine track + track_number += 1; + continue; + } + StopReason::EndOfStream | StopReason::ChannelClosed => { + // Fin de l'encodage + break; + } + } + } + + Ok(FlacCacheSinkStats { tracks: all_tracks }) + } +} + +/// Signal retourné par pump_segments indiquant pourquoi l'encodage s'est arrêté. +enum StopReason { + TrackBoundary(Arc>), + EndOfStream, + ChannelClosed, +} + +/// Attend et retourne le premier chunk audio avec les métadonnées du TrackBoundary si présent. +/// Retourne une erreur si EndOfStream est reçu avant tout audio. +async fn wait_for_first_audio_chunk_with_metadata( + rx: &mut mpsc::Receiver>, +) -> Result< + ( + Arc, + Option>>, + ), + AudioError, +> { + let mut track_metadata: Option>> = None; + + loop { + let segment = rx + .recv() + .await + .ok_or_else(|| AudioError::ProcessingError("No audio data received".into()))?; + + match &segment.segment { + _AudioSegment::Chunk(chunk) => { + if chunk.len() == 0 { + return Err(AudioError::ProcessingError("Received empty chunk".into())); + } + return Ok((segment, track_metadata)); + } + _AudioSegment::Sync(marker) => match **marker { + SyncMarker::TrackBoundary { ref metadata, .. } => { + // Capturer les métadonnées du TrackBoundary + track_metadata = Some(metadata.clone()); + continue; + } + SyncMarker::EndOfStream => { + return Err(AudioError::ProcessingError( + "EndOfStream received before any audio".into(), + )); + } + _ => { + // Ignorer TopZeroSync, Heartbeat, etc. + continue; + } + }, + } + } +} + +/// Pompe les segments pour une seule track (s'arrête au TrackBoundary). +async fn pump_track_segments( + first_segment: Arc, + rx: &mut mpsc::Receiver>, + pcm_tx: mpsc::Sender>, + bits_per_sample: u8, + expected_rate: u32, +) -> Result<(u64, u64, f64, StopReason), AudioError> { + let mut chunks = 0u64; + let mut samples = 0u64; + let mut duration_sec = 0.0f64; + + // Traiter le premier segment + if let Some(chunk) = first_segment.as_chunk() { + let pcm_bytes = chunk_to_pcm_bytes(chunk, bits_per_sample)?; + if !pcm_bytes.is_empty() { + 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; + } + } + + // Boucle sur les segments suivants + loop { + let segment = match rx.recv().await { + Some(seg) => seg, + None => { + drop(pcm_tx); // Fermer le channel PCM + return Ok((chunks, samples, duration_sec, StopReason::ChannelClosed)); + } + }; + + match &segment.segment { + _AudioSegment::Chunk(chunk) => { + // Vérifier la cohérence du sample rate + if chunk.sample_rate() != expected_rate { + return Err(AudioError::ProcessingError(format!( + "FlacCacheSink: inconsistent sample rate ({} vs {})", + chunk.sample_rate(), + expected_rate + ))); + } + + let pcm_bytes = chunk_to_pcm_bytes(chunk, bits_per_sample)?; + 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; + } + _AudioSegment::Sync(marker) => match &**marker { + SyncMarker::TrackBoundary { metadata, .. } => { + drop(pcm_tx); // Fermer le channel PCM + return Ok(( + chunks, + samples, + duration_sec, + StopReason::TrackBoundary(metadata.clone()), + )); + } + SyncMarker::EndOfStream => { + drop(pcm_tx); // Fermer le channel PCM + return Ok((chunks, samples, duration_sec, StopReason::EndOfStream)); + } + _ => {} // Ignorer les autres syncmarkers + }, + } + } +} + +/// Détermine la profondeur de bit d'un chunk audio +fn get_chunk_bit_depth(chunk: &AudioChunk) -> u8 { + match chunk { + AudioChunk::I16(_) => 16, + AudioChunk::I24(_) => 24, + AudioChunk::I32(_) => 32, + AudioChunk::F32(_) => 32, // Les flottants seront convertis en 32-bit + AudioChunk::F64(_) => 32, // Les flottants seront convertis en 32-bit + } +} + +/// Convertit un chunk audio en bytes PCM avec la profondeur de bit spécifiée +fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result, AudioError> { + // Vérifier que le chunk est de type entier + match chunk { + AudioChunk::F32(_) | AudioChunk::F64(_) => { + return Err(AudioError::ProcessingError( + "FlacCacheSink only supports integer audio chunks (I16, I24, I32)".into(), + )); + } + _ => {} + } + + let len = chunk.len(); + let bytes_per_frame = (bits_per_sample / 8) as usize * 2; // 2 channels + let mut bytes = Vec::with_capacity(len * bytes_per_frame); + + // Convertir selon le type du chunk + match (chunk, bits_per_sample) { + // I16 source + (AudioChunk::I16(data), 16) => { + for frame in data.frames() { + bytes.extend_from_slice(&frame[0].to_le_bytes()); + bytes.extend_from_slice(&frame[1].to_le_bytes()); + } + } + (AudioChunk::I16(data), 24) => { + for frame in data.frames() { + let left = (frame[0] as i32) << 8; + let right = (frame[1] as i32) << 8; + bytes.extend_from_slice(&left.to_le_bytes()[..3]); + bytes.extend_from_slice(&right.to_le_bytes()[..3]); + } + } + (AudioChunk::I16(data), 32) => { + for frame in data.frames() { + let left = (frame[0] as i32) << 16; + let right = (frame[1] as i32) << 16; + bytes.extend_from_slice(&left.to_le_bytes()); + bytes.extend_from_slice(&right.to_le_bytes()); + } + } + + // I24 source + (AudioChunk::I24(data), 16) => { + for frame in data.frames() { + let left = (frame[0].as_i32() >> 8) as i16; + let right = (frame[1].as_i32() >> 8) as i16; + bytes.extend_from_slice(&left.to_le_bytes()); + bytes.extend_from_slice(&right.to_le_bytes()); + } + } + (AudioChunk::I24(data), 24) => { + for frame in data.frames() { + bytes.extend_from_slice(&frame[0].as_i32().to_le_bytes()[..3]); + bytes.extend_from_slice(&frame[1].as_i32().to_le_bytes()[..3]); + } + } + (AudioChunk::I24(data), 32) => { + for frame in data.frames() { + let left = frame[0].as_i32() << 8; + let right = frame[1].as_i32() << 8; + bytes.extend_from_slice(&left.to_le_bytes()); + bytes.extend_from_slice(&right.to_le_bytes()); + } + } + + // I32 source + (AudioChunk::I32(data), 16) => { + for frame in data.frames() { + let left = (frame[0] >> 16) as i16; + let right = (frame[1] >> 16) as i16; + bytes.extend_from_slice(&left.to_le_bytes()); + bytes.extend_from_slice(&right.to_le_bytes()); + } + } + (AudioChunk::I32(data), 24) => { + for frame in data.frames() { + let left = frame[0] >> 8; + let right = frame[1] >> 8; + bytes.extend_from_slice(&left.to_le_bytes()[..3]); + bytes.extend_from_slice(&right.to_le_bytes()[..3]); + } + } + (AudioChunk::I32(data), 32) => { + for frame in data.frames() { + bytes.extend_from_slice(&frame[0].to_le_bytes()); + bytes.extend_from_slice(&frame[1].to_le_bytes()); + } + } + + _ => { + return Err(AudioError::ProcessingError(format!( + "Unsupported bits_per_sample: {}", + bits_per_sample + ))); + } + } + + Ok(bytes) +} + +struct ByteStreamReader { + rx: mpsc::Receiver>, + buffer: VecDeque, + finished: bool, +} + +impl ByteStreamReader { + fn new(rx: mpsc::Receiver>) -> 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> { + 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 pour une track individuelle. +#[derive(Debug, Clone)] +pub struct TrackStats { + pub pk: String, + pub track_number: usize, + pub chunks_received: u64, + pub total_samples: u64, + pub total_duration_sec: f64, +} + +/// Statistiques produites par le `FlacCacheSink`. +#[derive(Debug, Clone)] +pub struct FlacCacheSinkStats { + pub tracks: Vec, +} + +impl TypedAudioNode for FlacCacheSink { + fn input_type(&self) -> Option { + // FlacCacheSink accepte n'importe quel type entier (I16, I24, I32) + // mais rejette les chunks flottants + Some(TypeRequirement::any_integer()) + } + + fn output_type(&self) -> Option { + // FlacCacheSink est un sink, il ne produit pas d'audio + None + } +} diff --git a/pmoaudiocache/src/nodes/mod.rs b/pmoaudiocache/src/nodes/mod.rs new file mode 100644 index 00000000..7d2d365d --- /dev/null +++ b/pmoaudiocache/src/nodes/mod.rs @@ -0,0 +1,8 @@ +//! Nodes audio pour pmoaudiocache +//! +//! Ce module fournit des nodes audio spécialisés qui étendent pmoaudio +//! pour intégrer le cache audio. + +pub mod flac_cache_sink; + +pub use flac_cache_sink::{FlacCacheSink, FlacCacheSinkStats}; diff --git a/pmoaudiocache/src/track_metadata.rs b/pmoaudiocache/src/track_metadata.rs new file mode 100644 index 00000000..e90d677e --- /dev/null +++ b/pmoaudiocache/src/track_metadata.rs @@ -0,0 +1,314 @@ +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use pmometadata::{MetadataError, MetadataResult, TrackMetadata}; +use serde_json::{Number, Value}; + +fn map_db_err(err: rusqlite::Error) -> MetadataError { + MetadataError::Backend(err.to_string()) +} + +pub struct AudioCacheTrackMetadata { + cache: Arc, + pk: String, +} + +impl AudioCacheTrackMetadata { + pub fn new(cache: Arc, pk: impl Into) -> Self { + Self { + cache, + pk: pk.into(), + } + } + + fn read_raw(&self, key: &str) -> Result, MetadataError> { + self.cache + .db + .get_a_metadata(&self.pk, key) + .map_err(map_db_err) + } + + fn write_raw(&self, key: &str, value: Value) -> Result<(), MetadataError> { + self.cache + .db + .set_a_metadata(&self.pk, key, value) + .map_err(map_db_err) + } + + fn read_string(&self, key: &str) -> Result, MetadataError> { + match self.read_raw(key)? { + Some(Value::String(s)) => Ok(Some(s)), + Some(Value::Null) | None => Ok(None), + Some(other) => Err(MetadataError::Backend(format!( + "metadata {key} for {} is not a string ({other})", + self.pk + ))), + } + } + + fn write_string(&self, key: &str, value: Option) -> Result<(), MetadataError> { + let json = value.map(Value::String).unwrap_or(Value::Null); + self.write_raw(key, json) + } + + fn read_number(&self, key: &str) -> Result, MetadataError> { + match self.read_raw(key)? { + Some(Value::Number(n)) => Ok(Some(n)), + Some(Value::Null) | None => Ok(None), + Some(other) => Err(MetadataError::Backend(format!( + "metadata {key} for {} is not a number ({other})", + self.pk + ))), + } + } + + fn write_number(&self, key: &str, value: Option) -> Result<(), MetadataError> { + let json = match value { + Some(n) => Value::Number(Number::from(n)), + None => Value::Null, + }; + self.write_raw(key, json) + } + + fn write_u64(&self, key: &str, value: Option) -> Result<(), MetadataError> { + let json = match value { + Some(n) => Value::Number(Number::from(n)), + None => Value::Null, + }; + self.write_raw(key, json) + } + + fn write_f64(&self, key: &str, value: Option) -> Result<(), MetadataError> { + let json = match value { + Some(v) => Number::from_f64(v) + .map(Value::Number) + .ok_or_else(|| MetadataError::Backend(format!("invalid float for {key}")))?, + None => Value::Null, + }; + self.write_raw(key, json) + } + + fn read_duration(&self) -> Result, MetadataError> { + match self.read_number("duration_secs")? { + Some(n) => match n.as_u64() { + Some(secs) => Ok(Some(Duration::from_secs(secs))), + None => Err(MetadataError::Backend(format!( + "duration_secs for {} out of range", + self.pk + ))), + }, + None => Ok(None), + } + } + + fn write_duration(&self, value: Option) -> Result<(), MetadataError> { + self.write_u64("duration_secs", value.map(|d| d.as_secs())) + } + + fn read_timestamp(&self) -> Result, MetadataError> { + match self.read_number("updated_at")? { + Some(n) => match n.as_u64() { + Some(secs) => Ok(Some(UNIX_EPOCH + Duration::from_secs(secs))), + None => Err(MetadataError::Backend(format!( + "updated_at for {} out of range", + self.pk + ))), + }, + None => Ok(None), + } + } + + fn write_timestamp(&self, when: SystemTime) -> Result<(), MetadataError> { + let secs = when + .duration_since(UNIX_EPOCH) + .map_err(|e| MetadataError::Backend(e.to_string()))? + .as_secs(); + self.write_u64("updated_at", Some(secs)) + } +} + +#[async_trait::async_trait] +impl TrackMetadata for AudioCacheTrackMetadata { + async fn get_title(&self) -> MetadataResult { + Ok(self.read_string("title")?) + } + + async fn set_title(&mut self, value: Option) -> MetadataResult<()> { + self.write_string("title", value)?; + let _ = self.touch().await?; + Ok(Some(())) + } + + async fn get_artist(&self) -> MetadataResult { + Ok(self.read_string("artist")?) + } + + async fn set_artist(&mut self, value: Option) -> MetadataResult<()> { + self.write_string("artist", value)?; + let _ = self.touch().await?; + Ok(Some(())) + } + + async fn get_album(&self) -> MetadataResult { + Ok(self.read_string("album")?) + } + + async fn set_album(&mut self, value: Option) -> MetadataResult<()> { + self.write_string("album", value)?; + let _ = self.touch().await?; + Ok(Some(())) + } + + async fn get_year(&self) -> MetadataResult { + Ok(match self.read_number("year")? { + Some(n) => n + .as_i64() + .and_then(|v| u32::try_from(v).ok()) + .map(Some) + .unwrap_or(None), + None => None, + }) + } + + async fn set_year(&mut self, value: Option) -> MetadataResult<()> { + self.write_number("year", value.map(|v| v as i64))?; + let _ = self.touch().await?; + Ok(Some(())) + } + + async fn get_duration(&self) -> MetadataResult { + Ok(self.read_duration()?) + } + + async fn set_duration(&mut self, value: Option) -> MetadataResult<()> { + self.write_duration(value)?; + let _ = self.touch().await?; + Ok(Some(())) + } + + async fn get_track_id(&self) -> MetadataResult { + Ok(self.read_string("track_id")?) + } + + async fn set_track_id(&mut self, value: Option) -> MetadataResult<()> { + self.write_string("track_id", value)?; + let _ = self.touch().await?; + Ok(Some(())) + } + + async fn get_channel_id(&self) -> MetadataResult { + Ok(self.read_string("channel_id")?) + } + + async fn set_channel_id(&mut self, value: Option) -> MetadataResult<()> { + self.write_string("channel_id", value)?; + let _ = self.touch().await?; + Ok(Some(())) + } + + async fn get_event(&self) -> MetadataResult { + Ok(self.read_string("event")?) + } + + async fn set_event(&mut self, value: Option) -> MetadataResult<()> { + self.write_string("event", value)?; + let _ = self.touch().await?; + Ok(Some(())) + } + + async fn get_rating(&self) -> MetadataResult { + Ok(match self.read_number("rating")? { + Some(n) => n.as_f64().map(|v| v as f32), + None => None, + }) + } + + async fn set_rating(&mut self, value: Option) -> MetadataResult<()> { + self.write_f64("rating", value.map(|v| v as f64))?; + let _ = self.touch().await?; + Ok(Some(())) + } + + async fn get_cover_url(&self) -> MetadataResult { + Ok(self.read_string("cover_url")?) + } + + async fn set_cover_url(&mut self, value: Option) -> MetadataResult<()> { + self.write_string("cover_url", value)?; + let _ = self.touch().await?; + Ok(Some(())) + } + + async fn get_cover_pk(&self) -> MetadataResult { + Ok(self.read_string("cover_pk")?) + } + + async fn set_cover_pk(&mut self, value: Option) -> MetadataResult<()> { + self.write_string("cover_pk", value)?; + let _ = self.touch().await?; + Ok(Some(())) + } + + async fn get_updated_at(&self) -> MetadataResult { + Ok(self.read_timestamp()?) + } + + async fn touch(&mut self) -> MetadataResult<()> { + self.write_timestamp(SystemTime::now())?; + Ok(Some(())) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::cache::new_cache; + use crate::metadata_ext::AudioTrackMetadataExt; + use std::sync::Arc; + use tempfile::tempdir; + + #[tokio::test] + async fn roundtrip_metadata() { + let dir = tempdir().unwrap(); + let cache = Arc::new(new_cache(dir.path().to_str().unwrap(), 4).unwrap()); + let pk = "track-test"; + cache.db.add(pk, None, None).unwrap(); + + let mut meta = cache.track_metadata(pk); + + meta.set_title(Some("Title".into())).await.unwrap(); + meta.set_artist(Some("Artist".into())).await.unwrap(); + meta.set_album(Some("Album".into())).await.unwrap(); + meta.set_year(Some(2024)).await.unwrap(); + meta.set_duration(Some(Duration::from_secs(90))) + .await + .unwrap(); + meta.set_track_id(Some("trk".into())).await.unwrap(); + meta.set_channel_id(Some("chn".into())).await.unwrap(); + meta.set_event(Some("event".into())).await.unwrap(); + meta.set_rating(Some(4.5)).await.unwrap(); + meta.set_cover_url(Some("http://cover".into())) + .await + .unwrap(); + meta.set_cover_pk(Some("cover123".into())).await.unwrap(); + + assert_eq!(meta.get_title().await.unwrap(), Some("Title".into())); + assert_eq!(meta.get_artist().await.unwrap(), Some("Artist".into())); + assert_eq!(meta.get_album().await.unwrap(), Some("Album".into())); + assert_eq!(meta.get_year().await.unwrap(), Some(2024)); + assert_eq!( + meta.get_duration().await.unwrap(), + Some(Duration::from_secs(90)) + ); + assert_eq!(meta.get_track_id().await.unwrap(), Some("trk".into())); + assert_eq!(meta.get_channel_id().await.unwrap(), Some("chn".into())); + assert_eq!(meta.get_event().await.unwrap(), Some("event".into())); + assert_eq!(meta.get_rating().await.unwrap(), Some(4.5)); + assert_eq!( + meta.get_cover_url().await.unwrap(), + Some("http://cover".into()) + ); + assert_eq!(meta.get_cover_pk().await.unwrap(), Some("cover123".into())); + assert!(meta.get_updated_at().await.unwrap().is_some()); + } +} diff --git a/pmocache/src/cache.rs b/pmocache/src/cache.rs index afc40f84..879cc8be 100644 --- a/pmocache/src/cache.rs +++ b/pmocache/src/cache.rs @@ -234,7 +234,7 @@ impl Cache { /// /// # Arguments /// - /// * `source_uri` - Identifiant logique du flux (pour traçabilité dans la DB) + /// * `source_uri` - Identifiant logique optionnel du flux (pour traçabilité dans la DB). Si None, l'origin_url ne sera pas sauvegardée. /// * `reader` - Flux asynchrone fournissant les données /// * `length` - Taille attendue (si connue) /// * `collection` - Collection optionnelle à laquelle appartient l'élément @@ -244,7 +244,7 @@ impl Cache { /// La clé primaire (pk) du fichier dans le cache, calculée à partir du contenu pub async fn add_from_reader( &self, - source_uri: &str, + source_uri: Option<&str>, mut reader: R, length: Option, collection: Option<&str>, @@ -259,7 +259,11 @@ impl Cache { // 2. Calculer le pk basé sur le contenu let pk = crate::cache_trait::pk_from_content_header(&header); - tracing::debug!("Computed pk {} for source_uri {}", pk, source_uri); + if let Some(uri) = source_uri { + tracing::debug!("Computed pk {} for source_uri {}", pk, uri); + } else { + tracing::debug!("Computed pk {} from reader", pk); + } // 3. Vérifier si le fichier est déjà en cache if self.db.get(&pk, false).is_ok() { @@ -300,7 +304,9 @@ impl Cache { } self.db.add(&pk, None, collection)?; - self.db.set_origin_url(&pk, source_uri); + if let Some(uri) = source_uri { + self.db.set_origin_url(&pk, uri)?; + } if let Err(e) = self.enforce_limit().await { tracing::warn!("Error enforcing cache limit: {}", e); @@ -354,7 +360,7 @@ impl Cache { let reader = tokio::fs::File::open(&canonical_path).await?; // add_from_reader() s'occupe de lire les 512 premiers octets et de calculer le pk - self.add_from_reader(&file_url, reader, length, collection) + self.add_from_reader(Some(&file_url), reader, length, collection) .await } diff --git a/pmoflac/src/encoder.rs b/pmoflac/src/encoder.rs index f0fd140b..2383e71d 100644 --- a/pmoflac/src/encoder.rs +++ b/pmoflac/src/encoder.rs @@ -2,12 +2,13 @@ use std::{ ffi::{c_void, CString}, io, pin::Pin, + sync::Arc, task::{Context, Poll}, }; use tokio::{ io::{AsyncRead, AsyncReadExt, AsyncWriteExt}, - sync::{mpsc, oneshot}, + sync::{mpsc, oneshot, RwLock}, }; use crate::{ @@ -85,8 +86,6 @@ impl tokio::io::AsyncRead for FlacEncodedStream { } } -use std::sync::Arc; - /// Extracted metadata values for FLAC encoding. /// /// This is a simple struct containing the extracted values from TrackMetadata, @@ -121,7 +120,7 @@ pub struct EncoderOptions { /// Metadata to embed in the FLAC file (Vorbis Comments). /// Default: None (no metadata) - pub metadata: Option>, + pub metadata: Option>>, } impl Default for EncoderOptions { @@ -246,7 +245,8 @@ where } // Extract metadata before spawn_blocking (since TrackMetadata has async methods) - let extracted_metadata = if let Some(metadata) = &options.metadata { + let extracted_metadata = if let Some(metadata_lock) = &options.metadata { + let metadata = metadata_lock.read().await; let title = metadata.get_title().await.ok().flatten(); let artist = metadata.get_artist().await.ok().flatten(); let album = metadata.get_album().await.ok().flatten(); diff --git a/pmosource/src/cache.rs b/pmosource/src/cache.rs index 09078879..f0e58b36 100644 --- a/pmosource/src/cache.rs +++ b/pmosource/src/cache.rs @@ -230,7 +230,7 @@ impl SourceCacheManager { { let pk = self .audio_cache - .add_from_reader(source_uri, reader, length, Some(&self.collection_id)) + .add_from_reader(Some(source_uri), reader, length, Some(&self.collection_id)) .await .map_err(|e| MusicSourceError::CacheError(e.to_string()))?; Ok(pk)