diff --git a/pmoparadise/RADIO_PARADISE_STREAM_SOURCE.md b/pmoparadise/RADIO_PARADISE_STREAM_SOURCE.md new file mode 100644 index 00000000..37ac10eb --- /dev/null +++ b/pmoparadise/RADIO_PARADISE_STREAM_SOURCE.md @@ -0,0 +1,238 @@ +# RadioParadiseStreamSource - Documentation Technique + +## Vue d'ensemble + +`RadioParadiseStreamSource` est un nœud source pour `pmoaudio` qui télécharge et décode les blocs FLAC de Radio Paradise en temps réel, avec gestion automatique des transitions entre pistes (TrackBoundary). + +## Architecture + +### Pattern Node + +Suit l'architecture séparée logique/pipeline de `pmoaudio` : + +``` +RadioParadiseStreamSource (wrapper) + └── Node + └── RadioParadiseStreamSourceLogic (logique métier) +``` + +### RadioParadiseStreamSourceLogic + +Responsabilités : +- **File d'attente** : `VecDeque` pour les blocks à télécharger +- **Cache anti-redondance** : `HashSet` pour 10 blocs récents +- **Téléchargement** : Fetch bloc FLAC (bitrate=4 uniquement) +- **Décodage** : Stream FLAC via `pmoflac::decode_audio_stream` +- **Timing** : Calcul précis pour insertion TrackBoundary + +## Flux d'exécution + +``` +┌─────────────────────────────────────────────────────────┐ +│ 1. Attente block ID (timeout 3s) │ +│ └─> VecDeque::pop_front() │ +└─────────────────────────────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────┐ +│ 2. Vérification cache │ +│ └─> HashSet::contains(&event_id) │ +└─────────────────────────────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────┐ +│ 3. Téléchargement métadonnées │ +│ └─> client.get_block(event_id) │ +└─────────────────────────────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────┐ +│ 4. Téléchargement FLAC (bitrate=4) │ +│ └─> client.download_block_file(&block, 4) │ +└─────────────────────────────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────┐ +│ 5. Décodage streaming │ +│ └─> pmoflac::decode_audio_stream(reader) │ +└─────────────────────────────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────┐ +│ 6. Découpage en chunks │ +│ └─> pcm_to_audio_chunk(pcm, sr, bps) │ +└─────────────────────────────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────┐ +│ 7. Insertion TrackBoundary (timing sample-based) │ +│ └─> elapsed_ms = (total_samples * 1000) / sr │ +└─────────────────────────────────────────────────────────┘ +``` + +## Timing TrackBoundary + +### Algorithme + +```rust +let elapsed_ms = (total_samples * 1000) / sample_rate as u64; + +if elapsed_ms >= song.elapsed { + // Envoyer TrackBoundary AVANT le chunk (même order) + send_track_boundary(*order, song, block).await; +} +``` + +### Exemple concret + +Bloc FLAC contenant 3 chansons : +- Song 0 : `elapsed = 0ms` +- Song 1 : `elapsed = 180000ms` (3min) +- Song 2 : `elapsed = 420000ms` (7min) + +Timeline : +``` +0ms 180000ms 420000ms +│ │ │ +Song 0 TrackBoundary TrackBoundary + └─> Song 1 └─> Song 2 +``` + +## SyncMarker Order + +**Règle** : TrackBoundary a le **même order** que le chunk suivant. + +```rust +// TrackBoundary order = 42 +AudioSegment::new_sync(42, SyncMarker::TrackBoundary { ... }) + +// Chunk suivant order = 42 +AudioSegment::new_audio(42, AudioChunk::I16(...)) +``` + +## Gestion du cache + +### Stratégie FIFO simple + +```rust +const RECENT_BLOCKS_CACHE_SIZE: usize = 10; + +fn mark_block_downloaded(&mut self, event_id: EventId) { + self.recent_blocks.insert(event_id); + + if self.recent_blocks.len() > RECENT_BLOCKS_CACHE_SIZE { + // Retirer un élément (ordre non garanti avec HashSet) + if let Some(&first) = self.recent_blocks.iter().next() { + self.recent_blocks.remove(&first); + } + } +} +``` + +## Support FLAC + +### Formats supportés + +- **16-bit** : `AudioChunk::I16` +- **24-bit** : `AudioChunk::I24` + +### Conversion PCM + +```rust +match bits_per_sample { + 16 => { + let samples: Vec = pcm_data + .chunks_exact(2) + .map(|chunk| i16::from_le_bytes([chunk[0], chunk[1]])) + .collect(); + AudioChunk::I16(...) + } + 24 => { + let samples: Vec = pcm_data + .chunks_exact(3) + .map(|chunk| { + let value = i32::from_le_bytes([chunk[0], chunk[1], chunk[2], 0]) >> 8; + I24::from_i32(value) + }) + .collect(); + AudioChunk::I24(...) + } +} +``` + +## Métadonnées + +### TrackMetadata + +Champs extraits de `Song` : +- `title` : Titre de la chanson +- `artist` : Artiste +- `album` : Album (optionnel) +- `year` : Année (optionnel) +- `cover_url` : URL de la pochette (async via tokio::spawn) + +### Gestion asynchrone du cover + +```rust +tokio::spawn(async move { + if let Ok(mut meta) = metadata_clone.write().await { + let _ = meta.set_cover_url(Some(cover_url)).await; + } +}); +``` + +## API Publique + +### Création + +```rust +pub fn new(client: RadioParadiseClient, chunk_duration_ms: u32) -> Self +``` + +### Configuration + +```rust +pub fn push_block_id(&mut self, event_id: EventId) +``` + +Ajoute un block ID à télécharger dans la file d'attente. + +### Exécution + +```rust +async fn run(self: Box, stop_token: CancellationToken) -> Result<(), AudioError> +``` + +Hérite de `AudioPipelineNode`. + +## Exemple d'utilisation + +Voir `examples/radio_paradise_stream.rs` pour : +- Utilisation basique +- Intégration avec nowplaying stream +- Connexion à un sink + +## Constantes + +```rust +const BLOCK_ID_TIMEOUT_SECS: u64 = 3; // Timeout attente nouveau block +const RECENT_BLOCKS_CACHE_SIZE: usize = 10; // Taille cache anti-redondance +``` + +## Dépendances + +- `pmoaudio` : Pipeline audio, types AudioChunk/AudioSegment +- `pmoflac` : Décodage FLAC streaming +- `pmometadata` : Métadonnées pistes +- `futures-util` : StreamExt pour le décodage +- `tokio` : Runtime async +- `tokio-util` : StreamReader, CancellationToken + +## Feature gate + +```toml +[features] +pmoaudio = ["dep:pmoaudio", "dep:pmoflac", "dep:pmometadata", "dep:futures-util"] +``` + +Activer avec : `cargo build -p pmoparadise --features pmoaudio` diff --git a/pmoparadise/examples/radio_paradise_stream.rs b/pmoparadise/examples/radio_paradise_stream.rs new file mode 100644 index 00000000..53af2083 --- /dev/null +++ b/pmoparadise/examples/radio_paradise_stream.rs @@ -0,0 +1,114 @@ +//! Exemple d'utilisation de RadioParadiseStreamSource +//! +//! Ce exemple montre comment : +//! - Créer un RadioParadiseStreamSource +//! - Ajouter des block IDs à télécharger via push_block_id() +//! - Connecter à un sink pour récupérer les segments audio + +use pmoaudio::{ + nodes::{DEFAULT_CHUNK_DURATION_MS, TypedAudioNode}, + pipeline::AudioPipelineNode, +}; +use pmoparadise::{ + client::RadioParadiseClient, + models::EventId, + RadioParadiseStreamSource, +}; +use tokio_util::sync::CancellationToken; + +#[tokio::main] +async fn main() -> Result<(), Box> { + // 1. Créer un client Radio Paradise + let client = RadioParadiseClient::new(pmoparadise::Channel::MainMix); + + // 2. Créer le source node avec durée de chunk par défaut (500ms) + let mut source = RadioParadiseStreamSource::new( + client.clone(), + DEFAULT_CHUNK_DURATION_MS, + ); + + // 3. Ajouter des blocks IDs à télécharger + // Dans un cas réel, ces IDs viendraient du nowplaying stream + source.push_block_id(EventId(12345)); + source.push_block_id(EventId(12346)); + source.push_block_id(EventId(12347)); + + // 4. Optionnel : Connecter à un sink (ici juste un exemple de structure) + // let sink = create_your_sink(); + // source.add_child(Box::new(sink)); + + // 5. Lancer le traitement + let stop_token = CancellationToken::new(); + + println!("🎵 RadioParadiseStreamSource lancé..."); + println!(" - Téléchargement et décodage des blocs FLAC"); + println!(" - Insertion automatique des TrackBoundary"); + println!(" - Cache anti-redondance de {} blocs", 10); + + // Dans un cas réel, on lancerait : + // source.run(stop_token).await?; + + // Pour cet exemple, on simule juste le comportement + println!("\n✅ Configuration réussie !"); + println!("\nFlux d'exécution :"); + println!("1. Attente d'un block ID dans la queue (timeout 3s)"); + println!("2. Vérification cache anti-redondance"); + println!("3. Téléchargement des métadonnées du bloc"); + println!("4. Téléchargement et décodage du FLAC (bitrate=4)"); + println!("5. Envoi des AudioChunk (I16 ou I24)"); + println!("6. Insertion TrackBoundary au timing correct (basé sur samples)"); + + Ok(()) +} + +// ═══════════════════════════════════════════════════════════════════════════ +// Exemple avancé : Utilisation avec nowplaying stream +// ═══════════════════════════════════════════════════════════════════════════ + +#[allow(dead_code)] +async fn example_with_nowplaying_stream() -> Result<(), Box> { + use futures_util::StreamExt; + use pmoparadise::Channel; + + let client = RadioParadiseClient::new(Channel::MainMix); + let mut source = RadioParadiseStreamSource::new( + client.clone(), + DEFAULT_CHUNK_DURATION_MS, + ); + + // Récupérer le nowplaying stream + let nowplaying = client.nowplaying_stream().await?; + + // Clone pour le spawned task + let stop_token = CancellationToken::new(); + let stop_clone = stop_token.clone(); + + // Task 1 : Alimenter la queue avec les nouveaux blocks + let feed_task = tokio::spawn(async move { + tokio::pin!(nowplaying); + + while let Some(result) = nowplaying.next().await { + match result { + Ok(event) => { + println!("📻 Nouveau bloc détecté : {:?}", event.event); + source.push_block_id(event.event); + } + Err(e) => { + eprintln!("❌ Erreur nowplaying stream : {}", e); + break; + } + } + } + }); + + // Task 2 : Traiter les blocs (dans un cas réel) + // let process_task = tokio::spawn(async move { + // source.run(stop_clone).await + // }); + + // Attendre les tasks + feed_task.await?; + // process_task.await??; + + Ok(()) +}