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

Incorpore les métadata dans les header flac

Browse Source
This commit is contained in:
Eric Coissac
2025-11-21 17:59:27 +01:00
parent f7c89d8000
commit f7f2364da6
7 changed files with 279 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
pmoaudio-ext/src/sinks/streaming_flac_sink.rs
View File

@@ -228,6 +228,9 @@ impl NodeLogic for StreamingFlacSinkLogic {
self.ctx.sample_rate = Some(sample_rate);
debug!("Detected sample rate: {} Hz", sample_rate);
// If a duration is already known for this track, fill total_samples now.
self.ctx.refresh_total_samples_with_sample_rate();
// Initialize the FLAC encoder now (first track starts at 0.0)
self.ctx
.initialize_encoder(
@@ -306,6 +309,15 @@ impl NodeLogic for StreamingFlacSinkLogic {
_AudioSegment::Sync(marker) => {
match marker.as_ref() {
SyncMarker::TrackBoundary { metadata } => {
// Prepare encoder options (metadata + duration) for the upcoming track.
if let Err(e) =
self.ctx.prepare_encoder_options_for_track(metadata).await
{
error!("Failed to prepare encoder options for new track: {}", e);
}
debug!("StreamingFlacSink: SyncMarker::TrackBoundary {:?}",metadata.read().await.get_duration().await);
// Only restart encoder if it's already initialized (not the first track)
if self.ctx.sample_rate.is_some() && self.ctx.encoder_state.is_some() {
// Restart encoder to emit new header and reset timestamps
@@ -464,6 +476,7 @@ impl StreamingFlacSink {
first_chunk_timestamp_checked: false,
timestamp_offset_sec: 0.0,
current_timestamp: Arc::new(RwLock::new(0.0)),
pending_track_duration: None,
},
};

37
pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs
View File

@@ -196,6 +196,9 @@ impl NodeLogic for StreamingOggFlacSinkLogic {
self.ctx.sample_rate = Some(sample_rate);
debug!("Detected sample rate: {} Hz", sample_rate);
// Populate total_samples if we already know the track duration.
self.ctx.refresh_total_samples_with_sample_rate();
// Initialize the FLAC encoder now (first track starts at 0.0)
self.ctx
.initialize_encoder(
@@ -263,6 +266,13 @@ impl NodeLogic for StreamingOggFlacSinkLogic {
_AudioSegment::Sync(marker) => {
match marker.as_ref() {
SyncMarker::TrackBoundary { metadata } => {
// Inject per-track metadata and duration into the next FLAC header.
if let Err(e) =
self.ctx.prepare_encoder_options_for_track(metadata).await
{
error!("Failed to prepare encoder options for new track: {}", e);
}
// Only restart encoder if it's already initialized (not the first track)
if self.ctx.sample_rate.is_some() && self.ctx.encoder_state.is_some() {
// Restart encoder to emit new OGG stream header and reset timestamps
@@ -421,6 +431,7 @@ impl StreamingOggFlacSink {
first_chunk_timestamp_checked: false,
timestamp_offset_sec: 0.0,
current_timestamp: Arc::new(RwLock::new(0.0)),
pending_track_duration: None,
},
};
@@ -515,8 +526,9 @@ async fn broadcast_ogg_flac_stream(
let bos_page = ogg_writer.create_page(&ogg_flac_id, true, false, false);
let bos_bytes = Bytes::from(bos_page);
// Step 3: Create Vorbis Comment page (empty for now, metadata comes from /metadata endpoint)
let vorbis_comment = create_empty_vorbis_comment();
// Step 3: Create Vorbis Comment page (reuse FLAC metadata blocks when available)
let vorbis_comment = extract_comment_packet_from_flac_header(&flac_header)
.unwrap_or_else(create_empty_vorbis_comment);
let comment_page = ogg_writer.create_page(&vorbis_comment, false, false, false);
let comment_bytes = Bytes::from(comment_page);
@@ -857,6 +869,27 @@ fn create_ogg_flac_identification(flac_header: &[u8]) -> Result<Vec<u8>, AudioEr
Ok(packet)
}
/// Extract the concatenated FLAC metadata blocks after STREAMINFO to use as the OGG comment packet.
/// Returns `None` if the FLAC header only contains STREAMINFO.
fn extract_comment_packet_from_flac_header(flac_header: &[u8]) -> Option<Vec<u8>> {
if flac_header.len() < 8 {
return None;
}
// STREAMINFO block length is stored in bytes 5-7 (after type byte at 4)
let block_length =
u32::from_be_bytes([0, flac_header[5], flac_header[6], flac_header[7]]) as usize;
let streaminfo_total = 4 + block_length; // block header + data
// Skip "fLaC" + STREAMINFO block.
let offset = 4 + streaminfo_total;
if flac_header.len() <= offset {
return None;
}
Some(flac_header[offset..].to_vec())
}
/// Create empty Vorbis Comment block as a proper FLAC metadata block
fn create_empty_vorbis_comment() -> Vec<u8> {
let mut vorbis_data = Vec::new();

45
pmoaudio-ext/src/sinks/streaming_sink_common.rs
View File

@@ -213,6 +213,7 @@ pub struct SharedSinkContext {
pub first_chunk_timestamp_checked: bool,
pub timestamp_offset_sec: f64,
pub current_timestamp: Arc<RwLock<f64>>,
pub pending_track_duration: Option<Duration>,
}
impl SharedSinkContext {
@@ -299,6 +300,50 @@ impl SharedSinkContext {
Ok(())
}
/// Prepare encoder options for a new track so the next FLAC header embeds up-to-date metadata
/// and duration (total_samples) when available.
pub async fn prepare_encoder_options_for_track(
&mut self,
metadata_lock: &Arc<RwLock<dyn TrackMetadata>>,
) -> Result<(), AudioError> {
debug!("Encoder metadata: preparing metadata");
// Always pass the metadata handle to the encoder so Vorbis comments are emitted.
self.encoder_options.metadata = Some(metadata_lock.clone());
// Capture duration (if any) to set total_samples.
let duration_opt = {
let metadata = metadata_lock.read().await;
metadata.get_duration().await.ok().flatten()
};
self.pending_track_duration = duration_opt;
// Compute total_samples only when we know the sample rate.
if let (Some(duration), Some(sr)) = (self.pending_track_duration, self.sample_rate) {
debug!("Encoder metadata: duration: {:3}s - rate: {}Hz", Duration::as_secs_f64(&duration),sr);
let samples = (duration.as_secs_f64() * sr as f64).round() as u64;
self.encoder_options.total_samples = Some(samples);
} else {
if self.pending_track_duration.is_none() {
debug!("Encoder metadata: duration: None");
}
if self.sample_rate.is_none() {
debug!("Encoder metadata: sample rate: None");
}
// Avoid leaking the previous track's length.
self.encoder_options.total_samples = None;
}
Ok(())
}
/// Refresh total_samples when the sample rate is learned after metadata was already set.
pub fn refresh_total_samples_with_sample_rate(&mut self) {
if let (Some(duration), Some(sr)) = (self.pending_track_duration, self.sample_rate) {
let samples = (duration.as_secs_f64() * sr as f64).round() as u64;
self.encoder_options.total_samples = Some(samples);
}
}
pub async fn restart_encoder_for_new_track<Fut, F>(
&mut self,
broadcaster: F,

101
pmoaudio-ext/src/sinks/timed_broadcast.rs
View File

@@ -5,13 +5,17 @@
//! - Propagation d’un compteur `epoch` incrémenté sur chaque TopZeroSync.
use std::{
collections::VecDeque, fmt, string, sync::{
Arc, Mutex, Weak, atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}
}, time::{Duration, Instant}
collections::VecDeque,
fmt, string,
sync::{
atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
Arc, Mutex, Weak,
},
time::{Duration, Instant},
};
use tokio::sync::Notify;
use tracing::{info, debug, trace, warn};
use tracing::{debug, info, trace, warn};
/// Tolérance pour détecter un timestamp à zéro (TopZero).
const TOP_ZERO_EPSILON: f64 = 1e-9;
@@ -93,9 +97,9 @@ struct State<T> {
}
impl<T> State<T> {
fn new(name: &str,capacity: usize, epoch_start: Instant) -> Self {
fn new(name: &str, capacity: usize, epoch_start: Instant) -> Self {
Self {
name: name.to_string() ,
name: name.to_string(),
buffer: VecDeque::with_capacity(capacity),
head_seq: 0,
next_seq: 0,
@@ -120,7 +124,13 @@ impl<T> State<T> {
while let Some(entry) = self.buffer.front() {
if entry.expires_at <= now {
let delta = now - entry.expires_at;
debug!("TimedBroadcast[{}]: purging expired packet (@{} epoch={},delta={})", self.name,entry.seq, entry.epoch, delta.as_millis());
trace!(
"TimedBroadcast[{}]: purging expired packet (@{} epoch={},delta={})",
self.name,
entry.seq,
entry.epoch,
delta.as_millis()
);
self.buffer.pop_front();
self.head_seq += 1;
purged += 1;
@@ -130,7 +140,8 @@ impl<T> State<T> {
}
if purged > 0 {
trace!(
"TimedBroadcast: purged {} expired packet(s) (head_seq={})",
"TimedBroadcast[{}]: purged {} expired packet(s) (head_seq={})",
self.name,
purged,
self.head_seq
);
@@ -167,8 +178,11 @@ impl<T> State<T> {
for _ in 0..removable {
let oentry = self.buffer.pop_front();
if oentry.is_some() {
let entry= oentry.unwrap();
debug!("TimedBroadcast[{}]: pruning played packet (@{} epoch={})", self.name,entry.seq, entry.epoch);
let entry = oentry.unwrap();
trace!(
"TimedBroadcast[{}]: pruning played packet (@{} epoch={})",
self.name, entry.seq, entry.epoch
);
self.head_seq += 1;
}
@@ -188,9 +202,9 @@ struct Inner<T> {
}
impl<T> Inner<T> {
fn new(name: &str,capacity: usize) -> Self {
fn new(name: &str, capacity: usize) -> Self {
Self {
state: Mutex::new(State::new(name,capacity, Instant::now())),
state: Mutex::new(State::new(name, capacity, Instant::now())),
data_notify: Notify::new(),
space_notify: Notify::new(),
capacity,
@@ -214,7 +228,7 @@ impl<T> Inner<T> {
/// Créé un channel broadcast temporisé.
pub fn channel<T>(name: &str, capacity: usize) -> (Sender<T>, Receiver<T>) {
assert!(capacity > 0, "capacity must be > 0");
let inner = Arc::new(Inner::new(name,capacity));
let inner = Arc::new(Inner::new(name, capacity));
let next_seq = {
let state = inner.state.lock().expect("timed broadcast mutex poisoned");
state.next_seq
@@ -297,44 +311,62 @@ impl<T> Sender<T> {
}
// 2. Vérifier si un slot est disponible et insérer
let is_top_zero = audio_timestamp.abs() < TOP_ZERO_EPSILON;
let is_top_zero =
audio_timestamp.abs() < TOP_ZERO_EPSILON
&& segment_duration >= TOP_ZERO_EPSILON;
let is_zero_header =
audio_timestamp.abs() < TOP_ZERO_EPSILON
&& segment_duration < TOP_ZERO_EPSILON;
if state.buffer.len() < self.inner.capacity {
if !state.initialized {
if !is_top_zero {
if !state.initialized {
if !is_top_zero && segment_duration >= TOP_ZERO_EPSILON {
warn!(
"TimedBroadcast: First packet has non-zero timestamp {:.3}s, treating as epoch start anyway",
audio_timestamp
"TimedBroadcast[{}]: First packet has non-zero timestamp {:.1}ms - Duration={:.1}ms, treating as epoch start anyway",
state.name,
audio_timestamp*1000.0,
segment_duration*1000.0
);
}
state.epoch_start = now;
state.epoch = 0;
state.initialized = true;
info!(
"TimedBroadcast: initialized (epoch=0, ts={:.3}s)",
audio_timestamp
"TimedBroadcast[{}]: initialized (epoch=0, ts={:.1}ms - Duration={:.1}ms)",
state.name,
audio_timestamp*1000.0,
segment_duration*1000.0
);
} else if is_top_zero {
} else if is_top_zero || is_zero_header {
// Restart epoch on TopZero relative to current wall-clock time to avoid
// expired packets when there's a long gap between tracks.
state.epoch_start = state.last_segment_end.unwrap_or(now);
// expired packets when there's a long gap between tracks. Also trigger
// on zero-duration headers (OGG BOS/comment) so the epoch is reset
// before testing expiration.
state.epoch_start = state
.last_segment_end
.map(|end| end.max(now))
.unwrap_or(now);
// state.epoch_start = now;
state.epoch = state.epoch.wrapping_add(1);
info!(
"TimedBroadcast: new epoch={} (continuous={})",
"TimedBroadcast[{}]: new epoch={} (continuous={} - Duration={}ms)",
state.name,
state.epoch,
state.last_segment_end.is_some()
state.last_segment_end.is_some(),
segment_duration*1000.0
);
}
let expires_at =
state.epoch_start + Duration::from_secs_f64(audio_timestamp + segment_duration);
let expires_at = state.epoch_start
+ Duration::from_secs_f64(audio_timestamp
+ segment_duration);
let is_first_packet = state.next_seq == 0;
if !is_first_packet && !is_top_zero && expires_at <= now {
if !is_first_packet && !is_top_zero && !is_zero_header && expires_at <= now {
let grace_period = Duration::from_millis(50);
if now > expires_at + grace_period {
warn!(
"TimedBroadcast: rejecting already expired packet (ts={:.3}s, epoch={}, delta={}ms)",
"TimedBroadcast[{}]: rejecting already expired packet (ts={:.3}s, epoch={}, delta={}ms)",
state.name,
audio_timestamp,
state.epoch,
now.duration_since(expires_at).as_millis()
@@ -355,10 +387,13 @@ impl<T> Sender<T> {
state.next_seq += 1;
state.buffer.push_back(entry);
// 5. Mettre à jour la fin du segment SEULEMENT pour les paquets non-TopZero
// (pour que le prochain segment commence à la fin du dernier paquet de données)
if !is_top_zero {
state.last_segment_end = Some(expires_at);
// 5. Only advance segment end for real audio (skip 0-duration metadata)
if segment_duration >= TOP_ZERO_EPSILON {
let new_end = expires_at;
state.last_segment_end = Some(match state.last_segment_end.take() {
Some(prev) => prev.max(new_end),
None => new_end,
});
}
let receivers = self.inner.receiver_count.load(Ordering::SeqCst);