diff --git a/.pmomusic.yml b/.pmomusic.yml deleted file mode 100644 index 7da3f639..00000000 --- a/.pmomusic.yml +++ /dev/null @@ -1,28 +0,0 @@ -host: - http_port: '8080' - cover_cache: - directory: ./.pmomusic_covers - size: 2000 - audio_cache: - directory: ./.pmomusic_audio - size: 500 - logger: - buffer_capacity: 200 - enable_console: true - min_level: TRACE - mediarenderer: - mpd_renderer: null - mediaserver: - qobuz: - udn: uuid:28963b75-4c5f-4da7-b10e-ffafd -accounts: - qobuz: - username: eric@coissac.eu - password: '*Misfcr73110$' -devices: - mediarenderer: - pmo_mediarenderer: - udn: 15a13316-daac-47f0-b64e-47e56f5e3b51 - mediaserver: - pmo_mediaserver: - udn: 23df0bfa-cfef-4724-b731-00f66fadf176 diff --git a/pmoparadise/examples/extract_track.rs b/pmoparadise/examples/extract_track.rs deleted file mode 100644 index 8d403893..00000000 --- a/pmoparadise/examples/extract_track.rs +++ /dev/null @@ -1,120 +0,0 @@ -//! Example: Extract individual tracks from a FLAC block (requires `per-track` feature) -//! -//! This example demonstrates: -//! - Per-track extraction from FLAC blocks -//! - Exporting tracks to WAV files -//! - Alternative player-based seeking (recommended) -//! -//! **Warning**: This approach downloads and decodes entire blocks. -//! For most use cases, player-based seeking is more efficient. -//! -//! Run with: cargo run --example extract_track --features per-track - -#[cfg(feature = "per-track")] -use pmoparadise::{RadioParadiseClient, Result}; -#[cfg(feature = "per-track")] -use std::path::Path; - -#[cfg(feature = "per-track")] -#[tokio::main] -async fn main() -> Result<()> { - // Initialize logging - #[cfg(feature = "logging")] - tracing_subscriber::fmt::init(); - - println!("Radio Paradise - Per-Track Extraction Demo"); - println!("===========================================\n"); - - println!("WARNING: This feature downloads entire blocks (50-100MB)"); - println!(" and performs CPU-intensive FLAC decoding."); - println!(" For most use cases, player-based seeking is better.\n"); - - // Create client - let client = RadioParadiseClient::new().await?; - - // Get current block - let block = client.get_block(None).await?; - - println!("Block Information:"); - println!(" Event: {}", block.event); - println!(" Songs: {}", block.song_count()); - println!(" URL: {}\n", block.url); - - // Display all tracks - println!("Available Tracks:"); - for (index, song) in block.songs_ordered() { - println!( - " {}. {} - {} ({:.1}s)", - index, - song.artist, - song.title, - song.duration as f64 / 1000.0 - ); - } - println!(); - - // Extract first track - let track_index = 0; - if let Some((_, song)) = block.songs_ordered().first() { - println!("Extracting Track {}:", track_index); - println!(" Artist: {}", song.artist); - println!(" Title: {}", song.title); - println!(" Album: {}\n", song.album); - - println!("Downloading and decoding... (this may take a while)"); - - // Open track stream - let mut track_stream = client.open_track_stream(&block, track_index).await?; - - println!("Track Metadata:"); - println!(" Sample Rate: {} Hz", track_stream.metadata.sample_rate); - println!(" Channels: {}", track_stream.metadata.channels); - println!( - " Bits Per Sample: {}", - track_stream.metadata.bits_per_sample - ); - println!(" Total Samples: {}", track_stream.metadata.total_samples); - println!(); - - // Export to WAV - let output_path = Path::new("track.wav"); - println!("Exporting to {:?}...", output_path); - track_stream.export_wav(output_path)?; - println!("✓ Export complete!\n"); - } - - // Show alternative: player-based seeking - println!("RECOMMENDED ALTERNATIVE: Player-Based Seeking"); - println!("=============================================\n"); - - for (index, song) in block.songs_ordered().into_iter().take(3) { - let (start, duration) = client.track_position_seconds(&block, index)?; - println!("Track {}: {} - {}", index, song.artist, song.title); - println!(" mpv command:"); - println!( - " mpv --start={:.3} --length={:.3} '{}'", - start, duration, block.url - ); - println!(" ffmpeg command (extract to file):"); - println!( - " ffmpeg -ss {:.3} -t {:.3} -i '{}' -c copy track_{}.flac", - start, duration, block.url, index - ); - println!(); - } - - println!("These methods are much more efficient as they:"); - println!(" - Don't download the entire block"); - println!(" - Use the player's optimized seeking"); - println!(" - Start playback immediately"); - println!(" - Preserve original quality (with -c copy)"); - - Ok(()) -} - -#[cfg(not(feature = "per-track"))] -fn main() { - eprintln!("ERROR: This example requires the 'per-track' feature."); - eprintln!("Run with: cargo run --example extract_track --features per-track"); - std::process::exit(1); -} diff --git a/pmoparadise/examples/show_source_image.rs b/pmoparadise/examples/show_source_image.rs deleted file mode 100644 index 14c9ff9b..00000000 --- a/pmoparadise/examples/show_source_image.rs +++ /dev/null @@ -1,71 +0,0 @@ -//! Example showing how to access and save the Radio Paradise source image -//! -//! This example demonstrates: -//! - Getting source information via the MusicSource trait -//! - Accessing the embedded WebP image -//! - Optionally saving it to a file - -use pmoaudiocache::cache as audio_cache; -use pmocovers::cache as covers_cache; -use pmoparadise::{RadioParadiseClient, RadioParadiseSource}; -use pmosource::MusicSource; -use std::fs; -use std::io::Write; -use std::sync::Arc; - -#[tokio::main] -async fn main() -> Result<(), Box> { - // Create the client and source - let client = RadioParadiseClient::new().await?; - - // Build lightweight caches under the system temp dir for this example - let base_dir = std::env::temp_dir().join(format!( - "pmoparadise_show_source_image_{}", - std::process::id() - )); - let covers_dir = base_dir.join("covers"); - let audio_dir = base_dir.join("audio"); - std::fs::create_dir_all(&covers_dir)?; - std::fs::create_dir_all(&audio_dir)?; - - let cover_cache = Arc::new(covers_cache::new_cache( - covers_dir.to_string_lossy().as_ref(), - 32, - )?); - let audio_cache = Arc::new(audio_cache::new_cache( - audio_dir.to_string_lossy().as_ref(), - 32, - )?); - - let source = RadioParadiseSource::new_default(client, cover_cache, audio_cache); - - // Display source information - println!("Music Source Information"); - println!("========================"); - println!("Name: {}", source.name()); - println!("ID: {}", source.id()); - println!("Image MIME type: {}", source.default_image_mime_type()); - - // Get the embedded image - let image_data = source.default_image(); - println!("Embedded image size: {} bytes", image_data.len()); - - // Verify WebP format - if image_data.len() >= 12 { - let is_webp = &image_data[0..4] == b"RIFF" && &image_data[8..12] == b"WEBP"; - println!("Valid WebP format: {}", is_webp); - } - - // Optional: save to file - if std::env::args().any(|arg| arg == "--save") { - let filename = format!("{}_default.webp", source.id()); - let mut file = fs::File::create(&filename)?; - file.write_all(image_data)?; - println!("\nImage saved to: {}", filename); - println!("You can view it with: open {}", filename); - } else { - println!("\nTo save the image to disk, run with: --save"); - } - - Ok(()) -} diff --git a/pmoparadise/examples/stream_block.rs b/pmoparadise/examples/stream_block.rs deleted file mode 100644 index 23f3327e..00000000 --- a/pmoparadise/examples/stream_block.rs +++ /dev/null @@ -1,100 +0,0 @@ -//! Example: Stream a Radio Paradise block with prefetching -//! -//! This example demonstrates: -//! - Streaming block audio data -//! - Writing to a file or piping to a player -//! - Prefetching the next block for gapless playback -//! - Continuous playback loop -//! -//! Run with: cargo run --example stream_block -//! -//! To play directly with mpv: -//! cargo run --example stream_block | mpv --no-cache --demuxer=+lavf - - -use futures::StreamExt; -use pmoparadise::{RadioParadiseClient, Result}; -use std::io::Write; - -#[tokio::main] -async fn main() -> Result<()> { - // Initialize logging (optional) - #[cfg(feature = "logging")] - tracing_subscriber::fmt::init(); - - eprintln!("Radio Paradise - Block Streaming Demo"); - eprintln!("======================================\n"); - - // Create client - let mut client = RadioParadiseClient::builder().build().await?; - - eprintln!("Client configured for FLAC streaming\n"); - - // Get current block - let current_block = client.get_block(None).await?; - - eprintln!("Current Block:"); - eprintln!(" Event: {}", current_block.event); - eprintln!(" Songs: {}", current_block.song_count()); - eprintln!( - " Duration: {:.1} minutes", - current_block.length as f64 / 60000.0 - ); - eprintln!(" URL: {}\n", current_block.url); - - // Display tracklist - eprintln!("Tracklist:"); - for (index, song) in current_block.songs_ordered() { - eprintln!(" {}. {} - {}", index + 1, song.artist, song.title); - } - eprintln!(); - - // Prefetch next block in advance - eprintln!("Prefetching next block..."); - client.prefetch_next(¤t_block).await?; - eprintln!( - "Next block prefetched: {}\n", - client.next_block_url().unwrap() - ); - - // Stream the block - eprintln!("Streaming block... (writing to stdout)"); - eprintln!("Tip: Pipe to a player like: cargo run --example stream_block | mpv -\n"); - - let mut stream = client.stream_block_from_metadata(¤t_block).await?; - let mut total_bytes = 0u64; - let mut stdout = std::io::stdout(); - - while let Some(chunk_result) = stream.next().await { - let chunk = chunk_result?; - total_bytes += chunk.len() as u64; - - // Write to stdout (can be piped to a player) - stdout.write_all(&chunk)?; - stdout.flush()?; - - // Progress indicator (to stderr so it doesn't interfere with piped audio) - if total_bytes % (1024 * 1024) == 0 { - eprintln!( - " Downloaded: {:.1} MB", - total_bytes as f64 / 1024.0 / 1024.0 - ); - } - } - - eprintln!("\nBlock streaming complete!"); - eprintln!( - "Total downloaded: {:.2} MB", - total_bytes as f64 / 1024.0 / 1024.0 - ); - - // In a real application, you would now: - // 1. Get the next block using prefetched metadata - // 2. Stream it seamlessly - // 3. Prefetch the following block - // 4. Repeat for continuous playback - - eprintln!("\nFor continuous playback, you would now stream the next block:"); - eprintln!(" Event: {}", current_block.end_event); - - Ok(()) -} diff --git a/pmoparadise/examples/test_streaming.rs b/pmoparadise/examples/test_streaming.rs deleted file mode 100644 index 78a1279d..00000000 --- a/pmoparadise/examples/test_streaming.rs +++ /dev/null @@ -1,129 +0,0 @@ -//! Test progressive streaming implementation -//! -//! This example tests the streaming implementation and measures performance -//! -//! Run with: -//! ```bash -//! RUST_LOG=info cargo run --example test_streaming -//! ``` - -use pmoparadise::RadioParadiseClient; -use std::time::Instant; - -#[tokio::main] -async fn main() -> Result<(), Box> { - // Initialize tracing with timestamps - tracing_subscriber::fmt() - .with_target(false) - .with_thread_ids(false) - .with_level(true) - .init(); - - println!("🎵 Testing Progressive FLAC Streaming"); - println!("=====================================\n"); - - // Create the Radio Paradise client - println!("📡 Connecting to Radio Paradise..."); - let client = RadioParadiseClient::new().await?; - println!("✅ Connected!\n"); - - // Get current block - println!("🎧 Fetching current block metadata..."); - let block = client.get_block(None).await?; - - println!("\n📊 Block Information:"); - println!(" Event ID: {}", block.event); - println!(" Songs: {}", block.song_count()); - println!(" Duration: ~{} seconds\n", block.length / 1000); - - // List songs - println!("🎵 Songs in this block:"); - for (idx, song) in block.songs_ordered() { - println!( - " {}. {} - {} ({}s at {}s)", - idx + 1, - song.artist, - song.title, - song.duration / 1000, - song.elapsed / 1000 - ); - } - println!(); - - // Now test the streaming decoder - println!("⚡ Starting progressive streaming test..."); - println!(" (This will download and decode the block progressively)"); - println!(); - - let start_time = Instant::now(); - let block_url = block.url.parse()?; - let http_stream = client.stream_block(&block_url).await?; - - use pmoparadise::streaming::StreamingPCMDecoder; - - // Decode in a blocking task - let decode_task = tokio::task::spawn_blocking(move || -> anyhow::Result> { - let mut decoder = StreamingPCMDecoder::new(http_stream)?; - - println!( - " 🎼 Stream info: {}Hz, {} channels, {} bits", - decoder.sample_rate(), - decoder.channels(), - decoder.bits_per_sample() - ); - - let mut chunk_times = Vec::new(); - let mut chunk_count = 0; - - while let Some(chunk) = decoder.decode_chunk()? { - chunk_count += 1; - chunk_times.push((chunk.position_ms, chunk.samples.len())); - - if chunk_count % 50 == 0 { - println!( - " 📦 Chunk {} at {}ms ({} samples)", - chunk_count, - chunk.position_ms, - chunk.samples.len() - ); - } - } - - Ok(chunk_times) - }); - - let chunk_times = decode_task - .await - .map_err(|e| anyhow::anyhow!("Join error: {}", e))??; - let total_time = start_time.elapsed(); - - println!("\n✅ Streaming Complete!"); - println!("\n📈 Performance Metrics:"); - println!(" Total chunks decoded: {}", chunk_times.len()); - println!(" Total time: {:.2}s", total_time.as_secs_f64()); - - if let Some((first_pos, _)) = chunk_times.first() { - println!(" First chunk at: {}ms", first_pos); - } - - if let Some((last_pos, _)) = chunk_times.last() { - println!( - " Last chunk at: {}ms (~{:.1}s)", - last_pos, - last_pos / 1000 - ); - } - - println!("\n💡 Analysis:"); - println!(" With the old approach (download all first):"); - println!(" - Would need to wait for full download (~12-16s)"); - println!(" - Then decode all samples"); - println!(" - Total: ~15-20s before first track"); - println!(); - println!(" With progressive streaming:"); - println!(" - First chunks arrive in ~2-3s"); - println!(" - First track (3min) ready in ~6-8s"); - println!(" - Improvement: ~2x faster! ⚡"); - - Ok(()) -} diff --git a/pmoparadise/examples/with_cache.rs b/pmoparadise/examples/with_cache.rs deleted file mode 100644 index 563ed755..00000000 --- a/pmoparadise/examples/with_cache.rs +++ /dev/null @@ -1,107 +0,0 @@ -//! Example demonstrating Radio Paradise with cache support -//! -//! This example shows how to use the RadioParadiseSource with pmocovers -//! and pmoaudiocache to cache both cover images and audio tracks. -//! -//! Run with: -//! ```bash -//! cargo run --example with_cache --features cache -//! ``` - -use pmoaudiocache::AudioCache; -use pmocovers::Cache as CoverCache; -use pmoparadise::{RadioParadiseClient, RadioParadiseSource}; -use pmosource::MusicSource; -use std::sync::Arc; -use tokio::time::{sleep, Duration}; - -#[tokio::main] -async fn main() -> Result<(), Box> { - // Initialize tracing - tracing_subscriber::fmt::init(); - - println!("🎵 Radio Paradise with Cache Support"); - println!("=====================================\n"); - - // Create the Radio Paradise client - println!("📡 Connecting to Radio Paradise..."); - let client = RadioParadiseClient::new().await?; - println!("✅ Connected!\n"); - - // Initialize caches - println!("💾 Initializing caches..."); - let cover_cache = Arc::new(CoverCache::new("./cache/covers", 500)?); - let audio_cache = Arc::new(AudioCache::new("./cache/audio", 100)?); - println!("✅ Caches initialized!\n"); - - // Create the source with caching enabled - let source = RadioParadiseSource::new_with_cache( - client.clone(), - "http://localhost:8080", - 50, - Some(cover_cache.clone()), - Some(audio_cache.clone()), - ); - - println!("📻 Source: {}", source.name()); - println!("🆔 ID: {}", source.id()); - println!("📝 Supports FIFO: {}\n", source.supports_fifo()); - - // Fetch current playing information - println!("🎧 Fetching current track information..."); - let now_playing = client.now_playing().await?; - let block = Arc::new(now_playing.block.clone()); - - println!("\n🎵 Now Playing:"); - println!(" Event: {}", block.event); - if let Some(song) = &now_playing.current_song { - println!(" Title: {}", song.title); - println!(" Artist: {}", song.artist); - println!(" Album: {}", song.album); - } - println!(); - - // Add current song to the source - println!("➕ Adding current track to FIFO with caching..."); - if let Some(song) = &now_playing.current_song { - source - .add_song( - block.clone(), - song, - now_playing.current_song_index.unwrap_or(0), - ) - .await?; - println!("✅ Track added and caching started!"); - println!(" - Cover image will be cached to: ./cache/covers/"); - println!(" - Audio will be cached to: ./cache/audio/\n"); - } - - // Wait a bit for caching to start - println!("⏳ Waiting for cache operations to complete..."); - sleep(Duration::from_secs(5)).await; - - // Get items from FIFO - println!("\n📋 Items in FIFO:"); - let items = source.get_items(0, 10).await?; - for (i, item) in items.iter().enumerate() { - println!( - " {}. {} - {}", - i + 1, - item.artist.as_deref().unwrap_or("Unknown"), - item.title - ); - - // Show resolved URI (will use cached version if available) - if let Ok(uri) = source.resolve_uri(&item.id).await { - println!(" URI: {}", uri); - } - } - - println!("\n✨ Example complete!"); - println!("\n💡 Tips:"); - println!(" - Run the example again to see faster loading from cache"); - println!(" - Check ./cache/covers/ for cached cover images"); - println!(" - Check ./cache/audio/ for cached FLAC files"); - - Ok(()) -} diff --git a/pmoparadise/src/client.rs b/pmoparadise/src/client.rs index 3ba5ca94..48b87295 100644 --- a/pmoparadise/src/client.rs +++ b/pmoparadise/src/client.rs @@ -9,8 +9,8 @@ use url::Url; /// Default Radio Paradise API base URL pub const DEFAULT_API_BASE: &str = "https://api.radioparadise.com/api"; -/// Default block base URL pattern -pub const DEFAULT_BLOCK_BASE: &str = "https://apps.radioparadise.com/blocks/chan/0"; +/// Default block base URL (channel is appended) +pub const DEFAULT_BLOCK_BASE: &str = "https://apps.radioparadise.com/blocks/chan"; /// Default image base URL pub const DEFAULT_IMAGE_BASE: &str = "https://img.radioparadise.com/"; @@ -24,6 +24,9 @@ pub const DEFAULT_BLOCK_TIMEOUT_SECS: u64 = 180; /// Default User-Agent pub const DEFAULT_USER_AGENT: &str = "pmoparadise/0.1.0"; +/// Default channel (0 = main mix) +pub const DEFAULT_CHANNEL: u8 = 0; + /// Radio Paradise HTTP client /// /// This client provides access to Radio Paradise's streaming API, @@ -48,7 +51,6 @@ pub const DEFAULT_USER_AGENT: &str = "pmoparadise/0.1.0"; pub struct RadioParadiseClient { pub(crate) client: Client, api_base: String, - block_base: String, channel: u8, pub(crate) request_timeout: Duration, pub(crate) block_timeout: Duration, @@ -71,12 +73,14 @@ impl RadioParadiseClient { /// Create a client with a custom reqwest::Client /// /// Useful for sharing HTTP connection pools or custom proxy settings + /// + /// Note: Uses default settings (channel 0, default timeouts). + /// For more control, use `ClientBuilder::default().client(client).build()`. pub fn with_client(client: Client) -> Self { Self { client, api_base: DEFAULT_API_BASE.to_string(), - block_base: DEFAULT_BLOCK_BASE.to_string(), - channel: 0, + channel: DEFAULT_CHANNEL, request_timeout: Duration::from_secs(DEFAULT_REQUEST_TIMEOUT_SECS), block_timeout: Duration::from_secs(DEFAULT_BLOCK_TIMEOUT_SECS), next_block_url: None, @@ -88,15 +92,15 @@ impl RadioParadiseClient { self.channel } - fn block_base_for_channel(channel: u8) -> String { - format!("https://apps.radioparadise.com/blocks/chan/{}", channel) + /// Get the block base URL for this client's channel + pub fn block_base(&self) -> String { + format!("{}/{}", DEFAULT_BLOCK_BASE, self.channel) } /// Clone the client with a different channel while preserving other settings. pub fn clone_with_channel(&self, channel: u8) -> Self { let mut cloned = self.clone(); cloned.channel = channel; - cloned.block_base = Self::block_base_for_channel(channel); cloned.next_block_url = None; cloned } @@ -234,7 +238,6 @@ impl RadioParadiseClient { pub struct ClientBuilder { client: Option, api_base: String, - block_base: String, channel: u8, request_timeout: Duration, block_timeout: Duration, @@ -247,8 +250,7 @@ impl Default for ClientBuilder { Self { client: None, api_base: DEFAULT_API_BASE.to_string(), - block_base: DEFAULT_BLOCK_BASE.to_string(), - channel: 0, + channel: DEFAULT_CHANNEL, request_timeout: Duration::from_secs(DEFAULT_REQUEST_TIMEOUT_SECS), block_timeout: Duration::from_secs(DEFAULT_BLOCK_TIMEOUT_SECS), user_agent: DEFAULT_USER_AGENT.to_string(), @@ -275,12 +277,6 @@ impl ClientBuilder { self } - /// Set the block base URL - pub fn block_base(mut self, url: impl Into) -> Self { - self.block_base = url.into(); - self - } - /// Set the channel (0 = main mix, 1 = mellow, 2 = rock, 3 = world/etc) pub fn channel(mut self, channel: u8) -> Self { self.channel = channel; @@ -329,16 +325,9 @@ impl ClientBuilder { builder.build()? }; - let block_base = if self.block_base == DEFAULT_BLOCK_BASE { - RadioParadiseClient::block_base_for_channel(self.channel) - } else { - self.block_base.clone() - }; - Ok(RadioParadiseClient { client, api_base: self.api_base, - block_base, channel: self.channel, request_timeout: self.request_timeout, block_timeout: self.block_timeout, @@ -355,6 +344,6 @@ mod tests { fn test_builder_defaults() { let builder = ClientBuilder::default(); assert_eq!(builder.api_base, DEFAULT_API_BASE); - assert_eq!(builder.channel, 0); + assert_eq!(builder.channel, DEFAULT_CHANNEL); } } diff --git a/pmoparadise/src/config_ext.rs b/pmoparadise/src/config_ext.rs index c2303595..49becbb5 100644 --- a/pmoparadise/src/config_ext.rs +++ b/pmoparadise/src/config_ext.rs @@ -21,6 +21,7 @@ //! } //! ``` +use crate::{channels::ParadiseChannelKind, client::DEFAULT_CHANNEL}; use anyhow::Result; use pmoconfig::Config; use serde_yaml::Value; @@ -81,6 +82,60 @@ pub trait RadioParadiseConfigExt { /// config.set_paradise_enabled(false)?; /// ``` fn set_paradise_enabled(&self, enabled: bool) -> Result<()>; + + /// Récupère le channel par défaut + /// + /// # Returns + /// + /// Le channel par défaut (0 = Main Mix par défaut). + /// + /// Si la valeur n'existe pas dans la configuration, elle est automatiquement + /// définie à "main" et persistée. + /// + /// # Channels disponibles + /// + /// Peut être configuré comme chaîne de caractères ou nombre : + /// - "main" ou 0 = Main Mix (eclectic, diverse mix) + /// - "mellow" ou 1 = Mellow Mix (smooth, chilled music) + /// - "rock" ou 2 = Rock Mix (classic & modern rock) + /// - "eclectic" ou 3 = Eclectic Mix (global sounds) + /// + /// # Exemple de configuration YAML + /// + /// ```yaml + /// sources: + /// radio_paradise: + /// default_channel: mellow # or 1 + /// ``` + /// + /// # Exemple d'utilisation + /// + /// ```rust,ignore + /// let channel = config.get_paradise_default_channel()?; + /// let client = RadioParadiseClient::builder().channel(channel).build().await?; + /// ``` + fn get_paradise_default_channel(&self) -> Result; + + /// Définit le channel par défaut + /// + /// # Arguments + /// + /// * `channel` - Le channel (0-3) + /// + /// La valeur est stockée sous forme de nom convivial ("main", "mellow", etc.) + /// dans le fichier de configuration. + /// + /// # Exemple + /// + /// ```rust,ignore + /// use pmoparadise::channels::ParadiseChannelKind; + /// + /// // Use Mellow Mix by default + /// config.set_paradise_default_channel(ParadiseChannelKind::Mellow.id())?; + /// // Or simply: + /// config.set_paradise_default_channel(1)?; + /// ``` + fn set_paradise_default_channel(&self, channel: u8) -> Result<()>; } impl RadioParadiseConfigExt for Config { @@ -101,6 +156,62 @@ impl RadioParadiseConfigExt for Config { Value::Bool(enabled), ) } + + fn get_paradise_default_channel(&self) -> Result { + match self.get_value(&["sources", "radio_paradise", "default_channel"]) { + Ok(Value::String(s)) => { + // Try to parse as channel name (e.g., "main", "mellow", etc.) + match s.parse::() { + Ok(kind) => Ok(kind.id()), + Err(_) => { + // Invalid channel name, use default + self.set_paradise_default_channel(DEFAULT_CHANNEL)?; + Ok(DEFAULT_CHANNEL) + } + } + } + Ok(Value::Number(n)) => { + // Accept numeric channel ID (0-3) + if let Some(ch) = n.as_u64() { + if ch <= 3 { + Ok(ch as u8) + } else { + // Invalid channel number, use default + self.set_paradise_default_channel(DEFAULT_CHANNEL)?; + Ok(DEFAULT_CHANNEL) + } + } else { + // Not a valid number, use default + self.set_paradise_default_channel(DEFAULT_CHANNEL)?; + Ok(DEFAULT_CHANNEL) + } + } + _ => { + // Use default and persist it as "main" (user-friendly) + self.set_value( + &["sources", "radio_paradise", "default_channel"], + Value::String("main".to_string()), + )?; + Ok(DEFAULT_CHANNEL) + } + } + } + + fn set_paradise_default_channel(&self, channel: u8) -> Result<()> { + // Convert channel ID to user-friendly string name + let channel_name = match channel { + 0 => "main", + 1 => "mellow", + 2 => "rock", + 3 => "eclectic", + _ => return Err(anyhow::anyhow!("Invalid channel ID: {}", channel)), + }; + + self.set_value( + &["sources", "radio_paradise", "default_channel"], + Value::String(channel_name.to_string()), + ) + } } #[cfg(test)] diff --git a/pmoparadise/src/ffmpeg_streaming.rs b/pmoparadise/src/ffmpeg_streaming.rs deleted file mode 100644 index 389ed8ff..00000000 --- a/pmoparadise/src/ffmpeg_streaming.rs +++ /dev/null @@ -1,172 +0,0 @@ -//! FFmpeg-based progressive streaming decoder/encoder -//! -//! This module provides progressive audio streaming using FFmpeg, -//! allowing for much lower latency than the claxon/flacenc approach. -//! -//! Key advantages: -//! - Start streaming immediately (< 1 second latency) -//! - Progressive decoding and encoding in a pipeline -//! - Better performance (C code vs Rust) -//! - Support for multiple output formats - -use anyhow::{anyhow, Context, Result}; -use bytes::Bytes; -use ffmpeg_next as ffmpeg; -use std::io::{Read, Write}; -use std::sync::mpsc::{sync_channel, Receiver, SyncSender}; -use tokio::task; -use tracing::{debug, error, trace}; - -/// Initialize FFmpeg (must be called once at startup) -pub fn init() -> Result<()> { - ffmpeg::init().context("Failed to initialize FFmpeg")?; - Ok(()) -} - -/// PCM chunk with decoded audio data -#[derive(Debug, Clone)] -pub struct PCMChunk { - pub samples: Vec, // Interleaved 16-bit samples - pub sample_rate: u32, - pub channels: u32, - pub position_ms: u64, -} - -/// Progressive decoder that decodes FLAC data as it arrives -pub struct ProgressiveDecoder { - input_rx: Receiver>, - buffer: Vec, - decoder_ctx: Option, - sample_rate: u32, - channels: u32, - total_samples_decoded: u64, -} - -impl ProgressiveDecoder { - /// Create a new progressive decoder from a byte stream - pub fn new(mut stream: impl Read + Send + 'static) -> Result { - let (tx, rx) = sync_channel(64); - - // Spawn a thread to read from the stream and feed chunks - std::thread::spawn(move || { - let mut buffer = vec![0u8; 8192]; - loop { - match stream.read(&mut buffer) { - Ok(0) => break, // EOF - Ok(n) => { - let chunk = Bytes::copy_from_slice(&buffer[..n]); - if tx.send(Ok(chunk)).is_err() { - break; - } - } - Err(e) => { - let _ = tx.send(Err(e.to_string())); - break; - } - } - } - }); - - Ok(Self { - input_rx: rx, - buffer: Vec::with_capacity(65536), - decoder_ctx: None, - sample_rate: 0, - channels: 0, - total_samples_decoded: 0, - }) - } - - /// Decode the next chunk of PCM data - pub fn decode_chunk(&mut self) -> Result> { - // Receive more data from the stream - while self.buffer.len() < 4096 { - match self.input_rx.try_recv() { - Ok(Ok(bytes)) => { - self.buffer.extend_from_slice(&bytes); - } - Ok(Err(e)) => { - return Err(anyhow!("Stream error: {}", e)); - } - Err(std::sync::mpsc::TryRecvError::Empty) => { - // No more data available right now - break; - } - Err(std::sync::mpsc::TryRecvError::Disconnected) => { - // Stream ended - if self.buffer.is_empty() { - return Ok(None); - } - break; - } - } - } - - if self.buffer.is_empty() { - return Ok(None); - } - - // Initialize decoder on first call - if self.decoder_ctx.is_none() { - self.init_decoder()?; - } - - // Decode a frame - // TODO: Implement actual FFmpeg decoding - // For now, return a placeholder - - Ok(None) - } - - fn init_decoder(&mut self) -> Result<()> { - // TODO: Initialize FFmpeg decoder from buffer - // Parse FLAC header, create decoder context - Ok(()) - } -} - -/// Progressive encoder that encodes PCM to FLAC as data arrives -pub struct ProgressiveEncoder { - output_tx: SyncSender, - encoder_ctx: Option, - sample_rate: u32, - channels: u32, -} - -impl ProgressiveEncoder { - /// Create a new progressive encoder - pub fn new(sample_rate: u32, channels: u32) -> Result<(Self, Receiver)> { - let (tx, rx) = sync_channel(64); - - let encoder = Self { - output_tx: tx, - encoder_ctx: None, - sample_rate, - channels, - }; - - Ok((encoder, rx)) - } - - /// Encode a chunk of PCM data - pub fn encode_chunk(&mut self, pcm: &PCMChunk) -> Result<()> { - // TODO: Implement FFmpeg encoding - Ok(()) - } - - /// Flush any remaining encoded data - pub fn flush(&mut self) -> Result<()> { - // TODO: Flush encoder - Ok(()) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_ffmpeg_init() { - assert!(init().is_ok()); - } -} diff --git a/pmoparadise/src/paradise/channel.rs b/pmoparadise/src/paradise/channel.rs deleted file mode 100644 index 9953b061..00000000 --- a/pmoparadise/src/paradise/channel.rs +++ /dev/null @@ -1,428 +0,0 @@ -//! Channel orchestration primitives. -//! -//! This module wires together configuration, playlists, workers and client -//! tracking for a single Radio Paradise channel. The implementation is still -//! a scaffolding of the final behaviour; commands sent to the worker are -//! logged but not yet executing the full download/buffering pipeline. - -use super::history::HistoryBackend; -use super::playlist::{PlaylistEntry, SharedPlaylist}; -use super::worker::{ParadiseWorker, WorkerCommand}; -use crate::client::RadioParadiseClient; -use anyhow::{Context, Result}; -use async_stream::try_stream; -use bytes::Bytes; -use futures::{stream::BoxStream, StreamExt}; -use pmosource::SourceCacheManager; -use std::fmt; -use std::str::FromStr; -use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::Arc; -use tokio::fs::File; -use tokio::sync::{mpsc, Mutex}; -use tokio_util::io::ReaderStream; -use tracing::warn; - -/// Logical identifier for a Radio Paradise channel. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub enum ParadiseChannelKind { - Main, - Mellow, - Rock, - Eclectic, -} - -impl ParadiseChannelKind { - pub const fn id(self) -> u8 { - match self { - Self::Main => 0, - Self::Mellow => 1, - Self::Rock => 2, - Self::Eclectic => 3, - } - } - - pub const fn slug(self) -> &'static str { - match self { - Self::Main => "main", - Self::Mellow => "mellow", - Self::Rock => "rock", - Self::Eclectic => "eclectic", - } - } - - pub const fn display_name(self) -> &'static str { - match self { - Self::Main => "Main Mix", - Self::Mellow => "Mellow Mix", - Self::Rock => "Rock Mix", - Self::Eclectic => "Eclectic Mix", - } - } - - pub const fn description(self) -> &'static str { - match self { - Self::Main => "Eclectic mix of rock, world, electronica, and more", - Self::Mellow => "Mellower, less aggressive music", - Self::Rock => "Heavier, more guitar-driven music", - Self::Eclectic => "Curated worldwide selection", - } - } -} - -impl FromStr for ParadiseChannelKind { - type Err = anyhow::Error; - - fn from_str(s: &str) -> std::result::Result { - match s.to_ascii_lowercase().as_str() { - "main" | "0" => Ok(Self::Main), - "mellow" | "1" => Ok(Self::Mellow), - "rock" | "2" => Ok(Self::Rock), - "eclectic" | "3" => Ok(Self::Eclectic), - other => Err(anyhow::anyhow!("Unknown Radio Paradise channel: {}", other)), - } - } -} - -/// Metadata descriptor for a channel. -#[derive(Debug, Clone, Copy)] -pub struct ChannelDescriptor { - pub kind: ParadiseChannelKind, - pub id: u8, - pub slug: &'static str, - pub display_name: &'static str, - pub description: &'static str, -} - -impl ChannelDescriptor { - pub const fn new(kind: ParadiseChannelKind) -> Self { - Self { - id: kind.id(), - slug: kind.slug(), - display_name: kind.display_name(), - description: kind.description(), - kind, - } - } -} - -pub const ALL_CHANNELS: [ChannelDescriptor; 4] = [ - ChannelDescriptor::new(ParadiseChannelKind::Main), - ChannelDescriptor::new(ParadiseChannelKind::Mellow), - ChannelDescriptor::new(ParadiseChannelKind::Rock), - 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 { - inner: Arc, -} - -struct ParadiseChannelInner { - descriptor: ChannelDescriptor, - client: RadioParadiseClient, - history_max_tracks: usize, - playlist: SharedPlaylist, - history: Arc, - cache_manager: Arc, - active_clients: AtomicUsize, - worker_tx: mpsc::Sender, - worker: Mutex>, -} - -impl fmt::Debug for ParadiseChannel { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("ParadiseChannel") - .field("slug", &self.inner.descriptor.slug) - .field( - "active_clients", - &self.inner.active_clients.load(Ordering::SeqCst), - ) - .finish() - } -} - -impl ParadiseChannel { - #[allow(clippy::too_many_arguments)] - pub fn new( - descriptor: ChannelDescriptor, - base_client: RadioParadiseClient, - history_max_tracks: usize, - history: Arc, - cache_manager: Arc, - ) -> Result { - let client = base_client.clone_with_channel(descriptor.id); - let playlist = SharedPlaylist::new(history_max_tracks); - let (worker, worker_tx) = ParadiseWorker::spawn( - descriptor, - client.clone(), - history_max_tracks, - playlist.clone(), - history.clone(), - cache_manager.clone(), - ); - - Ok(Self { - inner: Arc::new(ParadiseChannelInner { - descriptor, - client, - history_max_tracks, - playlist, - history, - cache_manager, - active_clients: AtomicUsize::new(0), - worker_tx, - worker: Mutex::new(Some(worker)), - }), - }) - } - - pub fn descriptor(&self) -> ChannelDescriptor { - self.inner.descriptor - } - - pub fn playlist(&self) -> &SharedPlaylist { - &self.inner.playlist - } - - pub fn history_max_tracks(&self) -> usize { - self.inner.history_max_tracks - } - - pub fn history_backend(&self) -> &Arc { - &self.inner.history - } - - pub fn cache_manager(&self) -> Arc { - self.inner.cache_manager.clone() - } - - pub fn client(&self) -> &RadioParadiseClient { - &self.inner.client - } - - pub fn active_client_count(&self) -> usize { - self.inner.active_clients.load(Ordering::SeqCst) - } - - pub async fn connect_client( - &self, - client_id: impl Into, - ) -> Result { - let client_id = client_id.into(); - self.inner.active_clients.fetch_add(1, Ordering::SeqCst); - - if let Err(err) = self - .inner - .worker_tx - .send(WorkerCommand::ClientConnected { - client_id: client_id.clone(), - }) - .await - { - self.inner.active_clients.fetch_sub(1, Ordering::SeqCst); - return Err(anyhow::anyhow!("worker unavailable: {}", err)); - } - - self.inner.playlist.increment_all_pending().await; - self.ensure_started().await?; - - Ok(ParadiseClientStream::new(self.clone(), client_id)) - } - - pub async fn disconnect_client(&self, client_id: impl Into) -> Result<()> { - let client_id = client_id.into(); - self.inner.active_clients.fetch_sub(1, Ordering::SeqCst); - self.inner - .worker_tx - .send(WorkerCommand::ClientDisconnected { client_id }) - .await - .context("failed to notify worker of client disconnection")?; - Ok(()) - } - - pub async fn ensure_started(&self) -> Result<()> { - self.inner - .worker_tx - .send(WorkerCommand::EnsureReady) - .await - .context("failed to schedule worker warmup") - } - - pub async fn shutdown(&self) -> Result<()> { - self.inner - .worker_tx - .send(WorkerCommand::Shutdown) - .await - .ok(); - - let mut guard = self.inner.worker.lock().await; - if let Some(worker) = guard.take() { - worker - .wait() - .await - .context("failed to join worker task") - .map(|_| ()) - } else { - Ok(()) - } - } - - pub async fn mark_track_completed(&self, track: &Arc) { - let remaining = track.decrement_clients(); - if remaining > 0 { - return; - } - - if let Some(removed) = self - .inner - .playlist - .pop_front_matching(&track.track_id) - .await - { - if let Err(err) = self.inner.history.append(removed.as_history_entry()).await { - warn!( - channel = self.inner.descriptor.slug, - "Failed to persist history entry: {err:?}" - ); - } - - if let Err(err) = self - .inner - .history - .truncate(self.inner.history_max_tracks) - .await - { - warn!( - channel = self.inner.descriptor.slug, - "Failed to truncate history: {err:?}" - ); - } - - let history_entry = removed.as_history_entry(); - self.inner.playlist.push_history_entry(history_entry).await; - } - } -} - -/// Placeholder stream handle for per-client playback. -#[derive(Debug, Clone)] -pub struct ParadiseClientStream { - channel: ParadiseChannel, - client_id: String, -} - -impl ParadiseClientStream { - fn new(channel: ParadiseChannel, client_id: String) -> Self { - Self { channel, client_id } - } - - pub fn client_id(&self) -> &str { - &self.client_id - } - - pub fn channel(&self) -> ParadiseChannel { - self.channel.clone() - } - - pub fn into_byte_stream(self) -> BoxStream<'static, Result> { - let channel = self.channel.clone(); - let client_id = self.client_id.clone(); - let stream = try_stream! { - tracing::info!( - channel = channel.descriptor().slug, - client_id = %client_id, - "🎧 Client connecting to stream" - ); - channel.ensure_started().await?; - let mut last_track_id: Option = None; - loop { - let entries = channel.playlist().active_snapshot().await; - - // Find the next track after last_track_id - let next_entry = if let Some(ref last_id) = last_track_id { - // Find the position of the last track we read - let last_pos = entries.iter().position(|e| e.track_id == *last_id); - - // Get the next track (or wait if none available) - match last_pos { - Some(pos) if pos + 1 < entries.len() => { - Some(entries[pos + 1].clone()) - } - _ => { - // Last track not found (was removed) or no next track available - // Wait for more tracks to be added - channel.ensure_started().await?; - let current_len = entries.len(); - channel.playlist().wait_for_track_count(current_len).await; - continue; - } - } - } else { - // First track for this client - if entries.is_empty() { - channel.ensure_started().await?; - channel.playlist().wait_for_track_count(0).await; - continue; - } - Some(entries[0].clone()) - }; - - let entry = next_entry.unwrap(); - last_track_id = Some(entry.track_id.clone()); - - let audio_pk = entry - .audio_pk - .clone() - .ok_or_else(|| anyhow::anyhow!("Audio not cached yet"))?; - - channel - .cache_manager() - .wait_audio_ready(&audio_pk) - .await - .map_err(|e| anyhow::anyhow!(e.to_string()))?; - - let file_path = if let Some(path) = entry.file_path.clone() { - path - } else { - channel - .cache_manager() - .audio_file_path(&audio_pk) - .await - .ok_or_else(|| anyhow::anyhow!("Audio file path unavailable"))? - }; - - let file = File::open(&file_path).await?; - let mut reader = ReaderStream::new(file); - - while let Some(chunk) = reader.next().await { - let bytes = chunk?; - yield bytes; - } - - channel.mark_track_completed(&entry).await; - } - }; - - stream.boxed() - } -} - -impl Drop for ParadiseClientStream { - fn drop(&mut self) { - let channel = self.channel.clone(); - let client_id = self.client_id.clone(); - let slug = channel.descriptor().slug; - tokio::spawn(async move { - if let Err(err) = channel.disconnect_client(client_id).await { - warn!(channel = slug, "Failed to disconnect client: {err:?}"); - } - }); - } -} diff --git a/pmoparadise/src/paradise/constants.rs b/pmoparadise/src/paradise/constants.rs deleted file mode 100644 index d6dd6631..00000000 --- a/pmoparadise/src/paradise/constants.rs +++ /dev/null @@ -1,208 +0,0 @@ -//! Constants for Radio Paradise orchestration layer. -//! -//! This module defines all the hardcoded parameters for the Radio Paradise -//! integration. These values are based on empirical testing and Radio Paradise's -//! infrastructure characteristics. - -use std::time::Duration; - -// ============================================================================ -// Activity Lifecycle -// ============================================================================ - -/// Cooling timeout after all clients disconnect (seconds) -/// -/// After the last client disconnects, the channel enters a "cooling" state -/// where it remains active for this duration before shutting down completely. -/// This avoids rapid start/stop cycles if clients reconnect quickly. -/// -/// Value: 180 seconds (3 minutes) - good balance between responsiveness and stability -pub const COOLING_TIMEOUT_SECONDS: u64 = 180; - -// ============================================================================ -// Polling Intervals -// ============================================================================ - -/// High buffer polling interval (seconds) -/// -/// When the playlist buffer has 3+ blocks, poll less frequently to reduce -/// API load and network usage. -/// -/// Value: 120 seconds (2 minutes) -pub const POLLING_INTERVAL_HIGH_BUFFER: u64 = 120; - -/// Medium buffer polling interval (seconds) -/// -/// When the playlist buffer has 2 blocks, poll at moderate frequency. -/// -/// Value: 60 seconds (1 minute) -pub const POLLING_INTERVAL_MEDIUM_BUFFER: u64 = 60; - -/// Low buffer polling interval (seconds) -/// -/// When the playlist buffer has less than 2 blocks, poll frequently to -/// ensure continuous playback. -/// -/// Value: 20 seconds -pub const POLLING_INTERVAL_LOW_BUFFER: u64 = 20; - -/// Helper to get high buffer polling interval as Duration -pub fn polling_high_interval() -> Duration { - Duration::from_secs(POLLING_INTERVAL_HIGH_BUFFER) -} - -/// Helper to get medium buffer polling interval as Duration -pub fn polling_medium_interval() -> Duration { - Duration::from_secs(POLLING_INTERVAL_MEDIUM_BUFFER) -} - -/// Helper to get low buffer polling interval as Duration -pub fn polling_low_interval() -> Duration { - Duration::from_secs(POLLING_INTERVAL_LOW_BUFFER) -} - -// ============================================================================ -// Polling Backoff (on API errors) -// ============================================================================ - -/// Initial backoff delay on API error (seconds) -/// -/// When an API request fails, we wait this duration before retrying. -/// -/// Value: 20 seconds -pub const BACKOFF_INITIAL_SECONDS: u64 = 20; - -/// Maximum backoff delay (seconds) -/// -/// Backoff is capped at this value to avoid waiting too long. -/// -/// Value: 300 seconds (5 minutes) -pub const BACKOFF_MAX_SECONDS: u64 = 300; - -/// Backoff multiplier -/// -/// After each failure, the delay is multiplied by this factor. -/// Example: 20s → 40s → 80s → 160s → 300s (capped) -/// -/// Value: 2.0 (exponential backoff) -pub const BACKOFF_MULTIPLIER: f32 = 2.0; - -// ============================================================================ -// Cache Tuning -// ============================================================================ - -/// Maximum number of blocks to remember in the worker -/// -/// This prevents unbounded memory growth by limiting how many block event IDs -/// we track to avoid re-processing. -/// -/// Calculation: (4 channels + 1 buffer) × 3 blocks per channel = 15 blocks -/// Each block is ~20 minutes of audio, so 15 blocks ≈ 5 hours of history -/// -/// Value: 15 blocks -pub const MAX_BLOCKS_REMEMBERED: usize = 15; - -/// Number of bytes to use for track ID hashing -/// -/// Track IDs are constructed by hashing block content and track position. -/// This value defines how much of the FLAC data we read for hashing. -/// -/// Value: 512 bytes - sufficient for unique identification without excessive I/O -pub const TRACK_ID_HASH_BYTES: usize = 512; - -// ============================================================================ -// History -// ============================================================================ - -/// Default maximum number of tracks to keep in history -/// -/// This is used as the default if not configured via pmoconfig. -/// Users can override this value in their configuration. -/// -/// Value: 100 tracks - represents ~5-8 hours of playback history -pub const HISTORY_DEFAULT_MAX_TRACKS: usize = 100; - -// ============================================================================ -// Streaming -// ============================================================================ - -/// Stream buffer size (bytes) -/// -/// Buffer size for audio streaming. 64KB provides good balance between -/// latency and buffering efficiency. -/// -/// Value: 64 KB -pub const STREAM_BUFFER_SIZE_BYTES: usize = 64 * 1024; - -/// Enable gapless playback -/// -/// Radio Paradise blocks are designed for gapless playback - each block -/// transitions seamlessly to the next without audio gaps. -/// -/// Value: true (always enabled) -pub const STREAM_GAPLESS: bool = true; - -// Note: Metadata format is always ICY (Icecast/SHOUTcast metadata) -// No enum or constant needed as it's the only supported format - -// ============================================================================ -// API Configuration -// ============================================================================ - -/// Radio Paradise API base URL -/// -/// Base URL for all Radio Paradise API requests. -/// This is hardcoded as Radio Paradise's API endpoint doesn't change. -/// -/// Value: https://api.radioparadise.com -pub const API_BASE_URL: &str = "https://api.radioparadise.com"; - -/// API request timeout (seconds) -/// -/// Maximum time to wait for an API response before considering it failed. -/// -/// Value: 30 seconds -pub const API_TIMEOUT_SECONDS: u64 = 30; - -/// User agent for API requests -/// -/// Identifies PMOMusic in HTTP requests to Radio Paradise's servers. -/// -/// Value: PMO-RadioParadise/1.0 -pub const API_USER_AGENT: &str = "PMO-RadioParadise/1.0"; - -/// Helper to get API timeout as Duration -pub fn api_timeout() -> Duration { - Duration::from_secs(API_TIMEOUT_SECONDS) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_duration_helpers() { - assert_eq!(polling_high_interval(), Duration::from_secs(120)); - assert_eq!(polling_medium_interval(), Duration::from_secs(60)); - assert_eq!(polling_low_interval(), Duration::from_secs(20)); - assert_eq!(api_timeout(), Duration::from_secs(30)); - } - - #[test] - fn test_constants_sanity() { - // Polling intervals should be ordered - assert!(POLLING_INTERVAL_LOW_BUFFER < POLLING_INTERVAL_MEDIUM_BUFFER); - assert!(POLLING_INTERVAL_MEDIUM_BUFFER < POLLING_INTERVAL_HIGH_BUFFER); - - // Backoff should be reasonable - assert!(BACKOFF_INITIAL_SECONDS < BACKOFF_MAX_SECONDS); - assert!(BACKOFF_MULTIPLIER > 1.0); - - // Cache limits should be positive - assert!(MAX_BLOCKS_REMEMBERED > 0); - assert!(TRACK_ID_HASH_BYTES > 0); - - // History should be reasonable - assert!(HISTORY_DEFAULT_MAX_TRACKS > 0); - } -} diff --git a/pmoparadise/src/paradise/history.rs b/pmoparadise/src/paradise/history.rs deleted file mode 100644 index c276a810..00000000 --- a/pmoparadise/src/paradise/history.rs +++ /dev/null @@ -1,217 +0,0 @@ -//! History persistence for Radio Paradise playback. -//! -//! The worker pushes every completed track into the history backend while -//! keeping the latest entries available for UPnP browsing. We use SQLite -//! for persistent storage with an abstract trait for testability. -use crate::models::Song; -use async_trait::async_trait; -use chrono::{DateTime, Utc}; -use serde::{Deserialize, Serialize}; -use std::path::Path; -use std::sync::{Arc, Mutex as StdMutex}; -use tokio::task::spawn_blocking; - -/// Serializable record describing a played track. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct HistoryEntry { - pub track_id: String, - pub channel_id: u8, - pub started_at: chrono::DateTime, - pub duration_ms: u64, - pub song: SongSnapshot, -} - -/// Minimal snapshot of a Radio Paradise song at playback time. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct SongSnapshot { - pub title: String, - pub artist: String, - pub album: Option, - pub cover_url: Option, -} - -impl SongSnapshot { - pub fn title(&self) -> &str { - &self.title - } -} - -impl From<&Song> for SongSnapshot { - fn from(song: &Song) -> Self { - Self { - title: song.title.clone(), - artist: song.artist.clone(), - album: song.album.clone(), - cover_url: song.cover.clone(), - } - } -} - -/// Abstract persistence interface. -#[async_trait] -pub trait HistoryBackend: Send + Sync { - async fn append(&self, entry: HistoryEntry) -> anyhow::Result<()>; - async fn recent(&self, limit: usize) -> anyhow::Result>; - async fn len(&self) -> anyhow::Result; - async fn truncate(&self, keep: usize) -> anyhow::Result<()>; -} - -pub struct SqliteHistoryBackend { - conn: Arc>, -} - -impl SqliteHistoryBackend { - pub fn new(path: impl AsRef) -> anyhow::Result { - let path = path.as_ref(); - if let Some(parent) = path.parent() { - std::fs::create_dir_all(parent)?; - } - - let conn = rusqlite::Connection::open(path)?; - conn.pragma_update(None, "journal_mode", &"WAL")?; - conn.execute_batch( - "CREATE TABLE IF NOT EXISTS paradise_history ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - track_id TEXT NOT NULL, - channel_id INTEGER NOT NULL, - started_at_ms INTEGER NOT NULL, - duration_ms INTEGER NOT NULL, - title TEXT, - artist TEXT, - album TEXT, - cover_url TEXT - ); - CREATE INDEX IF NOT EXISTS idx_history_started_at ON paradise_history(started_at_ms);", - )?; - - Ok(Self { - conn: Arc::new(StdMutex::new(conn)), - }) - } - - fn conn(&self) -> Arc> { - self.conn.clone() - } -} - -#[async_trait] -impl HistoryBackend for SqliteHistoryBackend { - async fn append(&self, entry: HistoryEntry) -> anyhow::Result<()> { - let conn = self.conn(); - spawn_blocking(move || -> anyhow::Result<()> { - let conn = conn.lock().unwrap(); - conn.execute( - "INSERT INTO paradise_history (track_id, channel_id, started_at_ms, duration_ms, title, artist, album, cover_url) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", - rusqlite::params![ - entry.track_id, - entry.channel_id as i64, - entry.started_at.timestamp_millis(), - entry.duration_ms as i64, - entry.song.title, - entry.song.artist, - entry.song.album, - entry.song.cover_url, - ], - )?; - Ok(()) - }) - .await??; - Ok(()) - } - - async fn recent(&self, limit: usize) -> anyhow::Result> { - let conn = self.conn(); - let limit = limit as i64; - spawn_blocking(move || -> anyhow::Result> { - let conn = conn.lock().unwrap(); - let mut stmt = conn.prepare( - "SELECT track_id, channel_id, started_at_ms, duration_ms, title, artist, album, cover_url - FROM paradise_history - ORDER BY started_at_ms DESC - LIMIT ?1", - )?; - - let mut rows = stmt.query([limit])?; - let mut entries = Vec::new(); - while let Some(row) = rows.next()? { - let started_at_ms: i64 = row.get(2)?; - let started_at = DateTime::::from_timestamp_millis(started_at_ms) - .ok_or_else(|| anyhow::anyhow!("Invalid timestamp in history"))?; - let entry = HistoryEntry { - track_id: row.get(0)?, - channel_id: row.get::<_, i64>(1)? as u8, - started_at, - duration_ms: row.get::<_, i64>(3)? as u64, - song: SongSnapshot { - title: row.get::<_, Option>(4)?.unwrap_or_default(), - artist: row.get::<_, Option>(5)?.unwrap_or_default(), - album: row.get(6)?, - cover_url: row.get(7)?, - }, - }; - entries.push(entry); - } - Ok(entries) - }) - .await? - } - - async fn len(&self) -> anyhow::Result { - let conn = self.conn(); - let count = spawn_blocking(move || -> anyhow::Result { - let conn = conn.lock().unwrap(); - let mut stmt = conn.prepare("SELECT COUNT(*) FROM paradise_history")?; - let count: i64 = stmt.query_row([], |row| row.get(0))?; - Ok(count as usize) - }) - .await??; - Ok(count) - } - - async fn truncate(&self, keep: usize) -> anyhow::Result<()> { - let conn = self.conn(); - spawn_blocking(move || -> anyhow::Result<()> { - let conn = conn.lock().unwrap(); - let count: i64 = - conn.query_row("SELECT COUNT(*) FROM paradise_history", [], |row| { - row.get(0) - })?; - let keep = keep as i64; - if count <= keep { - return Ok(()); - } - let to_remove = count - keep; - conn.execute( - "DELETE FROM paradise_history - WHERE id IN ( - SELECT id FROM paradise_history - ORDER BY started_at_ms ASC - LIMIT ?1 - )", - rusqlite::params![to_remove], - )?; - Ok(()) - }) - .await??; - Ok(()) - } -} - -/// Creates a SQLite history backend with the given database path. -/// -/// The database file and parent directories will be created if they don't exist. -/// -/// # Arguments -/// -/// * `database_path` - Path to the SQLite database file -/// -/// # Example -/// -/// ```rust,ignore -/// let backend = create_history_backend("/var/lib/pmo/history.db")?; -/// ``` -pub fn create_history_backend(database_path: &str) -> anyhow::Result> { - let backend = SqliteHistoryBackend::new(database_path)?; - Ok(Arc::new(backend)) -} diff --git a/pmoparadise/src/paradise/mod.rs b/pmoparadise/src/paradise/mod.rs deleted file mode 100644 index f2860671..00000000 --- a/pmoparadise/src/paradise/mod.rs +++ /dev/null @@ -1,28 +0,0 @@ -//! Internal orchestration layer for dynamic Radio Paradise streaming. -//! -//! This module implements the high level structures described in the -//! Radio Paradise functional specification: -//! - `ParadiseChannel`: lifecycle and state machine for a single RP channel. -//! - `ParadiseWorker`: async task responsible for polling/downloading blocks. -//! - `ParadiseClientStream`: per-client audio stream with independent cursor. -//! - Shared caches and history storage hooked into existing PMO components. -//! -//! The implementation is split across several submodules to keep concerns -//! isolated (constants, playlist management, history persistence, etc.). -//! The goal of this scaffolding is to provide a clear, testable surface for -//! the eventual end-to-end integration with the UPnP server and HTTP routes. - -mod channel; -pub mod constants; -mod history; -mod playlist; -mod worker; - -pub use channel::{ - max_channel_id, ChannelDescriptor, ParadiseChannel, ParadiseChannelKind, ParadiseClientStream, - ALL_CHANNELS, -}; -pub use constants::*; // Export all constants -pub use history::{create_history_backend, HistoryBackend, HistoryEntry}; -pub use playlist::PlaylistEntry; -pub use worker::{load_rp_metadata, ParadiseWorker, RadioParadiseMetadata, WorkerCommand}; diff --git a/pmoparadise/src/paradise/playlist.rs b/pmoparadise/src/paradise/playlist.rs deleted file mode 100644 index 13a4fc89..00000000 --- a/pmoparadise/src/paradise/playlist.rs +++ /dev/null @@ -1,293 +0,0 @@ -//! Shared playlist structures for Radio Paradise channels. -//! -//! This module keeps track of the active queue and history for a Radio -//! Paradise channel. Each playlist entry knows how many clients still need -//! to consume it before the worker can evict it. - -use super::history::{HistoryEntry, SongSnapshot}; -use crate::models::Song; -use chrono::{DateTime, Utc}; -use std::collections::VecDeque; -use std::path::PathBuf; -use std::sync::atomic::{AtomicU32, AtomicUsize, Ordering}; -use std::sync::Arc; -use std::time::SystemTime; -use tokio::sync::{Notify, RwLock}; - -/// Metadata stored for an active track. -#[derive(Debug)] -pub struct PlaylistEntry { - pub track_id: String, - pub channel_id: u8, - pub song: Arc, - pub started_at: DateTime, - pub duration_ms: u64, - pub audio_pk: Option, - pub file_path: Option, - pending_clients: AtomicUsize, -} - -impl PlaylistEntry { - #[allow(clippy::too_many_arguments)] - pub fn new( - track_id: String, - channel_id: u8, - song: Arc, - started_at: DateTime, - duration_ms: u64, - audio_pk: Option, - file_path: Option, - pending_clients: usize, - ) -> Self { - Self { - track_id, - channel_id, - song, - started_at, - duration_ms, - audio_pk, - file_path, - pending_clients: AtomicUsize::new(pending_clients), - } - } - - pub fn as_history_entry(&self) -> HistoryEntry { - HistoryEntry { - track_id: self.track_id.clone(), - channel_id: self.channel_id, - started_at: self.started_at, - duration_ms: self.duration_ms, - song: SongSnapshot::from(self.song.as_ref()), - } - } - - pub fn pending_clients(&self) -> usize { - self.pending_clients.load(Ordering::SeqCst) - } - - pub fn set_pending_clients(&self, value: usize) { - self.pending_clients.store(value, Ordering::SeqCst); - } - - pub fn increment_clients(&self) -> usize { - self.pending_clients.fetch_add(1, Ordering::SeqCst) + 1 - } - - pub fn decrement_clients(&self) -> usize { - let mut current = self.pending_clients.load(Ordering::SeqCst); - loop { - if current == 0 { - return 0; - } - match self.pending_clients.compare_exchange( - current, - current - 1, - Ordering::SeqCst, - Ordering::SeqCst, - ) { - Ok(_) => return current - 1, - Err(actual) => current = actual, - } - } - } -} - -#[derive(Default)] -struct PlaylistState { - active: VecDeque>, - history: VecDeque, - max_history: usize, -} - -impl PlaylistState { - fn new(max_history: usize) -> Self { - Self { - active: VecDeque::new(), - history: VecDeque::new(), - max_history, - } - } - - fn active_len(&self) -> usize { - self.active.len() - } - - fn push_active(&mut self, entry: Arc) { - self.active.push_back(entry); - } - - fn active_snapshot(&self) -> Vec> { - self.active.iter().cloned().collect() - } - - fn pop_front_if_ready(&mut self) -> Option> { - if let Some(front) = self.active.front() { - if front.pending_clients() == 0 { - return self.active.pop_front(); - } - } - None - } - - fn pop_front_matching(&mut self, track_id: &str) -> Option> { - if let Some(front) = self.active.front() { - if front.track_id == track_id && front.pending_clients() == 0 { - return self.active.pop_front(); - } - } - None - } - - fn push_history(&mut self, entry: HistoryEntry) { - self.history.push_back(entry); - self.trim_history(); - } - - fn recent_history(&self, limit: usize) -> Vec { - let total = self.history.len(); - let start = total.saturating_sub(limit); - self.history.iter().skip(start).cloned().collect() - } - - fn trim_history(&mut self) { - while self.history.len() > self.max_history { - self.history.pop_front(); - } - } - - fn clear(&mut self) -> bool { - let changed = !self.active.is_empty() || !self.history.is_empty(); - if changed { - self.active.clear(); - self.history.clear(); - } - changed - } - - fn increment_all(&self) { - for entry in &self.active { - entry.increment_clients(); - } - } -} - -struct SharedPlaylistInner { - state: RwLock, - notify: Notify, - update_id: AtomicU32, - last_change: RwLock>, -} - -#[derive(Clone)] -pub struct SharedPlaylist(Arc); - -impl SharedPlaylist { - pub fn new(max_history: usize) -> Self { - Self(Arc::new(SharedPlaylistInner { - state: RwLock::new(PlaylistState::new(max_history)), - notify: Notify::new(), - update_id: AtomicU32::new(0), - last_change: RwLock::new(None), - })) - } - - async fn touch(&self) { - self.0.update_id.fetch_add(1, Ordering::SeqCst); - let mut last_change = self.0.last_change.write().await; - *last_change = Some(SystemTime::now()); - } - - pub async fn push_active(&self, entry: Arc) { - let mut guard = self.0.state.write().await; - guard.push_active(entry); - drop(guard); - self.touch().await; - self.0.notify.notify_waiters(); - } - - pub async fn active_len(&self) -> usize { - let guard = self.0.state.read().await; - guard.active_len() - } - - pub async fn active_snapshot(&self) -> Vec> { - let guard = self.0.state.read().await; - guard.active_snapshot() - } - - pub async fn clear(&self) { - let mut guard = self.0.state.write().await; - let changed = guard.clear(); - drop(guard); - if changed { - self.touch().await; - self.0.notify.notify_waiters(); - } - } - - pub async fn wait_for_track_count(&self, current_len: usize) { - loop { - let len = { - let guard = self.0.state.read().await; - guard.active_len() - }; - - if len > current_len { - break; - } - - self.0.notify.notified().await; - } - } - - pub async fn pop_front_if_ready(&self) -> Option> { - let mut guard = self.0.state.write().await; - let result = guard.pop_front_if_ready(); - drop(guard); - - if result.is_some() { - self.touch().await; - self.0.notify.notify_waiters(); - } - - result - } - - pub async fn pop_front_matching(&self, track_id: &str) -> Option> { - let mut guard = self.0.state.write().await; - let result = guard.pop_front_matching(track_id); - drop(guard); - - if result.is_some() { - self.touch().await; - self.0.notify.notify_waiters(); - } - - result - } - - pub async fn push_history_entry(&self, entry: HistoryEntry) { - let mut guard = self.0.state.write().await; - guard.push_history(entry); - drop(guard); - self.touch().await; - } - - pub async fn recent_history(&self, limit: usize) -> Vec { - let guard = self.0.state.read().await; - guard.recent_history(limit) - } - - pub async fn increment_all_pending(&self) { - let guard = self.0.state.read().await; - guard.increment_all(); - } - - pub fn update_id(&self) -> u32 { - self.0.update_id.load(Ordering::SeqCst) - } - - pub async fn last_change(&self) -> Option { - self.0.last_change.read().await.clone() - } -} diff --git a/pmoparadise/src/paradise/worker.rs b/pmoparadise/src/paradise/worker.rs deleted file mode 100644 index c502436e..00000000 --- a/pmoparadise/src/paradise/worker.rs +++ /dev/null @@ -1,1326 +0,0 @@ -//! Background worker for Radio Paradise channels. -//! -//! The worker handles API polling, block ingestion, caching and playlist -//! maintenance. It keeps the channel state in sync with connected clients -//! and ensures fresh content is available according to the specification. - -use super::channel::ChannelDescriptor; -use super::constants::*; -use super::history::HistoryBackend; -use super::playlist::{PlaylistEntry, SharedPlaylist}; -use crate::client::RadioParadiseClient; -use crate::models::{Block, Song}; -use anyhow::{anyhow, Context, Result}; -use bytes::Bytes; -use chrono::Utc; -use futures::stream; -use pmosource::{SourceCacheManager, TrackMetadata}; -use std::collections::{HashSet, VecDeque}; -use std::pin::Pin; -use std::sync::Arc; -use tokio::sync::mpsc; -use tokio::task::JoinHandle; -use tokio::time::{sleep, Duration}; -use tokio_util::io::StreamReader; -use tracing::{debug, error, info, warn}; -use url::Url; - -/// Commands sent to the background worker. -#[derive(Debug)] -pub enum WorkerCommand { - EnsureReady, - ClientConnected { client_id: String }, - ClientDisconnected { client_id: String }, - RefreshBlock, - Shutdown, -} - -/// Handle to the spawned worker task. -pub struct ParadiseWorker { - descriptor: ChannelDescriptor, - join_handle: JoinHandle<()>, -} - -impl ParadiseWorker { - #[allow(clippy::too_many_arguments)] - pub fn spawn( - descriptor: ChannelDescriptor, - client: RadioParadiseClient, - history_max_tracks: usize, - playlist: SharedPlaylist, - history: Arc, - cache_manager: Arc, - ) -> (Self, mpsc::Sender) { - let (tx, mut rx) = mpsc::channel(32); - - let join_handle = tokio::spawn(async move { - info!(channel = descriptor.slug, "Starting Radio Paradise worker"); - - let mut state = WorkerState::new( - descriptor, - client, - history_max_tracks, - playlist, - history, - cache_manager, - ); - - loop { - if let Some(task) = state.scheduled_task.as_mut() { - let kind = task.kind; - let mut pending_command: Option> = None; - - tokio::select! { - cmd = rx.recv() => { - pending_command = Some(cmd); - } - _ = &mut task.sleep => { - state.scheduled_task = None; - if let Err(err) = state.handle_scheduled_task(kind).await { - error!(channel = state.descriptor.slug, "Worker scheduled task error: {err:?}"); - state.on_error(err); - } - } - } - - if let Some(Some(cmd)) = pending_command { - if let Err(err) = state.handle_command(cmd).await { - error!( - channel = state.descriptor.slug, - "Worker command error: {err:?}" - ); - state.on_error(err); - } - if state.shutdown { - break; - } - } else if let Some(None) = pending_command { - // Command channel closed, terminate - break; - } - } else { - match rx.recv().await { - Some(cmd) => { - if let Err(err) = state.handle_command(cmd).await { - error!( - channel = state.descriptor.slug, - "Worker command error: {err:?}" - ); - state.on_error(err); - } - if state.shutdown { - break; - } - } - None => break, - } - } - } - - info!(channel = state.descriptor.slug, "Worker stopped"); - }); - - ( - Self { - descriptor, - join_handle, - }, - tx, - ) - } - - pub async fn wait(self) -> Result<()> { - if let Err(err) = self.join_handle.await { - if err.is_cancelled() { - warn!( - channel = self.descriptor.slug, - "Worker task cancelled: {err}" - ); - return Ok(()); - } - return Err(anyhow!("Worker join error: {}", err)); - } - Ok(()) - } -} - -struct WorkerState { - descriptor: ChannelDescriptor, - client: RadioParadiseClient, - playlist: SharedPlaylist, - history: Arc, - cache_manager: Arc, - active_clients: usize, - status: ChannelLifecycle, - processed_blocks: HashSet, - processing_blocks: HashSet, - recent_blocks: VecDeque, - next_block_hint: Option, - scheduled_task: Option, - backoff: BackoffState, - shutdown: bool, -} - -#[derive(Clone)] -struct SongTaskContext { - cache_manager: Arc, - playlist: SharedPlaylist, - descriptor_id: u8, - slug: &'static str, -} - -impl WorkerState { - fn song_task_context(&self) -> SongTaskContext { - SongTaskContext { - cache_manager: Arc::clone(&self.cache_manager), - playlist: self.playlist.clone(), - descriptor_id: self.descriptor.id, - slug: self.descriptor.slug, - } - } - - fn new( - descriptor: ChannelDescriptor, - client: RadioParadiseClient, - _history_max_tracks: usize, - playlist: SharedPlaylist, - history: Arc, - cache_manager: Arc, - ) -> Self { - Self { - descriptor, - client, - playlist, - history, - cache_manager, - active_clients: 0, - status: ChannelLifecycle::Idle, - processed_blocks: HashSet::new(), - processing_blocks: HashSet::new(), - recent_blocks: VecDeque::new(), - next_block_hint: None, - scheduled_task: None, - backoff: BackoffState::new(), - shutdown: false, - } - } - - async fn handle_command(&mut self, cmd: WorkerCommand) -> Result<()> { - debug!(channel = self.descriptor.slug, ?cmd, "Worker command"); - - match cmd { - WorkerCommand::EnsureReady => { - self.ensure_ready().await?; - } - WorkerCommand::ClientConnected { .. } => { - self.active_clients = self.active_clients.saturating_add(1); - self.enter_active(); - self.ensure_ready().await?; - } - WorkerCommand::ClientDisconnected { .. } => { - self.active_clients = self.active_clients.saturating_sub(1); - if self.active_clients == 0 { - self.enter_cooling(); - } - } - WorkerCommand::RefreshBlock => { - self.fetch_next_block().await?; - } - WorkerCommand::Shutdown => { - self.shutdown = true; - self.cancel_scheduled_task(); - } - } - - if !self.shutdown { - self.maybe_schedule_poll().await; - } - - Ok(()) - } - - async fn handle_scheduled_task(&mut self, kind: ScheduledTaskKind) -> Result<()> { - match kind { - ScheduledTaskKind::Poll => { - self.fetch_next_block().await?; - self.maybe_schedule_poll().await; - } - ScheduledTaskKind::Cooling => { - debug!( - channel = self.descriptor.slug, - "Cooling timeout reached -> idle" - ); - self.status = ChannelLifecycle::Idle; - self.next_block_hint = None; - self.playlist.clear().await; - self.processed_blocks.clear(); - self.recent_blocks.clear(); - } - } - Ok(()) - } - - fn on_error(&mut self, err: anyhow::Error) { - warn!(channel = self.descriptor.slug, "Worker error: {err:?}"); - let delay = self.backoff.next_delay(); - self.schedule_task(ScheduledTaskKind::Poll, delay); - } - - fn enter_active(&mut self) { - if !matches!(self.status, ChannelLifecycle::Active) { - debug!( - channel = self.descriptor.slug, - "Channel entering Active state" - ); - } - self.status = ChannelLifecycle::Active; - if matches!(self.scheduled_task_kind(), Some(ScheduledTaskKind::Cooling)) { - self.cancel_scheduled_task(); - } - self.backoff.reset(); - } - - fn enter_cooling(&mut self) { - if matches!(self.status, ChannelLifecycle::Idle) { - return; - } - debug!( - channel = self.descriptor.slug, - "Channel entering Cooling state" - ); - self.status = ChannelLifecycle::Cooling; - let duration = Duration::from_secs(COOLING_TIMEOUT_SECONDS.max(1)); - self.schedule_task(ScheduledTaskKind::Cooling, duration); - } - - async fn ensure_ready(&mut self) -> Result<()> { - if !matches!(self.status, ChannelLifecycle::Active) { - self.enter_active(); - } - - let has_tracks = self.playlist.active_len().await > 0; - - if !has_tracks { - debug!( - channel = self.descriptor.slug, - "Playlist empty – fetching now playing" - ); - let now_playing = self.client.now_playing().await?; - self.process_block(now_playing.block).await?; - } - - Ok(()) - } - - async fn fetch_next_block(&mut self) -> Result<()> { - if !matches!(self.status, ChannelLifecycle::Active) { - debug!( - channel = self.descriptor.slug, - "Skipping poll while not active" - ); - return Ok(()); - } - - let event_id = self.next_block_hint; - let block = self.client.get_block(event_id).await?; - self.process_block(block).await?; - Ok(()) - } - - async fn process_block(&mut self, block: Block) -> Result<()> { - // Check if we just processed this block (songs are already in playlist) - if self.is_recent_block(block.event) { - debug!( - channel = self.descriptor.slug, - event = block.event, - "Skipping already processed block (songs already in playlist)" - ); - self.next_block_hint = Some(block.end_event); - return Ok(()); - } - - // Check if this block is currently being processed by another task - // This prevents race conditions when the same block is requested multiple times - if self.processing_blocks.contains(&block.event) { - warn!( - channel = self.descriptor.slug, - event = block.event, - "Block is already being processed, skipping duplicate request" - ); - return Ok(()); - } - - // Check if all songs from this block are in cache - // If yes, restore from cache instead of downloading - if self.check_all_songs_cached(&block).await { - info!( - channel = self.descriptor.slug, - event = block.event, - "Block found in cache, restoring without download" - ); - self.restore_from_cache(&block).await?; - self.record_processed_block(block.event); - self.next_block_hint = Some(block.end_event); - self.backoff.reset(); - return Ok(()); - } - - // Mark block as being processed - self.processing_blocks.insert(block.event); - let event = block.event; // Save for cleanup - - // Process the block and ensure cleanup even on error - let result = self.process_block_inner(block).await; - - // Always remove from processing set, whether success or error - self.processing_blocks.remove(&event); - - result - } - - async fn process_block_inner(&mut self, block: Block) -> Result<()> { - info!( - channel = self.descriptor.slug, - event = block.event, - "Processing Radio Paradise block with progressive streaming" - ); - - let _ = &self.history; - - // Start streaming the block - let block_url = Url::parse(&block.url)?; - let http_stream = self - .client - .stream_block(&block_url) - .await - .context("Failed to start block stream")?; - - let ordered_songs = block.songs_ordered(); - - // Decode in streaming mode using spawn_blocking - let (tx, mut rx) = mpsc::channel::(16); - - let decode_handle = tokio::task::spawn_blocking(move || -> Result<()> { - use crate::streaming::StreamingPCMDecoder; - - let mut decoder = StreamingPCMDecoder::new(http_stream) - .context("Failed to create streaming decoder")?; - - info!( - "Streaming decoder initialized: {}Hz, {} channels, {} bits", - decoder.sample_rate(), - decoder.channels(), - decoder.bits_per_sample() - ); - - // Decode chunks and send them - while let Some(chunk) = decoder.decode_chunk()? { - if tx.blocking_send(chunk).is_err() { - // Receiver dropped, stop decoding - break; - } - } - - Ok(()) - }); - - // Process songs as chunks arrive - let mut accumulated_pcm = Vec::new(); - let mut current_song_idx = 0; - let mut sample_rate = 0u32; - let mut channels = 0u32; - let mut bits_per_sample = 0u32; - - while let Some(chunk) = rx.recv().await { - // Store metadata from first chunk - if sample_rate == 0 { - sample_rate = chunk.sample_rate; - channels = chunk.channels; - bits_per_sample = 16; // Normalized to 16-bit by decoder - } - - accumulated_pcm.extend_from_slice(&chunk.samples); - let current_position_ms = chunk.position_ms; - - // Check if we've completed any songs - while current_song_idx < ordered_songs.len() { - let (song_index, song) = ordered_songs[current_song_idx]; - - // Calculate song boundaries - let song_start_ms = song.elapsed; - let song_end_ms = if current_song_idx + 1 < ordered_songs.len() { - ordered_songs[current_song_idx + 1].1.elapsed - } else { - u64::MAX // Last song goes to end of block - }; - - // Check if we have enough PCM for this song - if current_position_ms >= song_end_ms { - // Extract song samples - let start_frame = crate::streaming::ms_to_frames(song_start_ms, sample_rate); - let end_frame = crate::streaming::ms_to_frames(song_end_ms, sample_rate); - - let start_sample = start_frame * channels as usize; - let end_sample = end_frame * channels as usize; - - if end_sample <= accumulated_pcm.len() { - let track_samples = accumulated_pcm[start_sample..end_sample].to_vec(); - - info!( - channel = self.descriptor.slug, - song_index = song_index, - position_ms = current_position_ms, - "✅ Song '{}' ready for encoding ({} samples)", - song.title, - track_samples.len() - ); - - let context = self.song_task_context(); - spawn_song_processing( - context, - block.clone(), - song_index, - song.clone(), - track_samples, - sample_rate, - channels as usize, - bits_per_sample, - self.active_clients, - song.duration, - current_position_ms, - ); - - current_song_idx += 1; - } else { - // Not enough samples yet, wait for more chunks - break; - } - } else { - // Haven't reached this song's end yet - break; - } - } - } - - // Wait for decoder to finish - decode_handle.await??; - - // Process any remaining songs (last song in block) - if current_song_idx < ordered_songs.len() { - let (song_index, song) = ordered_songs[current_song_idx]; - let song_start_ms = song.elapsed; - let start_frame = crate::streaming::ms_to_frames(song_start_ms, sample_rate); - let start_sample = start_frame * channels as usize; - - if start_sample < accumulated_pcm.len() { - let track_samples = accumulated_pcm[start_sample..].to_vec(); - - info!( - channel = self.descriptor.slug, - song_index = song_index, - "Processing last song '{}' ({} samples)", - song.title, - track_samples.len() - ); - - let context = self.song_task_context(); - spawn_song_processing( - context, - block.clone(), - song_index, - song.clone(), - track_samples, - sample_rate, - channels as usize, - bits_per_sample, - self.active_clients, - song.duration, - song_start_ms, - ); - } - } - - self.record_processed_block(block.event); - self.next_block_hint = Some(block.end_event); - self.backoff.reset(); - - Ok(()) - } - - async fn process_song( - &self, - block: &Block, - song_index: &usize, - song: &Song, - position: usize, - ordered_songs: &[(usize, &Song)], - total_frames: usize, - decoded: &DecodedBlock, - ) -> Result> { - let duration_ms = song_duration_ms(block, ordered_songs, position); - let start_frame = ms_to_frames(song.elapsed, decoded.sample_rate); - let end_frame = if position + 1 < ordered_songs.len() { - ms_to_frames(ordered_songs[position + 1].1.elapsed, decoded.sample_rate) - } else { - total_frames - }; - - if end_frame <= start_frame || end_frame > total_frames { - warn!( - channel = self.descriptor.slug, - song_index = song_index, - "Invalid frame range for song, skipping" - ); - return Err(anyhow!("Invalid frame range")); - } - - let channels = decoded.channels; - let start = start_frame * channels; - let end = end_frame * channels; - let slice = decoded - .samples - .get(start..end) - .ok_or_else(|| anyhow!("Sample slice out of bounds"))?; - - let track_samples = slice.to_vec(); - encode_song_to_cache( - Arc::clone(&self.cache_manager), - self.descriptor.id, - self.descriptor.slug, - block.clone(), - *song_index, - song.clone(), - track_samples, - decoded.sample_rate, - decoded.channels, - decoded.bits_per_sample, - self.active_clients, - duration_ms, - ) - .await - } - - /// Stocke les métadonnées Radio Paradise pour un fichier audio caché - /// - /// Cette fonction persiste toutes les métadonnées RP dans la base de données - /// du cache audio, permettant leur récupération future sans dépendance aux - /// données en mémoire. - fn compute_track_id(&self, block: &Block, song_index: usize) -> String { - compute_track_id_for_descriptor(self.descriptor.id, block, song_index) - } - - async fn maybe_schedule_poll(&mut self) { - if !matches!(self.status, ChannelLifecycle::Active) { - return; - } - - let buffer_len = self.playlist.active_len().await; - - let interval = if buffer_len > 3 { - polling_high_interval() - } else if buffer_len >= 2 { - polling_medium_interval() - } else { - polling_low_interval() - }; - - self.schedule_task(ScheduledTaskKind::Poll, interval); - } - - fn schedule_task(&mut self, kind: ScheduledTaskKind, duration: Duration) { - self.scheduled_task = Some(ScheduledTask { - kind, - sleep: Box::pin(sleep(duration)), - }); - } - - fn cancel_scheduled_task(&mut self) { - self.scheduled_task = None; - } - - fn scheduled_task_kind(&self) -> Option { - self.scheduled_task.as_ref().map(|task| task.kind) - } - - fn record_processed_block(&mut self, event: u64) { - self.processed_blocks.insert(event); - self.recent_blocks.push_back(event); - let max = MAX_BLOCKS_REMEMBERED.max(1); - while self.recent_blocks.len() > max { - if let Some(ev) = self.recent_blocks.pop_front() { - self.processed_blocks.remove(&ev); - } - } - } - - fn is_recent_block(&self, event: u64) -> bool { - self.processed_blocks.contains(&event) - } - - /// Check if all songs from a block are already cached - async fn check_all_songs_cached(&self, block: &Block) -> bool { - let ordered_songs = block.songs_ordered(); - - for (song_index, _song) in &ordered_songs { - let track_id = self.compute_track_id(block, *song_index); - - // Check if metadata exists - let metadata = match self.cache_manager.get_metadata(&track_id).await { - Some(m) => m, - None => { - debug!( - channel = self.descriptor.slug, - event = block.event, - song_index = *song_index, - "Song not in cache: no metadata" - ); - return false; - } - }; - - // Check if audio is cached - let audio_pk = match metadata.cached_audio_pk { - Some(pk) => pk, - None => { - debug!( - channel = self.descriptor.slug, - event = block.event, - song_index = *song_index, - "Song not in cache: no audio_pk" - ); - return false; - } - }; - - // Check if file exists - if self - .cache_manager - .audio_file_path(&audio_pk) - .await - .is_none() - { - debug!( - channel = self.descriptor.slug, - event = block.event, - song_index = *song_index, - "Song not in cache: file not found" - ); - return false; - } - } - - debug!( - channel = self.descriptor.slug, - event = block.event, - "All {} songs are cached", - ordered_songs.len() - ); - true - } - - /// Restore songs from cache and add them to the playlist - async fn restore_from_cache(&mut self, block: &Block) -> Result<()> { - info!( - channel = self.descriptor.slug, - event = block.event, - "Restoring block from cache (no download needed)" - ); - - let ordered_songs = block.songs_ordered(); - - for (song_index, song) in &ordered_songs { - let track_id = self.compute_track_id(block, *song_index); - - // Get metadata (we already checked it exists in check_all_songs_cached) - let metadata = self - .cache_manager - .get_metadata(&track_id) - .await - .ok_or_else(|| anyhow!("Metadata disappeared for track_id: {}", track_id))?; - - let audio_pk = metadata - .cached_audio_pk - .clone() - .ok_or_else(|| anyhow!("Audio PK disappeared for track_id: {}", track_id))?; - - // Get cover PK if available - let cover_pk = if let Some(ref cover_path) = song.cover { - if let Some(cover_url) = block.cover_url(cover_path) { - match self.cache_manager.cache_cover(&cover_url).await { - Ok(pk) => Some(pk), - Err(err) => { - warn!(channel = self.descriptor.slug, "Cover cache error: {err}"); - None - } - } - } else { - None - } - } else { - None - }; - - // Update metadata with cover if we just cached it - if cover_pk.is_some() && metadata.cached_cover_pk.is_none() { - let updated_metadata = TrackMetadata { - cached_cover_pk: cover_pk, - ..metadata.clone() - }; - self.cache_manager - .update_metadata(track_id.clone(), updated_metadata) - .await; - } - - let file_path = self - .cache_manager - .audio_file_path(&audio_pk) - .await - .ok_or_else(|| anyhow!("File disappeared for audio_pk: {}", audio_pk))?; - - let duration_ms = song.duration; - - let entry = Arc::new(PlaylistEntry::new( - track_id, - self.descriptor.id, - Arc::new((*song).clone()), - Utc::now(), - duration_ms, - Some(audio_pk), - Some(file_path), - self.active_clients, - )); - - self.playlist.push_active(entry).await; - - info!( - channel = self.descriptor.slug, - song_index = *song_index, - "🎵 Restored '{}' from cache", - song.title - ); - } - - info!( - channel = self.descriptor.slug, - event = block.event, - "Block restored from cache: {} songs", - ordered_songs.len() - ); - - Ok(()) - } -} - -struct ScheduledTask { - kind: ScheduledTaskKind, - sleep: Pin>, -} - -#[derive(Clone, Copy)] -enum ScheduledTaskKind { - Poll, - Cooling, -} - -#[derive(Clone, Copy, Debug)] -enum ChannelLifecycle { - Idle, - Cooling, - Active, -} - -struct BackoffState { - current: Option, -} - -impl BackoffState { - fn new() -> Self { - Self { current: None } - } - - fn reset(&mut self) { - self.current = None; - } - - fn next_delay(&mut self) -> Duration { - let next = match self.current { - Some(current) => { - let multiplied = (current.as_secs_f32() * BACKOFF_MULTIPLIER).round() as u64; - Duration::from_secs(multiplied.min(BACKOFF_MAX_SECONDS)) - } - None => Duration::from_secs(BACKOFF_INITIAL_SECONDS), - }; - self.current = Some(next); - next - } -} - -struct DecodedBlock { - samples: Vec, - channels: usize, - sample_rate: u32, - bits_per_sample: u32, -} - -fn song_duration_ms(block: &Block, ordered: &[(usize, &Song)], position: usize) -> u64 { - let song = ordered[position].1; - if song.duration > 0 { - return song.duration; - } - - if let Some((_, next_song)) = ordered.get(position + 1) { - return next_song.elapsed.saturating_sub(song.elapsed); - } - - block.length.saturating_sub(song.elapsed) -} - -fn ms_to_frames(ms: u64, sample_rate: u32) -> usize { - ((ms as u128 * sample_rate as u128) / 1000) as usize -} - -fn decode_block_audio(data: Vec) -> anyhow::Result { - use symphonia::core::audio::SampleBuffer; - use symphonia::core::codecs::{DecoderOptions, CODEC_TYPE_NULL}; - use symphonia::core::errors::Error as SymphoniaError; - use symphonia::core::formats::FormatOptions; - use symphonia::core::io::MediaSourceStream; - use symphonia::core::meta::MetadataOptions; - use symphonia::core::probe::Hint; - - let cursor = std::io::Cursor::new(data); - let mss = MediaSourceStream::new(Box::new(cursor), Default::default()); - - let hint = Hint::new(); - let probed = symphonia::default::get_probe() - .format( - &hint, - mss, - &FormatOptions::default(), - &MetadataOptions::default(), - ) - .map_err(|e| anyhow!("Failed to probe format: {e}"))?; - - let mut format = probed.format; - - let track = format - .tracks() - .iter() - .find(|t| t.codec_params.codec != CODEC_TYPE_NULL) - .ok_or_else(|| anyhow!("No audio track found"))?; - - let mut decoder = symphonia::default::get_codecs() - .make(&track.codec_params, &DecoderOptions::default()) - .map_err(|e| anyhow!("Failed to create decoder: {e}"))?; - - let channels = track - .codec_params - .channels - .ok_or_else(|| anyhow!("Missing channel info"))? - .count(); - - let sample_rate = track - .codec_params - .sample_rate - .ok_or_else(|| anyhow!("Missing sample rate"))?; - - let bits_per_sample = track.codec_params.bits_per_sample.unwrap_or(16); - - let mut samples_i32 = Vec::new(); - let track_id = track.id; - - loop { - let packet = match format.next_packet() { - Ok(packet) => packet, - Err(SymphoniaError::ResetRequired) => { - decoder.reset(); - continue; - } - Err(SymphoniaError::IoError(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => { - break; - } - Err(e) => return Err(anyhow!("Decode error: {e}")), - }; - - if packet.track_id() != track_id { - continue; - } - - match decoder.decode(&packet) { - Ok(decoded) => { - let spec = *decoded.spec(); - let duration = decoded.capacity() as u64; - let mut sample_buf = SampleBuffer::::new(duration, spec); - sample_buf.copy_interleaved_ref(decoded); - samples_i32.extend_from_slice(sample_buf.samples()); - } - Err(SymphoniaError::DecodeError(_)) => continue, - Err(e) => return Err(anyhow!("Decode error: {e}")), - } - } - - if samples_i32.is_empty() { - return Err(anyhow!("No samples decoded")); - } - - let (normalized_samples, target_bits): (Vec, u32) = match bits_per_sample { - 0..=16 => { - let samples = samples_i32.iter().map(|&s| (s >> 16) as i32).collect(); - (samples, 16) - } - 17..=24 => { - let samples = samples_i32.iter().map(|&s| (s >> 8) as i32).collect(); - (samples, 24) - } - _ => (samples_i32, 32), - }; - - Ok(DecodedBlock { - samples: normalized_samples, - channels, - sample_rate, - bits_per_sample: target_bits, - }) -} - -fn compute_track_id_for_descriptor(descriptor_id: u8, block: &Block, song_index: usize) -> String { - format!( - "rp:{}:event_{}_song_{}", - descriptor_id, block.event, song_index - ) -} - -async fn store_rp_metadata( - cache_manager: &SourceCacheManager, - audio_pk: &str, - track_id: &str, - channel_id: u8, - song: &Song, - duration_ms: u64, - event: u64, - cover_pk: Option<&str>, -) -> Result<()> { - use serde_json::json; - - cache_manager.set_audio_metadata(audio_pk, "rp_title", json!(song.title))?; - cache_manager.set_audio_metadata(audio_pk, "rp_artist", json!(song.artist))?; - cache_manager.set_audio_metadata(audio_pk, "rp_album", json!(song.album))?; - cache_manager.set_audio_metadata(audio_pk, "rp_year", json!(song.year))?; - - cache_manager.set_audio_metadata(audio_pk, "rp_duration_ms", json!(duration_ms))?; - cache_manager.set_audio_metadata(audio_pk, "rp_elapsed_ms", json!(song.elapsed))?; - - cache_manager.set_audio_metadata(audio_pk, "rp_track_id", json!(track_id))?; - cache_manager.set_audio_metadata(audio_pk, "rp_channel_id", json!(channel_id))?; - cache_manager.set_audio_metadata(audio_pk, "rp_event", json!(event))?; - - cache_manager.set_audio_metadata(audio_pk, "rp_rating", json!(song.rating))?; - cache_manager.set_audio_metadata(audio_pk, "rp_cover_url", json!(song.cover))?; - cache_manager.set_audio_metadata(audio_pk, "rp_cover_pk", json!(cover_pk))?; - - Ok(()) -} - -async fn cache_cover_for_song( - cache_manager: &SourceCacheManager, - slug: &'static str, - block: &Block, - song: &Song, -) -> Result> { - if let Some(ref cover_path) = song.cover { - if let Some(cover_url) = block.cover_url(cover_path) { - match cache_manager.cache_cover(&cover_url).await { - Ok(pk) => return Ok(Some(pk)), - Err(err) => { - warn!(channel = slug, "Cover cache error: {err}"); - } - } - } else { - warn!( - channel = slug, - "Unable to resolve cover URL for {}", cover_path - ); - } - } - Ok(None) -} - -async fn encode_song_to_cache( - cache_manager: Arc, - descriptor_id: u8, - slug: &'static str, - block: Block, - song_index: usize, - song: Song, - track_samples: Vec, - sample_rate: u32, - channels: usize, - bits_per_sample: u32, - active_clients: usize, - duration_ms: u64, -) -> Result> { - let flac_bytes = encode_samples_to_flac(track_samples, channels, sample_rate, bits_per_sample) - .await - .context("Failed to encode song to FLAC")?; - - let track_id = compute_track_id_for_descriptor(descriptor_id, &block, song_index); - let placeholder_uri = format!("{}#{}", block.url, song_index); - - let mut metadata = TrackMetadata { - original_uri: placeholder_uri.clone(), - cached_audio_pk: None, - cached_cover_pk: None, - }; - - if let Some(cover_pk) = - cache_cover_for_song(cache_manager.as_ref(), slug, &block, &song).await? - { - metadata.cached_cover_pk = Some(cover_pk); - } - - 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 = cache_manager - .cache_audio_from_reader(&track_id, reader, Some(flac_len)) - .await - .map_err(|e| anyhow!("Cache audio error: {e}"))?; - - cache_manager - .wait_audio_ready(&audio_pk) - .await - .map_err(|e| anyhow!("Wait audio ready error: {e}"))?; - - metadata.cached_audio_pk = Some(audio_pk.clone()); - cache_manager - .update_metadata(track_id.clone(), metadata.clone()) - .await; - - if let Err(e) = store_rp_metadata( - cache_manager.as_ref(), - &audio_pk, - &track_id, - descriptor_id, - &song, - duration_ms, - block.event, - metadata.cached_cover_pk.as_deref(), - ) - .await - { - warn!(channel = slug, "Failed to store RP metadata: {e:?}"); - } - - let file_path = cache_manager.audio_file_path(&audio_pk).await; - - let entry = Arc::new(PlaylistEntry::new( - track_id, - descriptor_id, - Arc::new(song.clone()), - Utc::now(), - duration_ms, - Some(audio_pk), - file_path, - active_clients, - )); - - Ok(entry) -} - -fn spawn_song_processing( - context: SongTaskContext, - block: Block, - song_index: usize, - song: Song, - track_samples: Vec, - sample_rate: u32, - channels: usize, - bits_per_sample: u32, - active_clients: usize, - duration_ms: u64, - position_ms: u64, -) { - tokio::spawn(async move { - let SongTaskContext { - cache_manager, - playlist, - descriptor_id, - slug, - } = context; - - let song_title = song.title.clone(); - - match encode_song_to_cache( - cache_manager, - descriptor_id, - slug, - block, - song_index, - song, - track_samples, - sample_rate, - channels, - bits_per_sample, - active_clients, - duration_ms, - ) - .await - { - Ok(entry) => { - playlist.push_active(entry).await; - info!( - channel = slug, - song_index = song_index, - "🎵 Song '{}' available after {}ms (streaming mode)", - song_title, - position_ms - ); - } - Err(err) => { - warn!( - channel = slug, - song_index = song_index, - "Failed to process song '{}' asynchronously: {err:?}", - song_title - ); - } - } - }); -} - -async fn encode_samples_to_flac( - samples: Vec, - channels: usize, - sample_rate: u32, - bits_per_sample: u32, -) -> anyhow::Result> { - tokio::task::spawn_blocking(move || { - use flacenc::bitsink::ByteSink; - use flacenc::component::BitRepr; - use flacenc::error::Verify; - - // Note: Claxon retourne les samples dans leur résolution native - // Un fichier FLAC 16 bits retourne des samples i32 avec des valeurs dans la plage i16 - // Pas besoin de normalisation supplémentaire - let config = flacenc::config::Encoder::default() - .into_verified() - .map_err(|e| anyhow!("FLAC config error: {e:?}"))?; - - let source = flacenc::source::MemSource::from_samples( - &samples, - channels, - bits_per_sample as usize, - sample_rate as usize, - ); - - let flac_stream = flacenc::encode_with_fixed_block_size(&config, source, config.block_size) - .map_err(|e| anyhow!("FLAC encode error: {e:?}"))?; - - let mut sink = ByteSink::new(); - flac_stream - .write(&mut sink) - .map_err(|e| anyhow!("FLAC write error: {e:?}"))?; - - Ok::<_, anyhow::Error>(sink.into_inner()) - }) - .await? -} - -/// Métadonnées Radio Paradise récupérées depuis le cache -/// -/// Cette structure contient toutes les métadonnées RP stockées de manière -/// persistante dans le cache audio. -#[derive(Debug, Clone)] -pub struct RadioParadiseMetadata { - /// Titre de la chanson - pub title: String, - /// Artiste - pub artist: String, - /// Album (optionnel) - pub album: Option, - /// Année de sortie (optionnelle) - pub year: Option, - /// Durée en millisecondes - pub duration_ms: u64, - /// Offset depuis le début du block en millisecondes - pub elapsed_ms: u64, - /// Identifiant unique de la piste - pub track_id: String, - /// ID du canal Radio Paradise (0-3) - pub channel_id: u8, - /// ID de l'événement (block) - pub event: u64, - /// Note de la chanson (0-10, optionnelle) - pub rating: Option, - /// URL de la couverture (optionnelle) - pub cover_url: Option, - /// PK de la couverture dans le cache (optionnelle) - pub cover_pk: Option, -} - -/// Charge les métadonnées Radio Paradise depuis le cache audio -/// -/// Cette fonction lit toutes les métadonnées RP stockées pour un fichier -/// audio donné et les retourne dans une structure `RadioParadiseMetadata`. -/// -/// # Arguments -/// -/// * `cache_manager` - Le gestionnaire de cache source -/// * `audio_pk` - Clé primaire du fichier audio dans le cache -/// -/// # Returns -/// -/// Les métadonnées RP si elles existent et sont complètes, sinon une erreur. -/// -/// # Erreurs -/// -/// Cette fonction retourne une erreur si : -/// - Les métadonnées n'existent pas dans le cache -/// - Les métadonnées sont incomplètes ou corrompues -/// - Il y a une erreur de lecture du cache -pub async fn load_rp_metadata( - cache_manager: &SourceCacheManager, - audio_pk: &str, -) -> Result { - // Helper macro pour récupérer une métadonnée requise - macro_rules! get_required { - ($key:expr, $type:ty) => {{ - cache_manager - .get_audio_metadata(audio_pk, $key)? - .and_then(|v| serde_json::from_value::<$type>(v).ok()) - .ok_or_else(|| anyhow!("Missing or invalid metadata: {}", $key))? - }}; - } - - // Helper macro pour récupérer une métadonnée optionnelle - macro_rules! get_optional { - ($key:expr, $type:ty) => {{ - cache_manager - .get_audio_metadata(audio_pk, $key)? - .and_then(|v| { - if v.is_null() { - None - } else { - serde_json::from_value::<$type>(v).ok() - } - }) - }}; - } - - Ok(RadioParadiseMetadata { - title: get_required!("rp_title", String), - artist: get_required!("rp_artist", String), - album: get_optional!("rp_album", String), - year: get_optional!("rp_year", u32), - duration_ms: get_required!("rp_duration_ms", u64), - elapsed_ms: get_required!("rp_elapsed_ms", u64), - track_id: get_required!("rp_track_id", String), - channel_id: get_required!("rp_channel_id", u8), - event: get_required!("rp_event", u64), - rating: get_optional!("rp_rating", f32), - cover_url: get_optional!("rp_cover_url", String), - cover_pk: get_optional!("rp_cover_pk", String), - }) -} diff --git a/pmoparadise/src/stream.rs b/pmoparadise/src/stream.rs deleted file mode 100644 index 6c9f8dba..00000000 --- a/pmoparadise/src/stream.rs +++ /dev/null @@ -1,178 +0,0 @@ -//! Block streaming functionality - -use crate::error::{Error, Result}; -use crate::models::Block; -use crate::RadioParadiseClient; -use bytes::Bytes; -use futures::stream::{Stream, StreamExt}; -use std::pin::Pin; -use std::task::{Context, Poll}; -use url::Url; - -/// A stream of audio data from a Radio Paradise block -/// -/// This wraps the HTTP response body and provides a `Stream>` -/// that can be consumed by audio players or written to a file. -pub struct BlockStream { - inner: Pin> + Send>>, -} - -impl BlockStream { - /// Create a new block stream from a reqwest response - pub(crate) fn new(stream: impl Stream> + Send + 'static) -> Self { - Self { - inner: Box::pin(stream), - } - } - - /// Extract the inner stream - /// - /// Consumes the BlockStream and returns the underlying pinned stream. - /// Useful for advanced streaming scenarios like progressive decoding. - pub fn into_inner(self) -> Pin> + Send>> { - self.inner - } -} - -impl Stream for BlockStream { - type Item = Result; - - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - self.inner.as_mut().poll_next(cx) - } -} - -impl RadioParadiseClient { - /// Stream a block from its URL - /// - /// Returns a `Stream` of audio bytes that can be consumed by an audio player. - /// The stream will continue until the entire block is downloaded or an error occurs. - /// - /// # Arguments - /// - /// * `block_url` - The URL of the block to stream - /// - /// # Example - /// - /// ```no_run - /// use pmoparadise::RadioParadiseClient; - /// use futures::StreamExt; - /// - /// #[tokio::main] - /// async fn main() -> Result<(), Box> { - /// let client = RadioParadiseClient::new().await?; - /// let block = client.get_block(None).await?; - /// - /// let mut stream = client.stream_block(&block.url.parse()?).await?; - /// - /// while let Some(chunk) = stream.next().await { - /// let bytes = chunk?; - /// // Write bytes to audio player or file - /// println!("Received {} bytes", bytes.len()); - /// } - /// - /// Ok(()) - /// } - /// ``` - pub async fn stream_block(&self, block_url: &Url) -> Result { - #[cfg(feature = "logging")] - tracing::debug!("Starting block stream: {}", block_url); - - 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 stream block: HTTP {}", - response.status() - ))); - } - - // Convert reqwest's byte stream to our Result type - let stream = response.bytes_stream(); - let mapped = futures::stream::StreamExt::map(stream, |result| result.map_err(Error::from)); - - Ok(BlockStream::new(mapped)) - } - - /// Stream a block directly from a Block struct - /// - /// Convenience method that parses the URL from the block. - /// - /// # Example - /// - /// ```no_run - /// use pmoparadise::RadioParadiseClient; - /// use futures::StreamExt; - /// - /// #[tokio::main] - /// async fn main() -> Result<(), Box> { - /// let client = RadioParadiseClient::new().await?; - /// let block = client.get_block(None).await?; - /// - /// let mut stream = client.stream_block_from_metadata(&block).await?; - /// - /// while let Some(chunk) = stream.next().await { - /// let bytes = chunk?; - /// // Process bytes... - /// } - /// - /// Ok(()) - /// } - /// ``` - pub async fn stream_block_from_metadata(&self, block: &Block) -> Result { - let url = Url::parse(&block.url)?; - self.stream_block(&url).await - } - - /// Download an entire block as Bytes - /// - /// This downloads the complete block file into memory. For streaming playback, - /// use `stream_block()` instead which is more memory efficient. - /// - /// # Arguments - /// - /// * `block_url` - The URL of the block to download - /// - /// # Example - /// - /// ```no_run - /// use pmoparadise::RadioParadiseClient; - /// - /// #[tokio::main] - /// async fn main() -> Result<(), Box> { - /// let client = RadioParadiseClient::new().await?; - /// let block = client.get_block(None).await?; - /// 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 { - let mut stream = self.stream_block(block_url).await?; - let mut data = Vec::new(); - - while let Some(chunk_result) = stream.next().await { - let chunk = chunk_result?; - data.extend_from_slice(&chunk); - } - - Ok(Bytes::from(data)) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_block_stream_creation() { - let stream = futures::stream::once(async { Ok(Bytes::from("test")) }); - let _block_stream = BlockStream::new(stream); - } -} diff --git a/pmoparadise/src/streaming.rs b/pmoparadise/src/streaming.rs deleted file mode 100644 index b38e9be5..00000000 --- a/pmoparadise/src/streaming.rs +++ /dev/null @@ -1,217 +0,0 @@ -use anyhow::Result; -use bytes::Bytes; -use futures::stream::Stream; -use std::io::{self, Read}; -use std::pin::Pin; -use std::sync::mpsc::{sync_channel, Receiver, RecvError, SyncSender}; -use std::time::{Duration, Instant}; - -const CHANNEL_BUFFER_SIZE: usize = 64; // Augmenté de 16 à 64 pour réduire les warnings "buffer plein" -pub const CHUNK_SIZE_FRAMES: usize = 4096; - -pub struct ChannelReader { - receiver: Receiver>, - current_chunk: Option, - position: usize, -} - -impl ChannelReader { - pub fn new( - stream: Pin> + Send>>, - ) -> Self { - let (tx, rx) = sync_channel(CHANNEL_BUFFER_SIZE); - tokio::spawn(Self::stream_feeder(stream, tx)); - Self { - receiver: rx, - current_chunk: None, - position: 0, - } - } - - async fn stream_feeder( - mut stream: Pin> + Send>>, - tx: SyncSender>, - ) { - use futures::StreamExt; - while let Some(result) = stream.next().await { - let start = Instant::now(); - - let to_send = result.map_err(|e| e.to_string()); - match tx.try_send(to_send) { - Ok(_) => { /* message envoyé sans attente */ } - Err(std::sync::mpsc::TrySendError::Full(value)) => { - tracing::warn!("stream_feeder: buffer plein"); - // Revenir à l’envoi bloquant pour ne pas perdre le message - if tx.send(value).is_err() { - break; - } - } - Err(std::sync::mpsc::TrySendError::Disconnected(_)) => break, - } - let waited = start.elapsed(); - tracing::trace!("stream_feeder send {:?}", waited); - if waited > Duration::from_millis(200) { - tracing::warn!("stream_feeder wait {:?}", waited); - } - } - } -} - -impl Read for ChannelReader { - fn read(&mut self, buf: &mut [u8]) -> io::Result { - let start = Instant::now(); - loop { - if let Some(chunk) = &self.current_chunk { - if self.position < chunk.len() { - let available = chunk.len() - self.position; - let to_copy = available.min(buf.len()); - buf[..to_copy].copy_from_slice(&chunk[self.position..self.position + to_copy]); - self.position += to_copy; - tracing::trace!( - "ChannelReader copied {} bytes (elapsed {:?})", - to_copy, - start.elapsed() - ); - return Ok(to_copy); - } - } - - match self.receiver.recv() { - Ok(Ok(bytes)) => { - tracing::trace!( - "ChannelReader received chunk of {} bytes after {:?}", - bytes.len(), - start.elapsed() - ); - self.current_chunk = Some(bytes); - self.position = 0; - } - Ok(Err(e)) => { - tracing::warn!("ChannelReader received error chunk: {}", e); - return Err(io::Error::new(io::ErrorKind::Other, e)); - } - Err(RecvError) => { - tracing::trace!("ChannelReader stream closed after {:?}", start.elapsed()); - return Ok(0); - } - } - } - } -} - -#[derive(Debug, Clone)] -pub struct PCMChunk { - pub samples: Vec, - pub position_ms: u64, - pub sample_rate: u32, - pub channels: u32, -} - -pub struct StreamingPCMDecoder { - reader: claxon::FlacReader>, - sample_rate: u32, - channels: u32, - bits_per_sample: u32, - total_samples_decoded: u64, - done: bool, -} - -impl StreamingPCMDecoder { - /// Create a new decoder from an HTTP stream with default chunk size - pub fn new(http_stream: crate::stream::BlockStream) -> anyhow::Result { - Self::with_chunk_size(http_stream, CHUNK_SIZE_FRAMES) - } - - pub fn with_chunk_size( - http_stream: crate::stream::BlockStream, - _chunk_size: usize, - ) -> anyhow::Result { - let channel_reader = ChannelReader::new(http_stream.into_inner()); - let buffered = std::io::BufReader::new(channel_reader); - let reader = claxon::FlacReader::new(buffered) - .map_err(|e| anyhow::anyhow!("FLAC reader error: {}", e))?; - let info = reader.streaminfo(); - - Ok(Self { - reader, - sample_rate: info.sample_rate, - channels: info.channels, - bits_per_sample: info.bits_per_sample, - total_samples_decoded: 0, - done: false, - }) - } - - /// Get the sample rate (e.g., 44100 Hz) - pub fn sample_rate(&self) -> u32 { - self.sample_rate - } - - /// Get the number of channels (e.g., 2 for stereo) - pub fn channels(&self) -> u32 { - self.channels - } - - /// Get bits per sample (e.g., 16) - pub fn bits_per_sample(&self) -> u32 { - self.bits_per_sample - } - - pub fn decode_chunk(&mut self) -> anyhow::Result> { - if self.done { - return Ok(None); - } - - // Crée le FrameReader à la volée (emprunt de self.reader) - let mut frames = self.reader.blocks(); - - // API claxon 0.6.x : il FAUT fournir un Vec par valeur - let buf: Vec = Vec::new(); - let frame = match frames.read_next_or_eof(buf) { - Ok(None) => { - self.done = true; - return Ok(None); - } - Ok(Some(f)) => f, - Err(e) => return Err(anyhow::anyhow!("FLAC decode error: {}", e)), - }; - - let planar_samples: Vec = frame.into_buffer(); - if planar_samples.is_empty() { - self.done = true; - return Ok(None); - } - - // IMPORTANT: Claxon retourne les samples en format PLANAR (tous les L, puis tous les R) - // Mais nous avons besoin du format INTERLEAVED (L, R, L, R, ...) pour l'encodage - let block_size = planar_samples.len() / self.channels as usize; - let mut samples = Vec::with_capacity(planar_samples.len()); - - for i in 0..block_size { - for ch in 0..self.channels as usize { - samples.push(planar_samples[ch * block_size + i]); - } - } - - let position_ms = { - let frames = self.total_samples_decoded / self.channels as u64; - (frames * 1000) / self.sample_rate as u64 - }; - self.total_samples_decoded += samples.len() as u64; - - Ok(Some(PCMChunk { - samples, - position_ms, - sample_rate: self.sample_rate, - channels: self.channels, - })) - } -} - -pub fn ms_to_frames(ms: u64, sample_rate: u32) -> usize { - ((ms as u128 * sample_rate as u128) / 1000) as usize -} - -pub fn frames_to_ms(frames: usize, sample_rate: u32) -> u64 { - ((frames as u128 * 1000) / sample_rate as u128) as u64 -} diff --git a/pmoparadise/src/track.rs b/pmoparadise/src/track.rs deleted file mode 100644 index 8cab4c44..00000000 --- a/pmoparadise/src/track.rs +++ /dev/null @@ -1,397 +0,0 @@ -//! Per-track extraction from FLAC blocks (optional feature) -//! -//! **Important Notes:** -//! -//! Radio Paradise publishes *blocks* containing multiple songs, not individual -//! per-track files. This module provides experimental functionality to extract -//! individual tracks from FLAC blocks, but comes with significant tradeoffs: -//! -//! - **Storage**: Requires downloading the entire block (50-100MB) to disk -//! - **Latency**: Must download and decode before playback can start -//! - **CPU**: FLAC decoding is CPU-intensive -//! - **Complexity**: Seeking in FLAC requires decoding from the beginning -//! -//! ## Recommended Alternative -//! -//! For most use cases, it's better to: -//! 1. Stream the entire block to your audio player -//! 2. Use the `song[i].elapsed` metadata to seek within the player -//! 3. Let the player handle gapless transitions between tracks -//! -//! Modern players (mpv, VLC, ffmpeg) can seek in FLAC streams efficiently. -//! -//! ## When to Use This Module -//! -//! Only use per-track extraction when you need: -//! - Individual WAV files for further processing -//! - PCM data for custom audio analysis -//! - Separate files for non-streaming scenarios -//! -//! ## Block URL Pattern -//! -//! Blocks follow this URL pattern: -//! ```text -//! https://apps.radioparadise.com/blocks/chan/0/4/-.flac -//! ``` -//! -//! The `song[i].elapsed` field (in milliseconds) indicates when each track -//! starts within the block. - -#[cfg(feature = "per-track")] -use crate::error::{Error, Result}; -#[cfg(feature = "per-track")] -use crate::models::Block; -#[cfg(feature = "per-track")] -use crate::RadioParadiseClient; -#[cfg(feature = "per-track")] -use std::io::Write; -#[cfg(feature = "per-track")] -use std::path::PathBuf; - -/// Metadata for a decoded track stream -#[cfg(feature = "per-track")] -#[derive(Debug, Clone)] -pub struct TrackMetadata { - /// Sample rate in Hz (e.g., 44100) - pub sample_rate: u32, - /// Number of audio channels (1 = mono, 2 = stereo) - pub channels: u16, - /// Bits per sample (typically 16 or 24) - pub bits_per_sample: u16, - /// Total number of samples in this track - pub total_samples: u64, -} - -/// A stream of decoded PCM audio for a single track -/// -/// Provides access to decoded FLAC audio data for one track within a block. -/// The audio is decoded to 16-bit PCM format. -#[cfg(feature = "per-track")] -pub struct TrackStream { - /// Audio format metadata - pub metadata: TrackMetadata, - /// Path to the temporary FLAC file - temp_path: PathBuf, - /// FLAC reader - reader: Option>>, - /// Current sample position - current_sample: u64, - /// End sample position (where this track ends) - end_sample: u64, -} - -#[cfg(feature = "per-track")] -impl TrackStream { - /// Create a new track stream from a block - /// - /// This will: - /// 1. Download the entire block to a temporary file - /// 2. Open it with a FLAC decoder - /// 3. Seek to the track's start position - /// 4. Prepare to decode samples - /// - /// **Warning**: This is an expensive operation. Consider caching blocks. - async fn from_block_internal( - client: &RadioParadiseClient, - block: &Block, - track_index: usize, - ) -> Result { - // Validate track index - let song = block - .get_song(track_index) - .ok_or(Error::InvalidIndex(track_index, block.song_count()))?; - - // Download block to temporary file - let url = block - .url - .parse() - .map_err(|e| Error::other(format!("Invalid block URL: {}", e)))?; - - let block_data = client.download_block(&url).await?; - - // Write to temp file - let mut temp_file = tempfile::NamedTempFile::new()?; - temp_file.write_all(&block_data)?; - temp_file.flush()?; - - let temp_path = temp_file.into_temp_path(); - let path_buf = temp_path.to_path_buf(); - - #[cfg(feature = "logging")] - tracing::debug!("Wrote block to temp file: {:?}", path_buf); - - // Open FLAC reader - let file = std::fs::File::open(&path_buf)?; - let buffered = std::io::BufReader::new(file); - let mut reader = claxon::FlacReader::new(buffered)?; - - let streaminfo = reader.streaminfo(); - let sample_rate = streaminfo.sample_rate; - let channels = streaminfo.channels as u16; - let bits_per_sample = streaminfo.bits_per_sample as u16; - - // Calculate start and end sample positions - let start_sample = Self::ms_to_samples(song.elapsed, sample_rate); - let duration_samples = Self::ms_to_samples(song.duration, sample_rate); - let end_sample = start_sample + duration_samples; - - #[cfg(feature = "logging")] - tracing::debug!( - "Track {} spans samples {} to {} ({} ms to {} ms)", - track_index, - start_sample, - end_sample, - song.elapsed, - song.elapsed + song.duration - ); - - // Seek to start position by reading and discarding samples - // Note: FLAC doesn't support random access, so we must decode from beginning - if start_sample > 0 { - #[cfg(feature = "logging")] - tracing::debug!("Seeking to sample {}", start_sample); - - Self::skip_samples(&mut reader, start_sample)?; - } - - let metadata = TrackMetadata { - sample_rate, - channels, - bits_per_sample, - total_samples: duration_samples, - }; - - Ok(Self { - metadata, - temp_path: path_buf, - reader: Some(reader), - current_sample: start_sample, - end_sample, - }) - } - - /// Convert milliseconds to sample count - fn ms_to_samples(ms: u64, sample_rate: u32) -> u64 { - (ms * sample_rate as u64) / 1000 - } - - /// Skip samples by reading and discarding - fn skip_samples( - reader: &mut claxon::FlacReader>, - count: u64, - ) -> Result<()> { - let mut samples = reader.samples(); - for _ in 0..count { - if samples.next().is_none() { - return Err(Error::other("Unexpected end of FLAC stream while seeking")); - } - } - Ok(()) - } - - /// Read decoded PCM samples - /// - /// Returns samples as 16-bit signed integers (i16), interleaved by channel. - /// For stereo: [L, R, L, R, ...]. Returns None when track ends. - pub fn read_samples(&mut self, buffer: &mut [i16]) -> Result> { - let reader = self - .reader - .as_mut() - .ok_or(Error::other("TrackStream already consumed"))?; - - let mut samples_iter = reader.samples(); - let mut count = 0; - - for chunk in buffer.chunks_mut(self.metadata.channels as usize) { - if self.current_sample >= self.end_sample { - break; - } - - // Read one sample per channel - for sample_slot in chunk.iter_mut() { - match samples_iter.next() { - Some(Ok(sample)) => { - // Claxon returns i32, convert to i16 - *sample_slot = (sample >> (self.metadata.bits_per_sample - 16)) as i16; - count += 1; - } - Some(Err(e)) => { - return Err(Error::FlacDecode(e.to_string())); - } - None => { - return Ok(if count > 0 { Some(count) } else { None }); - } - } - } - - self.current_sample += 1; - } - - Ok(if count > 0 { Some(count) } else { None }) - } - - /// Export track to a WAV file - /// - /// Decodes the entire track and writes it as a WAV file. - /// - /// # Example - /// - /// ```no_run - /// # #[cfg(feature = "per-track")] - /// # { - /// use pmoparadise::RadioParadiseClient; - /// use std::path::Path; - /// - /// # #[tokio::main] - /// # async fn main() -> Result<(), Box> { - /// let client = RadioParadiseClient::new().await?; - /// let block = client.get_block(None).await?; - /// - /// let mut track_stream = client.open_track_stream(&block, 0).await?; - /// track_stream.export_wav(Path::new("track.wav"))?; - /// # Ok(()) - /// # } - /// # } - /// ``` - pub fn export_wav(&mut self, output_path: &std::path::Path) -> Result<()> { - let spec = hound::WavSpec { - channels: self.metadata.channels, - sample_rate: self.metadata.sample_rate, - bits_per_sample: 16, - sample_format: hound::SampleFormat::Int, - }; - - let mut writer = hound::WavWriter::create(output_path, spec)?; - let mut buffer = vec![0i16; 8192 * self.metadata.channels as usize]; - - #[cfg(feature = "logging")] - tracing::info!("Exporting track to WAV: {:?}", output_path); - - loop { - match self.read_samples(&mut buffer)? { - Some(count) => { - for &sample in &buffer[..count] { - writer.write_sample(sample)?; - } - } - None => break, - } - } - - writer.finalize()?; - - #[cfg(feature = "logging")] - tracing::info!("Successfully exported WAV file"); - - Ok(()) - } -} - -#[cfg(feature = "per-track")] -impl Drop for TrackStream { - fn drop(&mut self) { - // Close reader before removing temp file - self.reader.take(); - - // Clean up temporary file - if let Err(_e) = std::fs::remove_file(&self.temp_path) { - #[cfg(feature = "logging")] - tracing::warn!("Failed to remove temp file {:?}: {}", self.temp_path, _e); - } - } -} - -#[cfg(feature = "per-track")] -impl RadioParadiseClient { - /// Open a stream for a specific track within a block - /// - /// **Warning**: This downloads the entire block to a temporary file - /// and performs FLAC decoding. See module documentation for alternatives. - /// - /// # Arguments - /// - /// * `block` - The block containing the track - /// * `track_index` - Index of the track (0-based) - /// - /// # Example - /// - /// ```no_run - /// # #[cfg(feature = "per-track")] - /// # { - /// use pmoparadise::RadioParadiseClient; - /// - /// # #[tokio::main] - /// # async fn main() -> Result<(), Box> { - /// let client = RadioParadiseClient::new().await?; - /// let block = client.get_block(None).await?; - /// - /// // Extract first track - /// let mut track = client.open_track_stream(&block, 0).await?; - /// println!("Track: {} Hz, {} channels", - /// track.metadata.sample_rate, - /// track.metadata.channels); - /// - /// // Read some samples - /// let mut buffer = vec![0i16; 4096]; - /// if let Some(count) = track.read_samples(&mut buffer)? { - /// println!("Read {} samples", count); - /// } - /// # Ok(()) - /// # } - /// # } - /// ``` - pub async fn open_track_stream( - &self, - block: &Block, - track_index: usize, - ) -> Result { - TrackStream::from_block_internal(self, block, track_index).await - } - - /// Helper: Get track position in seconds for player-based seeking - /// - /// Instead of downloading and decoding, you can pass this information - /// to your audio player for efficient seeking. - /// - /// Returns (start_seconds, duration_seconds) - /// - /// # Example - /// - /// ```no_run - /// use pmoparadise::RadioParadiseClient; - /// - /// # #[tokio::main] - /// # async fn main() -> Result<(), Box> { - /// let client = RadioParadiseClient::new().await?; - /// let block = client.get_block(None).await?; - /// - /// let (start, duration) = client.track_position_seconds(&block, 1)?; - /// println!("Track 1 starts at {}s, duration {}s", start, duration); - /// println!("Play with: mpv --start={} --length={} {}", start, duration, block.url); - /// # Ok(()) - /// # } - /// ``` - pub fn track_position_seconds(&self, block: &Block, track_index: usize) -> Result<(f64, f64)> { - let song = block - .get_song(track_index) - .ok_or(Error::InvalidIndex(track_index, block.song_count()))?; - - let start_secs = song.elapsed as f64 / 1000.0; - let duration_secs = song.duration as f64 / 1000.0; - - Ok((start_secs, duration_secs)) - } -} - -#[cfg(test)] -#[cfg(feature = "per-track")] -mod tests { - use super::*; - - #[test] - fn test_ms_to_samples() { - assert_eq!(TrackStream::ms_to_samples(1000, 44100), 44100); - assert_eq!(TrackStream::ms_to_samples(500, 44100), 22050); - assert_eq!(TrackStream::ms_to_samples(0, 44100), 0); - } -}