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

Changement du mécanisme d'attention sur les channels Radio Paradise.

Browse Source
This commit is contained in:
Eric Coissac
2025-11-29 14:19:16 +01:00
parent cf3f0afde4
commit 0a03f72467
44 changed files with 564 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
pmoparadise/Cargo.toml
View File

@@ -29,6 +29,7 @@ hex = "0.4"
tokio-util = { version = "0.7", features = ["io"] }
async-stream = "0.3"
rusqlite = { version = "0.37", features = ["bundled"] }
once_cell = "1.20"
# Gestion des erreurs
thiserror = "2.0.17"

32
pmoparadise/examples/single_channel_server.rs
View File

@@ -16,16 +16,18 @@ use axum::{
routing::get,
Json, Router,
};
use pmoaudio_ext::StreamingSinkOptions;
use pmoaudiocache::{
new_cache_with_consolidation as new_audio_cache,
register_audio_cache as register_global_audio_cache,
};
use pmocovers::{new_cache_with_consolidation as new_cover_cache, register_cover_cache, Cache as CoverCache};
use pmocovers::{
new_cache_with_consolidation as new_cover_cache, register_cover_cache, Cache as CoverCache,
};
use pmoparadise::{
channels::{ChannelDescriptor, ALL_CHANNELS},
ParadiseHistoryBuilder, ParadiseStreamChannel, ParadiseStreamChannelConfig,
};
use pmoaudio_ext::StreamingSinkOptions;
use pmoplaylist::register_audio_cache as register_playlist_audio_cache;
use std::{fs, net::SocketAddr, sync::Arc};
use tokio::net::TcpListener;
@@ -44,9 +46,7 @@ async fn main() -> anyhow::Result<()> {
let env_filter = tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info"));
tracing_subscriber::fmt()
.with_env_filter(env_filter)
.init();
tracing_subscriber::fmt().with_env_filter(env_filter).init();
let descriptor = pick_descriptor(std::env::args().nth(1))?;
info!(
@@ -206,22 +206,16 @@ async fn get_cover(
) -> Result<Response, StatusCode> {
// Récupérer le chemin de la cover depuis le cache
// Le cache retourne un PathBuf pointant vers le fichier .webp
let cover_path = state
.cover_cache
.get(&pk)
.await
.map_err(|e| {
tracing::error!("Failed to get cover path for {}: {}", pk, e);
StatusCode::NOT_FOUND
})?;
let cover_path = state.cover_cache.get(&pk).await.map_err(|e| {
tracing::error!("Failed to get cover path for {}: {}", pk, e);
StatusCode::NOT_FOUND
})?;
// Lire le fichier
let cover_data = tokio::fs::read(&cover_path)
.await
.map_err(|e| {
tracing::error!("Failed to read cover file {:?}: {}", cover_path, e);
StatusCode::NOT_FOUND
})?;
let cover_data = tokio::fs::read(&cover_path).await.map_err(|e| {
tracing::error!("Failed to read cover file {:?}: {}", cover_path, e);
StatusCode::NOT_FOUND
})?;
Response::builder()
.status(StatusCode::OK)

6
pmoparadise/src/pmoserver_ext.rs
View File

@@ -429,11 +429,7 @@ async fn get_cover_url(
})?;
let song = block.get_song(song_index).ok_or_else(|| {
tracing::warn!(
"Song index {} not found in block {}",
song_index,
event_id
);
tracing::warn!("Song index {} not found in block {}", song_index, event_id);
StatusCode::NOT_FOUND
})?;

153
pmoparadise/src/source.rs
View File

@@ -4,20 +4,26 @@
//! exposing live streams and historical playlists for all 4 channels.
use crate::channels::{ChannelDescriptor, ALL_CHANNELS};
use std::fmt;
use pmosource::pmodidl::{Container, Item, Resource};
use pmosource::{
async_trait, AudioFormat, BrowseResult, MusicSource, MusicSourceError, Result,
SourceCapabilities,
};
use std::fmt;
use std::sync::Arc;
use std::time::SystemTime;
use std::time::{Duration, Instant, SystemTime};
use tokio::sync::RwLock;
/// Default Radio Paradise image (embedded in binary)
const DEFAULT_IMAGE: &[u8] = include_bytes!("../assets/default.webp");
#[cfg(feature = "playlist")]
const LIVE_PLAYLIST_MIN_READY_ITEMS: usize = 5;
#[cfg(feature = "playlist")]
const LIVE_PLAYLIST_READY_TIMEOUT: Duration = Duration::from_secs(10);
#[cfg(feature = "playlist")]
const LIVE_PLAYLIST_READY_POLL: Duration = Duration::from_millis(200);
/// RadioParadiseSource - UPnP ContentDirectory source for Radio Paradise
///
/// Provides access to:
@@ -142,14 +148,21 @@ impl RadioParadiseSource {
ALL_CHANNELS
.iter()
.find(|ch| pid.ends_with(ch.slug))
.map(|ch| vec![format!("radio-paradise:channel:{}:history", ch.slug)])
.map(|ch| {
vec![format!("radio-paradise:channel:{}:history", ch.slug)]
})
.unwrap_or_default()
} else {
// live playlist -> container liveplaylist
ALL_CHANNELS
.iter()
.find(|ch| pid.ends_with(ch.slug))
.map(|ch| vec![format!("radio-paradise:channel:{}:liveplaylist", ch.slug)])
.map(|ch| {
vec![format!(
"radio-paradise:channel:{}:liveplaylist",
ch.slug
)]
})
.unwrap_or_default()
};
@@ -181,7 +194,10 @@ impl RadioParadiseSource {
match response.json::<serde_json::Value>().await {
Ok(json) => {
// Parse metadata from JSON and create an Item
let title = json["title"].as_str().unwrap_or("Unknown Title").to_string();
let title = json["title"]
.as_str()
.unwrap_or("Unknown Title")
.to_string();
let artist = json["artist"].as_str().map(|s| s.to_string());
let album = json["album"].as_str().map(|s| s.to_string());
let year = json["year"].as_u64().map(|y| y as u32);
@@ -201,10 +217,12 @@ impl RadioParadiseSource {
.or_else(|| json["duration"].as_f64())
.map(|secs| {
let total_secs = secs as u64;
format!("{}:{:02}:{:02}",
format!(
"{}:{:02}:{:02}",
total_secs / 3600,
(total_secs % 3600) / 60,
total_secs % 60)
total_secs % 60
)
});
// Create the item with current metadata
@@ -254,6 +272,42 @@ impl RadioParadiseSource {
format!("radio-paradise-live-{}", slug)
}
#[cfg(feature = "playlist")]
async fn wait_for_live_playlist_ready(&self, slug: &str) -> Result<()> {
let playlist_id = Self::live_playlist_id(slug);
let manager = pmoplaylist::PlaylistManager();
let reader = manager.get_read_handle(&playlist_id).await.map_err(|e| {
MusicSourceError::BrowseError(format!(
"Failed to get live playlist {}: {}",
playlist_id, e
))
})?;
let start = Instant::now();
loop {
match reader.remaining().await {
Ok(count) if count >= LIVE_PLAYLIST_MIN_READY_ITEMS => return Ok(()),
Ok(_) => {}
Err(e) => {
return Err(MusicSourceError::BrowseError(format!(
"Failed to inspect live playlist {}: {}",
playlist_id, e
)));
}
}
if start.elapsed() >= LIVE_PLAYLIST_READY_TIMEOUT {
tracing::warn!(
"Timeout waiting for live playlist {} to reach {} items",
playlist_id,
LIVE_PLAYLIST_MIN_READY_ITEMS
);
return Ok(());
}
tokio::time::sleep(LIVE_PLAYLIST_READY_POLL).await;
}
}
/// Get channel descriptor by slug
fn get_channel_by_slug(slug: &str) -> Option<&'static ChannelDescriptor> {
ALL_CHANNELS.iter().find(|ch| ch.slug == slug)
@@ -380,7 +434,10 @@ impl RadioParadiseSource {
/// Build a history container with accurate child count from playlist
#[cfg(feature = "playlist")]
async fn build_history_container_with_count(&self, descriptor: &ChannelDescriptor) -> Container {
async fn build_history_container_with_count(
&self,
descriptor: &ChannelDescriptor,
) -> Container {
let mut container = self.build_history_container(descriptor);
// Try to get actual count from playlist
@@ -466,32 +523,48 @@ impl RadioParadiseSource {
_offset: usize,
count: usize,
) -> Result<Vec<Item>> {
#[cfg(all(feature = "playlist", feature = "pmoaudio"))]
if let Some(descriptor) = Self::get_channel_by_slug(slug) {
if let Some(manager) = crate::stream_channel::get_global_channel_manager() {
if let Err(e) = manager.prefetch_until_horizon(descriptor.id).await {
tracing::warn!(
"Failed to prefetch live playlist for {}: {}",
descriptor.slug,
e
);
}
}
}
#[cfg(feature = "playlist")]
if let Err(e) = self.wait_for_live_playlist_ready(slug).await {
tracing::warn!(
"Failed to wait for live playlist readiness on {}: {}",
slug,
e
);
}
let playlist_id = Self::live_playlist_id(slug);
let manager = pmoplaylist::PlaylistManager();
let reader = manager.get_read_handle(&playlist_id).await.map_err(|e| {
MusicSourceError::BrowseError(format!("Failed to get live playlist {}: {}", playlist_id, e))
MusicSourceError::BrowseError(format!(
"Failed to get live playlist {}: {}",
playlist_id, e
))
})?;
let mut items = reader.to_items(count).await.map_err(|e| {
MusicSourceError::BrowseError(format!(
"Failed to read live playlist entries: {}",
e
))
MusicSourceError::BrowseError(format!("Failed to read live playlist entries: {}", e))
})?;
for item in items.iter_mut() {
// Ajuster id/parent/url pour coller au schéma Radio Paradise
if let Some(resource) = item.resources.first_mut() {
if let Some(pk) = resource.url.split('/').last() {
item.id = format!(
"radio-paradise:channel:{}:liveplaylist:track:{}",
slug, pk
);
item.parent_id = format!(
"radio-paradise:channel:{}:liveplaylist",
slug
);
item.id = format!("radio-paradise:channel:{}:liveplaylist:track:{}", slug, pk);
item.parent_id = format!("radio-paradise:channel:{}:liveplaylist", slug);
if resource.url.starts_with('/') {
resource.url = format!("{}{}", self.base_url, resource.url);
@@ -519,10 +592,7 @@ impl RadioParadiseSource {
#[cfg(feature = "playlist")]
async fn get_live_playlist_item(&self, slug: &str, pk: &str) -> Result<Item> {
let items = self.get_live_playlist_items(slug, 0, 1000).await?;
let expected_id = format!(
"radio-paradise:channel:{}:liveplaylist:track:{}",
slug, pk
);
let expected_id = format!("radio-paradise:channel:{}:liveplaylist:track:{}", slug, pk);
for item in items {
if item.id == expected_id {
return Ok(item);
@@ -533,7 +603,6 @@ impl RadioParadiseSource {
pk
)))
}
}
/// Types of object IDs in the Radio Paradise source
@@ -619,7 +688,8 @@ impl MusicSource for RadioParadiseSource {
#[cfg(feature = "playlist")]
{
let history_container = self.build_history_container_with_count(descriptor).await;
let history_container =
self.build_history_container_with_count(descriptor).await;
let items = self.get_history_items(&slug, 0, 100).await?;
Ok(BrowseResult::Mixed {
containers: vec![history_container],
@@ -821,14 +891,14 @@ impl MusicSource for RadioParadiseSource {
for mut item in items {
if let Some(resource) = item.resources.first_mut() {
if let Some(pk2) = resource.url.split('/').last() {
item.id =
format!("radio-paradise:channel:{}:history:track:{}", slug, pk2);
item.parent_id =
format!("radio-paradise:channel:{}:history", slug);
item.id = format!(
"radio-paradise:channel:{}:history:track:{}",
slug, pk2
);
item.parent_id = format!("radio-paradise:channel:{}:history", slug);
if resource.url.starts_with('/') {
resource.url =
format!("{}{}", self.base_url, resource.url);
resource.url = format!("{}{}", self.base_url, resource.url);
}
}
}
@@ -839,7 +909,8 @@ impl MusicSource for RadioParadiseSource {
}
// Find the item matching this pk in the item ID
let expected_id = format!("radio-paradise:channel:{}:history:track:{}", slug, pk);
let expected_id =
format!("radio-paradise:channel:{}:history:track:{}", slug, pk);
for item in adjusted {
if item.id == expected_id {
return Ok(item);
@@ -887,10 +958,8 @@ impl MusicSource for RadioParadiseSource {
"radio-paradise:channel:{}:liveplaylist:track:{}",
slug, pk2
);
item.parent_id = format!(
"radio-paradise:channel:{}:liveplaylist",
slug
);
item.parent_id =
format!("radio-paradise:channel:{}:liveplaylist", slug);
if resource.url.starts_with('/') {
resource.url = format!("{}{}", self.base_url, resource.url);
@@ -910,10 +979,8 @@ impl MusicSource for RadioParadiseSource {
item.album_art = Some(self.default_cover_url());
}
let expected_id = format!(
"radio-paradise:channel:{}:liveplaylist:track:{}",
slug, pk
);
let expected_id =
format!("radio-paradise:channel:{}:liveplaylist:track:{}", slug, pk);
if item.id == expected_id {
return Ok(item);
}

114
pmoparadise/src/stream_channel.rs
View File

@@ -21,12 +21,13 @@ use crate::{
models::{Block, EventId},
playlist_feeder::RadioParadisePlaylistFeeder,
};
use anyhow::{anyhow, Result};
use anyhow::{anyhow, Context as AnyhowContext, Result};
use once_cell::sync::OnceCell;
use pmoaudio::{AudioError, AudioPipelineNode};
use pmoaudio_ext::{
FlacClientStream, IcyClientStream, MetadataSnapshot, OggFlacClientStream, OggFlacStreamHandle,
PlaylistSource, StreamHandle, StreamingFlacSink, StreamingOggFlacSink, TrackBoundaryCoverNode,
StreamingSinkOptions,
PlaylistSource, StreamHandle, StreamingFlacSink, StreamingOggFlacSink, StreamingSinkOptions,
TrackBoundaryCoverNode,
};
use pmoaudiocache::{get_audio_cache, Cache as AudioCache};
use pmocovers::{get_cover_cache, Cache as CoverCache};
@@ -203,9 +204,14 @@ impl ParadiseStreamChannel {
// Propager server_base_url dans les options pour que les encoders injectent les covers du cache
let mut config = config;
if let Some(ref base) = config.server_base_url {
config.flac_options =
config.flac_options.clone().with_server_base_url(Some(base.clone()));
config.ogg_options = config.ogg_options.clone().with_server_base_url(Some(base.clone()));
config.flac_options = config
.flac_options
.clone()
.with_server_base_url(Some(base.clone()));
config.ogg_options = config
.ogg_options
.clone()
.with_server_base_url(Some(base.clone()));
}
let cover_cache = cover_cache
.or_else(|| history.as_ref().map(|opts| opts.cover_cache.clone()))
@@ -312,6 +318,7 @@ impl ParadiseStreamChannel {
activity_notify: Notify::new(),
stop_token,
current_block: Mutex::new(None),
prefetch_lock: Mutex::new(()),
});
let pipeline_state = state.clone();
@@ -496,6 +503,12 @@ impl Drop for ParadiseStreamChannel {
const MAX_BLOCK_LEAD: Duration = Duration::from_secs(3600);
const BLOCK_LEAD_CHECK_CHUNK: Duration = Duration::from_secs(300);
const LIVE_PREFETCH_MIN_TRACKS: usize = 5;
const LIVE_PREFETCH_TIMEOUT: Duration = Duration::from_secs(10);
const LIVE_PREFETCH_POLL_INTERVAL: Duration = Duration::from_millis(200);
const LIVE_PREFETCH_MAX_BLOCKS: usize = 4;
static GLOBAL_CHANNEL_MANAGER: OnceCell<std::sync::Weak<ParadiseChannelManager>> = OnceCell::new();
struct ChannelState {
descriptor: ChannelDescriptor,
@@ -510,6 +523,7 @@ struct ChannelState {
activity_notify: Notify,
stop_token: CancellationToken,
current_block: Mutex<Option<EventId>>,
prefetch_lock: Mutex<()>,
}
impl ChannelState {
@@ -580,6 +594,66 @@ impl ChannelState {
}
}
fn live_playlist_id(&self) -> String {
format!("radio-paradise-live-{}", self.descriptor.slug)
}
async fn prefetch_until_horizon(&self) -> Result<()> {
let _guard = self.prefetch_lock.lock().await;
let playlist_id = self.live_playlist_id();
let manager = PlaylistManager::get();
let reader = manager
.get_read_handle(&playlist_id)
.await
.with_context(|| format!("Failed to get live playlist {}", playlist_id))?;
let start = Instant::now();
let mut next_event: Option<EventId> = None;
let mut attempts = 0usize;
loop {
let available = reader
.remaining()
.await
.with_context(|| format!("Failed to inspect playlist {}", playlist_id))?;
if available >= LIVE_PREFETCH_MIN_TRACKS {
return Ok(());
}
if start.elapsed() >= LIVE_PREFETCH_TIMEOUT {
warn!(
"Prefetch timeout for channel {} ({} tracks available)",
self.descriptor.display_name, available
);
return Ok(());
}
if attempts >= LIVE_PREFETCH_MAX_BLOCKS {
warn!(
"Prefetch block limit reached for channel {} ({} tracks available)",
self.descriptor.display_name, available
);
return Ok(());
}
match self.client.get_block(next_event).await {
Ok(block) => {
attempts += 1;
next_event = Some(block.end_event);
self.feeder.push_block_id(block.event).await;
}
Err(e) => {
warn!(
"Failed to fetch block during prefetch for channel {}: {}",
self.descriptor.display_name, e
);
return Ok(());
}
}
tokio::time::sleep(LIVE_PREFETCH_POLL_INTERVAL).await;
}
}
async fn set_current_block(&self, event_id: EventId) {
let mut guard = self.current_block.lock().await;
*guard = Some(event_id);
@@ -866,12 +940,7 @@ impl ParadiseChannelManager {
);
let channel = match tokio::time::timeout(
Duration::from_secs(20),
ParadiseStreamChannel::new(
descriptor,
config,
cover_cache.clone(),
history_opts,
),
ParadiseStreamChannel::new(descriptor, config, cover_cache.clone(), history_opts),
)
.await
{
@@ -918,4 +987,25 @@ impl ParadiseChannelManager {
pub fn iter(&self) -> impl Iterator<Item = &Arc<ParadiseStreamChannel>> {
self.channels.values()
}
pub async fn prefetch_until_horizon(&self, channel_id: u8) -> Result<()> {
let channel = self
.get(channel_id)
.ok_or_else(|| anyhow!("Unknown channel id {}", channel_id))?;
channel.prefetch_until_horizon().await
}
}
pub fn register_global_channel_manager(manager: Arc<ParadiseChannelManager>) {
let _ = GLOBAL_CHANNEL_MANAGER.set(Arc::downgrade(&manager));
}
pub fn get_global_channel_manager() -> Option<Arc<ParadiseChannelManager>> {
GLOBAL_CHANNEL_MANAGER.get().and_then(|weak| weak.upgrade())
}
impl ParadiseStreamChannel {
pub async fn prefetch_until_horizon(&self) -> Result<()> {
self.state.prefetch_until_horizon().await
}
}