From 6c40b93089ead98b874f4af7a8dc6491d6f45c99 Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Sat, 25 Oct 2025 21:01:28 +0200 Subject: [PATCH] on retravaille les sources et pmoparadise en particulier --- pmomediaserver/src/sources_api.rs | 36 +- pmoparadise/Cargo.toml | 8 +- pmoparadise/STREAMING_IMPLEMENTATION_GUIDE.md | 610 ++++++++++++++++++ pmoparadise/src/client.rs | 100 +-- pmoparadise/src/paradise/channel.rs | 5 + pmoparadise/src/paradise/mod.rs | 3 +- pmoparadise/src/paradise/worker.rs | 42 +- pmoparadise/src/pmoserver_ext.rs | 42 +- pmoparadise/src/source.rs | 7 +- pmoparadise/src/stream.rs | 46 +- 10 files changed, 682 insertions(+), 217 deletions(-) create mode 100644 pmoparadise/STREAMING_IMPLEMENTATION_GUIDE.md diff --git a/pmomediaserver/src/sources_api.rs b/pmomediaserver/src/sources_api.rs index b26c5586..ee4b8193 100644 --- a/pmomediaserver/src/sources_api.rs +++ b/pmomediaserver/src/sources_api.rs @@ -151,31 +151,17 @@ async fn register_paradise(Json(params): Json) -> impl IntoRespo }; // Créer et enregistrer la source depuis le registry - let source = if let Some(capacity) = params.fifo_capacity { - match RadioParadiseSource::from_registry(client, capacity) { - Ok(s) => Arc::new(s), - Err(e) => { - return ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ErrorResponse { - error: format!("Failed to create source: {}", e), - }), - ) - .into_response(); - } - } - } else { - match RadioParadiseSource::from_registry_default(client) { - Ok(s) => Arc::new(s), - Err(e) => { - return ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ErrorResponse { - error: format!("Failed to create source: {}", e), - }), - ) - .into_response(); - } + // Note: params.fifo_capacity is currently not used by from_registry + let source = match RadioParadiseSource::from_registry(client) { + Ok(s) => Arc::new(s), + Err(e) => { + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(ErrorResponse { + error: format!("Failed to create source: {}", e), + }), + ) + .into_response(); } }; diff --git a/pmoparadise/Cargo.toml b/pmoparadise/Cargo.toml index 5de8ff9b..424e320f 100644 --- a/pmoparadise/Cargo.toml +++ b/pmoparadise/Cargo.toml @@ -48,8 +48,10 @@ url = "2.5" # Audio decoding/encoding symphonia = { version = "0.5", features = ["all"] } +# Audio decoding - claxon for FLAC streaming +claxon = "0.4" + # Per-track feature dependencies -claxon = { version = "0.4", optional = true } hound = { version = "3.5", optional = true } tempfile = { version = "3.8", optional = true } @@ -78,8 +80,8 @@ axum = { version = "0.8.4", optional = true } default = ["metadata-only"] # Mode métadonnées seules (pas de décodage FLAC) metadata-only = [] -# Active le décodage FLAC par-track -per-track = ["dep:claxon", "dep:hound", "dep:tempfile"] +# Active l'extraction par-track (WAV export, etc.) +per-track = ["dep:hound", "dep:tempfile"] # Active l'API REST pmoserver pmoserver = ["dep:pmoserver", "dep:utoipa", "dep:axum", "server"] # Active le media server UPnP (includes pmoserver) diff --git a/pmoparadise/STREAMING_IMPLEMENTATION_GUIDE.md b/pmoparadise/STREAMING_IMPLEMENTATION_GUIDE.md new file mode 100644 index 00000000..3b285747 --- /dev/null +++ b/pmoparadise/STREAMING_IMPLEMENTATION_GUIDE.md @@ -0,0 +1,610 @@ +# Guide d'Implémentation - Streaming Progressif avec Claxon + +## Objectif +Transformer le worker pour qu'il décode le FLAC en streaming au fur et à mesure du téléchargement HTTP, afin d'envoyer le premier morceau au cache en **~6-8 secondes** au lieu de 12-16 secondes. + +--- + +## Vue d'Ensemble de l'Architecture + +### Architecture Actuelle (LENTE - 12-16s) +``` +HTTP Request → Télécharger TOUT le block (75-100 MB) → Symphonia (Cursor) + ↓ +Décoder TOUT en PCM + ↓ +Pour chaque morceau: + - Découper PCM + - Encoder FLAC + - Envoyer au cache +``` + +**Problème** : On attend le téléchargement complet avant de commencer quoi que ce soit. + +### Architecture Cible (RAPIDE - 6-8s) +``` +HTTP Stream → StreamReader (adapt async → sync) + ↓ +claxon::FlacReader (lit frame par frame SANS Seek) + ↓ +Accumule PCM dans buffer + ↓ +Dès que buffer.samples >= durée_morceau_1: + - Découper buffer + - Encoder FLAC + - Envoyer au cache (morceau 1 disponible!) + ↓ +Continue streaming pour morceaux 2, 3, ... +``` + +--- + +## Étape 1 : Créer AsyncReadAdapter + +### But +Convertir `Stream>` (async) en `impl Read` (sync) pour claxon. + +### Localisation +Ajouter au début de `paradise/worker.rs`, après les imports. + +### Code Complet + +```rust +use std::pin::Pin; +use std::task::{Context, Poll}; +use futures::stream::Stream; +use std::io::{self, Read}; +use std::collections::VecDeque; +use tokio::runtime::Handle; +use bytes::Bytes; + +/// Adapte un Stream async en impl Read synchrone +/// +/// Utilise le runtime tokio courant pour bloquer sur le stream async. +/// ATTENTION: Doit être appelé depuis un contexte tokio (spawn_blocking). +struct AsyncReadAdapter { + stream: Pin> + Send>>, + buffer: VecDeque, + runtime: Handle, + done: bool, +} + +impl AsyncReadAdapter { + fn new(stream: Pin> + Send>>) -> Self { + Self { + stream, + buffer: VecDeque::new(), + runtime: Handle::current(), + done: false, + } + } + + fn fill_buffer(&mut self) -> io::Result<()> { + if self.done { + return Ok(()); + } + + // Bloquer pour récupérer le prochain chunk du stream + let next_chunk = self.runtime.block_on(async { + use futures::StreamExt; + self.stream.next().await + }); + + match next_chunk { + Some(Ok(bytes)) => { + self.buffer.extend(bytes.iter()); + Ok(()) + } + Some(Err(e)) => { + self.done = true; + Err(io::Error::new(io::ErrorKind::Other, e)) + } + None => { + self.done = true; + Ok(()) + } + } + } +} + +impl Read for AsyncReadAdapter { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + // Si buffer vide et stream pas terminé, remplir + while self.buffer.is_empty() && !self.done { + self.fill_buffer()?; + } + + // Copier du buffer vers buf + let to_read = buf.len().min(self.buffer.len()); + for i in 0..to_read { + buf[i] = self.buffer.pop_front().unwrap(); + } + + Ok(to_read) + } +} +``` + +### Pièges à Éviter + +1. **Context Tokio** : `AsyncReadAdapter` DOIT être créé dans un contexte tokio (utilisez `tokio::task::spawn_blocking`) +2. **Deadlock** : Ne jamais appeler depuis le même thread qui exécute le stream +3. **Buffer Size** : VecDeque peut grossir - surveiller la mémoire + +--- + +## Étape 2 : Remplacer process_block() + +### Architecture de la Nouvelle Fonction + +```rust +async fn process_block(&mut self, block: Block) -> Result<()> { + // 1. Vérifications habituelles + if self.is_recent_block(block.event) { ... } + + // 2. Lancer le streaming HTTP + let block_url = Url::parse(&block.url)?; + let http_stream = self.client.stream_block(&block_url).await?; + + // 3. Spawn un thread bloquant pour le décodage + let songs_ordered = block.songs_ordered(); + let channel_id = self.descriptor.id; + let sample_rate = 44100; // Sera détecté par claxon + + let tracks = tokio::task::spawn_blocking(move || { + decode_and_split_streaming( + http_stream, + songs_ordered, + sample_rate + ) + }).await??; + + // 4. Pour chaque track décodé, envoyer au cache + for (track_pcm, song_index, song) in tracks { + let entry = self.process_song_from_pcm( + &block, + song_index, + song, + track_pcm + ).await?; + + self.playlist.push_active(entry).await; + } + + // 5. Mise à jour + self.record_processed_block(block.event); + self.next_block_hint = Some(block.end_event); + Ok(()) +} +``` + +--- + +## Étape 3 : Fonction de Décodage Streaming + +### Pseudo-Code Détaillé + +```rust +fn decode_and_split_streaming( + http_stream: BlockStream, // Le stream de client.stream_block() + songs: Vec<(usize, &Song)>, + expected_sample_rate: u32, +) -> Result, usize, Song)>> { + // 1. Convertir BlockStream en AsyncReadAdapter + let adapter = AsyncReadAdapter::new(http_stream.into_inner()); + let buffered = std::io::BufReader::new(adapter); + + // 2. Créer le FlacReader de claxon + let mut reader = claxon::FlacReader::new(buffered) + .map_err(|e| anyhow!("Failed to create FLAC reader: {e}"))?; + + let streaminfo = reader.streaminfo(); + let channels = streaminfo.channels as usize; + let sample_rate = streaminfo.sample_rate; + let bits_per_sample = streaminfo.bits_per_sample; + + // 3. Buffer PCM accumulé + let mut accumulated_samples: Vec = Vec::new(); + let mut current_frame = 0; // Nombre de frames PCM lues + let mut tracks = Vec::new(); + let mut next_song_idx = 0; + + // 4. Lire frame par frame + loop { + // Lire une frame FLAC + let frame = match reader.read_next_or_eof(/* buffer */) { + Ok(Some(frame_data)) => frame_data, + Ok(None) => break, // EOF + Err(e) => return Err(anyhow!("FLAC decode error: {e}")), + }; + + // Convertir frame en i32 et accumuler + // NOTE: claxon retourne des samples par canal, il faut entrelacer + let samples_in_frame = frame.len() / channels; + for sample_idx in 0..samples_in_frame { + for ch in 0..channels { + let sample = frame[ch * samples_in_frame + sample_idx]; + // Normaliser selon bits_per_sample + let normalized = normalize_sample(sample, bits_per_sample); + accumulated_samples.push(normalized); + } + } + + current_frame += samples_in_frame; + + // 5. Vérifier si on a atteint la fin du morceau courant + if next_song_idx < songs.len() { + let (song_index, song) = &songs[next_song_idx]; + let song_end_frame = if next_song_idx + 1 < songs.len() { + // Fin = début du prochain morceau + ms_to_frames(songs[next_song_idx + 1].1.elapsed, sample_rate) + } else { + // Dernier morceau = fin du block + usize::MAX // On prendra tout jusqu'à la fin + }; + + if current_frame >= song_end_frame { + // 6. Découper le buffer + let song_start_frame = ms_to_frames(song.elapsed, sample_rate); + let start_sample = song_start_frame * channels; + let end_sample = song_end_frame * channels; + + let track_samples = accumulated_samples[start_sample..end_sample.min(accumulated_samples.len())] + .to_vec(); + + tracks.push((track_samples, *song_index, (*song).clone())); + + next_song_idx += 1; + + // IMPORTANT: Premier morceau envoyé ici! + // Les suivants continueront pendant que le premier est traité + } + } + } + + // 7. Traiter le dernier morceau si nécessaire + if next_song_idx < songs.len() { + let (song_index, song) = &songs[next_song_idx]; + let song_start_frame = ms_to_frames(song.elapsed, sample_rate); + let start_sample = song_start_frame * channels; + let track_samples = accumulated_samples[start_sample..].to_vec(); + tracks.push((track_samples, *song_index, (*song).clone())); + } + + Ok(tracks) +} + +fn normalize_sample(sample: i32, bits_per_sample: u32) -> i32 { + match bits_per_sample { + 0..=16 => sample << 16, // Shift to 32-bit range + 17..=24 => sample << 8, + _ => sample, + } +} + +fn ms_to_frames(ms: u64, sample_rate: u32) -> usize { + ((ms as u128 * sample_rate as u128) / 1000) as usize +} +``` + +--- + +## Étape 4 : Adapter process_song + +### Nouvelle Signature + +```rust +async fn process_song_from_pcm( + &self, + block: &Block, + song_index: usize, + song: &Song, + track_samples: Vec, // PCM déjà découpé +) -> Result> +``` + +### Changements + +1. **Supprimer** le découpage (déjà fait dans decode_and_split_streaming) +2. **Garder** l'encodage FLAC +3. **Garder** le cache audio/cover +4. **Garder** la création de PlaylistEntry + +```rust +async fn process_song_from_pcm( + &self, + block: &Block, + song_index: usize, + song: &Song, + track_samples: Vec, +) -> Result> { + // 1. Encoder PCM → FLAC (déjà existant) + let flac_bytes = encode_samples_to_flac( + track_samples, + 2, // channels - TODO: passer en paramètre + 44100, // sample_rate - TODO: passer en paramètre + 16, // bits - TODO: passer en paramètre + ).await?; + + // 2. Calculer track_id + let track_id = self.compute_track_id(&flac_bytes); + let placeholder_uri = format!("{}#{}", block.url, song_index); + + // 3. Cache cover (inchangé) + let cover_pk = self.cache_cover(block, song).await?; + + // 4. Cache audio (inchangé) + let flac_len = flac_bytes.len() as u64; + let reader = StreamReader::new(stream::iter(vec![Ok::<_, std::io::Error>( + Bytes::from(flac_bytes) + )])); + let audio_pk = self.cache_manager + .cache_audio_from_reader(&track_id, reader, Some(flac_len)) + .await?; + + // 5. Metadata (inchangé) + let metadata = TrackMetadata { + original_uri: placeholder_uri, + cached_audio_pk: Some(audio_pk.clone()), + cached_cover_pk: cover_pk, + }; + self.cache_manager.update_metadata(track_id.clone(), metadata).await; + + // 6. Créer PlaylistEntry (inchangé) + let duration_ms = song.duration; + let file_path = self.cache_manager.audio_file_path(&audio_pk).await; + let entry = Arc::new(PlaylistEntry::new( + track_id, + self.descriptor.id, + Arc::new(song.clone()), + Utc::now(), + duration_ms, + Some(audio_pk), + file_path, + self.active_clients, + )); + + Ok(entry) +} +``` + +--- + +## Étape 5 : API de claxon + +### Documentation Claxon + +```rust +// Créer un reader +let mut reader = claxon::FlacReader::new(buffered_reader)?; + +// Obtenir les infos du stream +let info = reader.streaminfo(); +// info.channels: u32 +// info.sample_rate: u32 +// info.bits_per_sample: u32 +// info.samples: Option (peut être None pour streams) + +// Lire des samples +// Option 1: Frame par frame (recommandé pour streaming) +let mut samples = vec![0i32; info.channels as usize * 4096]; +loop { + match reader.read_next_or_eof(samples.as_mut_slice()) { + Ok(Some(n)) => { + // n samples lus, entrelacer si nécessaire + } + Ok(None) => break, // EOF + Err(e) => return Err(e), + } +} + +// Option 2: Iterator (plus simple mais moins contrôle) +for sample in reader.samples() { + let s = sample?; + // Traiter sample par sample +} +``` + +### Entrelacement des Samples + +Claxon retourne les samples **par canal** : +``` +Buffer claxon: [L0, L1, L2, ..., Ln, R0, R1, R2, ..., Rn] +``` + +Il faut entrelacer pour PCM standard : +``` +Buffer PCM: [L0, R0, L1, R1, L2, R2, ..., Ln, Rn] +``` + +```rust +fn interleave_samples(frame: &[i32], channels: usize) -> Vec { + let samples_per_channel = frame.len() / channels; + let mut interleaved = Vec::with_capacity(frame.len()); + + for i in 0..samples_per_channel { + for ch in 0..channels { + interleaved.push(frame[ch * samples_per_channel + i]); + } + } + + interleaved +} +``` + +--- + +## Étape 6 : Gestion d'Erreurs + +### Erreurs Potentielles + +1. **Stream HTTP interrompu** : Gérer les EOF prématurés +2. **Mauvais timing** : Vérifier que `song.elapsed` < durée totale +3. **Corruption FLAC** : claxon peut échouer sur frames corrompues + +### Pattern de Gestion + +```rust +match reader.read_next_or_eof(buffer) { + Ok(Some(n)) => { + // Traiter n samples + } + Ok(None) => { + // EOF normal + break; + } + Err(claxon::Error::FormatError(msg)) => { + // Frame corrompue, continuer ou abandonner? + warn!("FLAC format error: {}", msg); + continue; // Ou break selon la criticité + } + Err(e) => { + // Erreur fatale + return Err(anyhow!("FLAC decode error: {}", e)); + } +} +``` + +--- + +## Étape 7 : Tests Recommandés + +### Test 1 : AsyncReadAdapter + +```rust +#[tokio::test] +async fn test_async_read_adapter() { + let data = vec![ + Ok(Bytes::from_static(b"Hello ")), + Ok(Bytes::from_static(b"World")), + ]; + let stream = futures::stream::iter(data); + + let mut adapter = AsyncReadAdapter::new(Box::pin(stream)); + let mut buf = [0u8; 11]; + let n = adapter.read(&mut buf).unwrap(); + + assert_eq!(n, 11); + assert_eq!(&buf, b"Hello World"); +} +``` + +### Test 2 : Décodage d'un Petit FLAC + +Créer un fichier FLAC de test (1 morceau, 10 secondes) et vérifier : +1. Le stream est lu progressivement +2. Le morceau est correctement découpé +3. Le FLAC réencodé est valide + +### Test 3 : Integration Complète + +1. Télécharger un vrai block Radio Paradise +2. Chronométrer le temps jusqu'au premier morceau disponible +3. Vérifier que les morceaux suivants arrivent bien + +--- + +## Étape 8 : Optimisations Futures + +### Buffer Size Tuning + +```rust +// Ajuster selon le réseau +const STREAM_BUFFER_SIZE: usize = 64 * 1024; // 64 KB +``` + +### Parallélisation + +Une fois le premier morceau envoyé, les suivants peuvent être traités en parallèle : + +```rust +let mut tasks = Vec::new(); +for (track_pcm, song_index, song) in tracks { + let task = tokio::spawn(async move { + // Encoder + envoyer au cache + }); + tasks.push(task); +} + +// Attendre tous en parallèle +futures::future::join_all(tasks).await; +``` + +--- + +## Pièges Critiques à Éviter + +### 1. Seek dans claxon +**ERREUR** : claxon::FlacReader n'a **PAS** de méthode `seek()` ! +- Ne tentez pas `reader.seek_to(position)` (compile pas) +- Le streaming est **séquentiel uniquement** + +### 2. Thread Blocking +**ERREUR** : Créer AsyncReadAdapter dans un contexte async +```rust +// ❌ MAUVAIS +async fn foo() { + let adapter = AsyncReadAdapter::new(stream); // Deadlock! +} + +// ✅ BON +tokio::task::spawn_blocking(move || { + let adapter = AsyncReadAdapter::new(stream); + // ... +}) +``` + +### 3. Normalisation des Samples +**ERREUR** : Ne pas normaliser selon bits_per_sample +- claxon retourne des samples **natifs** (16-bit → i32 avec shift) +- flacenc attend des samples dans la plage correcte +- **Toujours normaliser** selon les bits réels + +### 4. Accumulation Mémoire +**ATTENTION** : `accumulated_samples` peut devenir ÉNORME (100 MB+) +- **Solution** : Ne garder que le nécessaire, supprimer les samples déjà traités +- Ou: Traiter morceau par morceau sans accumuler tout le block + +--- + +## Mesures de Performance Attendues + +### Avant (Architecture Actuelle) +- Téléchargement block : 12-16 secondes (75-100 MB @ 50 Mbps) +- Premier morceau disponible : **12-16 secondes** + +### Après (Streaming Progressif) +- Temps pour 1er morceau (3 min, ~30 MB) : **~6-8 secondes** +- Amélioration : **2x plus rapide** ⚡ + +### Métriques à Surveiller +1. Temps entre `get_block()` et premier `push_active()` +2. Débit du stream HTTP (surveiller throttling) +3. Utilisation mémoire de `accumulated_samples` + +--- + +## Checklist d'Implémentation + +- [ ] Créer `AsyncReadAdapter` avec tests unitaires +- [ ] Remplacer `decode_block_audio()` par `decode_and_split_streaming()` +- [ ] Adapter `process_block()` pour utiliser streaming +- [ ] Créer `process_song_from_pcm()` +- [ ] Tester avec un petit FLAC local +- [ ] Tester avec un vrai block Radio Paradise +- [ ] Mesurer les performances (avant/après) +- [ ] Vérifier pas de régression sur la qualité audio +- [ ] Vérifier pas de fuite mémoire +- [ ] Ajouter logs de debug pour troubleshooting + +--- + +## Ressources + +- **claxon docs** : https://docs.rs/claxon/latest/claxon/ +- **Radio Paradise API** : https://api.radioparadise.com/api +- **FLAC spec** : https://xiph.org/flac/format.html + +--- + +Bon courage pour l'implémentation ! 🚀 diff --git a/pmoparadise/src/client.rs b/pmoparadise/src/client.rs index 95ab5777..2a875489 100644 --- a/pmoparadise/src/client.rs +++ b/pmoparadise/src/client.rs @@ -25,45 +25,6 @@ pub const DEFAULT_BLOCK_TIMEOUT_SECS: u64 = 180; pub const DEFAULT_USER_AGENT: &str = "pmoparadise/0.1.0"; -fn normalize_cover_base_url(base: &str) -> String { - let mut normalized = base.trim().to_string(); - - if normalized.is_empty() { - return DEFAULT_IMAGE_BASE.to_string(); - } - - if normalized.starts_with("//") { - normalized = format!("https:{}", normalized); - } else if !(normalized.starts_with("http://") || normalized.starts_with("https://")) { - normalized = format!("https://{}", normalized.trim_start_matches('/')); - } - - if !normalized.ends_with('/') { - normalized.push('/'); - } - - normalized -} - -fn resolve_cover_with_base(base: &str, cover_path: &str) -> Result { - let cover_path = cover_path.trim(); - - if cover_path.starts_with("http://") || cover_path.starts_with("https://") { - return Ok(Url::parse(cover_path)?); - } - - if cover_path.starts_with("//") { - let url = format!("https:{}", cover_path); - return Ok(Url::parse(&url)?); - } - - let base = normalize_cover_base_url(base); - let base_url = Url::parse(&base)?; - - Ok(base_url.join(cover_path)?) -} - - /// Radio Paradise HTTP client /// /// This client provides access to Radio Paradise's streaming API, @@ -89,7 +50,6 @@ pub struct RadioParadiseClient { pub(crate) client: Client, api_base: String, block_base: String, - image_base: String, bitrate: Bitrate, channel: u8, pub(crate) request_timeout: Duration, @@ -118,7 +78,6 @@ impl RadioParadiseClient { client, api_base: DEFAULT_API_BASE.to_string(), block_base: DEFAULT_BLOCK_BASE.to_string(), - image_base: normalize_cover_base_url(DEFAULT_IMAGE_BASE), bitrate: Bitrate::default(), channel: 0, request_timeout: Duration::from_secs(DEFAULT_REQUEST_TIMEOUT_SECS), @@ -158,13 +117,6 @@ impl RadioParadiseClient { cloned } - /// Clone the client with an updated channel and bitrate. - pub fn clone_with_channel_and_bitrate(&self, channel: u8, bitrate: Bitrate) -> Self { - let mut cloned = self.clone_with_channel(channel); - cloned.bitrate = bitrate; - cloned - } - /// Get a block by event ID /// /// If `event` is None, returns the current block. @@ -223,11 +175,14 @@ impl RadioParadiseClient { let mut block: Block = response.json().await?; - // Set image_base if not provided - if let Some(ref mut base) = block.image_base { - *base = normalize_cover_base_url(base); + // Normalize protocol-relative URLs from API (//img.radioparadise.com/) + if let Some(ref base) = block.image_base { + if base.starts_with("//") { + block.image_base = Some(format!("https:{}", base)); + } } else { - block.image_base = Some(self.image_base.clone()); + // Fallback if API doesn't provide image_base (should never happen) + block.image_base = Some(DEFAULT_IMAGE_BASE.to_string()); } #[cfg(feature = "logging")] @@ -252,28 +207,6 @@ impl RadioParadiseClient { Ok(NowPlaying::from_block(block)) } - /// Get the full URL for a cover image - /// - /// # Arguments - /// - /// * `cover_path` - The cover filename/path from song metadata - /// - /// # Example - /// - /// ```no_run - /// # use pmoparadise::RadioParadiseClient; - /// # #[tokio::main] - /// # async fn main() -> Result<(), Box> { - /// let client = RadioParadiseClient::new().await?; - /// let url = client.cover_url("B00000I0JF.jpg")?; - /// println!("Cover URL: {}", url); - /// # Ok(()) - /// # } - /// ``` - pub fn cover_url(&self, cover_path: &str) -> Result { - resolve_cover_with_base(&self.image_base, cover_path) - } - /// Prefetch metadata for the next block /// /// Stores the next block URL internally for seamless transitions. @@ -318,7 +251,6 @@ pub struct ClientBuilder { client: Option, api_base: String, block_base: String, - image_base: String, bitrate: Bitrate, channel: u8, request_timeout: Duration, @@ -333,7 +265,6 @@ impl Default for ClientBuilder { client: None, api_base: DEFAULT_API_BASE.to_string(), block_base: DEFAULT_BLOCK_BASE.to_string(), - image_base: DEFAULT_IMAGE_BASE.to_string(), bitrate: Bitrate::default(), channel: 0, request_timeout: Duration::from_secs(DEFAULT_REQUEST_TIMEOUT_SECS), @@ -368,12 +299,6 @@ impl ClientBuilder { self } - /// Set the image base URL - pub fn image_base(mut self, url: impl Into) -> Self { - self.image_base = url.into(); - self - } - /// Set the bitrate/quality level /// /// # Example @@ -441,13 +366,11 @@ impl ClientBuilder { } else { self.block_base.clone() }; - let image_base = normalize_cover_base_url(&self.image_base); Ok(RadioParadiseClient { client, api_base: self.api_base, block_base, - image_base, bitrate: self.bitrate, channel: self.channel, request_timeout: self.request_timeout, @@ -469,13 +392,4 @@ mod tests { assert_eq!(builder.channel, 0); } - #[test] - fn test_cover_url() { - let client = RadioParadiseClient::with_client(Client::new()); - let url = client.cover_url("test.jpg").unwrap(); - assert_eq!( - url.as_str(), - "https://img.radioparadise.com/covers/l/test.jpg" - ); - } } diff --git a/pmoparadise/src/paradise/channel.rs b/pmoparadise/src/paradise/channel.rs index e6ce98f5..0585baa3 100644 --- a/pmoparadise/src/paradise/channel.rs +++ b/pmoparadise/src/paradise/channel.rs @@ -114,6 +114,11 @@ pub const ALL_CHANNELS: [ChannelDescriptor; 4] = [ ChannelDescriptor::new(ParadiseChannelKind::Eclectic), ]; +/// Returns the maximum valid channel ID +pub const fn max_channel_id() -> u8 { + (ALL_CHANNELS.len() - 1) as u8 +} + /// Public handle to interact with a channel. #[derive(Clone)] pub struct ParadiseChannel { diff --git a/pmoparadise/src/paradise/mod.rs b/pmoparadise/src/paradise/mod.rs index fb6e8192..dc389a3d 100644 --- a/pmoparadise/src/paradise/mod.rs +++ b/pmoparadise/src/paradise/mod.rs @@ -19,7 +19,8 @@ mod playlist; mod worker; pub use channel::{ - ChannelDescriptor, ParadiseChannel, ParadiseChannelKind, ParadiseClientStream, ALL_CHANNELS, + max_channel_id, ChannelDescriptor, ParadiseChannel, ParadiseChannelKind, ParadiseClientStream, + ALL_CHANNELS, }; pub use config::{ ActivityConfig, ApiConfig, CacheConfig, HistoryConfig, PollingConfig, RadioParadiseConfig, diff --git a/pmoparadise/src/paradise/worker.rs b/pmoparadise/src/paradise/worker.rs index b4cdf4dc..2f36ff24 100644 --- a/pmoparadise/src/paradise/worker.rs +++ b/pmoparadise/src/paradise/worker.rs @@ -452,14 +452,15 @@ impl WorkerState { async fn cache_cover(&self, block: &Block, song: &Song) -> Result> { if let Some(ref cover_path) = song.cover { - let cover_url = - resolve_cover_url(block.image_base.as_deref(), &self.client, cover_path) - .context("Invalid cover URL")?; - match self.cache_manager.cache_cover(cover_url.as_str()).await { - Ok(pk) => return Ok(Some(pk)), - Err(err) => { - warn!(channel = self.descriptor.slug, "Cover cache error: {err}"); + if let Some(cover_url) = block.cover_url(cover_path) { + match self.cache_manager.cache_cover(&cover_url).await { + Ok(pk) => return Ok(Some(pk)), + Err(err) => { + warn!(channel = self.descriptor.slug, "Cover cache error: {err}"); + } } + } else { + warn!(channel = self.descriptor.slug, "Unable to resolve cover URL for {}", cover_path); } } Ok(None) @@ -729,30 +730,3 @@ async fn encode_samples_to_flac( .await? } -fn resolve_cover_url( - image_base: Option<&str>, - client: &RadioParadiseClient, - cover: &str, -) -> Result { - if cover.starts_with("http://") || cover.starts_with("https://") { - return Url::parse(cover).map_err(|e| anyhow!("Invalid cover URL '{cover}': {e}")); - } - - if cover.starts_with("//") { - let url = format!("https:{cover}"); - return Url::parse(&url).map_err(|e| anyhow!("Invalid cover URL '{cover}': {e}")); - } - - if let Some(base) = image_base { - match Url::parse(base).and_then(|base_url| base_url.join(cover)) { - Ok(url) => return Ok(url), - Err(err) => { - debug!("Failed to join cover '{cover}' with base '{base}': {err}"); - } - } - } - - client - .cover_url(cover) - .map_err(|e| anyhow!("Invalid cover URL '{cover}': {e}")) -} diff --git a/pmoparadise/src/pmoserver_ext.rs b/pmoparadise/src/pmoserver_ext.rs index 931c2406..d18a9b49 100644 --- a/pmoparadise/src/pmoserver_ext.rs +++ b/pmoparadise/src/pmoserver_ext.rs @@ -3,7 +3,7 @@ //! Ce module fournit un trait d'extension pour ajouter facilement l'API Radio Paradise //! à un serveur pmoserver. -use crate::paradise::{ParadiseChannel, PlaylistEntry}; +use crate::paradise::{max_channel_id, ParadiseChannel, PlaylistEntry, ALL_CHANNELS}; use crate::{models::Bitrate, Block, NowPlaying, RadioParadiseClient, RadioParadiseSource}; use axum::{ body::Body, @@ -30,8 +30,6 @@ pub struct RadioParadiseState { source: Arc, } -const MAX_CHANNEL_ID: u8 = 3; - #[derive(Debug, Default, Deserialize)] #[serde(default)] struct ParadiseQuery { @@ -101,7 +99,7 @@ impl RadioParadiseState { let mut client = base_client; if let Some(channel) = params.channel { - if channel > MAX_CHANNEL_ID { + if channel > max_channel_id() { tracing::warn!("Invalid Radio Paradise channel requested: {}", channel); return Err(StatusCode::BAD_REQUEST); } @@ -120,7 +118,7 @@ impl RadioParadiseState { } fn channel_for_id(&self, channel_id: u8) -> Result, StatusCode> { - if channel_id > MAX_CHANNEL_ID { + if channel_id > max_channel_id() { return Err(StatusCode::BAD_REQUEST); } self.source @@ -144,6 +142,16 @@ pub struct ChannelInfo { pub description: String, } +impl From<&crate::paradise::ChannelDescriptor> for ChannelInfo { + fn from(descriptor: &crate::paradise::ChannelDescriptor) -> Self { + Self { + id: descriptor.id, + name: descriptor.display_name.to_string(), + description: descriptor.description.to_string(), + } + } +} + /// Réponse avec informations étendues sur le morceau en cours #[derive(Debug, Clone, Serialize, ToSchema)] pub struct NowPlayingResponse { @@ -372,29 +380,7 @@ async fn get_block_by_id( tag = "Radio Paradise" )] async fn get_channels() -> Json> { - let channels = vec![ - ChannelInfo { - id: 0, - name: "Main Mix".to_string(), - description: "Eclectic mix of rock, world, electronica, and more".to_string(), - }, - ChannelInfo { - id: 1, - name: "Mellow Mix".to_string(), - description: "Mellower, less aggressive music".to_string(), - }, - ChannelInfo { - id: 2, - name: "Rock Mix".to_string(), - description: "Heavier, more guitar-driven music".to_string(), - }, - ChannelInfo { - id: 3, - name: "World/Etc Mix".to_string(), - description: "Global beats and world music".to_string(), - }, - ]; - + let channels: Vec = ALL_CHANNELS.iter().map(Into::into).collect(); Json(channels) } diff --git a/pmoparadise/src/source.rs b/pmoparadise/src/source.rs index 7b6f00f9..d87284bb 100644 --- a/pmoparadise/src/source.rs +++ b/pmoparadise/src/source.rs @@ -75,7 +75,7 @@ impl std::fmt::Debug for RadioParadiseSource { impl RadioParadiseSource { #[cfg(feature = "server")] - pub fn from_registry(client: RadioParadiseClient, _legacy_capacity: usize) -> Result { + pub fn from_registry(client: RadioParadiseClient) -> Result { let config = Arc::new(RadioParadiseConfig::load_from_pmoconfig().unwrap_or_default()); let history_backend = history_backend_from_config(&config.history).map_err(|e| { MusicSourceError::SourceUnavailable(format!( @@ -114,12 +114,11 @@ impl RadioParadiseSource { #[cfg(feature = "server")] pub fn from_registry_default(client: RadioParadiseClient) -> Result { - Self::from_registry(client, 0) + Self::from_registry(client) } pub fn new( client: RadioParadiseClient, - _legacy_capacity: usize, cover_cache: Arc, audio_cache: Arc, ) -> Self { @@ -166,7 +165,7 @@ impl RadioParadiseSource { cover_cache: Arc, audio_cache: Arc, ) -> Self { - Self::new(client, 0, cover_cache, audio_cache) + Self::new(client, cover_cache, audio_cache) } pub fn client_for_channel(&self, channel: u8) -> Option { diff --git a/pmoparadise/src/stream.rs b/pmoparadise/src/stream.rs index 95f2e881..0aeef97d 100644 --- a/pmoparadise/src/stream.rs +++ b/pmoparadise/src/stream.rs @@ -4,7 +4,7 @@ use crate::error::{Error, Result}; use crate::models::Block; use crate::RadioParadiseClient; use bytes::Bytes; -use futures::stream::Stream; +use futures::stream::{Stream, StreamExt}; use std::pin::Pin; use std::task::{Context, Poll}; use url::Url; @@ -121,12 +121,14 @@ impl RadioParadiseClient { self.stream_block(&url).await } - /// Download a complete block to memory + /// Download an entire block as Bytes /// - /// **Warning**: Blocks can be large (50-100MB for FLAC). Use streaming - /// for playback instead of downloading the entire block to memory. + /// This downloads the complete block file into memory. For streaming playback, + /// use `stream_block()` instead which is more memory efficient. /// - /// This is useful for the per-track feature which needs random access. + /// # Arguments + /// + /// * `block_url` - The URL of the block to download /// /// # Example /// @@ -137,38 +139,24 @@ impl RadioParadiseClient { /// async fn main() -> Result<(), Box> { /// let client = RadioParadiseClient::new().await?; /// let block = client.get_block(None).await?; - /// - /// let data = client.download_block(&block.url.parse()?).await?; - /// println!("Downloaded {} bytes", data.len()); - /// + /// let url = block.url.parse()?; + /// let bytes = client.download_block(&url).await?; + /// println!("Downloaded {} bytes", bytes.len()); /// Ok(()) /// } /// ``` pub async fn download_block(&self, block_url: &Url) -> Result { - #[cfg(feature = "logging")] - tracing::debug!("Downloading complete block: {}", block_url); + let mut stream = self.stream_block(block_url).await?; + let mut data = Vec::new(); - let response = self - .client - .get(block_url.clone()) - .timeout(self.block_timeout) - .send() - .await?; - - if !response.status().is_success() { - return Err(Error::other(format!( - "Failed to download block: HTTP {}", - response.status() - ))); + while let Some(chunk_result) = stream.next().await { + let chunk = chunk_result?; + data.extend_from_slice(&chunk); } - let bytes = response.bytes().await?; - - #[cfg(feature = "logging")] - tracing::debug!("Downloaded {} bytes", bytes.len()); - - Ok(bytes) + Ok(Bytes::from(data)) } + } #[cfg(test)]