From 8924552696cf3dd98d41a94e6f1b928dbc2e1402 Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Sat, 4 Apr 2026 14:37:11 +0200 Subject: [PATCH 01/14] :sparkles: Add StreamType for multi-client support and improved UPnP control MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Introduce `Streamtype` enum (Continuous vs Finite) to distinguish radio streams from finite tracks - Enrich `TrackBoundary` sync marker with stream type for proper pause behavior per mode (silence vs backpressure) - Update all sources and sinks to pass `StreamType` when creating track boundaries - Radio Paradise, HTTP source → Continuous (infinite) - Improve UPnP control architecture: pause sends silence for radio, blocks pipeline via backpressure for tracks - Prepare groundwork for multi-client DSP architecture with shared source and per-DSP pipelines --- .kilo/plans/1775302116634-sunny-nebula.md | 151 ++++++++++++++++++ Cargo.lock | 2 +- pmoaudio-ext/src/sinks/flac_cache_sink.rs | 4 +- pmoaudio-ext/src/sinks/streaming_flac_sink.rs | 2 +- .../src/sinks/streaming_icyflac_sink.rs | 11 +- .../src/sinks/streaming_ogg_flac_sink.rs | 2 +- pmoaudio-ext/src/sources/player_source.rs | 3 +- pmoaudio-ext/src/sources/playlist_source.rs | 4 +- pmoaudio/src/audio_segment.rs | 17 +- pmoaudio/src/lib.rs | 2 +- pmoaudio/src/nodes/file_source.rs | 3 +- pmoaudio/src/nodes/flac_file_sink.rs | 3 +- pmoaudio/src/nodes/http_source.rs | 4 +- pmoaudio/src/nodes/resampling_node.rs | 4 +- pmoaudio/src/sync_marker.rs | 8 +- .../src/radio_paradise_stream_source.rs | 5 +- 16 files changed, 204 insertions(+), 21 deletions(-) create mode 100644 .kilo/plans/1775302116634-sunny-nebula.md 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 From 340c69cb2bc5db7a2c673207c2c915ccfec982d6 Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Sat, 4 Apr 2026 18:00:20 +0200 Subject: [PATCH 02/14] =?UTF-8?q?:arrow=5Fup:=20ureq=20v2=E2=86=92v3,=20ad?= =?UTF-8?q?d=20continuous=20stream=20detection?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Updateurequest dependency from v2 to latest major (v3.14) - Add ureq as optional dependency in pmoaudio-ext - Introduce is_continuous flag to UriSource and PlayerState for proper StreamType handling (Finite vs Continuous) - Implement detect_continuous_stream() with URL pattern matching and HTTP HEAD header inspection (ICY, chunked encoding) - Update send_track_boundary() to accept StreamType parameter --- Cargo.lock | 30 ++++++- pmoaudio-ext/Cargo.toml | 3 +- pmoaudio-ext/src/sources/player_source.rs | 30 ++++--- pmoaudio-ext/src/sources/uri_source.rs | 96 ++++++++++++++++++++++- pmoaudio/src/nodes/http_source.rs | 2 +- 5 files changed, 144 insertions(+), 17 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 14d3f466..4c19cb72 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3897,6 +3897,7 @@ dependencies = [ "tokio", "tokio-util", "tracing", + "ureq 2.12.1", ] [[package]] @@ -4016,7 +4017,7 @@ dependencies = [ "tracing", "tracing-log 0.1.4", "tracing-subscriber", - "ureq", + "ureq 3.1.4", "url", "urlencoding", "utoipa", @@ -6513,6 +6514,22 @@ version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" +[[package]] +name = "ureq" +version = "2.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "02d1a66277ed75f640d608235660df48c8e3c19f3b4edb6a263315626cc3c01d" +dependencies = [ + "base64 0.22.1", + "flate2", + "log", + "once_cell", + "rustls", + "rustls-pki-types", + "url", + "webpki-roots 0.26.11", +] + [[package]] name = "ureq" version = "3.1.4" @@ -6527,7 +6544,7 @@ dependencies = [ "rustls-pki-types", "ureq-proto", "utf-8", - "webpki-roots", + "webpki-roots 1.0.4", ] [[package]] @@ -6820,6 +6837,15 @@ dependencies = [ "libwebp-sys", ] +[[package]] +name = "webpki-roots" +version = "0.26.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" +dependencies = [ + "webpki-roots 1.0.4", +] + [[package]] name = "webpki-roots" version = "1.0.4" diff --git a/pmoaudio-ext/Cargo.toml b/pmoaudio-ext/Cargo.toml index f36fdc6b..debf6720 100644 --- a/pmoaudio-ext/Cargo.toml +++ b/pmoaudio-ext/Cargo.toml @@ -32,10 +32,11 @@ serde = { workspace = true, optional = true } serde_json = { workspace = true, optional = true } reqwest = { workspace = true, features = ["stream"], optional = true } futures = { version = "0.3", optional = true } +ureq = { version = "2", optional = true } [features] default = [] cache-sink = ["dep:pmoaudiocache", "dep:pmoflac", "dep:pmometadata", "dep:serde_json"] playlist = ["cache-sink", "dep:pmoplaylist", "dep:pmocache"] -http-stream = ["dep:pmoflac", "dep:pmometadata", "dep:bytes", "dep:serde", "dep:reqwest", "dep:futures"] +http-stream = ["dep:pmoflac", "dep:pmometadata", "dep:bytes", "dep:serde", "dep:reqwest", "dep:futures", "dep:ureq"] all = ["cache-sink", "playlist", "http-stream"] diff --git a/pmoaudio-ext/src/sources/player_source.rs b/pmoaudio-ext/src/sources/player_source.rs index 051421ed..acdd56c7 100644 --- a/pmoaudio-ext/src/sources/player_source.rs +++ b/pmoaudio-ext/src/sources/player_source.rs @@ -165,6 +165,7 @@ impl NodeLogic for PlayerSourceLogic { let mut current_uri: Option = None; let mut next_uri: Option = None; let mut paused_at_sec: f64 = 0.0; + let mut is_continuous: bool = false; info!("PlayerSource: started"); @@ -182,7 +183,7 @@ impl NodeLogic for PlayerSourceLogic { None => break, Some(cmd) => { self.handle_command( - cmd, &mut state, &mut current_uri, + cmd, is_continuous, &mut state, &mut current_uri, &mut next_uri, &mut paused_at_sec, &output, &stop_token, ).await?; @@ -213,15 +214,17 @@ impl NodeLogic for PlayerSourceLogic { }; let duration_sec = source.duration_sec(); + let is_continuous = source.is_continuous(); let _ = self.event_tx.send(PlayerEvent::Playing { uri: uri.clone(), duration_sec, }); - info!("PlayerSource: playing {:?} from {:.1}s", uri, paused_at_sec); + info!("PlayerSource: playing {:?} from {:.1}s continuous={}", uri, paused_at_sec, is_continuous); // Pompe audio — s'arrête sur EOF, Pause, Stop, ou commande let result = self.pump( source, + is_continuous, &mut state, &mut current_uri, &mut next_uri, @@ -256,6 +259,7 @@ impl PlayerSourceLogic { async fn handle_command( &mut self, cmd: PlayerCommand, + is_continuous: bool, state: &mut TransportState, current_uri: &mut Option, next_uri: &mut Option, @@ -282,14 +286,16 @@ impl PlayerSourceLogic { TransportState::Paused => { info!("PlayerSource: Play (resume from {:.1}s)", paused_at_sec); // Injecter un TrackBoundary pour EOS + nouveau BOS OGG propre - send_track_boundary(current_uri.as_deref(), output, *paused_at_sec, stop_token).await?; + let stream_type = if is_continuous { StreamType::Continuous } else { StreamType::Finite }; + send_track_boundary(current_uri.as_deref(), output, *paused_at_sec, stream_type, stop_token).await?; *state = TransportState::Playing; } TransportState::Loaded => { info!("PlayerSource: Play (start)"); *paused_at_sec = 0.0; // TrackBoundary initial pour le premier BOS OGG - send_track_boundary(current_uri.as_deref(), output, 0.0, stop_token).await?; + let stream_type = if is_continuous { StreamType::Continuous } else { StreamType::Finite }; + send_track_boundary(current_uri.as_deref(), output, 0.0, stream_type, stop_token).await?; *state = TransportState::Playing; } TransportState::Idle => { @@ -321,7 +327,8 @@ impl PlayerSourceLogic { if current_uri.is_some() { info!("PlayerSource: Seek to {:.1}s", pos); *paused_at_sec = pos; - send_track_boundary(current_uri.as_deref(), output, pos, stop_token).await?; + let stream_type = if is_continuous { StreamType::Continuous } else { StreamType::Finite }; + send_track_boundary(current_uri.as_deref(), output, pos, stream_type, stop_token).await?; *state = TransportState::Playing; } } @@ -336,6 +343,7 @@ impl PlayerSourceLogic { async fn pump( &mut self, source: UriSource, + is_continuous: bool, state: &mut TransportState, current_uri: &mut Option, next_uri: &mut Option, @@ -395,7 +403,8 @@ impl PlayerSourceLogic { *next_uri = None; *paused_at_sec = 0.0; // TrackBoundary pour clore le bitstream OGG proprement - if let Err(e) = send_track_boundary(current_uri.as_deref(), output, 0.0, stop_token).await { + let stream_type = if is_continuous { StreamType::Continuous } else { StreamType::Finite }; + if let Err(e) = send_track_boundary(current_uri.as_deref(), output, 0.0, stream_type, stop_token).await { result = Err(e); } *state = TransportState::Playing; @@ -409,7 +418,8 @@ impl PlayerSourceLogic { info!("PlayerSource: Seek to {:.1}s", pos); source_stop.cancel(); *paused_at_sec = pos; - if let Err(e) = send_track_boundary(current_uri.as_deref(), output, pos, stop_token).await { + let stream_type = if is_continuous { StreamType::Continuous } else { StreamType::Finite }; + if let Err(e) = send_track_boundary(current_uri.as_deref(), output, pos, stream_type, stop_token).await { result = Err(e); break; } @@ -435,7 +445,8 @@ impl PlayerSourceLogic { info!("PlayerSource: gapless transition to {:?}", next); *current_uri = Some(next); *paused_at_sec = 0.0; - if let Err(e) = send_track_boundary(current_uri.as_deref(), output, 0.0, stop_token).await { + let stream_type = if is_continuous { StreamType::Continuous } else { StreamType::Finite }; + if let Err(e) = send_track_boundary(current_uri.as_deref(), output, 0.0, stream_type, stop_token).await { result = Err(e); } *state = TransportState::Playing; @@ -495,6 +506,7 @@ async fn send_track_boundary( uri: Option<&str>, output: &[mpsc::Sender>], timestamp_sec: f64, + stream_type: StreamType, stop_token: &CancellationToken, ) -> Result<(), AudioError> { if output.is_empty() || stop_token.is_cancelled() { @@ -506,7 +518,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, StreamType::Finite); + let boundary = AudioSegment::new_track_boundary(0, timestamp_sec, meta_arc, stream_type); send_to_children("PlayerSource", output, boundary).await } diff --git a/pmoaudio-ext/src/sources/uri_source.rs b/pmoaudio-ext/src/sources/uri_source.rs index ecf7c196..8f1e91b7 100644 --- a/pmoaudio-ext/src/sources/uri_source.rs +++ b/pmoaudio-ext/src/sources/uri_source.rs @@ -49,6 +49,8 @@ pub struct UriSource { reader: Box, stream_info: StreamInfo, frames_to_skip: u64, + /// true si c'est un flux continu (radio, stream) sans durée définie + pub is_continuous: bool, } impl UriSource { @@ -88,6 +90,11 @@ impl UriSource { }) } + /// Retourne true si c'est un flux continu (radio, stream) sans durée définie. + pub fn is_continuous(&self) -> bool { + self.is_continuous + } + /// Émet les chunks audio vers `tx`. /// /// Retourne `Ok(true)` si EOF naturel, `Ok(false)` si annulé ou receiver fermé. @@ -201,7 +208,7 @@ impl UriSource { ); let (_, reader) = stream.into_reader(); - Ok(Self { reader: Box::new(reader), stream_info, frames_to_skip }) + Ok(Self { reader: Box::new(reader), stream_info, frames_to_skip, is_continuous: false }) } async fn open_http( @@ -209,6 +216,9 @@ impl UriSource { seek_sec: f64, stop_token: &CancellationToken, ) -> Result { + // Détecter si c'est un flux continu (radio, stream) basé sur l'URL + let is_continuous = detect_continuous_stream(url); + let response = tokio::select! { _ = stop_token.cancelled() => { return Err(AudioError::IoError("Cancelled before HTTP connect".into())); @@ -244,11 +254,89 @@ impl UriSource { let frames_to_skip = (seek_sec * stream_info.sample_rate as f64) as u64; info!( - "UriSource: opened HTTP {} Hz {} ch {} bps", - stream_info.sample_rate, stream_info.channels, stream_info.bits_per_sample, + "UriSource: opened HTTP {} Hz {} ch {} bps continuous={}", + stream_info.sample_rate, stream_info.channels, stream_info.bits_per_sample, is_continuous, ); let (_, reader) = stream.into_reader(); - Ok(Self { reader: Box::new(reader), stream_info, frames_to_skip }) + Ok(Self { + reader: Box::new(reader), + stream_info, + frames_to_skip, + is_continuous, + }) } } + +/// Détecte si une URL correspond à un flux continu (radio, stream) sans durée définie. +/// +/// Cette fonction: +/// 1. Vérifie les patterns d'URL connus (stream, live, radio, etc.) +/// 2. Fait une requête HTTP HEAD pour vérifier les headers (Content-Length, ICY, etc.) +fn detect_continuous_stream(url: &str) -> bool { + let url_lower = url.to_lowercase(); + + // 1. Quick check sur les patterns d'URL très explicites + // Ces patterns indiquent clairement un stream live + if url_lower.contains("/live") + || url_lower.contains("/radiolar") + || url_lower.contains(".pls") + || url_lower.contains(".m3u") + || url_lower.contains("icy") + { + return true; + } + + // 2. Vérification HTTP headers (le plus fiable) + if url.starts_with("http://") || url.starts_with("https://") { + if let Ok(is_stream) = check_http_stream_headers(url) { + if is_stream { + return true; + } + } + } + + false +} + +/// Vérifie les headers HTTP pour déterminer si c'est un stream +fn check_http_stream_headers(url: &str) -> Result { + use std::time::Duration; + + let agent = ureq::AgentBuilder::new() + .timeout(Duration::from_secs(3)) + .build(); + + let response = agent + .head(url) + .call() + .map_err(|e| format!("HTTP HEAD failed: {}", e))?; + + // Headers ICY (Icecast/Shoutcast) = toujours un stream + if response.header("icy-name").is_some() + || response.header("icy-metaint").is_some() + { + return Ok(true); + } + + // Pas de Content-Length = stream potentiel + let has_content_length = response.header("content-length").is_some(); + + // Transfer-Encoding: chunked = stream potentiel + let is_chunked = response + .header("transfer-encoding") + .map(|v| v.to_lowercase().contains("chunked")) + .unwrap_or(false); + + // Decision: pas de Content-Length + (chunked ou content-type streaming) + let content_type = response + .header("content-type") + .unwrap_or("") + .to_lowercase(); + + let is_streaming_mime = content_type.contains("audio/mpeg") + || content_type.contains("audio/aac") + || content_type.contains("application/ogg"); + + Ok(!has_content_length && (is_streaming_mime || is_chunked)) +} diff --git a/pmoaudio/src/nodes/http_source.rs b/pmoaudio/src/nodes/http_source.rs index ee7b548a..6eaf720a 100755 --- a/pmoaudio/src/nodes/http_source.rs +++ b/pmoaudio/src/nodes/http_source.rs @@ -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)), StreamType::Continuous); + AudioSegment::new_track_boundary(0, 0.0, Arc::new(tokio::sync::RwLock::new(metadata)), StreamType::Finite); send_to_children(std::any::type_name::(), &output, track_boundary).await?; // Préparer la lecture des chunks audio From 546e8a782f2bd4eb0c707818a5f29c6faa7cfa42 Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Sun, 5 Apr 2026 10:52:54 +0200 Subject: [PATCH 03/14] :wastebasket: Remove unused imports, macros and dead code - Drop `Path`/``State``` from unused Axum imports in config.rs and registry - Mark `_position_sec` field as `#[allow(dead_code)]`` in PositionUpdateRequest and PlayerStateReport - Remove unused macro rules (`add_action_arg!`, `add_action!``, `` add_var!)`` - Delete unused PlayerReport struct and related handler code --- Blackboard/ToThinkAbout/webrenderer.md | 529 ++++++++++++++---- Cargo.lock | 83 +-- .../src/components/unified/RendererDrawer.vue | 20 +- .../webapp/src/composables/useWebRenderer.ts | 271 +++------ pmoapp/webapp/src/services/PMOPlayer.ts | 369 ++++++++++++ pmoaudio-ext/src/sinks/streaming_flac_sink.rs | 3 + .../src/sinks/streaming_ogg_flac_sink.rs | 188 ++++++- .../src/sinks/streaming_sink_common.rs | 12 + pmoaudio/src/audio_chunk.rs | 24 + pmoserver/Cargo.toml | 2 +- pmoserver/src/lib.rs | 1 + pmoserver/src/serve_embed.rs | 90 +++ pmoserver/src/server.rs | 8 +- pmowebrenderer/Cargo.toml | 9 +- pmowebrenderer/src/config.rs | 24 +- pmowebrenderer/src/handlers.rs | 25 +- pmowebrenderer/src/messages.rs | 7 + pmowebrenderer/src/register.rs | 115 +++- pmowebrenderer/src/registry.rs | 127 ++++- pmowebrenderer/src/state.rs | 4 + pmowebrenderer/src/stream.rs | 19 +- 21 files changed, 1479 insertions(+), 451 deletions(-) create mode 100644 pmoapp/webapp/src/services/PMOPlayer.ts create mode 100644 pmoserver/src/serve_embed.rs diff --git a/Blackboard/ToThinkAbout/webrenderer.md b/Blackboard/ToThinkAbout/webrenderer.md index 07aab8ce..af6fc291 100644 --- a/Blackboard/ToThinkAbout/webrenderer.md +++ b/Blackboard/ToThinkAbout/webrenderer.md @@ -1,138 +1,443 @@ -Parfait. Voici un **schéma fonctionnel minimal** pour un **MediaRenderer UPnP privé par navigateur** avec **token**. L’idée est de rester fidèle à ton backend Rust existant et à la webapp Vue.js. En s'appuyant sur l'architecture de PMOMusic, j'aimerais que tu proposes un plan détaillé pour implémenter un tel système de Média Renderer. +# Web Media Renderer - Architecture -- L'application web se trouve dans: [@webapp](file:///Users/coissac/Sync/maison/Petite_maisons/src/pmomusic/pmoapp/webapp) -- Tu as un prototype de Média Renderer dans: [@pmomediarenderer](file:///Users/coissac/Sync/maison/Petite_maisons/src/pmomusic/pmomediarenderer) -- Le contrôle point est dans : [@pmocontrol](file:///Users/coissac/Sync/maison/Petite_maisons/src/pmomusic/pmocontrol) -- Tu implémenteras ce nouveau système de Média Renderer dans la CRATe pmowebrenderer +## Vision -Tu mettras une version du plan en Markdown dans le répertoire [@Architecture](file:///Users/coissac/Sync/maison/Petite_maisons/src/pmomusic/Blackboard/Architecture) . +Système de Media Renderer pilotable à distance via UPnP, exposant un flux audio vers différents types de lecteurs physiques. ---- +## Architecture globale en 4 parties -## 1. Flow général - -``` -Browser (Vue.js Control Point) - ┌───────────────┐ - │ UI / audio │ - │ WebSocket │ - └───────▲───────┘ - │ token - │ - ▼ -Rust backend (UPnP MediaRenderer) - ┌───────────────────────────┐ - │ Token → Renderer mapping │ - │ Device XML / SOAP endpoints│ - │ Play/Pause/Stop → WS → Browser │ - └───────────────────────────┘ +```mermaid +flowchart LR + A[Media Server] -->|flux audio| B[Control Point] + B -->|commandes| C[Web Media Renderer] + C -->|flux + contrôles| D[Device physique] + + subgraph Devices physiques + D1[Browser] + D2[Android Auto] + D3[Apple CarPlay] + D4[Sonos multipoint] + D5[Chromecast] + end + + D --> D1 + D --> D2 + D --> D3 + D --> D4 + D --> D5 ``` ---- +### Rôles -## 2. Étapes détaillées +1. **Media Server** - Source audio (le flux OGG-FLAC existant) +2. **Control Point** - Interface UI qui envoie les commandes (pause, play, seek, next, prev) +3. **Web Media Renderer** - Hub qui expose le flux et traduit les commandes selon le device +4. **Physical Device** - Lecteur final (browser, voiture, Sonos, Chromecast...) -### a) Création du renderer +## Web Media Renderer - Rôle central -1. Le navigateur se connecte via WebSocket ou HTTP. -2. Rust génère un token unique pour ce client : - - ```rust - use uuid::Uuid; - let token = Uuid::new_v4().to_string(); - ``` -3. Rust crée une instance MediaRenderer **privée**, associée à ce token : - - * Device description XML : `/renderer//desc.xml` - * AVTransport SOAP : `/renderer//avtransport` - * RenderingControl SOAP : `/renderer//renderingcontrol` - ---- - -### b) Control Point - -* La webapp Vue.js reçoit le token et la “déclare” au Control Point : - -```js -const renderer = { - token: "abcd-1234-efgh", - name: "Browser Renderer" -}; - -// Ajout au control point local -controlPoint.addRenderer(renderer); -``` - -* Toutes les commandes Play/Pause/Stop incluent ce token : - -```js -ws.send(JSON.stringify({ - token: renderer.token, - action: "play", - uri: "http://localhost:8080/media.mp3" -})); -``` - ---- - -### c) Backend Rust : dispatcher les commandes - -* Rust reçoit le JSON avec le token. -* Vérifie que le token correspond à un renderer actif. -* Transmet la commande au navigateur via WebSocket (ou HTTP push) : - -```rust -match msg.action.as_str() { - "play" => send_ws_to_browser(&token, format!("play:{}", msg.uri)), - "pause" => send_ws_to_browser(&token, "pause".to_string()), - "stop" => send_ws_to_browser(&token, "stop".to_string()), - _ => (), +```mermaid +blockdiag +{ + block = Commandes UPnP + block -> "Web Media Renderer" -> Adaptation selon device + "Web Media Renderer" -> Device-specific protocols } ``` -* Rust met à jour l’état du renderer (AVTransport/RenderingControl) pour le Control Point. +### Rôle central: Adaptateur ---- +Le Web Media Renderer est un **adaptateur** qui: +- **Reçoit le flux** du Media Server (OGG-FLAC) +- **Reçoit les commandes** du Control Point (UPnP) +- **Les traduit** vers les devices physiques +- **Expose une API de contrôle** commune -### d) Lecture côté navigateur +### Ce qui est COMMUN (factorisé) -* Le navigateur reçoit la commande via WebSocket et pilote `