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

Merge branch 'main' into claude/download-block-example-011CUpYvvxQzW5Hv2E4aL1nk

Browse Source
This commit is contained in:
coissac
2025-11-05 11:51:11 +01:00
committed by GitHub
parent 8ff74da191 4332aa9b89
commit 7cefc51020
16 changed files with 4019 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
.pmomusic.yml Normal file
View File

@@ -0,0 +1,28 @@
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 Normal file
View File

@@ -0,0 +1,120 @@
//! 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 Normal file
View File

@@ -0,0 +1,71 @@
//! 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 Normal file
View File

@@ -0,0 +1,100 @@
//! 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 Normal file
View File

@@ -0,0 +1,129 @@
//! 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 Normal file
View File

@@ -0,0 +1,107 @@
//! 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(())
}

172
pmoparadise/src/ffmpeg_streaming.rs Normal file
View File

@@ -0,0 +1,172 @@
//! 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 Normal file
View File

@@ -0,0 +1,428 @@
//! 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 Normal file
View File

@@ -0,0 +1,208 @@
//! 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 Normal file
View File

@@ -0,0 +1,217 @@
//! 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 Normal file
View File

@@ -0,0 +1,28 @@
//! 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 Normal file
View File

@@ -0,0 +1,293 @@
//! 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 Normal file
View File

File diff suppressed because it is too large Load Diff

178
pmoparadise/src/stream.rs Normal file
View File

@@ -0,0 +1,178 @@
//! 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 Normal file
View File

@@ -0,0 +1,217 @@
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 Normal file
View File

@@ -0,0 +1,397 @@
//! 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);
}
}