diff --git a/.pmomusic.yml.example b/.pmomusic.yml.example new file mode 100644 index 00000000..3f7c1c0d --- /dev/null +++ b/.pmomusic.yml.example @@ -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: your-email@example.com + password: 'YOUR_PASSWORD_HERE' +devices: + mediarenderer: + pmo_mediarenderer: + udn: 15a13316-daac-47f0-b64e-47e56f5e3b51 + mediaserver: + pmo_mediaserver: + udn: 23df0bfa-cfef-4724-b731-00f66fadf176 diff --git a/Cargo.lock b/Cargo.lock index a24b6de0..751c47c1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -441,6 +441,8 @@ dependencies = [ "cexpr", "clang-sys", "itertools 0.13.0", + "log", + "prettyplease", "proc-macro2", "quote", "regex", @@ -1093,31 +1095,6 @@ dependencies = [ "simd-adler32", ] -[[package]] -name = "ffmpeg-next" -version = "8.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d658424d233cbd993a972dd73a66ca733acd12a494c68995c9ac32ae1fe65b40" -dependencies = [ - "bitflags 2.10.0", - "ffmpeg-sys-next", - "libc", -] - -[[package]] -name = "ffmpeg-sys-next" -version = "8.0.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9bca20aa4ee774fe384c2490096c122b0b23cf524a9910add0686691003d797b" -dependencies = [ - "bindgen", - "cc", - "libc", - "num_cpus", - "pkg-config", - "vcpkg", -] - [[package]] name = "find-msvc-tools" version = "0.1.4" @@ -1495,12 +1472,6 @@ version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" -[[package]] -name = "hound" -version = "3.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "62adaabb884c94955b19907d60019f4e145d091c75345379e70d1ee696f7854f" - [[package]] name = "htmlescape" version = "0.3.1" @@ -2298,17 +2269,70 @@ dependencies = [ ] [[package]] -name = "netstat2" -version = "0.9.1" +name = "netlink-packet-core" +version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0faa3f4ad230fd2bf2a5dad71476ecbaeaed904b3c7e7e5b1f266c415c03761f" +checksum = "72724faf704479d67b388da142b186f916188505e7e0b26719019c525882eda4" dependencies = [ + "anyhow", + "byteorder", + "netlink-packet-utils", +] + +[[package]] +name = "netlink-packet-sock-diag" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a495cb1de50560a7cd12fdcf023db70eec00e340df81be31cedbbfd4aadd6b76" +dependencies = [ + "anyhow", "bitflags 1.3.2", "byteorder", "libc", + "netlink-packet-core", + "netlink-packet-utils", + "smallvec", +] + +[[package]] +name = "netlink-packet-utils" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ede8a08c71ad5a95cdd0e4e52facd37190977039a4704eb82a283f713747d34" +dependencies = [ + "anyhow", + "byteorder", + "paste", + "thiserror 1.0.69", +] + +[[package]] +name = "netlink-sys" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "16c903aa70590cb93691bf97a767c8d1d6122d2cc9070433deb3bbf36ce8bd23" +dependencies = [ + "bytes", + "libc", + "log", +] + +[[package]] +name = "netstat2" +version = "0.11.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "496f264d3ead4870d6b366deb9d20597592d64aac2a907f3e7d07c2325ba4663" +dependencies = [ + "bindgen", + "bitflags 2.10.0", + "byteorder", + "netlink-packet-core", + "netlink-packet-sock-diag", + "netlink-packet-utils", + "netlink-sys", "num-derive 0.3.3", "num-traits", - "thiserror 1.0.69", + "thiserror 2.0.17", ] [[package]] @@ -2850,14 +2874,16 @@ dependencies = [ "bytes", "chrono", "claxon", - "ffmpeg-next", "flacenc", "futures", + "futures-util", "hex", - "hound", + "pmoaudio", "pmoaudiocache", "pmoconfig", "pmocovers", + "pmoflac", + "pmometadata", "pmoplaylist", "pmoserver", "pmosource", @@ -2868,7 +2894,6 @@ dependencies = [ "serde_yaml", "sha2", "symphonia", - "tempfile", "thiserror 2.0.17", "tokio", "tokio-test", @@ -3072,6 +3097,16 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "prettyplease" +version = "0.2.37" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" +dependencies = [ + "proc-macro2", + "syn 2.0.108", +] + [[package]] name = "proc-macro2" version = "1.0.103" diff --git a/SECURITY_CONFIG.md b/SECURITY_CONFIG.md new file mode 100644 index 00000000..68e9ca78 --- /dev/null +++ b/SECURITY_CONFIG.md @@ -0,0 +1,27 @@ +# Configuration Sécurisée + +## Configuration de PMOMusic + +Le fichier `.pmomusic.yml` contient des informations sensibles (mots de passe, identifiants). + +### Installation + +1. Copiez le fichier exemple : + ```bash + cp .pmomusic.yml.example .pmomusic.yml + ``` + +2. Éditez `.pmomusic.yml` et remplacez les valeurs par vos véritables identifiants : + - `accounts.qobuz.username` : votre email Qobuz + - `accounts.qobuz.password` : votre mot de passe Qobuz + +3. **Important** : Ne commitez JAMAIS le fichier `.pmomusic.yml` dans git ! + - Il est déjà dans `.gitignore` + - Utilisez des variables d'environnement pour la production + +## Variables d'environnement (recommandé pour production) + +```bash +export QOBUZ_USERNAME="votre-email@example.com" +export QOBUZ_PASSWORD="votre-mot-de-passe" +``` diff --git a/pmoaudio-ext/src/sinks/flac_cache_sink.rs b/pmoaudio-ext/src/sinks/flac_cache_sink.rs index 2d349d66..f1924a9c 100755 --- a/pmoaudio-ext/src/sinks/flac_cache_sink.rs +++ b/pmoaudio-ext/src/sinks/flac_cache_sink.rs @@ -251,8 +251,6 @@ impl NodeLogic for FlacCacheSinkLogic { pub struct FlacCacheSink { inner: Node, - #[cfg(feature = "playlist")] - playlist_handle_pending: Option>, } impl FlacCacheSink { @@ -297,8 +295,6 @@ impl FlacCacheSink { let logic = FlacCacheSinkLogic::new(cache, covers, collection, encoder_options, 8); Self { inner: Node::new_with_input(logic, channel_size), - #[cfg(feature = "playlist")] - playlist_handle_pending: None, } } @@ -309,7 +305,7 @@ impl FlacCacheSink { /// * `handle` - WriteHandle de la playlist qui recevra les pk des tracks #[cfg(feature = "playlist")] pub fn register_playlist(&mut self, handle: pmoplaylist::WriteHandle) { - self.playlist_handle_pending = Some(Arc::new(handle)); + self.inner.logic_mut().set_playlist_handle(Arc::new(handle)); } } @@ -649,16 +645,7 @@ impl AudioPipelineNode for FlacCacheSink { panic!("FlacCacheSink is a terminal sink and cannot have children"); } - async fn run(mut self: Box, stop_token: CancellationToken) -> Result<(), AudioError> { - // Transférer le playlist_handle_pending à la logique si présent - #[cfg(feature = "playlist")] - if let Some(handle) = self.playlist_handle_pending.take() { - // FIXME: Node devrait exposer une méthode logic_mut() pour permettre - // la configuration post-construction. Pour l'instant, on ignore ce handle. - // L'utilisateur devra configurer la playlist avant construction. - let _ = handle; - } - + async fn run(self: Box, stop_token: CancellationToken) -> Result<(), AudioError> { Box::new(self.inner).run(stop_token).await } } diff --git a/pmoaudio/src/pipeline.rs b/pmoaudio/src/pipeline.rs index 5977b788..7cffac38 100755 --- a/pmoaudio/src/pipeline.rs +++ b/pmoaudio/src/pipeline.rs @@ -482,6 +482,15 @@ impl Node { pub fn logic(&self) -> &L { &self.logic } + + /// Retourne une référence mutable vers la logique métier du nœud + /// + /// Permet de configurer la logique après construction mais avant run(). + /// Utile pour définir des options qui ne peuvent pas être connues + /// au moment de la construction du nœud. + pub fn logic_mut(&mut self) -> &mut L { + &mut self.logic + } } #[async_trait::async_trait] diff --git a/pmocache/src/cache_trait.rs b/pmocache/src/cache_trait.rs index 536c3a0b..07909f2d 100644 --- a/pmocache/src/cache_trait.rs +++ b/pmocache/src/cache_trait.rs @@ -1,5 +1,4 @@ use anyhow::Result; -use sha1::{Digest, Sha1}; use std::{ path::{Path, PathBuf}, sync::Arc, diff --git a/pmoparadise/Cargo.toml b/pmoparadise/Cargo.toml index 546eacf3..fe7b770c 100644 --- a/pmoparadise/Cargo.toml +++ b/pmoparadise/Cargo.toml @@ -51,13 +51,6 @@ symphonia = { version = "0.5", features = ["all"] } # Audio decoding - claxon for FLAC streaming claxon = "0.4" -# FFmpeg for progressive streaming (decoding + encoding) - optional -ffmpeg-next = { version = "8.0", optional = true } - -# Per-track feature dependencies -hound = { version = "3.5", optional = true } -tempfile = { version = "3.8", optional = true } - # Common music source traits pmosource = { path = "../pmosource" } @@ -74,12 +67,16 @@ pmoserver = { path = "../pmoserver", optional = true } utoipa = { version = "5.4.0", optional = true } axum = { version = "0.8.4", optional = true } +# pmoaudio node support +pmoaudio = { path = "../pmoaudio", optional = true } +pmoflac = { path = "../pmoflac", optional = true } +pmometadata = { path = "../pmometadata", optional = true } +futures-util = { version = "0.3", optional = true } + [features] default = ["metadata-only", "pmoconfig"] # Mode métadonnées seules (pas de décodage FLAC) metadata-only = [] -# Active l'extraction par-track (WAV export, etc.) -per-track = ["dep:hound", "dep:tempfile"] # Active l'API REST pmoserver pmoserver = ["dep:pmoserver", "dep:utoipa", "dep:axum", "server"] # Feature pour activer le support serveur (cache registry) @@ -88,8 +85,8 @@ server = ["pmosource/server", "pmoconfig"] pmoconfig = ["dep:pmoconfig"] # Feature cache (deprecated - toujours actif maintenant) cache = [] -# Active le streaming progressif avec FFmpeg (latence réduite) -ffmpeg = ["dep:ffmpeg-next"] +# Active le support pmoaudio node (RadioParadiseStreamSource) +pmoaudio = ["dep:pmoaudio", "dep:pmoflac", "dep:pmometadata", "dep:futures-util"] [dev-dependencies] # Tests @@ -104,17 +101,3 @@ pmoaudiocache = { path = "../pmoaudiocache" } [[example]] name = "now_playing" path = "examples/now_playing.rs" - -[[example]] -name = "stream_block" -path = "examples/stream_block.rs" - -[[example]] -name = "extract_track" -path = "examples/extract_track.rs" -required-features = ["per-track"] - -[[example]] -name = "with_cache" -path = "examples/with_cache.rs" -required-features = ["cache"] diff --git a/pmoparadise/RADIO_PARADISE_STREAM_SOURCE.md b/pmoparadise/RADIO_PARADISE_STREAM_SOURCE.md new file mode 100644 index 00000000..35bb5bf5 --- /dev/null +++ b/pmoparadise/RADIO_PARADISE_STREAM_SOURCE.md @@ -0,0 +1,244 @@ +# RadioParadiseStreamSource - Documentation Technique + +## Vue d'ensemble + +`RadioParadiseStreamSource` est un nœud source pour `pmoaudio` qui télécharge et décode les blocs FLAC de Radio Paradise en temps réel, avec gestion automatique des transitions entre pistes (TrackBoundary). + +## Architecture + +### Pattern Node + +Suit l'architecture séparée logique/pipeline de `pmoaudio` : + +``` +RadioParadiseStreamSource (wrapper) + └── Node + └── RadioParadiseStreamSourceLogic (logique métier) +``` + +### RadioParadiseStreamSourceLogic + +Responsabilités : +- **File d'attente** : `VecDeque` pour les blocks à télécharger +- **Cache anti-redondance** : `VecDeque` pour 10 blocs récents (FIFO) +- **Téléchargement** : Fetch bloc FLAC (bitrate=4 uniquement) +- **Décodage** : Stream FLAC via `pmoflac::decode_audio_stream` +- **Timing** : Calcul précis pour insertion TrackBoundary + +## Flux d'exécution + +``` +┌─────────────────────────────────────────────────────────┐ +│ 1. Attente block ID (timeout 3s) │ +│ └─> VecDeque::pop_front() │ +└─────────────────────────────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────┐ +│ 2. Vérification cache │ +│ └─> VecDeque::contains(&event_id) │ +└─────────────────────────────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────┐ +│ 3. Téléchargement métadonnées │ +│ └─> client.get_block(event_id) │ +└─────────────────────────────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────┐ +│ 4. Téléchargement FLAC (bitrate=4) │ +│ └─> client.download_block_file(&block, 4) │ +└─────────────────────────────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────┐ +│ 5. Décodage streaming │ +│ └─> pmoflac::decode_audio_stream(reader) │ +└─────────────────────────────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────┐ +│ 6. Découpage en chunks │ +│ └─> pcm_to_audio_chunk(pcm, sr, bps) │ +└─────────────────────────────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────┐ +│ 7. Insertion TrackBoundary (timing sample-based) │ +│ └─> elapsed_ms = (total_samples * 1000) / sr │ +└─────────────────────────────────────────────────────────┘ +``` + +## Timing TrackBoundary + +### Algorithme + +```rust +let elapsed_ms = (total_samples * 1000) / sample_rate as u64; + +if elapsed_ms >= song.elapsed { + // Envoyer TrackBoundary AVANT le chunk (même order) + send_track_boundary(*order, song, block).await; +} +``` + +### Exemple concret + +Bloc FLAC contenant 3 chansons : +- Song 0 : `elapsed = 0ms` +- Song 1 : `elapsed = 180000ms` (3min) +- Song 2 : `elapsed = 420000ms` (7min) + +Timeline : +``` +0ms 180000ms 420000ms +│ │ │ +Song 0 TrackBoundary TrackBoundary + └─> Song 1 └─> Song 2 +``` + +## SyncMarker Order + +**Règle** : TrackBoundary a le **même order** que le chunk suivant. + +```rust +// TrackBoundary order = 42 +AudioSegment::new_sync(42, SyncMarker::TrackBoundary { ... }) + +// Chunk suivant order = 42 +AudioSegment::new_audio(42, AudioChunk::I16(...)) +``` + +## Gestion du cache + +### Stratégie FIFO simple + +```rust +const RECENT_BLOCKS_CACHE_SIZE: usize = 10; + +fn mark_block_downloaded(&mut self, event_id: EventId) { + // Retirer tous les éléments excédentaires (garantit <= CACHE_SIZE) + while self.recent_blocks.len() >= RECENT_BLOCKS_CACHE_SIZE { + self.recent_blocks.pop_front(); + } + + // Puis ajouter le nouveau bloc + self.recent_blocks.push_back(event_id); +} +``` + +**Avantages VecDeque** : +- ✅ Ordre FIFO garanti (le plus ancien est toujours retiré) +- ✅ Simple et prévisible +- ✅ Robuste : `while` garantit exactement 10 éléments max, même en cas d'état anormal +- ✅ Ne dépasse jamais la capacité pré-allouée (retire avant d'ajouter) +- ✅ Pour 10 éléments, `contains()` en O(n) reste très performant + +## Support FLAC + +### Formats supportés + +- **16-bit** : `AudioChunk::I16` +- **24-bit** : `AudioChunk::I24` + +### Conversion PCM + +```rust +match bits_per_sample { + 16 => { + let samples: Vec = pcm_data + .chunks_exact(2) + .map(|chunk| i16::from_le_bytes([chunk[0], chunk[1]])) + .collect(); + AudioChunk::I16(...) + } + 24 => { + let samples: Vec = pcm_data + .chunks_exact(3) + .map(|chunk| { + let value = i32::from_le_bytes([chunk[0], chunk[1], chunk[2], 0]) >> 8; + I24::from_i32(value) + }) + .collect(); + AudioChunk::I24(...) + } +} +``` + +## Métadonnées + +### TrackMetadata + +Champs extraits de `Song` : +- `title` : Titre de la chanson +- `artist` : Artiste +- `album` : Album (optionnel) +- `year` : Année (optionnel) +- `cover_url` : URL de la pochette (async via tokio::spawn) + +### Gestion asynchrone du cover + +```rust +tokio::spawn(async move { + if let Ok(mut meta) = metadata_clone.write().await { + let _ = meta.set_cover_url(Some(cover_url)).await; + } +}); +``` + +## API Publique + +### Création + +```rust +pub fn new(client: RadioParadiseClient, chunk_duration_ms: u32) -> Self +``` + +### Configuration + +```rust +pub fn push_block_id(&mut self, event_id: EventId) +``` + +Ajoute un block ID à télécharger dans la file d'attente. + +### Exécution + +```rust +async fn run(self: Box, stop_token: CancellationToken) -> Result<(), AudioError> +``` + +Hérite de `AudioPipelineNode`. + +## Exemple d'utilisation + +Voir `examples/radio_paradise_stream.rs` pour : +- Utilisation basique +- Intégration avec nowplaying stream +- Connexion à un sink + +## Constantes + +```rust +const BLOCK_ID_TIMEOUT_SECS: u64 = 3; // Timeout attente nouveau block +const RECENT_BLOCKS_CACHE_SIZE: usize = 10; // Taille cache anti-redondance +``` + +## Dépendances + +- `pmoaudio` : Pipeline audio, types AudioChunk/AudioSegment +- `pmoflac` : Décodage FLAC streaming +- `pmometadata` : Métadonnées pistes +- `futures-util` : StreamExt pour le décodage +- `tokio` : Runtime async +- `tokio-util` : StreamReader, CancellationToken + +## Feature gate + +```toml +[features] +pmoaudio = ["dep:pmoaudio", "dep:pmoflac", "dep:pmometadata", "dep:futures-util"] +``` + +Activer avec : `cargo build -p pmoparadise --features pmoaudio` diff --git a/pmoparadise/examples/download_block.rs b/pmoparadise/examples/download_block.rs new file mode 100644 index 00000000..47cdeda8 --- /dev/null +++ b/pmoparadise/examples/download_block.rs @@ -0,0 +1,205 @@ +//! Télécharge un bloc complet de Radio Paradise et sauvegarde toutes les pistes en FLAC +//! +//! Ce programme démontre l'utilisation de la chaîne : +//! 1. RadioParadiseStreamSource - Télécharge et décode un bloc FLAC de Radio Paradise +//! 2. FlacFileSink - Sauvegarde automatiquement chaque piste dans un fichier FLAC séparé +//! +//! La nouvelle architecture AudioPipelineNode permet de : +//! - Télécharger et décoder automatiquement les blocs FLAC de Radio Paradise +//! - Détecter les limites de pistes (TrackBoundary) +//! - Sauvegarder automatiquement chaque piste dans un fichier séparé +//! - Gérer proprement l'arrêt du pipeline avec un CancellationToken +//! +//! Usage: +//! cargo run --example download_block -- +//! +//! Exemple: +//! cargo run --example download_block -- 0 # Main Mix +//! cargo run --example download_block -- 1 # Mellow Mix +//! cargo run --example download_block -- 2 # Rock Mix +//! cargo run --example download_block -- 3 # World/Etc Mix + +use pmoaudio::{AudioPipelineNode, FlacFileSink}; +use pmoparadise::{RadioParadiseClient, RadioParadiseStreamSource}; +use std::env; +use tokio_util::sync::CancellationToken; + +#[tokio::main] +async fn main() -> Result<(), Box> { + // Initialiser tracing pour le debug + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::from_default_env() + .add_directive(tracing::Level::INFO.into()), + ) + .init(); + + // Récupérer les arguments + let args: Vec = env::args().collect(); + if args.len() != 2 { + eprintln!("Usage: {} ", args[0]); + eprintln!(); + eprintln!("Downloads a complete Radio Paradise block and saves all tracks as FLAC files."); + eprintln!(); + eprintln!("Channel IDs:"); + eprintln!(" 0 - Main Mix (eclectic, diverse mix)"); + eprintln!(" 1 - Mellow Mix (smooth, chilled music)"); + eprintln!(" 2 - Rock Mix (classic & modern rock)"); + eprintln!(" 3 - World/Etc Mix (global sounds)"); + eprintln!(); + eprintln!("Example:"); + eprintln!(" {} 0 # Download Main Mix", args[0]); + eprintln!(" {} 2 # Download Rock Mix", args[0]); + std::process::exit(1); + } + + let channel_id: u8 = match args[1].parse() { + Ok(id) => id, + Err(_) => { + eprintln!("Error: channel_id must be a number between 0 and 3"); + std::process::exit(1); + } + }; + + if channel_id > 3 { + eprintln!("Error: channel_id must be between 0 and 3"); + std::process::exit(1); + } + + println!("=== Radio Paradise Block Downloader ==="); + println!(); + println!("Channel ID: {}", channel_id); + println!(); + + // Créer le client Radio Paradise pour le channel spécifié + println!("Fetching current block metadata..."); + let client = RadioParadiseClient::builder() + .channel(channel_id) + .build() + .await?; + + // Récupérer le bloc actuel + let block = client.get_block(None).await?; + + println!("Block Information:"); + println!(" Event ID: {}", block.event); + println!(" Songs: {}", block.song_count()); + println!( + " Duration: {:.1} minutes", + block.length as f64 / 60000.0 + ); + println!(); + + // Afficher la liste des pistes + println!("Tracklist:"); + for (index, song) in block.songs_ordered() { + println!( + " {:2}. {} - {} ({})", + index + 1, + song.artist, + song.title, + song.album.as_deref().unwrap_or("Unknown Album") + ); + } + println!(); + + // Créer le répertoire de sortie + let output_dir = format!("./rp_channel_{}block{}", channel_id, block.event); + std::fs::create_dir_all(&output_dir)?; + println!("Output directory: {}", output_dir); + println!(); + + // Créer le pipeline: RadioParadiseStreamSource → FlacFileSink + let mut source = RadioParadiseStreamSource::new(client); + + // Ajouter le bloc à télécharger + source.push_block_id(block.event); + + // Créer le sink qui sauvegarde chaque piste dans un fichier séparé + let base_path = format!("{}/track.flac", output_dir); + let sink = FlacFileSink::new(&base_path); + + // Construire la chaîne: source → sink + source.register(Box::new(sink)); + + // Créer un token d'arrêt + let stop_token = CancellationToken::new(); + + // Gérer Ctrl+C pour arrêt propre + let stop_token_clone = stop_token.clone(); + tokio::spawn(async move { + tokio::signal::ctrl_c().await.ok(); + println!("\n\nReceived Ctrl+C, stopping..."); + stop_token_clone.cancel(); + }); + + // Lancer tout le pipeline + println!("Downloading and processing block..."); + println!("Press Ctrl+C to stop."); + println!(); + let start = std::time::Instant::now(); + + let result = Box::new(source).run(stop_token).await; + + let elapsed = start.elapsed(); + + // Vérifier le résultat + match result { + Ok(()) => { + println!(); + println!( + "✓ Download completed successfully in {:.2}s", + elapsed.as_secs_f64() + ); + println!(" Output directory: {}", output_dir); + println!(); + + // Afficher les fichiers créés + let entries = std::fs::read_dir(&output_dir)?; + let mut files: Vec<_> = entries + .filter_map(|e| e.ok()) + .filter(|e| { + e.path() + .extension() + .and_then(|s| s.to_str()) + .map(|s| s == "flac") + .unwrap_or(false) + }) + .collect(); + files.sort_by_key(|e| e.path()); + + println!("Files created:"); + for (i, entry) in files.iter().enumerate() { + let path = entry.path(); + let metadata = std::fs::metadata(&path)?; + let size_mb = metadata.len() as f64 / (1024.0 * 1024.0); + println!( + " {:2}. {} ({:.2} MB)", + i + 1, + path.file_name().unwrap().to_string_lossy(), + size_mb + ); + } + println!(); + + // Calculer la taille totale + let total_size: u64 = files + .iter() + .filter_map(|e| std::fs::metadata(e.path()).ok()) + .map(|m| m.len()) + .sum(); + println!( + "Total size: {:.2} MB", + total_size as f64 / (1024.0 * 1024.0) + ); + } + Err(e) => { + eprintln!(); + eprintln!("✗ Download error: {}", e); + eprintln!(); + return Err(e.into()); + } + } + + Ok(()) +} diff --git a/pmoparadise/src/channels.rs b/pmoparadise/src/channels.rs new file mode 100644 index 00000000..2c1779ef --- /dev/null +++ b/pmoparadise/src/channels.rs @@ -0,0 +1,143 @@ +//! Radio Paradise channel definitions +//! +//! This module defines the available Radio Paradise channels and their metadata. + +use std::str::FromStr; + +/// Logical identifier for a Radio Paradise channel. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum ParadiseChannelKind { + Main, + Mellow, + Rock, + Eclectic, +} + +impl ParadiseChannelKind { + pub const fn id(self) -> u8 { + match self { + Self::Main => 0, + Self::Mellow => 1, + Self::Rock => 2, + Self::Eclectic => 3, + } + } + + pub const fn slug(self) -> &'static str { + match self { + Self::Main => "main", + Self::Mellow => "mellow", + Self::Rock => "rock", + Self::Eclectic => "eclectic", + } + } + + pub const fn display_name(self) -> &'static str { + match self { + Self::Main => "Main Mix", + Self::Mellow => "Mellow Mix", + Self::Rock => "Rock Mix", + Self::Eclectic => "Eclectic Mix", + } + } + + pub const fn description(self) -> &'static str { + match self { + Self::Main => "Eclectic mix of rock, world, electronica, and more", + Self::Mellow => "Mellower, less aggressive music", + Self::Rock => "Heavier, more guitar-driven music", + Self::Eclectic => "Curated worldwide selection", + } + } +} + +impl FromStr for ParadiseChannelKind { + type Err = anyhow::Error; + + fn from_str(s: &str) -> std::result::Result { + match s.to_ascii_lowercase().as_str() { + "main" | "0" => Ok(Self::Main), + "mellow" | "1" => Ok(Self::Mellow), + "rock" | "2" => Ok(Self::Rock), + "eclectic" | "3" => Ok(Self::Eclectic), + other => Err(anyhow::anyhow!("Unknown Radio Paradise channel: {}", other)), + } + } +} + +/// Metadata descriptor for a channel. +#[derive(Debug, Clone, Copy)] +pub struct ChannelDescriptor { + pub kind: ParadiseChannelKind, + pub id: u8, + pub slug: &'static str, + pub display_name: &'static str, + pub description: &'static str, +} + +impl ChannelDescriptor { + pub const fn new(kind: ParadiseChannelKind) -> Self { + Self { + id: kind.id(), + slug: kind.slug(), + display_name: kind.display_name(), + description: kind.description(), + kind, + } + } +} + +/// All available Radio Paradise channels +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 +} + +/// Default maximum number of tracks to keep in history +/// +/// This is used as the default if not configured via pmoconfig. +/// Value: 100 tracks - represents ~5-8 hours of playback history +pub const HISTORY_DEFAULT_MAX_TRACKS: usize = 100; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_channel_ids() { + assert_eq!(ParadiseChannelKind::Main.id(), 0); + assert_eq!(ParadiseChannelKind::Mellow.id(), 1); + assert_eq!(ParadiseChannelKind::Rock.id(), 2); + assert_eq!(ParadiseChannelKind::Eclectic.id(), 3); + } + + #[test] + fn test_max_channel_id() { + assert_eq!(max_channel_id(), 3); + } + + #[test] + fn test_all_channels_length() { + assert_eq!(ALL_CHANNELS.len(), 4); + } + + #[test] + fn test_channel_from_str() { + assert!(matches!( + "main".parse::(), + Ok(ParadiseChannelKind::Main) + )); + assert!(matches!( + "0".parse::(), + Ok(ParadiseChannelKind::Main) + )); + assert!("invalid".parse::().is_err()); + } +} diff --git a/pmoparadise/src/config_ext.rs b/pmoparadise/src/config_ext.rs index 9a0a79f9..c2303595 100644 --- a/pmoparadise/src/config_ext.rs +++ b/pmoparadise/src/config_ext.rs @@ -5,11 +5,6 @@ //! //! La configuration est minimale - seulement ce qui doit vraiment être configurable : //! - Activation/désactivation de la source -//! - Chemin de la base de données d'historique -//! - Taille maximale de l'historique -//! -//! Tous les autres paramètres (polling, timeouts, etc.) sont des constantes -//! définies dans `paradise::constants`. //! //! # Exemple //! @@ -24,24 +19,11 @@ //! println!("Radio Paradise is disabled"); //! return Ok(()); //! } -//! -//! // Get configuration -//! let db_path = config.get_paradise_history_database()?; -//! let max_tracks = config.get_paradise_history_size()?; //! ``` -use std::path::PathBuf; - -use anyhow::{anyhow, Result}; +use anyhow::Result; use pmoconfig::Config; -use serde_yaml::{Number, Value}; - -use crate::paradise::constants; - -/// Nom du répertoire pour Radio Paradise (relatif au config_dir) -/// -/// La base de données sera stockée dans `/paradise/history.db` -const DEFAULT_HISTORY_DATABASE_DIR: &str = "paradise"; +use serde_yaml::Value; /// Trait d'extension pour gérer la configuration Radio Paradise dans pmoconfig /// @@ -50,7 +32,7 @@ const DEFAULT_HISTORY_DATABASE_DIR: &str = "paradise"; /// /// # Auto-persist des valeurs par défaut /// -/// Tous les getters persistent automatiquement la valeur par défaut dans la +/// Le getter persiste automatiquement la valeur par défaut dans la /// configuration si elle n'existe pas encore. Cela permet à l'utilisateur de /// voir la configuration effective dans le fichier YAML et de la modifier facilement. /// @@ -65,10 +47,7 @@ const DEFAULT_HISTORY_DATABASE_DIR: &str = "paradise"; /// // Premier appel : persiste "enabled: true" dans la config et retourne true /// let enabled = config.get_paradise_enabled()?; /// -/// // Premier appel : persiste "max_tracks: 100" dans la config et retourne 100 -/// let max_tracks = config.get_paradise_history_size()?; -/// -/// // L'utilisateur peut maintenant éditer ces valeurs dans le fichier YAML +/// // L'utilisateur peut maintenant éditer cette valeur dans le fichier YAML /// ``` pub trait RadioParadiseConfigExt { /// Vérifie si Radio Paradise est activé @@ -102,71 +81,6 @@ pub trait RadioParadiseConfigExt { /// config.set_paradise_enabled(false)?; /// ``` fn set_paradise_enabled(&self, enabled: bool) -> Result<()>; - - /// Récupère le chemin de la base de données d'historique - /// - /// Le chemin retourné est absolu et pointe vers `/paradise/history.db`. - /// Le répertoire `paradise` est créé automatiquement s'il n'existe pas. - /// - /// # Returns - /// - /// Le chemin absolu vers la base de données SQLite d'historique. - /// Exemple: `/home/user/.config/pmo/paradise/history.db` - /// - /// # Exemple - /// - /// ```rust,ignore - /// let db_path = config.get_paradise_history_database()?; - /// let backend = SqliteHistoryBackend::new(&db_path)?; - /// ``` - fn get_paradise_history_database(&self) -> Result; - - /// Définit le chemin de la base de données d'historique - /// - /// # Arguments - /// - /// * `path` - Chemin complet vers la base de données (doit inclure le nom du fichier) - /// - /// Le répertoire parent sera extrait et stocké dans la configuration. - /// - /// # Exemple - /// - /// ```rust,ignore - /// // Set custom path - /// config.set_paradise_history_database("/var/lib/pmo/paradise.db".to_string())?; - /// ``` - fn set_paradise_history_database(&self, path: String) -> Result<()>; - - /// Récupère le nombre maximal de pistes dans l'historique - /// - /// # Returns - /// - /// Le nombre maximal de pistes à conserver dans l'historique. - /// - /// Si la valeur n'existe pas dans la configuration, elle est automatiquement - /// définie à la constante `HISTORY_DEFAULT_MAX_TRACKS` (100) et persistée. - /// - /// # Exemple - /// - /// ```rust,ignore - /// let max_tracks = config.get_paradise_history_size()?; - /// println!("Keeping last {} tracks", max_tracks); - /// ``` - fn get_paradise_history_size(&self) -> Result; - - /// Définit le nombre maximal de pistes dans l'historique - /// - /// # Arguments - /// - /// * `size` - Nombre maximal de pistes à conserver - /// - /// # Exemple - /// - /// ```rust,ignore - /// // Keep last 200 tracks - /// config.set_paradise_history_size(200)?; - /// ``` - fn set_paradise_history_size(&self, size: usize) -> Result<()>; } impl RadioParadiseConfigExt for Config { @@ -187,55 +101,6 @@ impl RadioParadiseConfigExt for Config { Value::Bool(enabled), ) } - - fn get_paradise_history_database(&self) -> Result { - // Get managed directory: ~/.config/pmo/paradise/ - let dir = self.get_managed_dir( - &["sources", "radio_paradise", "database"], - DEFAULT_HISTORY_DATABASE_DIR, - )?; - - // Ensure directory exists - std::fs::create_dir_all(&dir)?; - - // Build full path: ~/.config/pmo/paradise/history.db - let mut path = PathBuf::from(dir); - path.push("history.db"); - - Ok(path.to_string_lossy().to_string()) - } - - fn set_paradise_history_database(&self, path: String) -> Result<()> { - // Extract parent directory from the full path - match PathBuf::from(&path).parent() { - Some(dir) => self.set_managed_dir( - &["sources", "radio_paradise", "database"], - dir.to_string_lossy().to_string(), - ), - None => Err(anyhow!("Invalid database path: no parent directory")), - } - } - - fn get_paradise_history_size(&self) -> Result { - match self.get_value(&["sources", "radio_paradise", "history", "max_tracks"]) { - Ok(Value::Number(n)) if n.is_u64() => Ok(n.as_u64().unwrap() as usize), - Ok(Value::Number(n)) if n.is_i64() => Ok(n.as_i64().unwrap() as usize), - _ => { - // Use default and persist it - let default = constants::HISTORY_DEFAULT_MAX_TRACKS; - self.set_paradise_history_size(default)?; - Ok(default) - } - } - } - - fn set_paradise_history_size(&self, size: usize) -> Result<()> { - let n = Number::from(size); - self.set_value( - &["sources", "radio_paradise", "history", "max_tracks"], - Value::Number(n), - ) - } } #[cfg(test)] @@ -243,21 +108,7 @@ mod tests { use super::*; #[test] - fn test_default_values() { - assert_eq!(DEFAULT_HISTORY_DATABASE_DIR, "paradise"); - assert_eq!(constants::HISTORY_DEFAULT_MAX_TRACKS, 100); - } - - #[test] - fn test_database_path_construction() { - // Simulating path construction - let base = "/home/user/.config/pmo/paradise"; - let mut path = PathBuf::from(base); - path.push("history.db"); - - assert_eq!( - path.to_string_lossy(), - "/home/user/.config/pmo/paradise/history.db" - ); + fn test_trait_exists() { + // Simple test to ensure the trait compiles } } diff --git a/pmoparadise/src/error.rs b/pmoparadise/src/error.rs index bbb75914..9ee77cd8 100644 --- a/pmoparadise/src/error.rs +++ b/pmoparadise/src/error.rs @@ -34,16 +34,6 @@ pub enum Error { #[error("Invalid event ID: {0}")] InvalidEvent(String), - /// FLAC decoding error (per-track feature) - #[cfg(feature = "per-track")] - #[error("FLAC decoding error: {0}")] - FlacDecode(String), - - /// WAV encoding error (per-track feature) - #[cfg(feature = "per-track")] - #[error("WAV encoding error: {0}")] - WavEncode(#[from] hound::Error), - /// Track not found in block #[error("Track not found at index {0}")] TrackNotFound(usize), @@ -67,11 +57,3 @@ impl Error { Self::Other(msg.into()) } } - -// Implement conversion from claxon errors for per-track feature -#[cfg(feature = "per-track")] -impl From for Error { - fn from(err: claxon::Error) -> Self { - Error::FlacDecode(err.to_string()) - } -} diff --git a/pmoparadise/src/lib.rs b/pmoparadise/src/lib.rs index 20b91780..960b3d94 100644 --- a/pmoparadise/src/lib.rs +++ b/pmoparadise/src/lib.rs @@ -164,73 +164,57 @@ //! } //! ``` //! -//! ## Caching Support (Feature: `cache`) +//! ## Audio Streaming (Feature: `pmoaudio`) //! -//! `pmoparadise` can optionally integrate with `pmocovers` and `pmoaudiocache` to cache -//! cover images and audio tracks locally: +//! For direct audio streaming and integration with pmoaudio pipelines, +//! use `RadioParadiseStreamSource`: //! //! ```no_run -//! # #[cfg(feature = "cache")] +//! # #[cfg(feature = "pmoaudio")] //! # { -//! use pmoparadise::{RadioParadiseClient, RadioParadiseSource}; -//! use std::sync::Arc; +//! use pmoparadise::{RadioParadiseClient, RadioParadiseStreamSource}; +//! use pmoaudio::pipeline::Node; //! //! #[tokio::main] //! async fn main() -> Result<(), Box> { -//! // Create caches -//! let cover_cache = Arc::new(pmocovers::cache::new_cache("./cache/covers", 500)?); -//! let audio_cache = Arc::new(pmoaudiocache::cache::new_cache("./cache/audio", 100)?); -//! -//! // Create client and source with caching //! let client = RadioParadiseClient::new().await?; -//! let source = RadioParadiseSource::new( -//! client, -//! 50, -//! cover_cache, -//! audio_cache, -//! ); +//! let stream_source = RadioParadiseStreamSource::new(client, None).await?; //! -//! println!("Source ready: {}", source.name()); +//! // Create audio node from stream source +//! let node = Node::from_logic(stream_source); +//! +//! // Use in pmoaudio pipeline... //! //! Ok(()) //! } //! # } //! ``` //! -//! **Benefits**: -//! - Cover images are automatically downloaded and converted to WebP -//! - Audio tracks are cached as FLAC with metadata preserved -//! - Subsequent access is instant (no re-download) -//! - URIs returned by `resolve_uri()` point to cached versions -//! -//! See the `with_cache` example for a complete demonstration. +//! **RadioParadiseStreamSource**: +//! - Downloads and decodes FLAC blocks in real-time +//! - Automatically detects bit depth (16/24/32-bit) +//! - Inserts track boundaries with metadata +//! - Integrates seamlessly with pmoaudio pipelines //! //! ## Cargo Features //! -//! - `default = ["metadata-only"]`: Standard metadata and streaming (no FLAC decoding) +//! - `default`: Standard metadata and streaming (no FLAC decoding) //! - `per-track`: Enable FLAC decoding and per-track extraction (adds `claxon`, `hound`, `tempfile`) //! - `pmoserver`: Enable REST API extension for pmoserver integration (adds `utoipa`, `axum`) -//! - `server`: Enable server-side features (cache registry integration) -//! - `cache`: Enable cover and audio caching support (adds `pmocovers`, `pmoaudiocache`) +//! - `pmoaudio`: Enable RadioParadiseStreamSource for pmoaudio integration +//! - `pmoconfig`: Enable configuration integration with pmoconfig +//! - `server`: Enable RadioParadiseSource stub for backward compatibility (deprecated) //! //! ## See Also //! //! - [Radio Paradise](https://radioparadise.com) - Official website //! - [Radio Paradise API](https://api.radioparadise.com) - API documentation +pub mod channels; pub mod client; pub mod error; pub mod models; -pub mod paradise; pub mod source; -pub mod stream; -pub mod streaming; - -#[cfg(feature = "per-track")] -pub mod track; - -#[cfg(feature = "ffmpeg")] -pub mod ffmpeg_streaming; #[cfg(feature = "pmoserver")] pub mod pmoserver_ext; @@ -238,15 +222,17 @@ pub mod pmoserver_ext; #[cfg(feature = "pmoconfig")] pub mod config_ext; +#[cfg(feature = "pmoaudio")] +pub mod radio_paradise_stream_source; + // Re-exports for convenience pub use client::{ClientBuilder, RadioParadiseClient}; pub use error::{Error, Result}; pub use models::{Block, DurationMs, EventId, NowPlaying, Song}; pub use source::RadioParadiseSource; -pub use stream::BlockStream; -#[cfg(feature = "per-track")] -pub use track::{TrackMetadata, TrackStream}; +#[cfg(feature = "pmoaudio")] +pub use radio_paradise_stream_source::RadioParadiseStreamSource; #[cfg(feature = "pmoserver")] pub use pmoserver_ext::{ diff --git a/pmoparadise/src/pmoserver_ext.rs b/pmoparadise/src/pmoserver_ext.rs index f21d7f16..5c19d5ad 100644 --- a/pmoparadise/src/pmoserver_ext.rs +++ b/pmoparadise/src/pmoserver_ext.rs @@ -3,31 +3,24 @@ //! Ce module fournit un trait d'extension pour ajouter facilement l'API Radio Paradise //! à un serveur pmoserver. -use crate::paradise::{max_channel_id, ParadiseChannel, PlaylistEntry, ALL_CHANNELS}; -use crate::{Block, NowPlaying, RadioParadiseClient, RadioParadiseSource}; +use crate::channels::{max_channel_id, ChannelDescriptor, ALL_CHANNELS}; +use crate::{Block, NowPlaying, RadioParadiseClient}; +use async_trait::async_trait; use axum::{ - body::Body, extract::{Path, Query, State}, - http::{HeaderMap, HeaderName, HeaderValue, StatusCode}, - response::IntoResponse, + http::StatusCode, routing::get, Json, Router, }; -use chrono::{DateTime, Utc}; -use futures::StreamExt; -use pmosource::api::CacheStatusInfo; -use pmosource::CacheStatus; use serde::{Deserialize, Serialize}; use std::sync::Arc; use tokio::sync::RwLock; -use tracing::error; -use utoipa::{IntoParams, OpenApi, ToSchema}; +use utoipa::{OpenApi, ToSchema}; /// État partagé pour l'API Radio Paradise #[derive(Clone)] pub struct RadioParadiseState { client: Arc>, - source: Arc, } #[derive(Debug, Default, Deserialize)] @@ -36,46 +29,14 @@ struct ParadiseQuery { channel: Option, } -#[derive(Debug, Default, Deserialize, IntoParams)] -#[serde(default)] -#[into_params(parameter_in = Query)] -struct ListLimitQuery { - /// Nombre maximum d'éléments à retourner (0 = tous) - #[serde(default)] - limit: Option, -} - impl RadioParadiseState { pub async fn new() -> anyhow::Result { let client = RadioParadiseClient::new() .await .map_err(|e| anyhow::anyhow!("Failed to create RadioParadise client: {}", e))?; - #[cfg(feature = "server")] - let source = RadioParadiseSource::from_registry_default(client.clone()) - .map_err(|e| anyhow::anyhow!(e.to_string()))?; - - #[cfg(not(feature = "server"))] - let source = { - let base_dir = std::env::temp_dir().join("pmoparadise_api"); - let cover_dir = base_dir.join("covers"); - let audio_dir = base_dir.join("audio"); - std::fs::create_dir_all(&cover_dir)?; - std::fs::create_dir_all(&audio_dir)?; - - let cover_cache = Arc::new(pmocovers::cache::new_cache( - cover_dir.to_string_lossy().as_ref(), - 256, - )?); - let audio_cache = Arc::new(pmoaudiocache::cache::new_cache( - audio_dir.to_string_lossy().as_ref(), - 256, - )?); - RadioParadiseSource::new_default(client.clone(), cover_cache, audio_cache) - }; Ok(Self { client: Arc::new(RwLock::new(client)), - source: Arc::new(source), }) } @@ -100,19 +61,6 @@ impl RadioParadiseState { Ok(client) } - - fn channel_for_id(&self, channel_id: u8) -> Result, StatusCode> { - if channel_id > max_channel_id() { - return Err(StatusCode::BAD_REQUEST); - } - self.source - .channel(channel_id) - .ok_or(StatusCode::SERVICE_UNAVAILABLE) - } - - pub fn source(&self) -> Arc { - self.source.clone() - } } /// Information sur un canal Radio Paradise @@ -126,8 +74,8 @@ pub struct ChannelInfo { pub description: String, } -impl From<&crate::paradise::ChannelDescriptor> for ChannelInfo { - fn from(descriptor: &crate::paradise::ChannelDescriptor) -> Self { +impl From<&ChannelDescriptor> for ChannelInfo { + fn from(descriptor: &ChannelDescriptor) -> Self { Self { id: descriptor.id, name: descriptor.display_name.to_string(), @@ -365,425 +313,28 @@ async fn get_channels() -> Json> { Json(channels) } -/// Statut opérationnel d'un canal Radio Paradise -#[derive(Debug, Clone, Serialize, ToSchema)] -pub struct ChannelStatusResponse { - /// ID numérique du canal - pub channel_id: u8, - /// Slug du canal (main, mellow, ...) - pub slug: String, - /// Nom complet du canal - pub name: String, - /// Description - pub description: String, - /// Nombre de clients connectés au flux - pub active_clients: usize, - /// Nombre de morceaux présents dans la file d'attente - pub queue_length: usize, - /// Valeur courante d'update_id - pub update_id: u32, - /// Dernière modification (RFC3339) - pub last_change: Option, - /// Nombre total d'entrées en historique (persisté) - pub history_entries: usize, - /// Limite configurée pour l'historique - pub history_max_tracks: usize, - /// Le canal est-il activé dans la configuration ? - pub configured: bool, - /// Identifiant de collection pour le cache - pub cache_collection_id: String, - /// Nombre total de pistes connues du cache - pub cache_total_tracks: usize, - /// Nombre de pistes déjà en cache - pub cache_cached_tracks: usize, -} - -/// Entrée détaillée de la file d'attente -#[derive(Debug, Clone, Serialize, ToSchema)] -pub struct ChannelPlaylistEntry { - /// Position dans la file - pub index: usize, - /// ID unique de la piste - pub track_id: String, - /// ID du canal - pub channel_id: u8, - /// Titre du morceau - pub title: String, - /// Artiste - pub artist: String, - /// Album - pub album: Option, - /// URL de couverture (si disponible) - pub cover_url: Option, - /// Durée du morceau en ms - pub duration_ms: u64, - /// Offset dans le block (ms) - pub elapsed_ms: u64, - /// Horodatage prévu/démarré (RFC3339) - pub started_at: String, - /// Nombre de clients restants à servir - pub pending_clients: usize, - /// Note éventuelle (0-10) - pub rating: Option, - /// Année éventuelle - pub year: Option, - /// Statut de cache - pub cache_status: CacheStatusInfo, -} - -impl ChannelPlaylistEntry { - fn from_entry(entry: &Arc, index: usize, cache_status: CacheStatusInfo) -> Self { - let song = entry.song.as_ref(); - Self { - index, - track_id: entry.track_id.clone(), - channel_id: entry.channel_id, - title: song.title.clone(), - artist: song.artist.clone(), - album: song.album.clone(), - cover_url: song.cover.clone(), - duration_ms: entry.duration_ms, - elapsed_ms: song.elapsed, - started_at: entry.started_at.to_rfc3339(), - pending_clients: entry.pending_clients(), - rating: song.rating, - year: song.year, - cache_status, - } - } -} - -/// Réponse pour la file d'attente d'un canal -#[derive(Debug, Clone, Serialize, ToSchema)] -pub struct ChannelPlaylistResponse { - /// ID du canal - pub channel_id: u8, - /// Slug du canal - pub slug: String, - /// Update ID du playlist - pub update_id: u32, - /// Taille totale de la file au moment de la capture - pub queue_length: usize, - /// Entrées retournées - pub items: Vec, -} - -/// Entrée d'historique d'écoute -#[derive(Debug, Clone, Serialize, ToSchema)] -pub struct ChannelHistoryEntry { - /// ID unique de la piste - pub track_id: String, - /// ID du canal - pub channel_id: u8, - /// Titre - pub title: String, - /// Artiste - pub artist: String, - /// Album - pub album: Option, - /// URL de couverture - pub cover_url: Option, - /// Début de lecture (RFC3339) - pub started_at: String, - /// Durée en ms - pub duration_ms: u64, -} - -/// Réponse pour l'historique d'un canal -#[derive(Debug, Clone, Serialize, ToSchema)] -pub struct ChannelHistoryResponse { - /// ID du canal - pub channel_id: u8, - /// Slug du canal - pub slug: String, - /// Nombre total d'entrées disponibles - pub total_available: usize, - /// Nombre d'entrées retournées dans cette réponse - pub returned: usize, - /// Entrées - pub entries: Vec, -} - -/// GET /channels/{channel_id}/status - Statut détaillé d'un canal -#[utoipa::path( - get, - path = "/channels/{channel_id}/status", - params( - ("channel_id" = u8, Path, description = "Channel ID (0-3)") - ), - responses( - (status = 200, description = "Statut du canal", body = ChannelStatusResponse), - (status = 400, description = "Canal invalide"), - (status = 503, description = "Canal indisponible"), - (status = 500, description = "Erreur interne lors de la récupération du statut") - ), - tag = "Radio Paradise" -)] -async fn get_channel_status( - State(state): State, - Path(channel_id): Path, -) -> Result, StatusCode> { - let channel = state.channel_for_id(channel_id)?; - let descriptor = channel.descriptor(); - - let playlist = channel.playlist(); - let queue_length = playlist.active_len().await; - let update_id = playlist.update_id(); - let last_change = playlist - .last_change() - .await - .map(|ts| DateTime::::from(ts).to_rfc3339()); - - let history_len = channel.history_backend().len().await.map_err(|e| { - error!( - channel = descriptor.slug, - "Failed to retrieve history size: {e:?}" - ); - StatusCode::INTERNAL_SERVER_ERROR - })?; - - let cache_stats = channel.cache_manager().statistics().await; - - let status = ChannelStatusResponse { - channel_id, - slug: descriptor.slug.to_string(), - name: descriptor.display_name.to_string(), - description: descriptor.description.to_string(), - active_clients: channel.active_client_count(), - queue_length, - update_id, - last_change, - history_entries: history_len, - history_max_tracks: channel.history_max_tracks(), - configured: true, // All channels are always available - cache_collection_id: cache_stats.collection_id, - cache_total_tracks: cache_stats.total_tracks, - cache_cached_tracks: cache_stats.cached_tracks, - }; - - Ok(Json(status)) -} - -/// GET /channels/{channel_id}/playlist - File d'attente du canal -#[utoipa::path( - get, - path = "/channels/{channel_id}/playlist", - params( - ("channel_id" = u8, Path, description = "Channel ID (0-3)"), - ListLimitQuery - ), - responses( - (status = 200, description = "File d'attente courante", body = ChannelPlaylistResponse), - (status = 400, description = "Canal invalide"), - (status = 503, description = "Canal indisponible"), - (status = 500, description = "Erreur lors de la récupération de la file d'attente") - ), - tag = "Radio Paradise" -)] -async fn get_channel_playlist( - State(state): State, - Path(channel_id): Path, - Query(query): Query, -) -> Result, StatusCode> { - let channel = state.channel_for_id(channel_id)?; - let descriptor = channel.descriptor(); - let playlist = channel.playlist(); - let snapshot = playlist.active_snapshot().await; - let total_len = snapshot.len(); - let limit = query.limit.filter(|limit| *limit > 0).unwrap_or(total_len); - - let cache_manager = channel.cache_manager(); - let mut items = Vec::new(); - - for (index, entry) in snapshot.into_iter().enumerate().take(limit) { - let cache_status = match cache_manager.get_cache_status(&entry.track_id).await { - Ok(status) => status, - Err(err) => CacheStatus::Failed { - error: err.to_string(), - }, - }; - - items.push(ChannelPlaylistEntry::from_entry( - &entry, - index, - CacheStatusInfo::from(cache_status), - )); - } - - let response = ChannelPlaylistResponse { - channel_id, - slug: descriptor.slug.to_string(), - update_id: playlist.update_id(), - queue_length: total_len, - items, - }; - - Ok(Json(response)) -} - -/// GET /channels/{channel_id}/history - Historique récent du canal -#[utoipa::path( - get, - path = "/channels/{channel_id}/history", - params( - ("channel_id" = u8, Path, description = "Channel ID (0-3)"), - ListLimitQuery - ), - responses( - (status = 200, description = "Historique récent", body = ChannelHistoryResponse), - (status = 400, description = "Canal invalide"), - (status = 503, description = "Canal indisponible"), - (status = 500, description = "Erreur lors de la récupération de l'historique") - ), - tag = "Radio Paradise" -)] -async fn get_channel_history( - State(state): State, - Path(channel_id): Path, - Query(query): Query, -) -> Result, StatusCode> { - let channel = state.channel_for_id(channel_id)?; - let descriptor = channel.descriptor(); - let backend = channel.history_backend().clone(); - let limit = query.limit.unwrap_or(50); - - let entries_raw = backend.recent(limit).await.map_err(|e| { - error!( - channel = descriptor.slug, - "Failed to retrieve channel history: {e:?}" - ); - StatusCode::INTERNAL_SERVER_ERROR - })?; - - let total_available = backend.len().await.map_err(|e| { - error!( - channel = descriptor.slug, - "Failed to count channel history entries: {e:?}" - ); - StatusCode::INTERNAL_SERVER_ERROR - })?; - - let entries: Vec = entries_raw - .into_iter() - .map(|entry| ChannelHistoryEntry { - track_id: entry.track_id, - channel_id: entry.channel_id, - title: entry.song.title, - artist: entry.song.artist, - album: entry.song.album, - cover_url: entry.song.cover_url, - started_at: entry.started_at.to_rfc3339(), - duration_ms: entry.duration_ms, - }) - .collect(); - - let response = ChannelHistoryResponse { - channel_id, - slug: descriptor.slug.to_string(), - total_available, - returned: entries.len(), - entries, - }; - - Ok(Json(response)) -} - -/// GET /channels/{channel_id}/stream/{connection_id} - Stream audio pour une connexion spécifique -#[utoipa::path( - get, - path = "/channels/{channel_id}/stream/{connection_id}", - params( - ("channel_id" = u8, Path, description = "Channel ID (0-3)"), - ("connection_id" = i32, Path, description = "Connection ID fourni par le media server") - ), - responses( - (status = 200, description = "Flux audio FLAC (gapless)", content_type = "audio/flac"), - (status = 400, description = "Canal invalide"), - (status = 503, description = "Canal indisponible") - ), - tag = "Radio Paradise" -)] -async fn stream_channel_by_connection( - State(state): State, - Path((channel_id, connection_id)): Path<(u8, i32)>, -) -> Result { - let channel = state.channel_for_id(channel_id)?; - - // Convertir connection_id en String pour l'utiliser comme client_id - let client_id = connection_id.to_string(); - - let client_stream = channel.connect_client(client_id).await.map_err(|e| { - error!("Failed to create streaming client: {e:?}"); - StatusCode::SERVICE_UNAVAILABLE - })?; - - let stream = client_stream - .into_byte_stream() - .map(|chunk| chunk.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))); - - let body = Body::from_stream(stream); - - let mut headers = HeaderMap::new(); - headers.insert( - axum::http::header::CONTENT_TYPE, - HeaderValue::from_static("audio/flac"), - ); - headers.insert( - axum::http::header::CACHE_CONTROL, - HeaderValue::from_static("no-cache"), - ); - headers.insert( - HeaderName::from_static("icy-name"), - HeaderValue::from_static("Radio Paradise"), - ); - headers.insert( - HeaderName::from_static("icy-genre"), - HeaderValue::from_static("Eclectic"), - ); - headers.insert( - HeaderName::from_static("icy-description"), - HeaderValue::from_static("PMO Radio Paradise relay"), - ); - headers.insert( - HeaderName::from_static("icy-metaint"), - HeaderValue::from_static("0"), - ); - - Ok((headers, body)) -} - /// Documentation OpenAPI pour l'API Radio Paradise #[derive(OpenApi)] #[openapi( info( title = "Radio Paradise API", version = "1.0.0", - description = "API REST pour accéder aux métadonnées et streams de Radio Paradise" + description = "API REST pour accéder aux métadonnées de Radio Paradise" ), paths( get_now_playing, get_current_block, get_block_by_id, - get_channels, - get_channel_status, - get_channel_playlist, - get_channel_history, - stream_channel_by_connection + get_channels ), components(schemas( NowPlayingResponse, BlockResponse, SongInfo, - ChannelInfo, - ChannelStatusResponse, - ChannelPlaylistEntry, - ChannelPlaylistResponse, - ChannelHistoryEntry, - ChannelHistoryResponse, - CacheStatusInfo + ChannelInfo )), tags( - (name = "Radio Paradise", description = "Endpoints pour Radio Paradise streaming") + (name = "Radio Paradise", description = "Endpoints pour Radio Paradise") ) )] pub struct RadioParadiseApiDoc; @@ -795,13 +346,6 @@ pub fn create_api_router(state: RadioParadiseState) -> Router { .route("/block/current", get(get_current_block)) .route("/block/{event_id}", get(get_block_by_id)) .route("/channels", get(get_channels)) - .route("/channels/{channel_id}/status", get(get_channel_status)) - .route("/channels/{channel_id}/playlist", get(get_channel_playlist)) - .route("/channels/{channel_id}/history", get(get_channel_history)) - .route( - "/channels/{channel_id}/stream/{connection_id}", - get(stream_channel_by_connection), - ) .with_state(state) } @@ -809,6 +353,7 @@ pub fn create_api_router(state: RadioParadiseState) -> Router { /// /// Permet d'initialiser Radio Paradise avec routes HTTP complètes #[cfg(feature = "pmoserver")] +#[async_trait] pub trait RadioParadiseExt { /// Initialise l'API Radio Paradise /// @@ -817,12 +362,13 @@ pub trait RadioParadiseExt { /// - API: `/api/radioparadise/*` /// - `/now-playing` /// - `/block/*` - /// - `/channels/{channel_id}/stream/{connection_id}` + /// - `/channels` /// - Swagger: `/swagger-ui/radioparadise` async fn init_radioparadise(&mut self) -> anyhow::Result; } #[cfg(feature = "pmoserver")] +#[async_trait] impl RadioParadiseExt for pmoserver::Server { async fn init_radioparadise(&mut self) -> anyhow::Result { let state = RadioParadiseState::new().await?; diff --git a/pmoparadise/src/radio_paradise_stream_source.rs b/pmoparadise/src/radio_paradise_stream_source.rs new file mode 100644 index 00000000..71c191b6 --- /dev/null +++ b/pmoparadise/src/radio_paradise_stream_source.rs @@ -0,0 +1,667 @@ +//! RadioParadiseStreamSource - Node audio pmoaudio pour Radio Paradise +//! +//! Ce node télécharge et décode les blocs FLAC de Radio Paradise en streaming, +//! avec insertion automatique des TrackBoundary au bon timing. + +use crate::{ + client::RadioParadiseClient, + models::{Block, EventId, Song}, +}; +use futures_util::StreamExt; +use pmoaudio::{ + nodes::{AudioError, TypedAudioNode, DEFAULT_CHUNK_DURATION_MS}, + pipeline::{Node, NodeLogic}, + type_constraints::TypeRequirement, + AudioPipelineNode, AudioSegment, SyncMarker, I24, +}; +use pmoflac::decode_audio_stream; +use pmometadata::{MemoryTrackMetadata, TrackMetadata}; +use std::{ + collections::VecDeque, + sync::Arc, + time::Duration, +}; +use tokio::io::AsyncReadExt; +use tokio::sync::{mpsc, RwLock}; +use tokio_util::{io::StreamReader, sync::CancellationToken}; + +/// Timeout pour attendre un nouveau block ID (radio en temps réel) +const BLOCK_ID_TIMEOUT_SECS: u64 = 3; + +/// Nombre de blocs récents à mémoriser pour éviter les re-téléchargements +const RECENT_BLOCKS_CACHE_SIZE: usize = 10; + +// ═══════════════════════════════════════════════════════════════════════════ +// RadioParadiseStreamSourceLogic - Logique métier pure +// ═══════════════════════════════════════════════════════════════════════════ + +/// Logique pure de téléchargement et décodage des blocs Radio Paradise +pub struct RadioParadiseStreamSourceLogic { + client: RadioParadiseClient, + chunk_frames: usize, + recent_blocks: VecDeque, + block_queue: VecDeque, +} + +impl RadioParadiseStreamSourceLogic { + pub fn new(client: RadioParadiseClient, chunk_duration_ms: u32) -> Self { + // Calculer chunk_frames pour la durée cible (on suppose 44.1kHz) + let chunk_frames = ((chunk_duration_ms as f64 / 1000.0) * 44100.0) as usize; + + Self { + client, + chunk_frames, + recent_blocks: VecDeque::with_capacity(RECENT_BLOCKS_CACHE_SIZE), + block_queue: VecDeque::new(), + } + } + + /// Ajoute un block ID à la file d'attente + pub fn push_block_id(&mut self, event_id: EventId) { + self.block_queue.push_back(event_id); + } + + /// Vérifie si un bloc a été téléchargé récemment + fn is_recent_block(&self, event_id: EventId) -> bool { + self.recent_blocks.contains(&event_id) + } + + /// Marque un bloc comme récemment téléchargé (FIFO) + fn mark_block_downloaded(&mut self, event_id: EventId) { + // Retirer tous les éléments excédentaires (garantit <= CACHE_SIZE) + while self.recent_blocks.len() >= RECENT_BLOCKS_CACHE_SIZE { + self.recent_blocks.pop_front(); + } + + // Puis ajouter le nouveau bloc + self.recent_blocks.push_back(event_id); + } + + /// Télécharge et décode un bloc FLAC + async fn download_and_decode_block( + &mut self, + block: &Block, + output: &[mpsc::Sender>], + stop_token: &CancellationToken, + order: &mut u64, + ) -> Result<(), AudioError> { + // Télécharger le FLAC + let response = self.client.client + .get(&block.url) + .timeout(self.client.block_timeout) + .send() + .await + .map_err(|e| AudioError::ProcessingError(format!("Block download failed: {}", e)))?; + + if !response.status().is_success() { + return Err(AudioError::ProcessingError(format!( + "Block download returned status {}", + response.status() + ))); + } + + // Créer un stream reader + let byte_stream = response.bytes_stream().map(|result| { + result.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e)) + }); + let stream_reader = StreamReader::new(byte_stream); + + // Décoder le FLAC + let mut decoder = decode_audio_stream(stream_reader) + .await + .map_err(|e| AudioError::ProcessingError(format!("FLAC decode failed: {}", e)))?; + + let stream_info = decoder.info().clone(); + let sample_rate = stream_info.sample_rate; + let bits_per_sample = stream_info.bits_per_sample; + + // Préparer les songs ordonnées pour tracking + let songs = block.songs_ordered(); + let mut song_index = 0; + let mut next_song: Option<(usize, &Song)> = songs.get(0).copied(); + let mut total_samples = 0u64; + + // Envoyer TopZeroSync au début du bloc + let top_zero = Arc::new(AudioSegment { + order: *order, + timestamp_sec: 0.0, + segment: pmoaudio::_AudioSegment::Sync(Arc::new(SyncMarker::TopZeroSync)), + }); + self.send_to_children(output, top_zero).await?; + + // Buffer pour lecture + let bytes_per_sample = (bits_per_sample / 8) as usize; + let frame_bytes = bytes_per_sample * 2; // stereo + let chunk_frames = self.chunk_frames; + let chunk_byte_len = chunk_frames * frame_bytes; + let mut read_buf = vec![0u8; chunk_byte_len * 2]; + let mut pending: Vec = Vec::with_capacity(chunk_byte_len * 2); + + // Traiter les chunks audio + loop { + // Vérifier stop_token + if stop_token.is_cancelled() { + return Ok(()); + } + + // Remplir le buffer + if pending.len() < chunk_byte_len { + let read = decoder.read(&mut read_buf).await + .map_err(|e| AudioError::ProcessingError(format!("Read error: {}", e)))?; + + if read == 0 { + break; // EOF + } + pending.extend_from_slice(&read_buf[..read]); + } + + if pending.is_empty() { + break; + } + + // Extraire un chunk + let frames_in_pending = pending.len() / frame_bytes; + let frames_to_emit = frames_in_pending.min(chunk_frames); + let take_bytes = frames_to_emit * frame_bytes; + let pcm_data = pending.drain(..take_bytes).collect::>(); + + // Calculer le nombre de frames (samples par canal) + let bytes_per_sample = (bits_per_sample / 8) as usize; + let chunk_len = (pcm_data.len() / (bytes_per_sample * 2)) as u64; // 2 = stereo + + // Vérifier si on doit insérer un TrackBoundary avant ce chunk + if let Some((_idx, song)) = next_song { + let elapsed_ms = (total_samples * 1000) / sample_rate as u64; + + if elapsed_ms >= song.elapsed { + // Envoyer TrackBoundary AVANT le chunk (avec le même order) + let metadata = song_to_metadata(song, block); + let timestamp_sec = total_samples as f64 / sample_rate as f64; + let track_boundary = AudioSegment::new_track_boundary( + *order, + timestamp_sec, + metadata, + ); + self.send_to_children(output, track_boundary).await?; + + // Passer à la song suivante + song_index += 1; + next_song = songs.get(song_index).copied(); + } + } + + // Envoyer le chunk audio + let timestamp_sec = total_samples as f64 / sample_rate as f64; + let audio_segment = pcm_to_audio_segment( + &pcm_data, + *order, + timestamp_sec, + sample_rate, + bits_per_sample, + )?; + self.send_to_children(output, audio_segment).await?; + + *order += 1; + total_samples += chunk_len; + } + + Ok(()) + } + + /// Envoie un segment à tous les enfants + async fn send_to_children( + &self, + output: &[mpsc::Sender>], + segment: Arc, + ) -> Result<(), AudioError> { + for tx in output { + tx.send(segment.clone()) + .await + .map_err(|_| AudioError::ChildDied)?; + } + Ok(()) + } +} + +/// Convertit PCM bytes en AudioSegment +fn pcm_to_audio_segment( + pcm_data: &[u8], + order: u64, + timestamp_sec: f64, + sample_rate: u32, + bits_per_sample: u8, +) -> Result, AudioError> { + use pmoaudio::{AudioChunk, AudioChunkData, _AudioSegment}; + + let bytes_per_sample = (bits_per_sample / 8) as usize; + let channels = 2; // Stereo + let frame_bytes = bytes_per_sample * channels; + let frames = pcm_data.len() / frame_bytes; + + // Valider que la taille des données est correcte + if pcm_data.len() % frame_bytes != 0 { + return Err(AudioError::ProcessingError(format!( + "Invalid PCM data size: {} bytes is not a multiple of frame size {} ({}bit, {} channels)", + pcm_data.len(), + frame_bytes, + bits_per_sample, + channels + ))); + } + + let chunk = match bits_per_sample { + 16 => { + // Type I16 + let mut stereo = Vec::with_capacity(frames); + for frame_idx in 0..frames { + let base = frame_idx * frame_bytes; + let left = i16::from_le_bytes([pcm_data[base], pcm_data[base + 1]]); + let right = i16::from_le_bytes([pcm_data[base + 2], pcm_data[base + 3]]); + stereo.push([left, right]); + } + let chunk_data = AudioChunkData::new(stereo, sample_rate, 0.0); + AudioChunk::I16(chunk_data) + } + 24 => { + // Type I24 avec sign extension correcte + let mut stereo = Vec::with_capacity(frames); + for frame_idx in 0..frames { + let base = frame_idx * frame_bytes; + + // Left channel (bytes 0,1,2) avec sign extension + let left_i32 = { + let mut buf = [0u8; 4]; + buf[..3].copy_from_slice(&pcm_data[base..base + 3]); + // Sign extend si négatif + if pcm_data[base + 2] & 0x80 != 0 { + buf[3] = 0xFF; + } + i32::from_le_bytes(buf) + }; + let left = I24::new(left_i32).ok_or_else(|| { + AudioError::ProcessingError(format!("Invalid I24 value: {}", left_i32)) + })?; + + // Right channel (bytes 3,4,5) avec sign extension + let right_i32 = { + let mut buf = [0u8; 4]; + buf[..3].copy_from_slice(&pcm_data[base + 3..base + 6]); + // Sign extend si négatif + if pcm_data[base + 5] & 0x80 != 0 { + buf[3] = 0xFF; + } + i32::from_le_bytes(buf) + }; + let right = I24::new(right_i32).ok_or_else(|| { + AudioError::ProcessingError(format!("Invalid I24 value: {}", right_i32)) + })?; + + stereo.push([left, right]); + } + let chunk_data = AudioChunkData::new(stereo, sample_rate, 0.0); + AudioChunk::I24(chunk_data) + } + 32 => { + // Type I32 + let mut stereo = Vec::with_capacity(frames); + for frame_idx in 0..frames { + let base = frame_idx * frame_bytes; + let left = i32::from_le_bytes([ + pcm_data[base], + pcm_data[base + 1], + pcm_data[base + 2], + pcm_data[base + 3], + ]); + let right = i32::from_le_bytes([ + pcm_data[base + 4], + pcm_data[base + 5], + pcm_data[base + 6], + pcm_data[base + 7], + ]); + stereo.push([left, right]); + } + let chunk_data = AudioChunkData::new(stereo, sample_rate, 0.0); + AudioChunk::I32(chunk_data) + } + _ => { + return Err(AudioError::ProcessingError(format!( + "Unsupported bit depth: {}", + bits_per_sample + ))) + } + }; + + Ok(Arc::new(AudioSegment { + order, + timestamp_sec, + segment: _AudioSegment::Chunk(Arc::new(chunk)), + })) +} + +/// Convertit Song en TrackMetadata +/// +/// Cette fonction est synchrone, donc on wrap la metadata dans Arc> +/// et on spawn une tâche async pour la configurer +fn song_to_metadata(song: &Song, block: &Block) -> Arc> { + let metadata = MemoryTrackMetadata::new(); + let metadata_arc = Arc::new(RwLock::new(metadata)) as Arc>; + let metadata_clone = metadata_arc.clone(); + + // Clone des données pour la task async + let title = song.title.clone(); + let artist = song.artist.clone(); + let album = song.album.clone(); + let year = song.year; + let cover_url = song.cover.as_ref().and_then(|cover| block.cover_url(cover)); + + // Configurer les métadonnées de manière asynchrone + tokio::spawn(async move { + let mut meta = metadata_clone.write().await; + + // Ces méthodes peuvent échouer (retournent Result), donc on propage avec ? + if let Err(e) = meta.set_title(Some(title)).await { + eprintln!("Warning: Failed to set title: {}", e); + } + if let Err(e) = meta.set_artist(Some(artist)).await { + eprintln!("Warning: Failed to set artist: {}", e); + } + if let Some(album) = album { + if let Err(e) = meta.set_album(Some(album)).await { + eprintln!("Warning: Failed to set album: {}", e); + } + } + if let Some(year) = year { + if let Err(e) = meta.set_year(Some(year)).await { + eprintln!("Warning: Failed to set year: {}", e); + } + } + if let Some(cover_url) = cover_url { + if let Err(e) = meta.set_cover_url(Some(cover_url)).await { + eprintln!("Warning: Failed to set cover_url: {}", e); + } + } + }); + + metadata_arc +} + +#[async_trait::async_trait] +impl NodeLogic for RadioParadiseStreamSourceLogic { + async fn process( + &mut self, + _input: Option>>, + output: Vec>>, + stop_token: CancellationToken, + ) -> Result<(), AudioError> { + let mut order = 0u64; + + loop { + // Attendre un block ID (timeout court pour une radio) + let event_id = match tokio::time::timeout( + Duration::from_secs(BLOCK_ID_TIMEOUT_SECS), + async { + while self.block_queue.is_empty() { + tokio::time::sleep(Duration::from_millis(100)).await; + + if stop_token.is_cancelled() { + return None; + } + } + self.block_queue.pop_front() + } + ).await { + Ok(Some(id)) => id, + Ok(None) => break, // Cancelled + Err(_) => { + // Timeout - pas de nouveau bloc, on termine + break; + } + }; + + // Vérifier si déjà téléchargé récemment + if self.is_recent_block(event_id) { + continue; + } + + // Récupérer les métadonnées du bloc + let block = self.client + .get_block(Some(event_id)) + .await + .map_err(|e| AudioError::ProcessingError(format!("Failed to get block: {}", e)))?; + + // Marquer comme téléchargé + self.mark_block_downloaded(event_id); + + // Télécharger et décoder le bloc + self.download_and_decode_block(&block, &output, &stop_token, &mut order) + .await?; + } + + // Envoyer EndOfStream + let eos = AudioSegment::new_end_of_stream(order, 0.0); + for tx in &output { + tx.send(eos.clone()) + .await + .map_err(|_| AudioError::ChildDied)?; + } + + Ok(()) + } +} + +// ═══════════════════════════════════════════════════════════════════════════ +// RadioParadiseStreamSource - Wrapper utilisant Node +// ═══════════════════════════════════════════════════════════════════════════ + +pub struct RadioParadiseStreamSource { + inner: Node, +} + +impl RadioParadiseStreamSource { + /// Crée une nouvelle source Radio Paradise avec durée de chunk par défaut + pub fn new(client: RadioParadiseClient) -> Self { + Self::with_chunk_duration(client, DEFAULT_CHUNK_DURATION_MS as u32) + } + + /// Crée une nouvelle source avec durée de chunk personnalisée + pub fn with_chunk_duration(client: RadioParadiseClient, chunk_duration_ms: u32) -> Self { + let logic = RadioParadiseStreamSourceLogic::new(client, chunk_duration_ms); + Self { + inner: Node::new_source(logic), + } + } + + /// Ajoute un block ID à la file d'attente de téléchargement + pub fn push_block_id(&mut self, event_id: EventId) { + self.inner.logic_mut().push_block_id(event_id); + } +} + +#[async_trait::async_trait] +impl AudioPipelineNode for RadioParadiseStreamSource { + fn get_tx(&self) -> Option>> { + self.inner.get_tx() + } + + fn register(&mut self, child: Box) { + self.inner.register(child); + } + + async fn run(self: Box, stop_token: CancellationToken) -> Result<(), AudioError> { + Box::new(self.inner).run(stop_token).await + } +} + +impl TypedAudioNode for RadioParadiseStreamSource { + fn input_type(&self) -> Option { + None // Source node + } + + fn output_type(&self) -> Option { + // Radio Paradise FLAC peut être 16-bit, 24-bit, ou 32-bit + // La profondeur est détectée automatiquement depuis le header FLAC + Some(TypeRequirement::any_integer()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn create_test_client() -> RadioParadiseClient { + RadioParadiseClient::with_client(reqwest::Client::new()) + } + + #[test] + fn test_cache_fifo_basic() { + let client = create_test_client(); + let mut logic = RadioParadiseStreamSourceLogic::new(client, DEFAULT_CHUNK_DURATION_MS as u32); + + // Ajouter 5 blocs + for i in 1..=5 { + logic.mark_block_downloaded(i); + } + + // Vérifier que tous sont dans le cache + for i in 1..=5 { + assert!(logic.is_recent_block(i), "Block {} should be in cache", i); + } + assert_eq!(logic.recent_blocks.len(), 5); + } + + #[test] + fn test_cache_fifo_exactly_10_elements() { + let client = create_test_client(); + let mut logic = RadioParadiseStreamSourceLogic::new(client, DEFAULT_CHUNK_DURATION_MS as u32); + + // Ajouter exactement 10 blocs + for i in 1..=10 { + logic.mark_block_downloaded(i); + } + + // Vérifier qu'on a exactement 10 éléments + assert_eq!(logic.recent_blocks.len(), 10, "Cache should have exactly 10 elements"); + + // Tous devraient être dans le cache + for i in 1..=10 { + assert!(logic.is_recent_block(i), "Block {} should be in cache", i); + } + } + + #[test] + fn test_cache_fifo_eviction_oldest() { + let client = create_test_client(); + let mut logic = RadioParadiseStreamSourceLogic::new(client, DEFAULT_CHUNK_DURATION_MS as u32); + + // Remplir le cache avec 10 éléments (1..=10) + for i in 1..=10 { + logic.mark_block_downloaded(i); + } + + // Ajouter un 11ème élément + logic.mark_block_downloaded(11); + + // Le cache doit toujours avoir 10 éléments + assert_eq!(logic.recent_blocks.len(), 10, "Cache should still have 10 elements"); + + // Le premier (plus ancien) doit avoir été évincé + assert!(!logic.is_recent_block(1), "Oldest block (1) should be evicted"); + + // Les éléments 2..=11 doivent être présents + for i in 2..=11 { + assert!(logic.is_recent_block(i), "Block {} should be in cache", i); + } + } + + #[test] + fn test_cache_fifo_multiple_evictions() { + let client = create_test_client(); + let mut logic = RadioParadiseStreamSourceLogic::new(client, DEFAULT_CHUNK_DURATION_MS as u32); + + // Remplir avec 10 éléments + for i in 1..=10 { + logic.mark_block_downloaded(i); + } + + // Ajouter 5 éléments supplémentaires + for i in 11..=15 { + logic.mark_block_downloaded(i); + } + + // Toujours 10 éléments + assert_eq!(logic.recent_blocks.len(), 10, "Cache should have 10 elements"); + + // Les 5 premiers doivent avoir été évincés + for i in 1..=5 { + assert!(!logic.is_recent_block(i), "Block {} should be evicted", i); + } + + // Les éléments 6..=15 doivent être présents + for i in 6..=15 { + assert!(logic.is_recent_block(i), "Block {} should be in cache", i); + } + } + + #[test] + fn test_cache_never_exceeds_capacity() { + let client = create_test_client(); + let mut logic = RadioParadiseStreamSourceLogic::new(client, DEFAULT_CHUNK_DURATION_MS as u32); + + // Vérifier la capacité pré-allouée + assert_eq!(logic.recent_blocks.capacity(), RECENT_BLOCKS_CACHE_SIZE); + + // Ajouter beaucoup d'éléments + for i in 1..=100 { + logic.mark_block_downloaded(i); + + // À chaque itération, vérifier qu'on ne dépasse jamais 10 + assert!( + logic.recent_blocks.len() <= RECENT_BLOCKS_CACHE_SIZE, + "Cache size {} exceeded max {}", + logic.recent_blocks.len(), + RECENT_BLOCKS_CACHE_SIZE + ); + } + + // Finalement, on doit avoir exactement 10 éléments + assert_eq!(logic.recent_blocks.len(), 10); + + // Ce doivent être les 10 derniers (91..=100) + for i in 91..=100 { + assert!(logic.is_recent_block(i), "Block {} should be in cache", i); + } + } + + #[test] + fn test_cache_fifo_order_preserved() { + let client = create_test_client(); + let mut logic = RadioParadiseStreamSourceLogic::new(client, DEFAULT_CHUNK_DURATION_MS as u32); + + // Ajouter 10 éléments + for i in 1..=10 { + logic.mark_block_downloaded(i); + } + + // Vérifier l'ordre dans la VecDeque (le front devrait être le plus ancien) + let front = logic.recent_blocks.front().copied(); + assert_eq!(front, Some(1), "Front should be the oldest element"); + + let back = logic.recent_blocks.back().copied(); + assert_eq!(back, Some(10), "Back should be the newest element"); + } + + #[test] + fn test_block_queue_push() { + let client = create_test_client(); + let mut logic = RadioParadiseStreamSourceLogic::new(client, DEFAULT_CHUNK_DURATION_MS as u32); + + // Tester push_block_id + logic.push_block_id(100); + logic.push_block_id(200); + logic.push_block_id(300); + + assert_eq!(logic.block_queue.len(), 3); + assert_eq!(logic.block_queue.front(), Some(&100)); + assert_eq!(logic.block_queue.back(), Some(&300)); + } +} diff --git a/pmoparadise/src/source.rs b/pmoparadise/src/source.rs index d6b6e8cb..946af700 100644 --- a/pmoparadise/src/source.rs +++ b/pmoparadise/src/source.rs @@ -1,395 +1,114 @@ -//! Music source implementation for Radio Paradise built on the new -//! `paradise` orchestration layer. +//! DEPRECATED: Stub implementation of RadioParadiseSource //! -//! The source exposes a DIDL-Lite hierarchy compatible with UPnP -//! ContentDirectory while delegating block ingestion, caching and -//! multi-client streaming to [`ParadiseChannel`]. +//! **⚠️ This module is deprecated and will be removed in a future version.** +//! +//! The orchestration-based RadioParadiseSource has been replaced by +//! `RadioParadiseStreamSource`, which integrates directly with the pmoaudio +//! pipeline for streaming and decoding. +//! +//! ## Migration Guide +//! +//! **Old approach** (deprecated): +//! ```rust,ignore +//! use pmoparadise::RadioParadiseSource; +//! let source = RadioParadiseSource::from_registry(client)?; +//! ``` +//! +//! **New approach** (recommended): +//! ```rust,ignore +//! use pmoparadise::RadioParadiseStreamSource; +//! use pmoaudio::pipeline::Node; +//! +//! let stream_source = RadioParadiseStreamSource::new(client, None).await?; +//! let node = Node::from_logic(stream_source); +//! // Use node in pmoaudio pipeline +//! ``` +//! +//! This stub implementation is provided only for backward compatibility with +//! existing code (e.g., pmomediaserver) until it can be updated to use +//! RadioParadiseStreamSource. use crate::client::RadioParadiseClient; -use crate::paradise::{ - create_history_backend, ChannelDescriptor, ParadiseChannel, PlaylistEntry, ALL_CHANNELS, -}; - -#[cfg(not(feature = "pmoconfig"))] -use crate::paradise::HISTORY_DEFAULT_MAX_TRACKS; -use anyhow::Result as AnyhowResult; -use pmoaudiocache::Cache as AudioCache; -use pmocovers::Cache as CoverCache; -use pmodidl::{Container, Item, Resource}; -use pmosource::pmodidl; -use pmosource::{ - async_trait, BrowseResult, CacheStatus, MusicSource, MusicSourceError, Result, - SourceCacheManager, SourceStatistics, -}; -use std::collections::HashMap; -use std::sync::Arc; +use pmosource::pmodidl::{Container, Item}; +use pmosource::{async_trait, BrowseResult, MusicSource, MusicSourceError, Result}; use std::time::SystemTime; -use tracing::warn; -/// Default image for Radio Paradise (300x300 WebP, embedded in binary) +/// Default Radio Paradise image (embedded in binary) const DEFAULT_IMAGE: &[u8] = include_bytes!("../assets/default.webp"); -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 parse_channel_container_id(object_id: &str) -> Option { - 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 parse_track_channel(track_id: &str) -> Option { - let mut parts = track_id.split(':'); - match (parts.next(), parts.next(), parts.next(), parts.next()) { - (Some("rp"), Some(channel_str), Some(_rest), None) => channel_str.parse().ok(), - _ => None, - } -} - -fn format_duration(duration_seconds: u64) -> String { - let hours = duration_seconds / 3600; - let minutes = (duration_seconds % 3600) / 60; - let seconds = duration_seconds % 60; - format!("{hours}:{minutes:02}:{seconds:02}") -} - -#[derive(Clone)] +/// DEPRECATED: Stub implementation of RadioParadiseSource +/// +/// This is a minimal stub that implements the MusicSource trait with no-op +/// implementations. It exists only to maintain API compatibility during the +/// migration to RadioParadiseStreamSource. +/// +/// **Do not use this in new code.** Use `RadioParadiseStreamSource` instead. +#[derive(Clone, Debug)] pub struct RadioParadiseSource { - inner: Arc, -} - -struct RadioParadiseSourceInner { - channels: HashMap>, -} - -impl std::fmt::Debug for RadioParadiseSource { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("RadioParadiseSource").finish() - } + _client: RadioParadiseClient, } impl RadioParadiseSource { + /// DEPRECATED: Create a new RadioParadiseSource from registry + /// + /// This method is deprecated and will always return an error indicating + /// that the orchestration-based source is no longer supported. + /// + /// Use `RadioParadiseStreamSource` instead for audio streaming. #[cfg(feature = "server")] - pub fn from_registry(client: RadioParadiseClient) -> Result { - // Load history configuration from pmoconfig using the config extension trait - #[cfg(feature = "pmoconfig")] - let (database_path, history_max_tracks) = { - use crate::config_ext::RadioParadiseConfigExt; - let cfg = pmoconfig::get_config(); - let database_path = cfg.get_paradise_history_database().map_err(|e| { - MusicSourceError::SourceUnavailable(format!( - "Failed to get history database path: {}", - e - )) - })?; - let max_tracks = cfg.get_paradise_history_size().map_err(|e| { - MusicSourceError::SourceUnavailable(format!("Failed to get history size: {}", e)) - })?; - (database_path, max_tracks) - }; - - #[cfg(not(feature = "pmoconfig"))] - let (database_path, history_max_tracks) = { - use std::path::PathBuf; - let mut path = PathBuf::from(std::env::var("HOME").unwrap_or_else(|_| ".".to_string())); - path.push(".config"); - path.push("pmo"); - path.push("paradise"); - std::fs::create_dir_all(&path).ok(); - path.push("history.db"); - ( - path.to_string_lossy().to_string(), - HISTORY_DEFAULT_MAX_TRACKS, - ) - }; - - let history_backend = create_history_backend(&database_path).map_err(|e| { - MusicSourceError::SourceUnavailable(format!( - "Failed to initialize history backend: {}", - e - )) - })?; - let mut channels = HashMap::new(); - - for descriptor in ALL_CHANNELS.iter() { - let cache_manager = Arc::new(SourceCacheManager::from_registry( - channel_collection_id(descriptor.id), - )?); - let channel = Arc::new( - ParadiseChannel::new( - *descriptor, - client.clone(), - history_max_tracks, - history_backend.clone(), - cache_manager, - ) - .map_err(|e| { - MusicSourceError::SourceUnavailable(format!( - "Failed to initialize channel {}: {e}", - descriptor.slug - )) - })?, - ); - channels.insert(descriptor.id, channel); - } - - Ok(Self { - inner: Arc::new(RadioParadiseSourceInner { channels }), - }) + pub fn from_registry(_client: RadioParadiseClient) -> Result { + Err(MusicSourceError::SourceUnavailable( + "RadioParadiseSource is deprecated. Use RadioParadiseStreamSource instead." + .to_string(), + )) } + /// DEPRECATED: Create a new RadioParadiseSource from registry with defaults + /// + /// This method creates a stub instance that will log deprecation warnings + /// but allows existing code to compile. + /// + /// Use `RadioParadiseStreamSource` instead for audio streaming. #[cfg(feature = "server")] - pub fn from_registry_default(client: RadioParadiseClient) -> Result { - Self::from_registry(client) + pub fn from_registry_default(client: RadioParadiseClient) -> Self { + tracing::warn!( + "RadioParadiseSource::from_registry_default is deprecated. \ + Use RadioParadiseStreamSource for audio streaming." + ); + Self { _client: client } } - pub fn new( - client: RadioParadiseClient, - cover_cache: Arc, - audio_cache: Arc, - ) -> Self { - // Load history configuration from pmoconfig using the config extension trait - #[cfg(feature = "pmoconfig")] - let (database_path, history_max_tracks) = { - use crate::config_ext::RadioParadiseConfigExt; - let cfg = pmoconfig::get_config(); - let database_path = cfg.get_paradise_history_database().unwrap_or_else(|e| { - panic!("Failed to get history database path: {e}"); - }); - let max_tracks = cfg.get_paradise_history_size().unwrap_or_else(|e| { - panic!("Failed to get history size: {e}"); - }); - (database_path, max_tracks) - }; - - #[cfg(not(feature = "pmoconfig"))] - let (database_path, history_max_tracks) = { - use std::path::PathBuf; - let mut path = PathBuf::from(std::env::var("HOME").unwrap_or_else(|_| ".".to_string())); - path.push(".config"); - path.push("pmo"); - path.push("paradise"); - std::fs::create_dir_all(&path).ok(); - path.push("history.db"); - ( - path.to_string_lossy().to_string(), - HISTORY_DEFAULT_MAX_TRACKS, - ) - }; - - let history_backend: Arc = - create_history_backend(&database_path).unwrap_or_else(|err| { - panic!("Failed to initialize history backend: {err}"); - }); - let mut channels = HashMap::new(); - - for descriptor in ALL_CHANNELS.iter() { - let cache_manager = Arc::new(SourceCacheManager::new( - channel_collection_id(descriptor.id), - Arc::clone(&cover_cache), - Arc::clone(&audio_cache), - )); - match ParadiseChannel::new( - *descriptor, - client.clone(), - history_max_tracks, - history_backend.clone(), - cache_manager, - ) { - Ok(channel) => { - channels.insert(descriptor.id, Arc::new(channel)); - } - Err(err) => { - warn!( - channel = descriptor.slug, - "Failed to initialize channel: {err:?}" - ); - } - } - } - - Self { - inner: Arc::new(RadioParadiseSourceInner { channels }), - } + /// DEPRECATED: Create a new RadioParadiseSource with default settings + /// + /// This method is deprecated and only exists for API compatibility. + pub fn new_default(client: RadioParadiseClient) -> Self { + tracing::warn!( + "RadioParadiseSource::new_default is deprecated. \ + Use RadioParadiseStreamSource for audio streaming." + ); + Self { _client: client } } - pub fn new_default( - client: RadioParadiseClient, - cover_cache: Arc, - audio_cache: Arc, - ) -> Self { - Self::new(client, cover_cache, audio_cache) - } - - pub fn client_for_channel(&self, channel: u8) -> Option { - self.inner - .channels - .get(&channel) - .map(|ch| ch.client().clone()) - } - - pub fn channel(&self, id: u8) -> Option> { - self.inner.channels.get(&id).cloned() - } - - 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(ALL_CHANNELS.len().to_string()), - searchable: Some("1".to_string()), - title: "Radio Paradise".to_string(), - class: "object.container".to_string(), - containers: vec![], - items: vec![], - } - } - - async fn build_channel_containers(&self) -> Vec { - let mut containers = Vec::new(); - for descriptor in ALL_CHANNELS.iter() { - if let Some(channel) = self.channel(descriptor.id) { - let len = channel.playlist().active_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(len.to_string()), - searchable: Some("1".to_string()), - title: descriptor.display_name.to_string(), - class: "object.container.playlistContainer".to_string(), - containers: vec![], - items: vec![], - }); - } - } - containers - } - - async fn channel_items( - &self, - descriptor: ChannelDescriptor, - offset: usize, - limit: Option, - ) -> Result> { - let channel = self - .channel(descriptor.id) - .ok_or_else(|| MusicSourceError::ObjectNotFound(descriptor.slug.to_string()))?; - - channel - .ensure_started() - .await - .map_err(|e| MusicSourceError::SourceUnavailable(e.to_string()))?; - - let entries = channel.playlist().active_snapshot().await; - if entries.is_empty() || offset >= entries.len() { - return Ok(Vec::new()); - } - - let end = limit - .map(|count| offset + count) - .unwrap_or(entries.len()) - .min(entries.len()); - - let parent_id = channel_container_id(descriptor.id); - - let mut items = Vec::with_capacity(end - offset); - for entry in entries.into_iter().skip(offset).take(end - offset) { - match self.entry_to_item(channel.clone(), &parent_id, entry).await { - Ok(item) => items.push(item), - Err(err) => warn!( - channel = descriptor.slug, - "Failed to build DIDL item: {err:?}" - ), - } - } - Ok(items) - } - - async fn entry_to_item( - &self, - channel: Arc, - parent_id: &str, - entry: Arc, - ) -> AnyhowResult { - let cache_manager = channel.cache_manager(); - let metadata = cache_manager.get_metadata(&entry.track_id).await; - - let resource_url = cache_manager - .resolve_uri(&entry.track_id) - .await - .or_else(|_| { - metadata - .as_ref() - .map(|meta| meta.original_uri.clone()) - .ok_or_else(|| MusicSourceError::ObjectNotFound(entry.track_id.clone())) - })?; - - let mut album_art = metadata - .as_ref() - .and_then(|meta| meta.cached_cover_pk.as_ref()) - .and_then(|pk| cache_manager.cover_url(pk, None).ok()); - - if album_art.is_none() { - album_art = entry.song.cover.clone(); - } - - let duration_seconds = entry.duration_ms / 1000; - let duration_str = if duration_seconds > 0 { - Some(format_duration(duration_seconds as u64)) - } else { - None - }; - - let resource = Resource { - protocol_info: "http-get:*:audio/flac:*".to_string(), - bits_per_sample: None, - sample_frequency: None, - nr_audio_channels: None, - duration: duration_str.clone(), - url: resource_url, - }; - - Ok(Item { - id: entry.track_id.clone(), - parent_id: parent_id.to_string(), - restricted: Some("1".to_string()), - title: entry.song.title.clone(), - creator: Some(entry.song.artist.clone()), - class: "object.item.audioItem.musicTrack".to_string(), - artist: Some(entry.song.artist.clone()), - album: entry.song.album.clone(), - genre: None, - album_art, - album_art_pk: None, - date: None, - original_track_number: None, - resources: vec![resource], - descriptions: vec![], - }) - } - - fn channels_iter(&self) -> impl Iterator)> { - self.inner.channels.iter() + /// DEPRECATED: Create a new RadioParadiseSource with cache + /// + /// This method is deprecated and only exists for API compatibility. + pub fn new_with_cache(client: RadioParadiseClient, _cache_size: usize) -> Self { + tracing::warn!( + "RadioParadiseSource::new_with_cache is deprecated. \ + Use RadioParadiseStreamSource for audio streaming." + ); + Self { _client: client } } } #[async_trait] impl MusicSource for RadioParadiseSource { fn name(&self) -> &str { - "Radio Paradise" + "Radio Paradise (DEPRECATED)" } fn id(&self) -> &str { - "radio-paradise" + "radio-paradise-deprecated" } fn default_image(&self) -> &[u8] { @@ -397,43 +116,32 @@ impl MusicSource for RadioParadiseSource { } async fn root_container(&self) -> Result { - Ok(self.build_root_container()) + Ok(Container { + id: "radio-paradise-deprecated".to_string(), + parent_id: "0".to_string(), + restricted: Some("1".to_string()), + child_count: Some("0".to_string()), + searchable: Some("0".to_string()), + title: "Radio Paradise (DEPRECATED)".to_string(), + class: "object.container".to_string(), + containers: vec![], + items: vec![], + }) } - async fn browse(&self, object_id: &str) -> Result { - 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 descriptor = ALL_CHANNELS - .iter() - .find(|desc| desc.id == channel_id) - .ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?; - - let items = self.channel_items(*descriptor, 0, None).await?; - Ok(BrowseResult::Items(items)) - } else { - Err(MusicSourceError::ObjectNotFound(object_id.to_string())) - } - } - } + async fn browse(&self, _object_id: &str) -> Result { + tracing::warn!("RadioParadiseSource::browse called but source is deprecated"); + Ok(BrowseResult::Mixed { + containers: vec![], + items: vec![], + }) } - async fn resolve_uri(&self, object_id: &str) -> Result { - let channel_id = parse_track_channel(object_id) - .ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?; - let channel = self - .channel(channel_id) - .ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?; - channel - .cache_manager() - .resolve_uri(object_id) - .await - .map_err(|e| MusicSourceError::CacheError(e.to_string())) + async fn resolve_uri(&self, _object_id: &str) -> Result { + Err(MusicSourceError::SourceUnavailable( + "RadioParadiseSource is deprecated. Use RadioParadiseStreamSource instead." + .to_string(), + )) } fn supports_fifo(&self) -> bool { @@ -441,171 +149,25 @@ impl MusicSource for RadioParadiseSource { } async fn append_track(&self, _track: Item) -> Result<()> { - Err(MusicSourceError::FifoNotSupported) + Err(MusicSourceError::SourceUnavailable( + "RadioParadiseSource is deprecated and does not support FIFO operations." + .to_string(), + )) } async fn remove_oldest(&self) -> Result> { - Err(MusicSourceError::FifoNotSupported) + Ok(None) } async fn update_id(&self) -> u32 { - self.channels_iter() - .map(|(_, channel)| channel.playlist().update_id()) - .max() - .unwrap_or(0) + 0 } async fn last_change(&self) -> Option { - let mut latest: Option = None; - for (_, channel) in self.channels_iter() { - if let Some(change) = channel.playlist().last_change().await { - latest = Some(match latest { - Some(current) if change <= current => current, - _ => change, - }); - } - } - latest + None } - async fn get_items(&self, offset: usize, count: usize) -> Result> { - let mut all = Vec::new(); - for descriptor in ALL_CHANNELS.iter() { - let mut items = self.channel_items(*descriptor, 0, None).await?; - all.append(&mut items); - } - - if offset >= all.len() { - return Ok(Vec::new()); - } - - let end = if count == 0 { - all.len() - } else { - (offset + count).min(all.len()) - }; - - Ok(all.into_iter().skip(offset).take(end - offset).collect()) - } - - async fn get_available_formats(&self, _object_id: &str) -> Result> { - Ok(vec![pmosource::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 { - let channel_id = parse_track_channel(object_id) - .ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?; - let channel = self - .channel(channel_id) - .ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?; - channel - .cache_manager() - .get_cache_status(object_id) - .await - .map_err(|e| MusicSourceError::CacheError(e.to_string())) - } - - async fn cache_item(&self, object_id: &str) -> Result { - let channel_id = parse_track_channel(object_id) - .ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?; - let channel = self - .channel(channel_id) - .ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?; - channel - .cache_manager() - .get_cache_status(object_id) - .await - .map_err(|e| MusicSourceError::CacheError(e.to_string())) - } - - async fn browse_paginated( - &self, - object_id: &str, - offset: usize, - limit: usize, - ) -> Result { - 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 descriptor = ALL_CHANNELS - .iter() - .find(|desc| desc.id == channel_id) - .ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?; - - let items = self.channel_items(*descriptor, offset, Some(limit)).await?; - Ok(BrowseResult::Items(items)) - } else { - Err(MusicSourceError::ObjectNotFound(object_id.to_string())) - } - } - } - } - - async fn get_item_count(&self, object_id: &str) -> Result { - match object_id { - "0" => Ok(1), - "radio-paradise" => Ok(ALL_CHANNELS.len()), - _ => { - if let Some(channel_id) = parse_channel_container_id(object_id) { - let channel = self - .channel(channel_id) - .ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?; - Ok(channel.playlist().active_len().await) - } else { - Err(MusicSourceError::ObjectNotFound(object_id.to_string())) - } - } - } - } - - async fn statistics(&self) -> Result { - let mut total_tracks = 0usize; - let mut cached_tracks = 0usize; - - for (_, channel) in self.channels_iter() { - total_tracks += channel.playlist().active_len().await; - let stats = channel.cache_manager().statistics().await; - cached_tracks += stats.cached_tracks; - } - - Ok(SourceStatistics { - total_items: Some(total_tracks), - total_containers: Some(ALL_CHANNELS.len() + 1), - cached_items: Some(cached_tracks), - cache_size_bytes: None, - }) + async fn get_items(&self, _offset: usize, _count: usize) -> Result> { + Ok(vec![]) } } diff --git a/pmoplaylist/src/handle/read.rs b/pmoplaylist/src/handle/read.rs index dec1433b..ffd762a6 100644 --- a/pmoplaylist/src/handle/read.rs +++ b/pmoplaylist/src/handle/read.rs @@ -4,7 +4,7 @@ use crate::playlist::Playlist; use crate::track::PlaylistTrack; use crate::Result; use pmocache::cache_trait::FileCache; -use pmodidl::{Container, Item, Resource}; +use pmodidl::{Container, Item}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; diff --git a/pmoplaylist/src/handle/write.rs b/pmoplaylist/src/handle/write.rs index 0d02aec9..1469c905 100644 --- a/pmoplaylist/src/handle/write.rs +++ b/pmoplaylist/src/handle/write.rs @@ -1,6 +1,5 @@ //! WriteHandle : accès exclusif en écriture à une playlist -use crate::playlist::core::PlaylistConfig; use crate::playlist::record::Record; use crate::playlist::Playlist; use crate::Result; @@ -185,7 +184,7 @@ impl WriteHandle { // Créer la nouvelle playlist persistante let manager = crate::manager::PlaylistManager(); - let mut new_handle = manager.create_persistent_playlist(new_id).await?; + let new_handle = manager.create_persistent_playlist(new_id).await?; // Copier le titre et la config new_handle.set_title(title).await?; diff --git a/pmoplaylist/src/playlist/mod.rs b/pmoplaylist/src/playlist/mod.rs index 1e9cc5be..a1e21025 100644 --- a/pmoplaylist/src/playlist/mod.rs +++ b/pmoplaylist/src/playlist/mod.rs @@ -4,7 +4,6 @@ pub mod core; pub mod record; use self::core::{PlaylistConfig, PlaylistCore}; -use self::record::Record; use std::sync::atomic::{AtomicU8, Ordering}; use std::sync::{Arc, Weak}; use std::time::SystemTime; diff --git a/pmoserver/src/logs/mod.rs b/pmoserver/src/logs/mod.rs index 919a9fa2..80a4b3e2 100644 --- a/pmoserver/src/logs/mod.rs +++ b/pmoserver/src/logs/mod.rs @@ -25,11 +25,10 @@ use tracing::Level; use tracing_subscriber::{ Registry, filter::LevelFilter, - layer::{Filter, SubscriberExt}, + layer::SubscriberExt, reload, util::SubscriberInitExt, }; -use utoipa::OpenApi; /// Représente une entrée de log #[derive(Debug, Clone, Serialize)] @@ -440,7 +439,7 @@ fn level_to_levelfilter(level: Level) -> LevelFilter { /// Crée le router pour l'API de gestion des logs pub fn create_logs_router(log_state: LogState) -> axum::Router { - use axum::routing::{get, post}; + use axum::routing::get; axum::Router::new() .route("/log_setup", get(log_setup_get).post(log_setup_post)) .with_state(log_state) diff --git a/pmosource/src/cache.rs b/pmosource/src/cache.rs index f0e58b36..852575c0 100755 --- a/pmosource/src/cache.rs +++ b/pmosource/src/cache.rs @@ -122,10 +122,10 @@ impl SourceCacheManager { let cache = self.track_cache.read().await; if let Some(metadata) = cache.get(object_id) { - if let Some(ref pk) = metadata.cached_audio_pk { + if let Some(ref _pk) = metadata.cached_audio_pk { #[cfg(feature = "server")] { - let url = pmoupnp::cache_registry::build_audio_url(pk, Some("stream")) + let url = pmoupnp::cache_registry::build_audio_url(_pk, Some("stream")) .map_err(|e| MusicSourceError::CacheError(e.to_string()))?; return Ok(url); } @@ -147,7 +147,7 @@ impl SourceCacheManager { let cache = self.track_cache.read().await; if let Some(metadata) = cache.get(object_id) { - if let Some(ref pk) = metadata.cached_audio_pk { + if let Some(ref _pk) = metadata.cached_audio_pk { // TODO: Ajouter get_info() à AudioCache // Pour l'instant, on retourne juste Cached sans taille return Ok(CacheStatus::Cached { size_bytes: 0 }); @@ -181,10 +181,10 @@ impl SourceCacheManager { /// # Returns /// /// L'URL complète de l'image - pub fn cover_url(&self, pk: &str, size: Option) -> Result { + pub fn cover_url(&self, _pk: &str, _size: Option) -> Result { #[cfg(feature = "server")] { - pmoupnp::cache_registry::build_cover_url(pk, size) + pmoupnp::cache_registry::build_cover_url(_pk, _size) .map_err(|e| MusicSourceError::CacheError(e.to_string())) } #[cfg(not(feature = "server"))] diff --git a/pmoupnp/src/actions/action_instance.rs b/pmoupnp/src/actions/action_instance.rs index 733accbe..a08d00fe 100644 --- a/pmoupnp/src/actions/action_instance.rs +++ b/pmoupnp/src/actions/action_instance.rs @@ -1,10 +1,8 @@ use std::{ collections::{HashMap, HashSet}, - env::var, sync::Arc, }; -use bevy_reflect::Reflect; use xmltree::{Element, XMLNode}; use crate::actions::{Action, ActionData, ActionInstance, ArgInstanceSet}; diff --git a/pmoupnp/src/actions/action_methods.rs b/pmoupnp/src/actions/action_methods.rs index 236c336c..3110c0d9 100644 --- a/pmoupnp/src/actions/action_methods.rs +++ b/pmoupnp/src/actions/action_methods.rs @@ -1,6 +1,6 @@ use std::sync::Arc; -use tracing::{info, trace}; +use tracing::info; use xmltree::{Element, XMLNode}; use crate::actions::{Action, ActionHandler, ActionInstance, Argument, ArgumentSet}; diff --git a/pmoupnp/src/actions/errors.rs b/pmoupnp/src/actions/errors.rs index f332e772..37d785c7 100644 --- a/pmoupnp/src/actions/errors.rs +++ b/pmoupnp/src/actions/errors.rs @@ -17,21 +17,3 @@ impl From for ActionError { ActionError::GeneralError(format!("IO error: {}", err)) } } - -#[derive(Error, Debug)] -pub enum ArgumentError { - #[error("Argument error: {0}")] - GeneralError(String), - - #[error("Argument error: {0}")] - ArgumentError(String), - - #[error("Set operation error: {0}")] - SetError(String), -} - -impl From for ArgumentError { - fn from(err: std::io::Error) -> Self { - ArgumentError::GeneralError(format!("IO error: {}", err)) - } -} diff --git a/pmoupnp/src/services/service_methods.rs b/pmoupnp/src/services/service_methods.rs index 4d4a6120..3d3e5374 100644 --- a/pmoupnp/src/services/service_methods.rs +++ b/pmoupnp/src/services/service_methods.rs @@ -47,21 +47,21 @@ impl UpnpObject for Service { elem.children.push(XMLNode::Element(service_id)); // SCPDURL - let mut SCPDURL = Element::new("SCPDURL"); - SCPDURL.children.push(XMLNode::Text(self.scpd_route())); - elem.children.push(XMLNode::Element(SCPDURL)); + let mut scpdurl = Element::new("SCPDURL"); + scpdurl.children.push(XMLNode::Text(self.scpd_route())); + elem.children.push(XMLNode::Element(scpdurl)); // controlURL - let mut controlURL = Element::new("controlURL"); - controlURL + let mut control_url = Element::new("controlURL"); + control_url .children .push(XMLNode::Text(self.control_route())); - elem.children.push(XMLNode::Element(controlURL)); + elem.children.push(XMLNode::Element(control_url)); // eventSubURL - let mut eventSubURL = Element::new("eventSubURL"); - eventSubURL.children.push(XMLNode::Text(self.event_route())); - elem.children.push(XMLNode::Element(eventSubURL)); + let mut event_sub_url = Element::new("eventSubURL"); + event_sub_url.children.push(XMLNode::Text(self.event_route())); + elem.children.push(XMLNode::Element(event_sub_url)); elem } diff --git a/pmoutils/Cargo.toml b/pmoutils/Cargo.toml index 9280a7c1..2c9da0e5 100644 --- a/pmoutils/Cargo.toml +++ b/pmoutils/Cargo.toml @@ -6,6 +6,6 @@ edition = "2024" [dependencies] get_if_addrs = "0.5.3" os_info = "3.8" -netstat2 = "0.9" +netstat2 = "0.11.2" sysinfo = "0.30" users = "0.11"