From 58c4383023b18d4dcac0375ccbfb6b30f36dcbe1 Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Sat, 15 Nov 2025 15:03:08 +0100 Subject: [PATCH] =?UTF-8?q?Ajout=20d'un=20n=C5=93ud=20de=20cache=20des=20i?= =?UTF-8?q?mages=20dans=20les=20trackboundary?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pmoaudio-ext/src/lib.rs | 6 + pmoaudio-ext/src/nodes/mod.rs | 5 + .../src/nodes/track_boundary_cover_node.rs | 170 ++++++++++++++++++ pmoparadise/src/stream_channel.rs | 35 +++- 4 files changed, 208 insertions(+), 8 deletions(-) create mode 100644 pmoaudio-ext/src/nodes/mod.rs create mode 100644 pmoaudio-ext/src/nodes/track_boundary_cover_node.rs diff --git a/pmoaudio-ext/src/lib.rs b/pmoaudio-ext/src/lib.rs index 7b9727a9..3828c88d 100755 --- a/pmoaudio-ext/src/lib.rs +++ b/pmoaudio-ext/src/lib.rs @@ -25,6 +25,9 @@ #[cfg(any(feature = "cache-sink", feature = "http-stream"))] pub mod sinks; +#[cfg(any(feature = "cache-sink", feature = "http-stream"))] +pub mod nodes; + #[cfg(feature = "playlist")] pub mod sources; @@ -32,5 +35,8 @@ pub mod sources; #[cfg(any(feature = "cache-sink", feature = "http-stream"))] pub use sinks::*; +#[cfg(any(feature = "cache-sink", feature = "http-stream"))] +pub use nodes::*; + #[cfg(feature = "playlist")] pub use sources::*; diff --git a/pmoaudio-ext/src/nodes/mod.rs b/pmoaudio-ext/src/nodes/mod.rs new file mode 100644 index 00000000..17a38e71 --- /dev/null +++ b/pmoaudio-ext/src/nodes/mod.rs @@ -0,0 +1,5 @@ +#[cfg(any(feature = "cache-sink", feature = "http-stream"))] +pub mod track_boundary_cover_node; + +#[cfg(any(feature = "cache-sink", feature = "http-stream"))] +pub use track_boundary_cover_node::TrackBoundaryCoverNode; diff --git a/pmoaudio-ext/src/nodes/track_boundary_cover_node.rs b/pmoaudio-ext/src/nodes/track_boundary_cover_node.rs new file mode 100644 index 00000000..c0303389 --- /dev/null +++ b/pmoaudio-ext/src/nodes/track_boundary_cover_node.rs @@ -0,0 +1,170 @@ +//! Node de conversion qui s'assure que chaque `TrackBoundary` possède un `cover_pk`. +//! +//! Il laisse passer tous les segments audio de manière transparente. Lorsqu'un +//! `TrackBoundary` est détecté, il vérifie si ses métadonnées contiennent déjà +//! un `cover_pk`. Si ce n'est pas le cas mais qu'une `cover_url` est disponible, +//! l'image est sauvegardée dans le cache de couvertures puis la clé primaire est +//! écrite dans les métadonnées avant de poursuivre la propagation. + +use std::sync::Arc; + +use pmoaudio::{ + nodes::{AudioError, DEFAULT_CHANNEL_SIZE}, + pipeline::{AudioPipelineNode, Node, NodeLogic, PipelineHandle}, + AudioSegment, TypeRequirement, TypedAudioNode, +}; +use pmocovers::Cache as CoverCache; +use pmometadata::TrackMetadata; +use tokio::select; +use tokio::sync::{mpsc, RwLock}; +use tokio_util::sync::CancellationToken; +use tracing::{debug, warn}; + +/// Node enveloppe qui applique [`TrackBoundaryCoverLogic`]. +pub struct TrackBoundaryCoverNode { + inner: Node, +} + +impl TrackBoundaryCoverNode { + /// Crée un nouveau node. + pub fn new(cover_cache: Arc) -> Self { + let logic = TrackBoundaryCoverLogic::new(cover_cache); + Self { + inner: Node::new_with_input(logic, DEFAULT_CHANNEL_SIZE), + } + } +} + +impl AudioPipelineNode for TrackBoundaryCoverNode { + fn get_tx(&self) -> Option>> { + self.inner.get_tx() + } + + fn register(&mut self, child: Box) { + self.inner.register(child); + } + + async fn run(self: Box, stop_token: CancellationToken) -> Result<(), AudioError> { + Box::new(self.inner).run(stop_token).await + } + + fn start(self: Box) -> PipelineHandle { + Box::new(self.inner).start() + } +} + +impl TypedAudioNode for TrackBoundaryCoverNode { + fn input_type(&self) -> Option { + Some(TypeRequirement::any()) + } + + fn output_type(&self) -> Option { + Some(TypeRequirement::any()) + } +} + +struct TrackBoundaryCoverLogic { + cover_cache: Arc, +} + +impl TrackBoundaryCoverLogic { + fn new(cover_cache: Arc) -> Self { + Self { cover_cache } + } + + async fn ensure_cover_pk(&self, metadata: Arc>) { + let cover_url = { + let guard = metadata.read().await; + + match guard.get_cover_pk().await { + Ok(Some(pk)) => { + debug!("TrackBoundaryCoverNode: cover_pk already set ({})", pk); + return; + } + Ok(None) => {} + Err(err) => warn!("TrackBoundaryCoverNode: cannot read cover_pk: {}", err), + } + + match guard.get_cover_url().await { + Ok(url) => url, + Err(err) => { + warn!("TrackBoundaryCoverNode: cannot read cover_url: {}", err); + return; + } + } + }; + + let cover_url = match cover_url { + Some(url) => url, + None => { + debug!("TrackBoundaryCoverNode: no cover_url present, skipping cache"); + return; + } + }; + + match self.cover_cache.add_from_url(&cover_url, None).await { + Ok(pk) => { + debug!( + "TrackBoundaryCoverNode: cached cover for url={}, pk={}", + cover_url, pk + ); + let mut guard = metadata.write().await; + if let Err(err) = guard.set_cover_pk(Some(pk.clone())).await { + warn!( + "TrackBoundaryCoverNode: failed to set cover_pk {}: {}", + pk, err + ); + } + } + Err(err) => { + warn!( + "TrackBoundaryCoverNode: failed to cache cover from {}: {}", + cover_url, err + ); + } + } + } +} + +#[async_trait::async_trait] +impl NodeLogic for TrackBoundaryCoverLogic { + async fn process( + &mut self, + input: Option>>, + output: Vec>>, + stop_token: CancellationToken, + ) -> Result<(), AudioError> { + let mut input = input.ok_or_else(|| { + AudioError::ProcessingError( + "TrackBoundaryCoverNode requires an upstream input channel".into(), + ) + })?; + + loop { + let segment = select! { + _ = stop_token.cancelled() => { + debug!("TrackBoundaryCoverNode: stop requested"); + break; + } + segment = input.recv() => segment, + }; + + let Some(segment) = segment else { + debug!("TrackBoundaryCoverNode: upstream closed"); + break; + }; + + if let Some(metadata) = segment.as_track_metadata() { + self.ensure_cover_pk(Arc::clone(metadata)).await; + } + + for tx in &output { + if tx.send(segment.clone()).await.is_err() { + return Err(AudioError::ChildDied); + } + } + } + + Ok(()) + } +} diff --git a/pmoparadise/src/stream_channel.rs b/pmoparadise/src/stream_channel.rs index e554cb71..c5ad6f31 100644 --- a/pmoparadise/src/stream_channel.rs +++ b/pmoparadise/src/stream_channel.rs @@ -18,8 +18,9 @@ use anyhow::Result; use pmoaudio::AudioPipelineNode; use pmoaudio_ext::{ FlacClientStream, IcyClientStream, MetadataSnapshot, OggFlacClientStream, OggFlacStreamHandle, - StreamHandle, StreamingFlacSink, StreamingOggFlacSink, + StreamHandle, StreamingFlacSink, StreamingOggFlacSink, TrackBoundaryCoverNode, }; +use pmocovers::Cache as CoverCache; use pmoflac::EncoderOptions; use tokio::io::{AsyncRead, ReadBuf}; use tokio::sync::Notify; @@ -101,6 +102,7 @@ impl ParadiseStreamChannel { descriptor: ChannelDescriptor, client: RadioParadiseClient, config: ParadiseStreamChannelConfig, + cover_cache: Option>, ) -> Self { let mut source = RadioParadiseStreamSource::new(client.clone()); let block_handle = source.block_handle(); @@ -116,8 +118,15 @@ impl ParadiseStreamChannel { config.max_lead_seconds, ); - source.register(Box::new(flac_sink)); - source.register(Box::new(ogg_sink)); + if let Some(cache) = cover_cache { + let mut cover_node = TrackBoundaryCoverNode::new(cache); + cover_node.register(Box::new(flac_sink)); + cover_node.register(Box::new(ogg_sink)); + source.register(Box::new(cover_node)); + } else { + source.register(Box::new(flac_sink)); + source.register(Box::new(ogg_sink)); + } stream_handle.set_auto_stop(false); ogg_handle.set_auto_stop(false); @@ -165,12 +174,13 @@ impl ParadiseStreamChannel { pub async fn new( descriptor: ChannelDescriptor, config: ParadiseStreamChannelConfig, + cover_cache: Option>, ) -> Result { let client = RadioParadiseClient::builder() .channel(descriptor.id) .build() .await?; - Ok(Self::with_client(descriptor, client, config)) + Ok(Self::with_client(descriptor, client, config, cover_cache)) } /// S'abonne au flux FLAC pur. @@ -360,17 +370,26 @@ impl ParadiseChannelManager { Self { channels } } - pub async fn with_defaults() -> Result { + pub async fn with_defaults_with_cover_cache( + cover_cache: Option>, + ) -> Result { let mut map = HashMap::new(); for descriptor in ALL_CHANNELS.iter().copied() { - let channel = - ParadiseStreamChannel::new(descriptor, ParadiseStreamChannelConfig::default()) - .await?; + let channel = ParadiseStreamChannel::new( + descriptor, + ParadiseStreamChannelConfig::default(), + cover_cache.clone(), + ) + .await?; map.insert(descriptor.id, Arc::new(channel)); } Ok(Self { channels: map }) } + pub async fn with_defaults() -> Result { + Self::with_defaults_with_cover_cache(None).await + } + pub fn get(&self, id: u8) -> Option> { self.channels.get(&id).cloned() }