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

Implémentation de la détection de flux continu et gestion de la barre de progression

Browse Source 
Cette mise à jour implémente la détection des flux continus (radio, streaming) et améliore la gestion de la barre de progression pour ces flux. Les changements incluent : 
- Ajout d'une méthode predicat `is_playing_a_stream()` au niveau de `MusicRenderer`
- Développement d'une fonction utilitaire `is_continuous_stream_url()` pour analyser les headers HTTP
- Mise à jour des backends (UPnP, OpenHome, LinkPlay, ArylicTcp, Chromecast) pour détecter les flux continus
- Ajout d'un flag `continuous_stream` dans chaque backend
- Export du module de détection de flux
- Mise à jour de l'API REST et des événements SSE pour transmettre l'état du flux
- Correction de la gestion de la position de lecture pour les flux continus

Les flux continus sont maintenant correctement détectés via une analyse HTTP des headers, permettant une gestion appropriée de la barre de progression dans l'interface web.
This commit is contained in:
Eric Coissac
2026-01-30 17:35:54 +01:00
parent 2d13556b17
commit ea5328450b
20 changed files with 789 additions and 373 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
pmocontrol/src/model.rs
View File

@@ -435,6 +435,10 @@ pub enum RendererEvent {
id: DeviceId,
binding: Option<PlaylistBinding>,
},
StreamStateChanged {
id: DeviceId,
is_stream: bool,
},
TimerStarted {
id: DeviceId,
duration_seconds: u32,

10
pmocontrol/src/music_renderer/arylic_tcp.rs
View File

@@ -45,6 +45,8 @@ pub struct ArylicTcpRenderer {
port: u16,
timeout: Duration,
queue: Arc<Mutex<MusicQueue>>,
/// Flag indicating if currently playing a continuous stream (radio without duration)
continuous_stream: Arc<Mutex<bool>>,
}
impl ArylicTcpRenderer {
@@ -123,6 +125,7 @@ impl RendererFromMediaRendererInfo for ArylicTcpRenderer {
port: ARYLIC_TCP_PORT,
timeout: Duration::from_secs(DEFAULT_TIMEOUT_SECS),
queue,
continuous_stream: Arc::new(Mutex::new(false)),
})
}
@@ -131,6 +134,13 @@ impl RendererFromMediaRendererInfo for ArylicTcpRenderer {
}
}
impl ArylicTcpRenderer {
/// Returns true if currently playing a continuous stream (radio without duration)
pub fn is_continuous_stream(&self) -> bool {
*self.continuous_stream.lock().unwrap()
}
}
impl TransportControl for ArylicTcpRenderer {
fn play_uri(&self, _uri: &str, _meta: &str) -> Result<(), ControlPointError> {
Err(ControlPointError::upnp_operation_not_supported(

19
pmocontrol/src/music_renderer/chromecast_renderer.rs
View File

@@ -59,6 +59,8 @@ pub struct ChromecastRenderer {
/// Wrapped in Arc<Mutex> to allow cloning and proper thread lifecycle management.
thread_handle: Arc<Mutex<Option<JoinHandle<()>>>>,
queue: Arc<Mutex<MusicQueue>>,
/// Flag indicating if currently playing a continuous stream (radio without duration)
continuous_stream: Arc<Mutex<bool>>,
}
impl std::fmt::Debug for ChromecastRenderer {
@@ -156,6 +158,7 @@ impl RendererFromMediaRendererInfo for ChromecastRenderer {
stop_signal,
thread_handle,
queue,
continuous_stream: Arc::new(Mutex::new(false)),
})
}
@@ -164,10 +167,26 @@ impl RendererFromMediaRendererInfo for ChromecastRenderer {
}
}
impl ChromecastRenderer {
/// Returns true if currently playing a continuous stream (radio without duration)
pub fn is_continuous_stream(&self) -> bool {
*self.continuous_stream.lock().unwrap()
}
}
impl TransportControl for ChromecastRenderer {
fn play_uri(&self, uri: &str, meta: &str) -> Result<(), ControlPointError> {
debug!("ChromecastRenderer: play_uri({})", uri);
// Détecte si l'URL est un flux continu
let is_stream = crate::music_renderer::is_continuous_stream_url(uri);
*self.continuous_stream.lock().unwrap() = is_stream;
tracing::debug!(
"ChromecastRenderer play_uri: URI={}, continuous_stream={}",
uri,
is_stream
);
// Signal any existing play thread to stop
if let Ok(mut stop) = self.stop_signal.lock() {
*stop = true;

19
pmocontrol/src/music_renderer/linkplay_renderer.rs
View File

@@ -28,6 +28,8 @@ pub struct LinkPlayRenderer {
host: String,
timeout: Duration,
queue: Arc<Mutex<MusicQueue>>,
/// Flag indicating if currently playing a continuous stream (radio without duration)
continuous_stream: Arc<Mutex<bool>>,
}
impl fmt::Debug for LinkPlayRenderer {
@@ -77,6 +79,7 @@ impl RendererFromMediaRendererInfo for LinkPlayRenderer {
host,
timeout: Duration::from_secs(DEFAULT_HTTP_TIMEOUT_SECS),
queue,
continuous_stream: Arc::new(Mutex::new(false)),
})
}
@@ -85,8 +88,24 @@ impl RendererFromMediaRendererInfo for LinkPlayRenderer {
}
}
impl LinkPlayRenderer {
/// Returns true if currently playing a continuous stream (radio without duration)
pub fn is_continuous_stream(&self) -> bool {
*self.continuous_stream.lock().unwrap()
}
}
impl TransportControl for LinkPlayRenderer {
fn play_uri(&self, uri: &str, _meta: &str) -> Result<(), ControlPointError> {
// Détecte si l'URL est un flux continu
let is_stream = crate::music_renderer::is_continuous_stream_url(uri);
*self.continuous_stream.lock().unwrap() = is_stream;
tracing::debug!(
"LinkPlayRenderer play_uri: URI={}, continuous_stream={}",
uri,
is_stream
);
let encoded = percent_encode(uri);
self.send_player_command(&format!("play:{}", encoded))
}

2
pmocontrol/src/music_renderer/mod.rs
View File

@@ -11,6 +11,7 @@ mod chromecast_renderer;
mod musicrenderer;
mod sleep_timer;
mod stream_detection;
pub mod time_utils;
pub mod watcher;
@@ -21,6 +22,7 @@ pub use crate::music_renderer::capabilities::{
};
pub use crate::music_renderer::musicrenderer::{MusicRenderer, PlaylistBinding};
pub use crate::music_renderer::sleep_timer::SleepTimer;
pub use crate::music_renderer::stream_detection::is_continuous_stream_url;
use crate::{
RendererInfo, errors::ControlPointError, music_renderer::musicrenderer::MusicRendererBackend,
};

207
pmocontrol/src/music_renderer/musicrenderer.rs
View File

@@ -82,7 +82,7 @@ pub enum MusicRendererBackend {
}
/// Internal state for tracking playback and control flow.
#[derive(Debug, Clone, Default)]
#[derive(Debug, Clone)]
struct MusicRendererState {
/// Last known track metadata (cached to avoid repeated queries).
last_metadata: Option<TrackMetadata>,
@@ -96,6 +96,22 @@ struct MusicRendererState {
/// This prevents auto-advance on transient STOPPED states during track initialization.
/// Auto-advance is only allowed when this flag is true.
has_played_since_track_start: bool,
/// Timestamp when the current track started playing.
/// Used to calculate elapsed time when renderer returns unreliable position info.
track_start_time: Option<SystemTime>,
}
impl Default for MusicRendererState {
fn default() -> Self {
Self {
last_metadata: None,
playback_source: PlaybackSource::default(),
user_stop_requested: false,
sleep_timer: SleepTimer::default(),
has_played_since_track_start: false,
track_start_time: None,
}
}
}
#[derive(Clone)]
@@ -323,7 +339,72 @@ impl MusicRenderer {
.expect("WatchedState mutex poisoned");
let prev_position = watched.position.clone();
// Poll position every tick
// First poll to check for metadata changes BEFORE getting patched position
let raw_position = self
.lock_backend_for("poll_position")
.playback_position()
.ok();
// Detect and handle metadata/track changes FIRST
if let Some(ref raw_pos) = raw_position {
if let Some(metadata) = extract_track_metadata(raw_pos) {
let metadata_changed = watched
.metadata
.as_ref()
.map(|prev| prev != &metadata)
.unwrap_or(true);
if metadata_changed {
// Check if this is a track change (title/artist/album) to reset track_start_time
// Also initialize track_start_time on first metadata detection
let is_first_metadata = watched.metadata.is_none();
let track_changed = watched
.metadata
.as_ref()
.map(|prev| {
let title_changed = prev.title != metadata.title;
let artist_changed = prev.artist != metadata.artist;
let album_changed = prev.album != metadata.album;
if !is_first_metadata {
tracing::info!(
"MusicRenderer [{}]: Comparing - title_changed={} ({:?} vs {:?}), artist_changed={} ({:?} vs {:?})",
self.info.friendly_name(),
title_changed, prev.title, metadata.title,
artist_changed, prev.artist, metadata.artist
);
}
title_changed || artist_changed || album_changed
})
.unwrap_or(false); // Only true if there was previous metadata AND it differs
if is_first_metadata || track_changed {
tracing::info!(
"MusicRenderer [{}]: {} ({:?} -> {:?}), resetting track_start_time",
self.info.friendly_name(),
if is_first_metadata {
"First metadata"
} else {
"Track changed"
},
watched.metadata.as_ref().and_then(|m| m.title.as_ref()),
metadata.title
);
// This resets track_start_time BEFORE we calculate position
self.set_last_metadata(Some(metadata.clone()));
}
self.emit_event(RendererEvent::MetadataChanged {
id: self.id(),
metadata: metadata.clone(),
});
watched.metadata = Some(metadata);
}
}
}
// Now get the fully patched position (with correct track_start_time and rel_time)
if let Ok(position) = self.playback_position() {
let changed = watched
.position
@@ -338,29 +419,6 @@ impl MusicRenderer {
});
}
// Extract and emit metadata changes
if let Some(metadata) = extract_track_metadata(&position) {
let metadata_changed = watched
.metadata
.as_ref()
.map(|prev| prev != &metadata)
.unwrap_or(true);
if metadata_changed {
debug!(
renderer = self.info.friendly_name(),
title = metadata.title.as_deref(),
artist = metadata.artist.as_deref(),
"Emitting metadata changed event"
);
self.emit_event(RendererEvent::MetadataChanged {
id: self.id(),
metadata: metadata.clone(),
});
watched.metadata = Some(metadata);
}
}
watched.position = Some(position);
}
@@ -419,6 +477,16 @@ impl MusicRenderer {
watched.mute = Some(mute);
}
}
// Check stream state (every other tick to avoid excessive polling)
let is_stream = self.is_playing_a_stream();
if watched.is_stream != Some(is_stream) {
self.emit_event(RendererEvent::StreamStateChanged {
id: self.id(),
is_stream,
});
watched.is_stream = Some(is_stream);
}
}
}
@@ -573,6 +641,43 @@ impl MusicRenderer {
self.info.capabilities().supports_set_next()
}
/// Returns true if currently playing a continuous stream (radio without duration).
///
/// This method queries the backend to determine if the current playback is a continuous
/// stream. The detection is based on HTTP headers analysis performed when the URL was
/// set via play_uri or when the track changed (for OpenHome).
///
/// # Returns
///
/// `true` if:
/// - The renderer is currently playing AND
/// - The current track is a continuous stream (radio, live broadcast, etc.)
///
/// `false` otherwise (not playing, or playing a bounded media file)
pub fn is_playing_a_stream(&self) -> bool {
let backend = self.lock_backend_for("is_playing_a_stream");
// Check if currently playing
let is_playing = match backend.playback_state() {
Ok(PlaybackState::Playing) => true,
_ => false,
};
if !is_playing {
return false;
}
// Query backend for stream status
match &*backend {
MusicRendererBackend::Upnp(upnp) => upnp.is_continuous_stream(),
MusicRendererBackend::OpenHome(oh) => oh.is_continuous_stream(),
MusicRendererBackend::LinkPlay(lp) => lp.is_continuous_stream(),
MusicRendererBackend::ArylicTcp(ary) => ary.is_continuous_stream(),
MusicRendererBackend::Chromecast(cc) => cc.is_continuous_stream(),
MusicRendererBackend::HybridUpnpArylic { upnp, .. } => upnp.is_continuous_stream(),
}
}
/// Prepare the renderer for attaching a new playlist by clearing the queue and stopping playback.
pub fn clear_for_playlist_attach(&self) -> Result<(), ControlPointError> {
let mut backend = self.lock_backend_for("clear_for_playlist_attach");
@@ -755,19 +860,19 @@ impl MusicRenderer {
}
/// Get playback position
///
/// This method patches the position info from the backend:
/// - If track_duration is None, try to extract it from DIDL metadata
/// - Calculates rel_time from track_start_time (backend values are unreliable)
///
/// Note: track_start_time is updated by poll_and_emit_changes when track changes
pub fn playback_position(&self) -> Result<PlaybackPositionInfo, ControlPointError> {
let mut position_info = self
.lock_backend_for("playback_position")
.playback_position()?;
// Si track_duration est absent ou invalide, essayer de le parser depuis le DIDL metadata
let needs_duration_fix = position_info
.track_duration
.as_ref()
.map(|d| d == "00:00:00" || d == "0:00:00")
.unwrap_or(true); // None = true
if needs_duration_fix {
// Si track_duration est absent, essayer de le parser depuis le DIDL metadata
if position_info.track_duration.is_none() {
if let Some(ref metadata_xml) = position_info.track_metadata {
if let Some(duration) = parse_didl_duration(metadata_xml) {
tracing::debug!(
@@ -779,6 +884,29 @@ impl MusicRenderer {
}
}
// Calculate rel_time from track_start_time if available (backend values are unreliable)
if let Some(start_time) = self.track_start_time() {
if let Ok(elapsed) = start_time.elapsed() {
let secs = elapsed.as_secs() as u32;
let hours = secs / 3600;
let minutes = (secs % 3600) / 60;
let seconds = secs % 60;
let new_rel_time = format!("{:02}:{:02}:{:02}", hours, minutes, seconds);
tracing::info!(
"MusicRenderer: Patching rel_time: backend={:?} -> calculated={} (elapsed={}s)",
position_info.rel_time,
new_rel_time,
secs
);
position_info.rel_time = Some(new_rel_time);
}
} else {
tracing::info!(
"MusicRenderer: No track_start_time, keeping backend rel_time={:?}",
position_info.rel_time
);
}
Ok(position_info)
}
@@ -1076,8 +1204,19 @@ impl MusicRenderer {
}
/// Sets the last known track metadata.
/// Updates track_start_time only if the metadata actually changes.
pub fn set_last_metadata(&self, metadata: Option<TrackMetadata>) {
self.state.lock().unwrap().last_metadata = metadata;
let mut state = self.state.lock().unwrap();
let metadata_changed = state.last_metadata != metadata;
state.last_metadata = metadata;
if metadata_changed {
state.track_start_time = Some(SystemTime::now());
}
}
/// Gets the timestamp when the current track started playing.
pub fn track_start_time(&self) -> Option<SystemTime> {
self.state.lock().unwrap().track_start_time
}
/// Gets the current playback source.

43
pmocontrol/src/music_renderer/openhome_renderer.rs
View File

@@ -32,6 +32,10 @@ pub struct OpenHomeRenderer {
#[allow(dead_code)]
radio_client: Option<OhRadioClient>,
queue: Arc<Mutex<MusicQueue>>,
/// Flag indicating if currently playing a continuous stream (radio without duration)
continuous_stream: Arc<Mutex<bool>>,
/// Cached current track URI to detect track changes
current_track_uri: Arc<Mutex<Option<String>>>,
}
impl OpenHomeRenderer {
@@ -52,9 +56,16 @@ impl OpenHomeRenderer {
product_client,
radio_client,
queue,
continuous_stream: Arc::new(Mutex::new(false)),
current_track_uri: Arc::new(Mutex::new(None)),
}
}
/// Returns true if currently playing a continuous stream (radio without duration)
pub fn is_continuous_stream(&self) -> bool {
*self.continuous_stream.lock().unwrap()
}
pub fn has_playlist(&self) -> bool {
self.playlist.is_some()
}
@@ -316,8 +327,27 @@ impl PlaybackPosition for OpenHomeRenderer {
if let Some(info_client) = &self.info_client {
match info_client.track() {
Ok(track) => {
track_uri = Some(track.uri);
track_uri = Some(track.uri.clone());
track_metadata_xml = track.metadata_xml;
// Check if the URI has changed to detect track changes
let mut cached_uri = self.current_track_uri.lock().unwrap();
let uri_changed = cached_uri.as_ref() != Some(&track.uri);
if uri_changed {
tracing::debug!(
"OpenHome track URI changed: {:?} -> {:?}",
cached_uri,
track.uri
);
// Détecte si la nouvelle URL est un flux continu
let is_stream = crate::music_renderer::is_continuous_stream_url(&track.uri);
*self.continuous_stream.lock().unwrap() = is_stream;
tracing::debug!("OpenHome URI changed, continuous_stream={}", is_stream);
*cached_uri = Some(track.uri);
}
}
Err(err) => debug!(
// renderer = self.info.id.0.as_str(),
@@ -327,12 +357,9 @@ impl PlaybackPosition for OpenHomeRenderer {
}
}
// Get duration from Time service, but fall back to DIDL metadata if duration is 0
// Get duration from Time service - duration_secs=0 means stream (no duration)
let track_duration = if time_info.duration_secs == 0 {
// Try to extract duration from DIDL metadata
track_metadata_xml
.as_ref()
.and_then(|xml| parse_didl_duration_openhome(xml))
None
} else {
Some(format_hhmmss_u32(time_info.duration_secs))
};
@@ -366,10 +393,6 @@ fn parse_didl_duration_openhome(didl: &str) -> Option<String> {
let duration_offset = duration_start + "duration=\"".len();
if let Some(duration_end) = tag_attrs[duration_offset..].find('"') {
let duration = &tag_attrs[duration_offset..duration_offset + duration_end];
tracing::info!(
"OpenHome: Extracted duration from DIDL metadata: {}",
duration
);
return Some(duration.to_string());
}
}

148
pmocontrol/src/music_renderer/stream_detection.rs Normal file
View File

@@ -0,0 +1,148 @@
//! Stream detection utilities for identifying continuous streams (radio) vs bounded media.
//!
//! This module provides utilities to detect whether a given URL points to a continuous
//! stream (like a radio station) or a bounded media file by analyzing HTTP headers.
use std::time::Duration;
use tracing::{debug, trace};
use ureq::Agent;
/// Default timeout for HTTP HEAD requests when detecting stream type
const DEFAULT_STREAM_DETECTION_TIMEOUT_SECS: u64 = 3;
/// Détecte si une URL correspond à un flux continu (radio sans durée définie).
///
/// Cette fonction effectue une requête HTTP HEAD sur l'URL fournie et analyse les headers
/// de la réponse pour déterminer si c'est un flux continu ou un fichier avec durée définie.
///
/// # Critères de détection d'un flux continu :
///
/// - Absence de header `Content-Length` (pas de taille définie)
/// - OU présence de `Transfer-Encoding: chunked` sans `Content-Length`
/// - OU `Content-Type` indiquant un stream (audio/mpeg avec icy-*, application/ogg, etc.)
/// - OU présence de headers ICY (Icecast/Shoutcast) qui indiquent toujours un stream
///
/// # Arguments
///
/// * `url` - L'URL à analyser
///
/// # Returns
///
/// `true` si l'URL correspond à un flux continu, `false` sinon.
/// En cas d'erreur de connexion, retourne `false` par défaut (considéré comme non-stream).
pub fn is_continuous_stream_url(url: &str) -> bool {
// Quick checks on URL pattern before making HTTP request
if is_known_stream_pattern(url) {
debug!("URL {} matches known stream pattern", url);
return true;
}
// Make HTTP HEAD request to analyze headers
match check_stream_headers(url) {
Ok(is_stream) => {
trace!("Stream detection for {}: {}", url, is_stream);
is_stream
}
Err(e) => {
debug!(
"Failed to detect stream type for {}: {}, assuming non-stream",
url, e
);
false
}
}
}
/// Vérifie si l'URL correspond à un pattern connu de streaming
fn is_known_stream_pattern(url: &str) -> bool {
let url_lower = url.to_lowercase();
// Common streaming endpoints
url_lower.contains("/stream")
|| url_lower.contains("/live")
|| url_lower.contains("/radio")
|| url_lower.contains(".pls")
|| url_lower.contains(".m3u")
|| url_lower.ends_with(":8000")
|| url_lower.ends_with(":8080")
}
/// Effectue une requête HTTP HEAD et analyse les headers
fn check_stream_headers(url: &str) -> Result<bool, String> {
let agent: Agent = Agent::config_builder()
.timeout_global(Some(Duration::from_secs(
DEFAULT_STREAM_DETECTION_TIMEOUT_SECS,
)))
.build()
.into();
let response = agent
.head(url)
.call()
.map_err(|e| format!("HTTP HEAD request failed: {}", e))?;
// Check for ICY headers (Icecast/Shoutcast) - always indicates streaming
if response.headers().get("icy-name").is_some()
|| response.headers().get("icy-metaint").is_some()
|| response.headers().get("ice-audio-info").is_some()
{
debug!("ICY headers detected for {}, this is a stream", url);
return Ok(true);
}
// Check Content-Length
let has_content_length = response.headers().get("content-length").is_some();
// Check Transfer-Encoding
let is_chunked = response
.headers()
.get("transfer-encoding")
.and_then(|v| v.to_str().ok())
.map(|v| v.to_lowercase().contains("chunked"))
.unwrap_or(false);
// Check Content-Type for streaming indicators
let content_type = response
.headers()
.get("content-type")
.and_then(|v| v.to_str().ok())
.unwrap_or("");
let content_type_lower = content_type.to_lowercase();
let is_streaming_mime = content_type_lower.contains("audio/mpeg")
|| content_type_lower.contains("audio/aac")
|| content_type_lower.contains("audio/aacp")
|| content_type_lower.contains("application/ogg")
|| content_type_lower.contains("audio/ogg");
// Decision logic:
// - No Content-Length + streaming MIME = stream
// - Chunked encoding without Content-Length = likely stream
// - Has Content-Length = bounded media (not a stream)
let is_stream = !has_content_length && (is_streaming_mime || is_chunked);
trace!(
"Stream detection for {}: content-length={}, chunked={}, streaming_mime={}, is_stream={}",
url, has_content_length, is_chunked, is_streaming_mime, is_stream
);
Ok(is_stream)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_known_stream_patterns() {
assert!(is_known_stream_pattern("http://example.com/stream"));
assert!(is_known_stream_pattern("http://example.com/live"));
assert!(is_known_stream_pattern("http://example.com/radio.mp3"));
assert!(is_known_stream_pattern("http://example.com:8000/"));
assert!(is_known_stream_pattern("http://example.com/playlist.m3u"));
assert!(!is_known_stream_pattern("http://example.com/music.mp3"));
assert!(!is_known_stream_pattern("http://example.com/file.flac"));
}
}

60
pmocontrol/src/music_renderer/upnp_renderer.rs
View File

@@ -25,6 +25,8 @@ pub struct UpnpRenderer {
queue: Arc<Mutex<MusicQueue>>,
/// Durée extraite du DIDL-Lite (fallback si l'ampli ne la retourne pas)
cached_duration: Arc<Mutex<Option<String>>>,
/// Flag indicating if currently playing a continuous stream (radio without duration)
continuous_stream: Arc<Mutex<bool>>,
}
impl UpnpRenderer {
@@ -106,8 +108,14 @@ impl UpnpRenderer {
has_avtransport_set_next,
queue,
cached_duration: Arc::new(Mutex::new(None)),
continuous_stream: Arc::new(Mutex::new(false)),
}
}
/// Returns true if currently playing a continuous stream (radio without duration)
pub fn is_continuous_stream(&self) -> bool {
*self.continuous_stream.lock().unwrap()
}
}
impl RendererFromMediaRendererInfo for UpnpRenderer {
@@ -157,6 +165,7 @@ impl RendererFromMediaRendererInfo for UpnpRenderer {
has_avtransport_set_next: info.capabilities().has_avtransport_set_next(),
queue,
cached_duration: Arc::new(Mutex::new(None)),
continuous_stream: Arc::new(Mutex::new(false)),
})
}
@@ -206,15 +215,19 @@ impl QueueTransportControl for UpnpRenderer {
)
};
tracing::info!(
"play_from_queue DIDL metadata (first 800 chars):\n{}",
&metadata[..metadata.len().min(800)]
// Détecte si l'URL est un flux continu en interrogeant le serveur HTTP
let is_stream = crate::music_renderer::is_continuous_stream_url(&item.uri);
*self.continuous_stream.lock().unwrap() = is_stream;
tracing::debug!(
"UpnpRenderer play_from_queue: URI={}, continuous_stream={}",
item.uri,
is_stream
);
// Parse et cache la durée du DIDL
// Parse et cache la durée du DIDL (fallback pour certains amplis)
let duration = parse_didl_duration(&metadata);
if let Some(ref dur) = duration {
tracing::info!("Caching duration from queue DIDL: {}", dur);
tracing::debug!("Caching duration from queue DIDL: {}", dur);
*self.cached_duration.lock().unwrap() = Some(dur.clone());
} else {
tracing::debug!("No duration to cache from queue DIDL");
@@ -346,7 +359,6 @@ fn parse_didl_duration(didl: &str) -> Option<String> {
let duration_offset = duration_start + "duration=\"".len();
if let Some(duration_end) = tag_attrs[duration_offset..].find('"') {
let duration = &tag_attrs[duration_offset..duration_offset + duration_end];
tracing::info!("Extracted duration from DIDL: {}", duration);
return Some(duration.to_string());
}
}
@@ -369,15 +381,22 @@ impl TransportControl for UpnpRenderer {
);
}
// Parse le DIDL pour extraire la durée
// Détecte si l'URL est un flux continu en interrogeant le serveur HTTP
let is_stream = crate::music_renderer::is_continuous_stream_url(uri);
*self.continuous_stream.lock().unwrap() = is_stream;
tracing::debug!(
"UpnpRenderer play_uri: URI={}, continuous_stream={}",
uri,
is_stream
);
// Parse le DIDL pour extraire la durée (fallback pour certains amplis)
let duration = parse_didl_duration(meta);
if let Some(ref dur) = duration {
tracing::info!("Caching duration from DIDL: {}", dur);
tracing::debug!("Caching duration from DIDL: {}", dur);
*self.cached_duration.lock().unwrap() = Some(dur.clone());
} else {
tracing::warn!(
"No duration to cache from DIDL (this may be expected for streams without duration)"
);
tracing::debug!("No duration to cache from DIDL");
*self.cached_duration.lock().unwrap() = None;
}
@@ -465,23 +484,8 @@ impl PlaybackPosition for UpnpRenderer {
}
});
// Si l'ampli ne retourne pas de durée, utilise la durée cachée du DIDL
let track_duration = if normalized_duration.is_none() {
let cached = self.cached_duration.lock().unwrap();
if let Some(ref duration) = *cached {
tracing::debug!("Using cached duration from DIDL as fallback: {}", duration);
Some(duration.clone())
} else {
tracing::warn!("No track_duration from renderer and no cached duration available!");
None
}
} else {
tracing::debug!(
"Using track_duration from renderer: {:?}",
normalized_duration
);
normalized_duration
};
// duration=00:00:00 means stream (no duration) - don't fallback to cached DIDL
let track_duration = normalized_duration;
tracing::trace!(
"Final PlaybackPositionInfo: track_duration={:?}, rel_time={:?}",

2
pmocontrol/src/music_renderer/watcher.rs
View File

@@ -89,6 +89,8 @@ pub struct WatchedState {
pub mute: Option<bool>,
/// Last known track metadata
pub metadata: Option<TrackMetadata>,
/// Last known stream state (continuous stream vs bounded media)
pub is_stream: Option<bool>,
}
// ============================================================================

10
pmocontrol/src/openapi.rs
View File

@@ -302,6 +302,16 @@ pub struct SleepTimerState {
pub remaining_seconds: Option<u32>,
}
/// État du flux (stream vs morceau délimité)
#[cfg(feature = "pmoserver")]
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct StreamState {
/// true si lecture en cours d'un flux continu (radio), false sinon
pub is_stream: bool,
/// true si actuellement en lecture, false sinon
pub is_playing: bool,
}
/// Réponse générique de succès
#[cfg(feature = "pmoserver")]
#[derive(Debug, Clone, Serialize, ToSchema)]

57
pmocontrol/src/pmoserver_ext.rs
View File

@@ -14,8 +14,8 @@ use crate::openapi::{
AttachPlaylistRequest, AttachedPlaylistInfo, BrowseResponse, ContainerEntry, ErrorResponse,
FullRendererSnapshot, MediaServerSummary, PlayContentRequest, QueueSnapshot,
RendererCapabilitiesSummary, RendererProtocolSummary, RendererState, RendererSummary,
SeekQueueRequest, SeekRequest, SleepTimerRequest, SleepTimerState, SuccessResponse,
TransferQueueRequest, VolumeSetRequest,
SeekQueueRequest, SeekRequest, SleepTimerRequest, SleepTimerState, StreamState,
SuccessResponse, TransferQueueRequest, VolumeSetRequest,
};
#[cfg(feature = "pmoserver")]
use crate::queue::PlaybackItem;
@@ -1333,6 +1333,54 @@ async fn get_sleep_timer_state(
}))
}
// ============================================================================
// HANDLERS - STREAM STATE
// ============================================================================
/// GET /control/renderers/{renderer_id}/stream-state - Récupère l'état stream du renderer
#[cfg(feature = "pmoserver")]
#[utoipa::path(
get,
path = "/renderers/{renderer_id}/stream-state",
params(
("renderer_id" = String, Path, description = "ID unique du renderer")
),
responses(
(status = 200, description = "État du flux", body = StreamState),
(status = 404, description = "Renderer non trouvé", body = ErrorResponse)
),
tag = "control"
)]
async fn get_stream_state(
State(state): State<ControlPointState>,
Path(renderer_id): Path<String>,
) -> Result<Json<StreamState>, (StatusCode, Json<ErrorResponse>)> {
let rid = DeviceId(renderer_id.clone());
let renderer = state
.control_point
.music_renderer_by_id(&rid)
.ok_or_else(|| {
(
StatusCode::NOT_FOUND,
Json(ErrorResponse {
error: format!("Renderer {} not found", renderer_id),
}),
)
})?;
let is_stream = renderer.is_playing_a_stream();
let is_playing = renderer
.playback_state()
.map(|s| matches!(s, crate::model::PlaybackState::Playing))
.unwrap_or(false);
Ok(Json(StreamState {
is_stream,
is_playing,
}))
}
// ============================================================================
// HANDLERS - QUEUE SHUFFLE
// ============================================================================
@@ -2290,6 +2338,11 @@ pub fn create_api_router(state: ControlPointState, control_point: Arc<ControlPoi
"/renderers/{renderer_id}/timer/cancel",
post(cancel_sleep_timer),
)
// Stream state
.route(
"/renderers/{renderer_id}/stream-state",
get(get_stream_state),
)
// Playlist binding
.route(
"/renderers/{renderer_id}/binding/attach",

444
pmocontrol/src/sse.rs
View File

@@ -85,6 +85,11 @@ pub enum RendererEventPayload {
container_id: Option<String>,
timestamp: chrono::DateTime<chrono::Utc>,
},
StreamStateChanged {
renderer_id: String,
is_stream: bool,
timestamp: chrono::DateTime<chrono::Utc>,
},
TimerStarted {
renderer_id: String,
duration_seconds: u32,
@@ -160,6 +165,158 @@ pub enum UnifiedEventPayload {
MediaServer(MediaServerEventPayload),
}
// ============================================================================
// HELPER FUNCTIONS
// ============================================================================
/// Convertit un RendererEvent en RendererEventPayload pour SSE
///
/// Cette fonction centralise la conversion pour éviter la duplication de code
/// entre les différents streams SSE (renderers-only et all-events).
#[cfg(feature = "pmoserver")]
fn renderer_event_to_payload(
event: RendererEvent,
timestamp: chrono::DateTime<chrono::Utc>,
) -> RendererEventPayload {
match event {
RendererEvent::StateChanged { id, state } => RendererEventPayload::StateChanged {
renderer_id: id.0,
state: state.as_str().to_string(),
timestamp,
},
RendererEvent::PositionChanged { id, position } => RendererEventPayload::PositionChanged {
renderer_id: id.0,
track: position.track,
rel_time: position.rel_time,
track_duration: position.track_duration,
timestamp,
},
RendererEvent::VolumeChanged { id, volume } => RendererEventPayload::VolumeChanged {
renderer_id: id.0,
volume,
timestamp,
},
RendererEvent::MuteChanged { id, mute } => RendererEventPayload::MuteChanged {
renderer_id: id.0,
mute,
timestamp,
},
RendererEvent::MetadataChanged { id, metadata } => RendererEventPayload::MetadataChanged {
renderer_id: id.0,
title: metadata.title,
artist: metadata.artist,
album: metadata.album,
album_art_uri: metadata.album_art_uri,
timestamp,
},
RendererEvent::QueueUpdated { id, queue_length } => RendererEventPayload::QueueUpdated {
renderer_id: id.0,
queue_length,
timestamp,
},
RendererEvent::BindingChanged { id, binding } => RendererEventPayload::BindingChanged {
renderer_id: id.0,
server_id: binding.as_ref().map(|b| b.server_id.0.clone()),
container_id: binding.as_ref().map(|b| b.container_id.clone()),
timestamp,
},
RendererEvent::StreamStateChanged { id, is_stream } => {
RendererEventPayload::StreamStateChanged {
renderer_id: id.0,
is_stream,
timestamp,
}
}
RendererEvent::TimerStarted {
id,
duration_seconds,
remaining_seconds,
} => RendererEventPayload::TimerStarted {
renderer_id: id.0,
duration_seconds,
remaining_seconds,
timestamp,
},
RendererEvent::TimerUpdated {
id,
duration_seconds,
remaining_seconds,
} => RendererEventPayload::TimerUpdated {
renderer_id: id.0,
duration_seconds,
remaining_seconds,
timestamp,
},
RendererEvent::TimerTick {
id,
remaining_seconds,
} => RendererEventPayload::TimerTick {
renderer_id: id.0,
remaining_seconds,
timestamp,
},
RendererEvent::TimerExpired { id } => RendererEventPayload::TimerExpired {
renderer_id: id.0,
timestamp,
},
RendererEvent::TimerCancelled { id } => RendererEventPayload::TimerCancelled {
renderer_id: id.0,
timestamp,
},
RendererEvent::Online { id, info } => RendererEventPayload::Online {
renderer_id: id.0,
friendly_name: info.friendly_name,
model_name: info.model_name,
manufacturer: info.manufacturer,
timestamp,
},
RendererEvent::Offline { id } => RendererEventPayload::Offline {
renderer_id: id.0,
timestamp,
},
}
}
/// Convertit un MediaServerEvent en MediaServerEventPayload pour SSE
///
/// Cette fonction centralise la conversion pour éviter la duplication de code
/// entre les différents streams SSE (servers-only et all-events).
#[cfg(feature = "pmoserver")]
fn media_server_event_to_payload(
event: MediaServerEvent,
timestamp: chrono::DateTime<chrono::Utc>,
) -> MediaServerEventPayload {
match event {
MediaServerEvent::GlobalUpdated {
server_id,
system_update_id,
} => MediaServerEventPayload::GlobalUpdated {
server_id: server_id.0,
system_update_id,
timestamp,
},
MediaServerEvent::ContainersUpdated {
server_id,
container_ids,
} => MediaServerEventPayload::ContainersUpdated {
server_id: server_id.0,
container_ids,
timestamp,
},
MediaServerEvent::Online { server_id, info } => MediaServerEventPayload::Online {
server_id: server_id.0,
friendly_name: info.friendly_name,
model_name: info.model_name,
manufacturer: info.manufacturer,
timestamp,
},
MediaServerEvent::Offline { server_id } => MediaServerEventPayload::Offline {
server_id: server_id.0,
timestamp,
},
}
}
// ============================================================================
// HANDLERS SSE
// ============================================================================
@@ -241,114 +398,7 @@ pub async fn renderer_events_sse(
// Regular events from the control point
Some(event) = rx_tokio.recv() => {
let timestamp = chrono::Utc::now();
let payload = match event {
RendererEvent::StateChanged { id, state } => {
RendererEventPayload::StateChanged {
renderer_id: id.0,
state: state.as_str().to_string(),
timestamp,
}
}
RendererEvent::PositionChanged { id, position } => {
RendererEventPayload::PositionChanged {
renderer_id: id.0,
track: position.track,
rel_time: position.rel_time,
track_duration: position.track_duration,
timestamp,
}
}
RendererEvent::VolumeChanged { id, volume } => {
RendererEventPayload::VolumeChanged {
renderer_id: id.0,
volume,
timestamp,
}
}
RendererEvent::MuteChanged { id, mute } => {
RendererEventPayload::MuteChanged {
renderer_id: id.0,
mute,
timestamp,
}
}
RendererEvent::MetadataChanged { id, metadata } => {
RendererEventPayload::MetadataChanged {
renderer_id: id.0,
title: metadata.title,
artist: metadata.artist,
album: metadata.album,
album_art_uri: metadata.album_art_uri,
timestamp,
}
}
RendererEvent::QueueUpdated { id, queue_length } => {
RendererEventPayload::QueueUpdated {
renderer_id: id.0,
queue_length,
timestamp,
}
}
RendererEvent::BindingChanged { id, binding } => {
RendererEventPayload::BindingChanged {
renderer_id: id.0,
server_id: binding.as_ref().map(|b| b.server_id.0.clone()),
container_id: binding.as_ref().map(|b| b.container_id.clone()),
timestamp,
}
}
RendererEvent::TimerStarted { id, duration_seconds, remaining_seconds } => {
RendererEventPayload::TimerStarted {
renderer_id: id.0,
duration_seconds,
remaining_seconds,
timestamp,
}
}
RendererEvent::TimerUpdated { id, duration_seconds, remaining_seconds } => {
RendererEventPayload::TimerUpdated {
renderer_id: id.0,
duration_seconds,
remaining_seconds,
timestamp,
}
}
RendererEvent::TimerTick { id, remaining_seconds } => {
RendererEventPayload::TimerTick {
renderer_id: id.0,
remaining_seconds,
timestamp,
}
}
RendererEvent::TimerExpired { id } => {
RendererEventPayload::TimerExpired {
renderer_id: id.0,
timestamp,
}
}
RendererEvent::TimerCancelled { id } => {
RendererEventPayload::TimerCancelled {
renderer_id: id.0,
timestamp,
}
}
RendererEvent::Online { id, info } => {
RendererEventPayload::Online {
renderer_id: id.0,
friendly_name: info.friendly_name.clone(),
model_name: info.model_name.clone(),
manufacturer: info.manufacturer.clone(),
timestamp,
}
}
RendererEvent::Offline { id } => {
RendererEventPayload::Offline {
renderer_id: id.0,
timestamp,
}
}
};
let payload = renderer_event_to_payload(event, timestamp);
if let Ok(json) = serde_json::to_string(&payload) {
yield Ok::<_, axum::Error>(Event::default().event("renderer").data(json));
@@ -467,39 +517,8 @@ pub async fn media_server_events_sse(
tokio::select! {
// Regular events from the control point
Some(event) = rx_tokio.recv() => {
let timestamp = chrono::Utc::now();
let payload = match event {
MediaServerEvent::GlobalUpdated { server_id, system_update_id } => {
MediaServerEventPayload::GlobalUpdated {
server_id: server_id.0,
system_update_id,
timestamp,
}
}
MediaServerEvent::ContainersUpdated { server_id, container_ids } => {
MediaServerEventPayload::ContainersUpdated {
server_id: server_id.0,
container_ids,
timestamp,
}
}
MediaServerEvent::Online { server_id, info } => {
MediaServerEventPayload::Online {
server_id: server_id.0,
friendly_name: info.friendly_name.clone(),
model_name: info.model_name.clone(),
manufacturer: info.manufacturer.clone(),
timestamp,
}
}
MediaServerEvent::Offline { server_id } => {
MediaServerEventPayload::Offline {
server_id: server_id.0,
timestamp,
}
}
};
let timestamp = chrono::Utc::now();
let payload = media_server_event_to_payload(event, timestamp);
if let Ok(json) = serde_json::to_string(&payload) {
yield Ok::<_, axum::Error>(Event::default().event("media_server").data(json));
@@ -650,114 +669,7 @@ pub async fn all_events_sse(State(control_point): State<Arc<ControlPoint>>) -> i
tokio::select! {
Some(event) = renderer_rx_tokio.recv() => {
let timestamp = chrono::Utc::now();
let renderer_payload = match event {
RendererEvent::StateChanged { id, state } => {
RendererEventPayload::StateChanged {
renderer_id: id.0,
state: state.as_str().to_string(),
timestamp,
}
}
RendererEvent::PositionChanged { id, position } => {
RendererEventPayload::PositionChanged {
renderer_id: id.0,
track: position.track,
rel_time: position.rel_time,
track_duration: position.track_duration,
timestamp,
}
}
RendererEvent::VolumeChanged { id, volume } => {
RendererEventPayload::VolumeChanged {
renderer_id: id.0,
volume,
timestamp,
}
}
RendererEvent::MuteChanged { id, mute } => {
RendererEventPayload::MuteChanged {
renderer_id: id.0,
mute,
timestamp,
}
}
RendererEvent::MetadataChanged { id, metadata } => {
RendererEventPayload::MetadataChanged {
renderer_id: id.0,
title: metadata.title,
artist: metadata.artist,
album: metadata.album,
album_art_uri: metadata.album_art_uri,
timestamp,
}
}
RendererEvent::QueueUpdated { id, queue_length } => {
RendererEventPayload::QueueUpdated {
renderer_id: id.0,
queue_length,
timestamp,
}
}
RendererEvent::BindingChanged { id, binding } => {
RendererEventPayload::BindingChanged {
renderer_id: id.0,
server_id: binding.as_ref().map(|b| b.server_id.0.clone()),
container_id: binding.as_ref().map(|b| b.container_id.clone()),
timestamp,
}
}
RendererEvent::TimerStarted { id, duration_seconds, remaining_seconds } => {
RendererEventPayload::TimerStarted {
renderer_id: id.0,
duration_seconds,
remaining_seconds,
timestamp,
}
}
RendererEvent::TimerUpdated { id, duration_seconds, remaining_seconds } => {
RendererEventPayload::TimerUpdated {
renderer_id: id.0,
duration_seconds,
remaining_seconds,
timestamp,
}
}
RendererEvent::TimerTick { id, remaining_seconds } => {
RendererEventPayload::TimerTick {
renderer_id: id.0,
remaining_seconds,
timestamp,
}
}
RendererEvent::TimerExpired { id } => {
RendererEventPayload::TimerExpired {
renderer_id: id.0,
timestamp,
}
}
RendererEvent::TimerCancelled { id } => {
RendererEventPayload::TimerCancelled {
renderer_id: id.0,
timestamp,
}
}
RendererEvent::Online { id, info } => {
RendererEventPayload::Online {
renderer_id: id.0,
friendly_name: info.friendly_name.clone(),
model_name: info.model_name.clone(),
manufacturer: info.manufacturer.clone(),
timestamp,
}
}
RendererEvent::Offline { id } => {
RendererEventPayload::Offline {
renderer_id: id.0,
timestamp,
}
}
};
let renderer_payload = renderer_event_to_payload(event, timestamp);
let payload = UnifiedEventPayload::Renderer(renderer_payload);
@@ -767,39 +679,7 @@ pub async fn all_events_sse(State(control_point): State<Arc<ControlPoint>>) -> i
}
Some(event) = server_rx_tokio.recv() => {
let timestamp = chrono::Utc::now();
let server_payload = match event {
MediaServerEvent::GlobalUpdated { server_id, system_update_id } => {
MediaServerEventPayload::GlobalUpdated {
server_id: server_id.0,
system_update_id,
timestamp,
}
}
MediaServerEvent::ContainersUpdated { server_id, container_ids } => {
MediaServerEventPayload::ContainersUpdated {
server_id: server_id.0,
container_ids,
timestamp,
}
}
MediaServerEvent::Online { server_id, info } => {
MediaServerEventPayload::Online {
server_id: server_id.0,
friendly_name: info.friendly_name.clone(),
model_name: info.model_name.clone(),
manufacturer: info.manufacturer.clone(),
timestamp,
}
}
MediaServerEvent::Offline { server_id } => {
MediaServerEventPayload::Offline {
server_id: server_id.0,
timestamp,
}
}
};
let server_payload = media_server_event_to_payload(event, timestamp);
let payload = UnifiedEventPayload::MediaServer(server_payload);
if let Ok(json) = serde_json::to_string(&payload) {

1
pmocontrol/src/upnp_clients/openhome_client.rs
View File

@@ -599,7 +599,6 @@ impl OhTimeClient {
.map_err(|_| {
ControlPointError::UpnpBadReturnValue("Second".to_string(), "".to_string())
})?;
Ok(OhTimePosition {
track_count,
duration_secs,