diff --git a/.kilo/plans/1775302116634-sunny-nebula.md b/.kilo/plans/1775302116634-sunny-nebula.md new file mode 100644 index 00000000..11d01b94 --- /dev/null +++ b/.kilo/plans/1775302116634-sunny-nebula.md @@ -0,0 +1,151 @@ +# Plan: pmowebrenderer - Améliorations et Multi-client avec DSP + +## Objectifs + +1. **Améliorer l'intégration UPnP Control** - Meilleur fonctionnement des commandes Play/Pause/Seek + - Contrôle piloté près de la sortie (streaming) plutôt qu'au début du pipeline + - Pour Pause: latence actuelle trop importante + - Play/Pause/Seek doivent fonctionner simultanément sur tous les clients + +2. **Améliorer la performance - Latence** - Réduire le délai entre l'envoi et la lecture + +3. **Ajouter le support multi-client avec DSP** - Chaque client peut avoir son propre pipeline DSP + +## Comportement UPnP Control + +### Mode Radio (flux infini) +- Pas de pause possible, hanya next ou stop +- Seek n'a pas de sens + + +### Architecture +``` +PlayerSource → ResamplingNode → ToI24Node + ├──→ [DSP Client 1] → StreamingOggFlacSink 1 + ├──→ [DSP Client 2] → StreamingOggFlacSink 2 + └──→ ... (dynamique) +``` + +Le control point UPnP voit UN seul Media Renderer. Les commandes Play/Pause/Seek affectent TOUTES les sorties client simultanément. + +## État Actuel + +Le pipeline actuel est linéaire pour un seul client: +``` +PlayerSource → ResamplingNode (96kHz) → ToI24Node → StreamingOggFlacSink +``` + +**Note importante:** Utiliser les crates pmoaudio et pmoaudio-ext existantes. Il est possible d'avoir plusieurs `StreamingOggFlacSink` consommant le même flux. Après ToI24Node, brancher en étoiles les différents DSP pour les différents clients. + +## Plan d'Implémentation + +### Phase 1: Amélioration UPnP Control + +1. **Analyser les handlers existants** dans `handlers.rs` +2. **Identifier les problèmes** avec Play/Pause/Seek: + - Timing des transitions d'état + - Gestion des erreurs + - Synchronisation entre clients HTTP et état UPnP +3. **Améliorer la fiabilité** des commandes + - Piloter le contrôle près de la sortie (streaming) + - Différerencier le comportement radio vs piste finie + +### Phase 2: Amélioration Latence + +1. **Réduire le buffer** dans `StreamingOggFlacSink` +2. **Optimiser le pacing** (actuellement max 0.5s ahead) +3. **Améliorer la directité** du chemin audio + +### Phase 3: Architecture Multi-client avec DSP + +1. **Refactorer le pipeline** pour supporter plusieurs clients comme décrit ci-dessus + +2. **Créer un système de DSP** dans pmoaudio ou pmoaudio-ext: + - Interface commune pour les effets audio + - Config des DSP via PMOconfig + - Room correction: equalizer, FIR filter, delay, gain + +3. **Gérer le cycle de vie**: + - Création du pipeline par client + - Nettoyage lors de la déconnexion + - Partage de la source commune entre clients + +## Fichiers à Modifier + +- `pipeline.rs` - Refactoring pour multi-client +- `handlers.rs` - Amélioration UPnP control +- `stream.rs` - Gestion multi-client +- `state.rs` - État par client +- pmoaudio ou pmoaudio-ext pour les mécanismes DSP + +## Défis Potentiels + +- Performance CPU avec plusieurs clients +- Synchronisation des clients avec le même contenu +- Gestion du gapless entre les pistes avec multi-client + +### Gestion de la Pause + +**Option recommandée: Silence (zéros)** +- Pendant la pause, continuer à envoyer des zéros encodés en FLAC +- Le client HTTP maintient sa connexion TCP alive +- Pas de reconnexion nécessaire quand on reprend la lecture +- Avantage: Seamless pour le client + +**Pourquoi pas réduction du sample rate:** +- Le header FLAC définit le sample rate en固定entête +- Changer le sample rate en cours de flux invalidate le flux entier +- Rebuild du flux serait plus complexe que le gain obtenu +- FLAC compresse très bien les zéros de toute façon (beaucoup de répétitions) + +**Autre option envisagée mais non recommandée:** +- Suspendre l'envoi: Le client HTTP va timeout et se déconnecter +- Segment OGG avec metadata: Complexe à implémenter, nécessite modification du client + +### Phase 0: StreamingOggFlacSink avec contrôle Pause + +**Distinction Radio vs Pistes finies:** + +| Mode | Comportement pendant Pause | +|------|---------------------------| +| **Radio (flux infini)** | Les chunks qui arrivent sont ignorés/perdus. On envoie du silence. La source continue à produire mais on n'en tient pas compte. | +| **Pistes finies** | On bloque la consommation des chunks. Par backpressure, le pipeline en amont s'arrête (TimerBufferNode arrête d'envoyer). La lecture est truly arrêtée. | + +**Architecture actuelle analysée:** +``` +AudioSegment → StreamingOggFlacSink → FLAC encoder → OGG wrapper → timed_broadcast → clients +``` + +**Implémentation suggérée:** + +1. **État de lecture distingué:** + - `PlaybackMode::Radio` - ignore les chunks entrants pendant pause + - `PlaybackMode::Track` - bloque la consommation (backpressure) + +2. **Dans SharedSinkContext:** + ```rust + pub enum PlaybackMode { + Radio, // Flux infini - ignore chunks pendant pause + Track, // Piste finie - block par backpressure + } + + pub playback_mode: PlaybackMode, + pub is_paused: Arc, + ``` + +3. **Traitement différent selon le mode:** + - **Radio**: Si `is_paused`, envoyer silence (zéros) mais perdre les chunks entrants + - **Track**: Si `is_paused`, ne pas consommer les chunks → backpressure → arrêt du pipeline en amont + +4. **Transition automatique:** + - Détecter le type de contenu via les métadonnées du TrackBoundary + - **Enrichir TrackBoundary** avec un champ `stream_type`: + ```rust + pub enum StreamType { + Continuous, // Radio/webcast - flux infini + Finite, // Piste/album - flux avec fin définie + } + + pub stream_type: StreamType, + ``` + - Si durée inconnue = Radio (Continuous), si durée connue = Track (Finite) diff --git a/Cargo.lock b/Cargo.lock index 66277c1f..14d3f466 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4,7 +4,7 @@ version = 4 [[package]] name = "PMOMusic" -version = "0.3.32" +version = "0.3.33" dependencies = [ "axum 0.8.7", "console-subscriber", diff --git a/pmoaudio-ext/src/sinks/flac_cache_sink.rs b/pmoaudio-ext/src/sinks/flac_cache_sink.rs index 32a99b42..d5fb4b4e 100755 --- a/pmoaudio-ext/src/sinks/flac_cache_sink.rs +++ b/pmoaudio-ext/src/sinks/flac_cache_sink.rs @@ -233,7 +233,7 @@ impl NodeLogic for FlacCacheSinkLogic { } } _AudioSegment::Sync(marker) => match &**marker { - SyncMarker::TrackBoundary { metadata } => { + SyncMarker::TrackBoundary { metadata, .. } => { // TrackBoundary pendant le prebuffer - track courte (< 512KB) tracing::warn!( "FlacCacheSink: TrackBoundary received before prebuffer complete - track shorter than 512KB, closing pump and waiting for ingestion" @@ -589,7 +589,7 @@ impl NodeLogic for FlacCacheSinkLogic { // Si pump_closed, ignorer silencieusement le chunk } _AudioSegment::Sync(marker) => match &**marker { - SyncMarker::TrackBoundary { metadata } => { + SyncMarker::TrackBoundary { metadata, .. } => { // Nouveau morceau - fermer le pump si pas déjà fermé tracing::debug!("FlacCacheSink: TrackBoundary received, closing pump and storing metadata for next track"); // Stocker les métadonnées pour la prochaine track diff --git a/pmoaudio-ext/src/sinks/streaming_flac_sink.rs b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs index fa46346d..21b9a739 100644 --- a/pmoaudio-ext/src/sinks/streaming_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs @@ -307,7 +307,7 @@ impl NodeLogic for StreamingFlacSinkLogic { _AudioSegment::Sync(marker) => { match marker.as_ref() { - SyncMarker::TrackBoundary { metadata } => { + SyncMarker::TrackBoundary { metadata, .. } => { // Prepare encoder options (metadata + duration) for the upcoming track. if let Err(e) = self.ctx.prepare_encoder_options_for_track(metadata).await diff --git a/pmoaudio-ext/src/sinks/streaming_icyflac_sink.rs b/pmoaudio-ext/src/sinks/streaming_icyflac_sink.rs index dda47325..7c304287 100644 --- a/pmoaudio-ext/src/sinks/streaming_icyflac_sink.rs +++ b/pmoaudio-ext/src/sinks/streaming_icyflac_sink.rs @@ -88,8 +88,17 @@ impl IcyClientStream { // Add cover URL if we have a cover_pk if let Some(pk) = &meta.cover_pk { + #[cfg(feature = "playlist")] let cover_url = pmocache::covers_absolute_url_for_upnp(pk, None); - metadata_str.push_str(&format!("StreamUrl='{}';", cover_url)); + #[cfg(feature = "playlist")] + { + metadata_str.push_str(&format!("StreamUrl='{}';", cover_url)); + } + #[cfg(not(feature = "playlist"))] + { + // When playlist feature is not enabled, use relative URL + metadata_str.push_str(&format!("StreamUrl='/covers/image/{}/256';", pk)); + } } else if let Some(url) = &meta.cover_url { // Fallback to external cover URL if no local pk metadata_str.push_str(&format!("StreamUrl='{}';", url)); diff --git a/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs index 73d233a3..6430dc04 100644 --- a/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs @@ -265,7 +265,7 @@ impl NodeLogic for StreamingOggFlacSinkLogic { _AudioSegment::Sync(marker) => { match marker.as_ref() { - SyncMarker::TrackBoundary { metadata } => { + SyncMarker::TrackBoundary { metadata, .. } => { // Inject per-track metadata and duration into the next FLAC header. if let Err(e) = self.ctx.prepare_encoder_options_for_track(metadata).await diff --git a/pmoaudio-ext/src/sources/player_source.rs b/pmoaudio-ext/src/sources/player_source.rs index b4a366d4..051421ed 100644 --- a/pmoaudio-ext/src/sources/player_source.rs +++ b/pmoaudio-ext/src/sources/player_source.rs @@ -35,6 +35,7 @@ use pmoaudio::{ AudioSegment, nodes::AudioError, pipeline::{AudioPipelineNode, Node, NodeLogic, send_to_children}, + StreamType, }; use tokio::sync::{broadcast, mpsc}; use tokio_util::sync::CancellationToken; @@ -505,7 +506,7 @@ async fn send_track_boundary( let _ = meta.set_title(Some(u.to_string())).await; } let meta_arc = Arc::new(tokio::sync::RwLock::new(meta)); - let boundary = AudioSegment::new_track_boundary(0, timestamp_sec, meta_arc); + let boundary = AudioSegment::new_track_boundary(0, timestamp_sec, meta_arc, StreamType::Finite); send_to_children("PlayerSource", output, boundary).await } diff --git a/pmoaudio-ext/src/sources/playlist_source.rs b/pmoaudio-ext/src/sources/playlist_source.rs index f8d1629b..d7ca21f8 100644 --- a/pmoaudio-ext/src/sources/playlist_source.rs +++ b/pmoaudio-ext/src/sources/playlist_source.rs @@ -113,7 +113,7 @@ use pmoaudio::{ nodes::{AudioError, TypedAudioNode, DEFAULT_CHUNK_DURATION_MS}, pipeline::{send_to_children, AudioPipelineNode, Node, NodeLogic}, type_constraints::TypeRequirement, - AudioSegment, + AudioSegment, StreamType, }; use pmoaudiocache::Cache as AudioCache; use pmoflac::decode_audio_stream; @@ -279,7 +279,7 @@ impl NodeLogic for PlaylistSourceLogic { let track_start = std::time::Instant::now(); tracing::debug!("PlaylistSourceLogic: emitting TrackBoundary"); let metadata_for_boundary = metadata.clone(); - let boundary = AudioSegment::new_track_boundary(0, 0.0, metadata_for_boundary); + let boundary = AudioSegment::new_track_boundary(0, 0.0, metadata_for_boundary, StreamType::Finite); send_to_children(node_name, &output, boundary).await?; // Obtenir le chemin du fichier diff --git a/pmoaudio/src/audio_segment.rs b/pmoaudio/src/audio_segment.rs index 1121c27b..558776ae 100755 --- a/pmoaudio/src/audio_segment.rs +++ b/pmoaudio/src/audio_segment.rs @@ -3,7 +3,7 @@ use tokio::sync::RwLock; use pmometadata::TrackMetadata; -use crate::{gain_db_from_linear, AudioChunk, AudioChunkData, BitDepth, SyncMarker}; +use crate::{gain_db_from_linear, AudioChunk, AudioChunkData, BitDepth, StreamType, SyncMarker}; pub enum _AudioSegment { Chunk(Arc), @@ -160,9 +160,11 @@ impl AudioSegment { order: u64, timestamp_sec: f64, metadata: Arc>, + stream_type: StreamType, ) -> Arc { let marker = Arc::new(SyncMarker::TrackBoundary { metadata: Arc::clone(&metadata), + stream_type, }); Arc::new(Self { order, @@ -303,7 +305,18 @@ impl AudioSegment { pub fn as_track_metadata(&self) -> Option<&Arc>> { match &self.segment { _AudioSegment::Sync(marker) => match &**marker { - SyncMarker::TrackBoundary { metadata } => Some(metadata), + SyncMarker::TrackBoundary { metadata, .. } => Some(metadata), + _ => None, + }, + _ => None, + } + } + + /// Récupère le stream_type si c'est un TrackBoundary + pub fn stream_type(&self) -> Option { + match &self.segment { + _AudioSegment::Sync(marker) => match &**marker { + SyncMarker::TrackBoundary { stream_type, .. } => Some(*stream_type), _ => None, }, _ => None, diff --git a/pmoaudio/src/lib.rs b/pmoaudio/src/lib.rs index ac92ce00..4f783f9d 100755 --- a/pmoaudio/src/lib.rs +++ b/pmoaudio/src/lib.rs @@ -95,7 +95,7 @@ pub mod bit_depth; pub mod dsp; pub use audio_segment::{AudioSegment, _AudioSegment}; -pub use sync_marker::SyncMarker; +pub use sync_marker::{StreamType, SyncMarker}; pub use audio_chunk::{ gain_db_from_linear, gain_linear_from_db, AudioChunk, AudioChunkData, AudioFloatChunk, diff --git a/pmoaudio/src/nodes/file_source.rs b/pmoaudio/src/nodes/file_source.rs index ff5de77c..424864f6 100755 --- a/pmoaudio/src/nodes/file_source.rs +++ b/pmoaudio/src/nodes/file_source.rs @@ -2,7 +2,7 @@ use crate::{ nodes::{AudioError, TypedAudioNode, DEFAULT_CHUNK_DURATION_MS}, pipeline::{send_to_children, AudioPipelineNode, Node, NodeLogic}, type_constraints::TypeRequirement, - AudioChunk, AudioChunkData, AudioSegment, I24, + AudioChunk, AudioChunkData, AudioSegment, I24, StreamType, }; use pmoflac::{decode_audio_stream, AudioFileMetadata, StreamInfo}; use pmometadata::{MemoryTrackMetadata, TrackMetadata}; @@ -98,6 +98,7 @@ impl NodeLogic for FileSourceLogic { 0, 0.0, Arc::new(tokio::sync::RwLock::new(metadata)), + StreamType::Finite, ); send_to_children(std::any::type_name::(), &output, track_boundary).await?; } diff --git a/pmoaudio/src/nodes/flac_file_sink.rs b/pmoaudio/src/nodes/flac_file_sink.rs index 204d6bbc..45e612de 100755 --- a/pmoaudio/src/nodes/flac_file_sink.rs +++ b/pmoaudio/src/nodes/flac_file_sink.rs @@ -2,7 +2,7 @@ use crate::{ nodes::{AudioError, TypedAudioNode, DEFAULT_CHANNEL_SIZE}, pipeline::{Node, NodeLogic}, type_constraints::TypeRequirement, - AudioChunk, AudioPipelineNode, AudioSegment, SyncMarker, + AudioChunk, AudioPipelineNode, AudioSegment, SyncMarker, StreamType, }; use pmoflac::{encode_flac_stream, EncoderOptions, PcmFormat}; use std::{ @@ -822,6 +822,7 @@ mod tests { 0, 0.0, std::sync::Arc::new(tokio::sync::RwLock::new(metadata)), + crate::StreamType::Finite, ); tx.send(track_boundary).await.unwrap(); diff --git a/pmoaudio/src/nodes/http_source.rs b/pmoaudio/src/nodes/http_source.rs index 3ca51ae3..ee7b548a 100755 --- a/pmoaudio/src/nodes/http_source.rs +++ b/pmoaudio/src/nodes/http_source.rs @@ -2,7 +2,7 @@ use crate::{ nodes::{AudioError, TypedAudioNode, DEFAULT_CHUNK_DURATION_MS}, pipeline::{send_to_children, Node, NodeLogic}, type_constraints::TypeRequirement, - AudioChunk, AudioChunkData, AudioPipelineNode, AudioSegment, I24, + AudioChunk, AudioChunkData, AudioPipelineNode, AudioSegment, I24, StreamType, }; use futures_util::StreamExt; use pmoflac::{decode_audio_stream, StreamInfo}; @@ -178,7 +178,7 @@ impl NodeLogic for HttpSourceLogic { // Émettre TrackBoundary avec les métadonnées HTTP let track_boundary = - AudioSegment::new_track_boundary(0, 0.0, Arc::new(tokio::sync::RwLock::new(metadata))); + AudioSegment::new_track_boundary(0, 0.0, Arc::new(tokio::sync::RwLock::new(metadata)), StreamType::Continuous); send_to_children(std::any::type_name::(), &output, track_boundary).await?; // Préparer la lecture des chunks audio diff --git a/pmoaudio/src/nodes/resampling_node.rs b/pmoaudio/src/nodes/resampling_node.rs index de17e5bb..b7b15467 100644 --- a/pmoaudio/src/nodes/resampling_node.rs +++ b/pmoaudio/src/nodes/resampling_node.rs @@ -33,7 +33,7 @@ use crate::{ nodes::{AudioError, TypedAudioNode}, pipeline::{send_to_children, AudioPipelineNode, Node, NodeLogic}, type_constraints::TypeRequirement, - AudioChunk, AudioChunkData, AudioSegment, BitDepth, I24, + AudioChunk, AudioChunkData, AudioSegment, BitDepth, I24, StreamType, }; use std::sync::Arc; use tokio::sync::mpsc; @@ -495,7 +495,7 @@ mod tests { let metadata = Arc::new(tokio::sync::RwLock::new( pmometadata::MemoryTrackMetadata::new(), )); - let boundary = AudioSegment::new_track_boundary(0, 0.0, metadata); + let boundary = AudioSegment::new_track_boundary(0, 0.0, metadata, StreamType::Finite); // Envoyer le boundary input_tx.send(boundary.clone()).await.unwrap(); diff --git a/pmoaudio/src/sync_marker.rs b/pmoaudio/src/sync_marker.rs index 57ba0ec0..7b70a0fe 100755 --- a/pmoaudio/src/sync_marker.rs +++ b/pmoaudio/src/sync_marker.rs @@ -3,9 +3,16 @@ use tokio::sync::RwLock; use pmometadata::TrackMetadata; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum StreamType { + Continuous, + Finite, +} + pub enum SyncMarker { TrackBoundary { metadata: Arc>, + stream_type: StreamType, }, StreamMetadata { key: String, @@ -15,5 +22,4 @@ pub enum SyncMarker { Heartbeat, EndOfStream, Error(String), - // autres cas à venir… } diff --git a/pmoparadise/src/radio_paradise_stream_source.rs b/pmoparadise/src/radio_paradise_stream_source.rs index b046b41f..92a5874e 100644 --- a/pmoparadise/src/radio_paradise_stream_source.rs +++ b/pmoparadise/src/radio_paradise_stream_source.rs @@ -13,7 +13,7 @@ use pmoaudio::{ nodes::{AudioError, TypedAudioNode, DEFAULT_CHUNK_DURATION_MS}, pipeline::{send_to_children, send_to_children_with_timing, Node, NodeLogic}, type_constraints::TypeRequirement, - AudioPipelineNode, AudioSegment, SyncMarker, I24, + AudioPipelineNode, AudioSegment, SyncMarker, I24, StreamType, }; use pmoflac::decode_audio_stream; use pmometadata::{MemoryTrackMetadata, TrackMetadata}; @@ -243,6 +243,7 @@ impl RadioParadiseStreamSourceLogic { let track_boundary = AudioSegment::new_track_boundary( *order, 0.0, // timestamp = 0 au début du stream metadata, + StreamType::Continuous, ); self.send_to_children(output, track_boundary).await?; song_index = 1; @@ -336,7 +337,7 @@ impl RadioParadiseStreamSourceLogic { let metadata = song_to_metadata(song, block).await; let timestamp_sec = total_samples as f64 / sample_rate as f64; let track_boundary = - AudioSegment::new_track_boundary(*order, timestamp_sec, metadata); + AudioSegment::new_track_boundary(*order, timestamp_sec, metadata, StreamType::Continuous); self.send_to_children(output, track_boundary).await?; // Passer à la song suivante