eric/pmomusic
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Merge pull request #8 from coissac/claude/download-block-example-011CUpYvvxQzW5Hv2E4aL1nk

Browse Source 
Ok maintenant tout devrait rouler
This commit is contained in:
coissac
2025-11-05 13:39:54 +01:00
committed by GitHub
parent 6dd3ef6c61 33fefa7d33
commit aa916fa1ac
18 changed files with 125 additions and 4044 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
.pmomusic.yml
View File

@@ -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

120
pmoparadise/examples/extract_track.rs
View File

@@ -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);
}

71
pmoparadise/examples/show_source_image.rs
View File

@@ -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<dyn std::error::Error>> {
// 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(())
}

100
pmoparadise/examples/stream_block.rs
View File

@@ -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(&current_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(&current_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(())
}

129
pmoparadise/examples/test_streaming.rs
View File

@@ -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<dyn std::error::Error>> {
// 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<Vec<(u64, usize)>> {
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(())
}

107
pmoparadise/examples/with_cache.rs
View File

@@ -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<dyn std::error::Error>> {
// 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(())
}

39
pmoparadise/src/client.rs
View File

@@ -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<Client>,
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<String>) -> 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);
}
}

111
pmoparadise/src/config_ext.rs
View File

@@ -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<u8>;
/// 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<u8> {
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::<ParadiseChannelKind>() {
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)]

172
pmoparadise/src/ffmpeg_streaming.rs
View File

@@ -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<i16>, // 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<Result<Bytes, String>>,
buffer: Vec<u8>,
decoder_ctx: Option<ffmpeg::codec::context::Context>,
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<Self> {
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<Option<PCMChunk>> {
// 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<Bytes>,
encoder_ctx: Option<ffmpeg::codec::context::Context>,
sample_rate: u32,
channels: u32,
}
impl ProgressiveEncoder {
/// Create a new progressive encoder
pub fn new(sample_rate: u32, channels: u32) -> Result<(Self, Receiver<Bytes>)> {
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());
}
}

428
pmoparadise/src/paradise/channel.rs
View File

@@ -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<Self, Self::Err> {
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<ParadiseChannelInner>,
}
struct ParadiseChannelInner {
descriptor: ChannelDescriptor,
client: RadioParadiseClient,
history_max_tracks: usize,
playlist: SharedPlaylist,
history: Arc<dyn HistoryBackend>,
cache_manager: Arc<SourceCacheManager>,
active_clients: AtomicUsize,
worker_tx: mpsc::Sender<WorkerCommand>,
worker: Mutex<Option<ParadiseWorker>>,
}
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<dyn HistoryBackend>,
cache_manager: Arc<SourceCacheManager>,
) -> Result<Self> {
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<dyn HistoryBackend> {
&self.inner.history
}
pub fn cache_manager(&self) -> Arc<SourceCacheManager> {
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<String>,
) -> Result<ParadiseClientStream> {
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<String>) -> 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<PlaylistEntry>) {
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<Bytes, anyhow::Error>> {
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<String> = 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:?}");
}
});
}
}

208
pmoparadise/src/paradise/constants.rs
View File

@@ -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);
}
}

217
pmoparadise/src/paradise/history.rs
View File

@@ -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<chrono::Utc>,
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<String>,
pub cover_url: Option<String>,
}
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<Vec<HistoryEntry>>;
async fn len(&self) -> anyhow::Result<usize>;
async fn truncate(&self, keep: usize) -> anyhow::Result<()>;
}
pub struct SqliteHistoryBackend {
conn: Arc<StdMutex<rusqlite::Connection>>,
}
impl SqliteHistoryBackend {
pub fn new(path: impl AsRef<Path>) -> anyhow::Result<Self> {
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<StdMutex<rusqlite::Connection>> {
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<Vec<HistoryEntry>> {
let conn = self.conn();
let limit = limit as i64;
spawn_blocking(move || -> anyhow::Result<Vec<HistoryEntry>> {
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::<Utc>::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<String>>(4)?.unwrap_or_default(),
artist: row.get::<_, Option<String>>(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<usize> {
let conn = self.conn();
let count = spawn_blocking(move || -> anyhow::Result<usize> {
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<Arc<dyn HistoryBackend>> {
let backend = SqliteHistoryBackend::new(database_path)?;
Ok(Arc::new(backend))
}

28
pmoparadise/src/paradise/mod.rs
View File

@@ -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};

293
pmoparadise/src/paradise/playlist.rs
View File

@@ -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<Song>,
pub started_at: DateTime<Utc>,
pub duration_ms: u64,
pub audio_pk: Option<String>,
pub file_path: Option<PathBuf>,
pending_clients: AtomicUsize,
}
impl PlaylistEntry {
#[allow(clippy::too_many_arguments)]
pub fn new(
track_id: String,
channel_id: u8,
song: Arc<Song>,
started_at: DateTime<Utc>,
duration_ms: u64,
audio_pk: Option<String>,
file_path: Option<PathBuf>,
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<Arc<PlaylistEntry>>,
history: VecDeque<HistoryEntry>,
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<PlaylistEntry>) {
self.active.push_back(entry);
}
fn active_snapshot(&self) -> Vec<Arc<PlaylistEntry>> {
self.active.iter().cloned().collect()
}
fn pop_front_if_ready(&mut self) -> Option<Arc<PlaylistEntry>> {
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<Arc<PlaylistEntry>> {
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<HistoryEntry> {
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<PlaylistState>,
notify: Notify,
update_id: AtomicU32,
last_change: RwLock<Option<SystemTime>>,
}
#[derive(Clone)]
pub struct SharedPlaylist(Arc<SharedPlaylistInner>);
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<PlaylistEntry>) {
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<Arc<PlaylistEntry>> {
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<Arc<PlaylistEntry>> {
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<Arc<PlaylistEntry>> {
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<HistoryEntry> {
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<SystemTime> {
self.0.last_change.read().await.clone()
}
}

1326
pmoparadise/src/paradise/worker.rs
View File

File diff suppressed because it is too large Load Diff

178
pmoparadise/src/stream.rs
View File

@@ -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<Item = Result<Bytes>>`
/// that can be consumed by audio players or written to a file.
pub struct BlockStream {
inner: Pin<Box<dyn Stream<Item = Result<Bytes>> + Send>>,
}
impl BlockStream {
/// Create a new block stream from a reqwest response
pub(crate) fn new(stream: impl Stream<Item = Result<Bytes>> + 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<Box<dyn Stream<Item = Result<Bytes>> + Send>> {
self.inner
}
}
impl Stream for BlockStream {
type Item = Result<Bytes>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
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<dyn std::error::Error>> {
/// 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<BlockStream> {
#[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<dyn std::error::Error>> {
/// 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<BlockStream> {
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<dyn std::error::Error>> {
/// 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<Bytes> {
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);
}
}

217
pmoparadise/src/streaming.rs
View File

@@ -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<Result<Bytes, String>>,
current_chunk: Option<Bytes>,
position: usize,
}
impl ChannelReader {
pub fn new(
stream: Pin<Box<dyn Stream<Item = Result<Bytes, crate::error::Error>> + 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<Box<dyn Stream<Item = Result<Bytes, crate::error::Error>> + Send>>,
tx: SyncSender<Result<Bytes, String>>,
) {
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<usize> {
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<i32>,
pub position_ms: u64,
pub sample_rate: u32,
pub channels: u32,
}
pub struct StreamingPCMDecoder<R: Read> {
reader: claxon::FlacReader<std::io::BufReader<R>>,
sample_rate: u32,
channels: u32,
bits_per_sample: u32,
total_samples_decoded: u64,
done: bool,
}
impl StreamingPCMDecoder<ChannelReader> {
/// Create a new decoder from an HTTP stream with default chunk size
pub fn new(http_stream: crate::stream::BlockStream) -> anyhow::Result<Self> {
Self::with_chunk_size(http_stream, CHUNK_SIZE_FRAMES)
}
pub fn with_chunk_size(
http_stream: crate::stream::BlockStream,
_chunk_size: usize,
) -> anyhow::Result<Self> {
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<Option<PCMChunk>> {
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<i32> par valeur
let buf: Vec<i32> = 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<i32> = 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
}

397
pmoparadise/src/track.rs
View File

@@ -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/<start_event>-<end_event>.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<claxon::FlacReader<std::io::BufReader<std::fs::File>>>,
/// 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<Self> {
// 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<std::io::BufReader<std::fs::File>>,
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<Option<usize>> {
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<dyn std::error::Error>> {
/// 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<dyn std::error::Error>> {
/// 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> {
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<dyn std::error::Error>> {
/// 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);
}
}