Files
pmomusic/pmoparadise/src/source.rs

967 lines
32 KiB
Rust
Raw Normal View History

2025-10-16 22:00:35 +02:00
//! Music source implementation for Radio Paradise
//!
//! This module implements the [`pmosource::MusicSource`] trait for Radio Paradise,
2025-10-16 22:13:00 +02:00
//! providing a complete music source with FIFO playlist support, browsing, and caching.
2025-10-16 22:00:35 +02:00
2025-10-16 22:13:00 +02:00
use crate::client::RadioParadiseClient;
use crate::models::{Block, Song};
use anyhow::anyhow;
use pmoaudiocache::Cache as AudioCache;
2025-10-17 23:45:01 +02:00
use pmocovers::Cache as CoverCache;
use pmodidl::{Container, Item};
2025-10-16 22:13:00 +02:00
use pmoplaylist::{FifoPlaylist, Track};
2025-10-19 13:42:29 +02:00
use pmosource::SourceCacheManager;
use pmosource::{async_trait, pmodidl, BrowseResult, MusicSource, MusicSourceError, Result};
use std::collections::{HashMap, HashSet};
use std::io::Cursor;
2025-10-16 22:13:00 +02:00
use std::sync::Arc;
use std::time::SystemTime;
use tokio::sync::{Mutex, RwLock};
use url::Url;
2025-10-16 22:00:35 +02:00
/// Default image for Radio Paradise (300x300 WebP, embedded in binary)
const DEFAULT_IMAGE: &[u8] = include_bytes!("../assets/default.webp");
2025-10-16 22:13:00 +02:00
/// Default FIFO capacity (number of recent tracks to keep)
const DEFAULT_FIFO_CAPACITY: usize = 50;
#[derive(Clone, Copy)]
struct ChannelDescriptor {
id: u8,
name: &'static str,
description: &'static str,
}
const CHANNELS: [ChannelDescriptor; 4] = [
ChannelDescriptor {
id: 0,
name: "Main Mix",
description: "Eclectic mix of rock, world, electronica, and more",
},
ChannelDescriptor {
id: 1,
name: "Mellow Mix",
description: "Mellower, less aggressive music",
},
ChannelDescriptor {
id: 2,
name: "Rock Mix",
description: "Heavier, more guitar-driven music",
},
ChannelDescriptor {
id: 3,
name: "World Mix",
description: "Global beats and world music",
},
];
fn channel_collection_id(channel_id: u8) -> String {
format!("radio-paradise:{}", channel_id)
}
fn channel_container_id(channel_id: u8) -> String {
format!("radio-paradise:channel:{}", channel_id)
}
fn channel_playlist_id(channel_id: u8) -> String {
channel_container_id(channel_id)
}
fn parse_channel_container_id(object_id: &str) -> Option<u8> {
let mut parts = object_id.split(':');
match (parts.next(), parts.next(), parts.next(), parts.next()) {
(Some("radio-paradise"), Some("channel"), Some(id_str), None) => id_str.parse().ok(),
_ => None,
}
}
fn track_identifier(channel_id: u8, event: u64, song_index: usize) -> String {
format!("rp:{}:{}:{}", channel_id, event, song_index)
}
fn parse_track_identifier(track_id: &str) -> Option<(u8, u64, usize)> {
let mut parts = track_id.split(':');
match (
parts.next(),
parts.next(),
parts.next(),
parts.next(),
parts.next(),
) {
(Some("rp"), Some(channel_str), Some(event_str), Some(index_str), None) => {
let channel = channel_str.parse().ok()?;
let event = event_str.parse().ok()?;
let idx = index_str.parse().ok()?;
Some((channel, event, idx))
}
_ => None,
}
}
2025-10-16 22:13:00 +02:00
/// Radio Paradise music source with full MusicSource trait implementation
///
/// This struct combines a [`RadioParadiseClient`] for API access with a FIFO playlist
/// for dynamic track management, implementing the complete [`MusicSource`] trait.
2025-10-16 22:00:35 +02:00
///
2025-10-16 22:13:00 +02:00
/// # Features
///
/// - **FIFO Playlist**: Dynamic track management with configurable capacity
/// - **API Integration**: Fetches blocks and metadata from Radio Paradise
/// - **URI Resolution**: Resolves track URIs with optional cache support
/// - **Change Tracking**: Tracks update_id and last_change for UPnP notifications
/// - **DIDL-Lite Export**: Converts tracks and blocks to UPnP-compatible formats
2025-10-16 22:00:35 +02:00
///
/// # Examples
///
2025-10-16 22:13:00 +02:00
/// ```no_run
/// use pmoparadise::{RadioParadiseSource, RadioParadiseClient};
2025-10-16 22:00:35 +02:00
/// use pmosource::MusicSource;
///
2025-10-16 22:13:00 +02:00
/// #[tokio::main]
/// async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// let client = RadioParadiseClient::new().await?;
/// let source = RadioParadiseSource::new(client, "http://localhost:8080", 50);
///
/// println!("Source: {}", source.name());
/// println!("Supports FIFO: {}", source.supports_fifo());
2025-10-16 22:00:35 +02:00
///
2025-10-16 22:13:00 +02:00
/// // Start streaming and the FIFO will be populated
/// Ok(())
/// }
2025-10-16 22:00:35 +02:00
/// ```
2025-10-16 22:13:00 +02:00
#[derive(Clone)]
pub struct RadioParadiseSource {
inner: Arc<RadioParadiseSourceInner>,
}
struct ChannelState {
descriptor: ChannelDescriptor,
2025-10-16 22:13:00 +02:00
client: RadioParadiseClient,
playlist: FifoPlaylist,
2025-10-17 23:45:01 +02:00
cache_manager: SourceCacheManager,
processed_blocks: RwLock<HashSet<u64>>,
ingest_lock: Mutex<()>,
}
2025-10-16 22:13:00 +02:00
struct DecodedBlock {
samples: Vec<i32>,
channels: usize,
sample_rate: u32,
bits_per_sample: u32,
}
struct RadioParadiseSourceInner {
channels: HashMap<u8, Arc<ChannelState>>,
2025-10-16 22:13:00 +02:00
}
impl std::fmt::Debug for RadioParadiseSource {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
2025-10-19 13:42:29 +02:00
f.debug_struct("RadioParadiseSource").finish()
2025-10-16 22:13:00 +02:00
}
}
impl RadioParadiseSource {
2025-10-18 09:58:39 +02:00
/// Create a new Radio Paradise source from the cache registry
///
/// This is the recommended way to create a source when using the UPnP server.
/// The caches are automatically retrieved from the global registry.
///
/// # Arguments
///
/// * `client` - Radio Paradise API client
/// * `fifo_capacity` - Maximum number of tracks in the FIFO
///
/// # Errors
///
/// Returns an error if the caches are not initialized in the registry
#[cfg(feature = "server")]
pub fn from_registry(client: RadioParadiseClient, fifo_capacity: usize) -> Result<Self> {
let mut channels = HashMap::new();
for descriptor in CHANNELS.iter() {
let channel_id = descriptor.id;
let channel_client = client.clone_with_channel(channel_id);
let playlist = FifoPlaylist::new(
channel_playlist_id(channel_id),
descriptor.name.to_string(),
fifo_capacity,
DEFAULT_IMAGE,
);
let cache_manager =
SourceCacheManager::from_registry(channel_collection_id(channel_id))?;
channels.insert(
channel_id,
Arc::new(ChannelState {
descriptor: *descriptor,
client: channel_client,
playlist,
cache_manager,
processed_blocks: RwLock::new(HashSet::new()),
ingest_lock: Mutex::new(()),
}),
);
}
2025-10-18 09:58:39 +02:00
Ok(Self {
inner: Arc::new(RadioParadiseSourceInner { channels }),
2025-10-18 09:58:39 +02:00
})
}
/// Create with default FIFO capacity from the cache registry
#[cfg(feature = "server")]
pub fn from_registry_default(client: RadioParadiseClient) -> Result<Self> {
Self::from_registry(client, DEFAULT_FIFO_CAPACITY)
}
/// Create a new Radio Paradise source with explicit caches (for tests)
2025-10-16 22:13:00 +02:00
///
/// # Arguments
///
/// * `client` - Radio Paradise API client
/// * `fifo_capacity` - Maximum number of tracks in the FIFO
2025-10-17 23:45:01 +02:00
/// * `cover_cache` - Cover image cache (required)
/// * `audio_cache` - Audio cache (required)
2025-10-16 22:13:00 +02:00
pub fn new(
client: RadioParadiseClient,
fifo_capacity: usize,
2025-10-17 23:45:01 +02:00
cover_cache: Arc<CoverCache>,
audio_cache: Arc<AudioCache>,
2025-10-16 22:13:00 +02:00
) -> Self {
let mut channels = HashMap::new();
for descriptor in CHANNELS.iter() {
let channel_id = descriptor.id;
let channel_client = client.clone_with_channel(channel_id);
let playlist = FifoPlaylist::new(
channel_playlist_id(channel_id),
descriptor.name.to_string(),
fifo_capacity,
DEFAULT_IMAGE,
);
let cache_manager = SourceCacheManager::new(
channel_collection_id(channel_id),
Arc::clone(&cover_cache),
Arc::clone(&audio_cache),
);
channels.insert(
channel_id,
Arc::new(ChannelState {
descriptor: *descriptor,
client: channel_client,
playlist,
cache_manager,
processed_blocks: RwLock::new(HashSet::new()),
ingest_lock: Mutex::new(()),
}),
);
}
2025-10-17 23:45:01 +02:00
2025-10-16 22:13:00 +02:00
Self {
inner: Arc::new(RadioParadiseSourceInner { channels }),
2025-10-16 22:13:00 +02:00
}
}
2025-10-18 09:58:39 +02:00
/// Create with default FIFO capacity (for tests)
2025-10-17 23:45:01 +02:00
pub fn new_default(
2025-10-16 22:13:00 +02:00
client: RadioParadiseClient,
2025-10-17 23:45:01 +02:00
cover_cache: Arc<CoverCache>,
audio_cache: Arc<AudioCache>,
2025-10-16 22:13:00 +02:00
) -> Self {
2025-10-18 09:58:39 +02:00
Self::new(client, DEFAULT_FIFO_CAPACITY, cover_cache, audio_cache)
2025-10-16 22:13:00 +02:00
}
/// Get the Radio Paradise client for a given channel
pub fn client_for_channel(&self, channel: u8) -> Option<RadioParadiseClient> {
self.inner
.channels
.get(&channel)
.map(|state| state.client.clone())
}
2025-10-16 22:13:00 +02:00
fn channel_state(&self, channel_id: u8) -> Option<Arc<ChannelState>> {
self.inner.channels.get(&channel_id).cloned()
}
2025-10-16 22:13:00 +02:00
fn build_root_container(&self) -> Container {
Container {
id: "radio-paradise".to_string(),
parent_id: "0".to_string(),
restricted: Some("1".to_string()),
child_count: Some(CHANNELS.len().to_string()),
title: "Radio Paradise".to_string(),
class: "object.container".to_string(),
containers: vec![],
items: vec![],
2025-10-16 22:13:00 +02:00
}
}
2025-10-16 22:13:00 +02:00
async fn build_channel_containers(&self) -> Vec<Container> {
let mut containers = Vec::new();
for descriptor in CHANNELS.iter() {
if let Some(channel) = self.channel_state(descriptor.id) {
let child_count = channel.playlist.len().await;
containers.push(Container {
id: channel_container_id(descriptor.id),
parent_id: "radio-paradise".to_string(),
restricted: Some("1".to_string()),
child_count: Some(child_count.to_string()),
title: descriptor.name.to_string(),
class: "object.container.playlistContainer".to_string(),
containers: vec![],
items: vec![],
});
2025-10-19 01:21:31 +02:00
}
2025-10-16 22:13:00 +02:00
}
containers
}
async fn ensure_channel_ready(&self, channel: Arc<ChannelState>) -> Result<()> {
if channel.playlist.len().await > 0 {
return Ok(());
}
let guard = channel.ingest_lock.lock().await;
if channel.playlist.len().await == 0 {
drop(guard);
self.populate_channel_locked(channel.clone()).await?;
} else {
drop(guard);
}
Ok(())
}
async fn populate_channel_locked(&self, channel: Arc<ChannelState>) -> Result<()> {
tracing::info!(
"📻 Fetching Radio Paradise block for channel {}",
channel.descriptor.name
);
2025-10-16 22:13:00 +02:00
let now_playing = channel
.client
.now_playing()
.await
.map_err(|e| MusicSourceError::SourceUnavailable(e.to_string()))?;
let block = Arc::new(now_playing.block);
self.ingest_block(channel, block).await
}
async fn ingest_block(&self, channel: Arc<ChannelState>, block: Arc<Block>) -> Result<()> {
{
let mut processed = channel.processed_blocks.write().await;
if !processed.insert(block.event) {
tracing::debug!(
"Channel {} already processed block {}",
channel.descriptor.name,
block.event
);
return Ok(());
}
2025-10-16 22:13:00 +02:00
}
let block_url = Url::parse(&block.url)
.map_err(|e| MusicSourceError::BrowseError(format!("Invalid block URL: {}", e)))?;
let block_bytes = channel
.client
.download_block(&block_url)
.await
.map_err(|e| {
MusicSourceError::BrowseError(format!("Failed to download block: {}", e))
})?;
let decoded = decode_block_audio(block_bytes.to_vec())
.map_err(|e| MusicSourceError::BrowseError(e.to_string()))?;
let ordered_songs = block.songs_ordered();
let total_frames = decoded.samples.len() / decoded.channels;
for (position, (song_index, song)) in ordered_songs.iter().enumerate() {
let track_id = track_identifier(channel.descriptor.id, block.event, *song_index);
if channel
.cache_manager
.get_metadata(&track_id)
.await
.is_some()
{
continue;
}
let duration_ms = song_duration_ms(&block, &ordered_songs, position);
if duration_ms == 0 {
tracing::debug!(
"Skipping track {} with zero duration on channel {}",
track_id,
channel.descriptor.name
);
continue;
}
let start_frame = ms_to_frames(song.elapsed, decoded.sample_rate);
let end_frame =
ms_to_frames(song.elapsed + duration_ms, decoded.sample_rate).min(total_frames);
if start_frame >= end_frame {
tracing::debug!(
"Invalid frame range for track {} (start {} >= end {})",
track_id,
start_frame,
end_frame
);
continue;
}
let start_index = start_frame * decoded.channels;
let end_index = end_frame * decoded.channels;
let song_samples = decoded.samples[start_index..end_index].to_vec();
let flac_data = encode_samples_to_flac(
song_samples,
decoded.channels,
decoded.sample_rate,
decoded.bits_per_sample,
)
.await
.map_err(|e| MusicSourceError::CacheError(e.to_string()))?;
let audio_source_uri = format!("{}#{}", block.url, song_index);
let reader = Cursor::new(flac_data.clone());
let audio_pk: String = channel
.cache_manager
.cache_audio_from_reader(&audio_source_uri, reader, Some(flac_data.len() as u64))
.await?;
channel.cache_manager.wait_audio_ready(&audio_pk).await?;
let cached_cover_pk = if let Some(ref image_base) = block.image_base {
if let Some(ref cover) = song.cover {
let image_url = format!("{}{}", image_base, cover);
match channel.cache_manager.cache_cover(&image_url).await {
Ok(pk) => Some(pk),
Err(e) => {
tracing::warn!(
"Failed to cache cover {} on channel {}: {}",
image_url,
channel.descriptor.name,
e
);
None
2025-10-18 09:58:39 +02:00
}
2025-10-17 23:45:01 +02:00
}
} else {
None
2025-10-16 22:13:00 +02:00
}
} else {
None
};
2025-10-16 22:13:00 +02:00
let playback_url = channel.cache_manager.resolve_uri(&track_id).await?;
2025-10-16 22:13:00 +02:00
let mut track = Track::new(track_id.clone(), song.title.clone(), playback_url);
if !song.artist.is_empty() {
track = track.with_artist(song.artist.clone());
2025-10-17 23:45:01 +02:00
}
if let Some(ref album) = song.album {
if !album.is_empty() {
track = track.with_album(album.clone());
}
2025-10-16 22:13:00 +02:00
}
track = track.with_duration((duration_ms / 1000) as u32);
2025-10-17 23:45:01 +02:00
if let Some(ref cover_pk) = cached_cover_pk {
if let Ok(url) = channel.cache_manager.cover_url(cover_pk, None) {
track = track.with_image(url);
}
} else if let Some(ref cover) = song.cover {
if let Some(ref image_base) = block.image_base {
track = track.with_image(format!("{}{}", image_base, cover));
}
}
channel
.cache_manager
.update_metadata(
track_id.clone(),
pmosource::TrackMetadata {
original_uri: block.url.clone(),
cached_audio_pk: Some(audio_pk.clone()),
cached_cover_pk,
},
)
.await;
channel.playlist.append_track(track).await;
2025-10-16 22:13:00 +02:00
}
tracing::info!(
"Channel {} now has {} tracks",
channel.descriptor.name,
channel.playlist.len().await
);
2025-10-16 22:13:00 +02:00
Ok(())
}
}
fn song_duration_ms(block: &Block, ordered: &[(usize, &Song)], position: usize) -> u64 {
let song = ordered[position].1;
if song.duration > 0 {
return song.duration;
}
if let Some((_, next_song)) = ordered.get(position + 1) {
return next_song.elapsed.saturating_sub(song.elapsed);
}
2025-10-16 22:13:00 +02:00
block.length.saturating_sub(song.elapsed)
}
fn ms_to_frames(ms: u64, sample_rate: u32) -> usize {
((ms as u128 * sample_rate as u128) / 1000) as usize
}
fn decode_block_audio(data: Vec<u8>) -> anyhow::Result<DecodedBlock> {
use symphonia::core::audio::SampleBuffer;
use symphonia::core::codecs::{DecoderOptions, CODEC_TYPE_NULL};
use symphonia::core::errors::Error as SymphoniaError;
use symphonia::core::formats::FormatOptions;
use symphonia::core::io::MediaSourceStream;
use symphonia::core::meta::MetadataOptions;
use symphonia::core::probe::Hint;
let cursor = Cursor::new(data);
let mss = MediaSourceStream::new(Box::new(cursor), Default::default());
let hint = Hint::new();
let probed = symphonia::default::get_probe()
.format(
&hint,
mss,
&FormatOptions::default(),
&MetadataOptions::default(),
)
.map_err(|e| anyhow!("Failed to probe format: {}", e))?;
let mut format = probed.format;
let track = format
.tracks()
.iter()
.find(|t| t.codec_params.codec != CODEC_TYPE_NULL)
.ok_or_else(|| anyhow!("No audio track found"))?;
let mut decoder = symphonia::default::get_codecs()
.make(&track.codec_params, &DecoderOptions::default())
.map_err(|e| anyhow!("Failed to create decoder: {}", e))?;
let channels = track
.codec_params
.channels
.ok_or_else(|| anyhow!("Missing channel info"))?
.count();
let sample_rate = track
.codec_params
.sample_rate
.ok_or_else(|| anyhow!("Missing sample rate"))?;
let bits_per_sample = track.codec_params.bits_per_sample.unwrap_or(16);
let mut samples_i32 = Vec::new();
let track_id = track.id;
loop {
let packet = match format.next_packet() {
Ok(packet) => packet,
Err(SymphoniaError::ResetRequired) => {
decoder.reset();
continue;
}
Err(SymphoniaError::IoError(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
break;
}
Err(e) => return Err(anyhow!("Decode error: {}", e)),
2025-10-16 22:13:00 +02:00
};
if packet.track_id() != track_id {
continue;
}
match decoder.decode(&packet) {
Ok(decoded) => {
let spec = *decoded.spec();
let duration = decoded.capacity() as u64;
let mut sample_buf = SampleBuffer::<i32>::new(duration, spec);
sample_buf.copy_interleaved_ref(decoded);
samples_i32.extend_from_slice(sample_buf.samples());
}
Err(SymphoniaError::DecodeError(_)) => continue,
Err(e) => return Err(anyhow!("Decode error: {}", e)),
2025-10-16 22:13:00 +02:00
}
}
if samples_i32.is_empty() {
return Err(anyhow!("No samples decoded"));
2025-10-16 22:13:00 +02:00
}
let (normalized_samples, target_bits): (Vec<i32>, u32) = match bits_per_sample {
0..=16 => {
let samples = samples_i32.iter().map(|&s| (s >> 16) as i32).collect();
(samples, 16)
}
17..=24 => {
let samples = samples_i32.iter().map(|&s| (s >> 8) as i32).collect();
(samples, 24)
}
_ => (samples_i32, 32),
};
Ok(DecodedBlock {
samples: normalized_samples,
channels,
sample_rate,
bits_per_sample: target_bits,
})
}
async fn encode_samples_to_flac(
samples: Vec<i32>,
channels: usize,
sample_rate: u32,
bits_per_sample: u32,
) -> anyhow::Result<Vec<u8>> {
tokio::task::spawn_blocking(move || {
use flacenc::bitsink::ByteSink;
use flacenc::component::BitRepr;
use flacenc::error::Verify;
let config = flacenc::config::Encoder::default()
.into_verified()
.map_err(|e| anyhow!("FLAC config error: {:?}", e))?;
let source = flacenc::source::MemSource::from_samples(
&samples,
channels,
bits_per_sample as usize,
sample_rate as usize,
);
let flac_stream = flacenc::encode_with_fixed_block_size(&config, source, config.block_size)
.map_err(|e| anyhow!("FLAC encode error: {:?}", e))?;
let mut sink = ByteSink::new();
flac_stream
.write(&mut sink)
.map_err(|e| anyhow!("FLAC write error: {:?}", e))?;
Ok::<_, anyhow::Error>(sink.into_inner())
})
.await?
2025-10-16 22:13:00 +02:00
}
#[async_trait]
2025-10-16 22:00:35 +02:00
impl MusicSource for RadioParadiseSource {
fn name(&self) -> &str {
"Radio Paradise"
}
fn id(&self) -> &str {
"radio-paradise"
}
fn default_image(&self) -> &[u8] {
DEFAULT_IMAGE
}
2025-10-16 22:13:00 +02:00
async fn root_container(&self) -> Result<Container> {
Ok(self.build_root_container())
2025-10-16 22:13:00 +02:00
}
async fn browse(&self, object_id: &str) -> Result<BrowseResult> {
match object_id {
"0" => Ok(BrowseResult::Containers(vec![self.build_root_container()])),
"radio-paradise" => {
let containers = self.build_channel_containers().await;
Ok(BrowseResult::Containers(containers))
}
_ => {
if let Some(channel_id) = parse_channel_container_id(object_id) {
let channel = self
.channel_state(channel_id)
.ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?;
self.ensure_channel_ready(channel.clone()).await?;
let len = channel.playlist.len().await;
let items = channel.playlist.as_objects(0, len, None).await;
Ok(BrowseResult::Items(items))
} else {
Err(MusicSourceError::ObjectNotFound(object_id.to_string()))
2025-10-18 21:39:12 +02:00
}
}
2025-10-16 22:13:00 +02:00
}
}
async fn resolve_uri(&self, object_id: &str) -> Result<String> {
let (channel_id, _, _) = parse_track_identifier(object_id)
.ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?;
let channel = self
.channel_state(channel_id)
.ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?;
channel.cache_manager.resolve_uri(object_id).await
2025-10-16 22:13:00 +02:00
}
fn supports_fifo(&self) -> bool {
false
}
2025-10-16 22:13:00 +02:00
async fn append_track(&self, _track: Item) -> Result<()> {
Err(MusicSourceError::FifoNotSupported)
2025-10-16 22:13:00 +02:00
}
async fn remove_oldest(&self) -> Result<Option<Item>> {
Err(MusicSourceError::FifoNotSupported)
2025-10-16 22:13:00 +02:00
}
async fn update_id(&self) -> u32 {
let mut max_id = 0;
for descriptor in CHANNELS.iter() {
if let Some(channel) = self.channel_state(descriptor.id) {
let id = channel.playlist.update_id().await;
max_id = max_id.max(id);
}
}
max_id
2025-10-16 22:13:00 +02:00
}
async fn last_change(&self) -> Option<SystemTime> {
let mut latest: Option<SystemTime> = None;
for descriptor in CHANNELS.iter() {
if let Some(channel) = self.channel_state(descriptor.id) {
let change = channel.playlist.last_change().await;
latest = Some(match latest {
Some(current) if change <= current => current,
_ => change,
});
}
}
latest
2025-10-16 22:13:00 +02:00
}
async fn get_items(&self, offset: usize, count: usize) -> Result<Vec<Item>> {
let mut all_items = Vec::new();
for descriptor in CHANNELS.iter() {
if let Some(channel) = self.channel_state(descriptor.id) {
self.ensure_channel_ready(channel.clone()).await?;
let len = channel.playlist.len().await;
let mut items = channel.playlist.as_objects(0, len, None).await;
all_items.append(&mut items);
}
}
let total = all_items.len();
if offset >= total {
return Ok(Vec::new());
}
let end = if count == 0 {
total
} else {
(offset + count).min(total)
};
Ok(all_items
.into_iter()
.skip(offset)
.take(end - offset)
.collect())
2025-10-16 22:13:00 +02:00
}
async fn search(&self, _query: &str) -> Result<BrowseResult> {
Err(MusicSourceError::SearchNotSupported)
}
2025-10-17 08:19:10 +02:00
fn capabilities(&self) -> pmosource::SourceCapabilities {
pmosource::SourceCapabilities {
supports_fifo: false,
2025-10-17 08:19:10 +02:00
supports_search: false,
supports_favorites: false,
supports_playlists: false,
supports_user_content: false,
supports_high_res_audio: true,
max_sample_rate: Some(96_000),
2025-10-17 08:19:10 +02:00
supports_multiple_formats: true,
supports_advanced_search: false,
supports_pagination: true,
2025-10-17 08:19:10 +02:00
}
}
async fn get_available_formats(&self, _object_id: &str) -> Result<Vec<pmosource::AudioFormat>> {
use pmosource::AudioFormat;
Ok(vec![
AudioFormat {
format_id: "mp3-128".to_string(),
mime_type: "audio/mpeg".to_string(),
sample_rate: Some(44100),
bit_depth: None,
bitrate: Some(128),
channels: Some(2),
},
AudioFormat {
format_id: "aac-64".to_string(),
mime_type: "audio/aac".to_string(),
sample_rate: Some(44100),
bit_depth: None,
bitrate: Some(64),
channels: Some(2),
},
AudioFormat {
format_id: "aac-128".to_string(),
mime_type: "audio/aac".to_string(),
sample_rate: Some(44100),
bit_depth: None,
bitrate: Some(128),
channels: Some(2),
},
AudioFormat {
format_id: "aac-320".to_string(),
mime_type: "audio/aac".to_string(),
sample_rate: Some(44100),
bit_depth: None,
bitrate: Some(320),
channels: Some(2),
},
AudioFormat {
format_id: "flac".to_string(),
mime_type: "audio/flac".to_string(),
sample_rate: Some(44100),
bit_depth: Some(16),
bitrate: None,
channels: Some(2),
},
])
}
async fn get_cache_status(&self, object_id: &str) -> Result<pmosource::CacheStatus> {
let (channel_id, _, _) = parse_track_identifier(object_id)
.ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?;
let channel = self
.channel_state(channel_id)
.ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?;
channel.cache_manager.get_cache_status(object_id).await
2025-10-17 08:19:10 +02:00
}
async fn cache_item(&self, object_id: &str) -> Result<pmosource::CacheStatus> {
let (channel_id, _, _) = parse_track_identifier(object_id)
2025-10-17 23:45:01 +02:00
.ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?;
let channel = self
.channel_state(channel_id)
2025-10-17 23:45:01 +02:00
.ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?;
self.ensure_channel_ready(channel.clone()).await?;
channel.cache_manager.get_cache_status(object_id).await
2025-10-17 08:19:10 +02:00
}
async fn browse_paginated(
&self,
object_id: &str,
offset: usize,
limit: usize,
) -> Result<BrowseResult> {
match object_id {
"0" => {
if offset == 0 {
Ok(BrowseResult::Containers(vec![self.build_root_container()]))
} else {
Ok(BrowseResult::Containers(Vec::new()))
}
}
"radio-paradise" => {
let containers = self.build_channel_containers().await;
let total = containers.len();
if offset >= total {
return Ok(BrowseResult::Containers(Vec::new()));
}
let end = if limit == 0 {
total
} else {
(offset + limit).min(total)
};
Ok(BrowseResult::Containers(
containers
.into_iter()
.skip(offset)
.take(end - offset)
.collect(),
))
}
_ => {
if let Some(channel_id) = parse_channel_container_id(object_id) {
let channel = self
.channel_state(channel_id)
.ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?;
self.ensure_channel_ready(channel.clone()).await?;
let len = channel.playlist.len().await;
if offset >= len {
return Ok(BrowseResult::Items(Vec::new()));
}
let count = if limit == 0 {
len - offset
} else {
limit.min(len - offset)
};
let items = channel.playlist.as_objects(offset, count, None).await;
Ok(BrowseResult::Items(items))
} else {
Err(MusicSourceError::ObjectNotFound(object_id.to_string()))
}
}
2025-10-17 08:19:10 +02:00
}
}
async fn get_item_count(&self, object_id: &str) -> Result<usize> {
match object_id {
"0" => Ok(1),
"radio-paradise" => Ok(CHANNELS.len()),
_ => {
if let Some(channel_id) = parse_channel_container_id(object_id) {
let channel = self
.channel_state(channel_id)
.ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?;
self.ensure_channel_ready(channel.clone()).await?;
Ok(channel.playlist.len().await)
} else {
Err(MusicSourceError::ObjectNotFound(object_id.to_string()))
}
}
2025-10-17 08:19:10 +02:00
}
}
async fn statistics(&self) -> Result<pmosource::SourceStatistics> {
let mut total_items = 0usize;
let mut cached_items = 0usize;
for descriptor in CHANNELS.iter() {
if let Some(channel) = self.channel_state(descriptor.id) {
total_items += channel.playlist.len().await;
let stats = channel.cache_manager.statistics().await;
cached_items += stats.cached_tracks;
}
}
2025-10-17 08:19:10 +02:00
2025-10-17 23:45:01 +02:00
Ok(pmosource::SourceStatistics {
total_items: Some(total_items),
total_containers: Some(CHANNELS.len() + 1),
cached_items: Some(cached_items),
cache_size_bytes: None,
2025-10-17 23:45:01 +02:00
})
2025-10-17 08:19:10 +02:00
}
2025-10-16 22:00:35 +02:00
}