sim_lib_audio_graph_live/
transport.rs

1use std::collections::BTreeMap;
2
3use sim_kernel::{Result, Symbol};
4use sim_lib_audio_graph_core::Transport;
5use sim_lib_stream_clock::Clock;
6use sim_lib_stream_core::{StreamDiagnostic, StreamEnvelope, StreamMedia, StreamPacket};
7
8/// Returns the symbol identifying the live audio graph clock.
9pub fn live_clock_symbol() -> Symbol {
10    Symbol::qualified("clock", "audio-graph-live")
11}
12
13/// Stream-clock-backed transport source for live audio blocks.
14#[derive(Clone, Debug, PartialEq, Eq)]
15pub struct LiveTransportClock {
16    clock: Clock,
17    sample_rate_hz: u32,
18}
19
20/// Reordering window that buffers LAN preview envelopes back into sequence
21/// order, emitting jitter/reorder/drop/late-packet diagnostics along the way.
22#[derive(Clone, Debug)]
23pub struct LanBufferedPreviewWindow {
24    next_sequence: u64,
25    reorder_depth: u64,
26    pending: BTreeMap<u64, StreamEnvelope>,
27    diagnostics: Vec<StreamPacket>,
28}
29
30impl LiveTransportClock {
31    /// Creates a sample-frame clock for the given sample rate.
32    pub fn sample_frame(sample_rate_hz: u32) -> Result<Self> {
33        Ok(Self {
34            clock: Clock::frame(live_clock_symbol(), u64::from(sample_rate_hz))?,
35            sample_rate_hz,
36        })
37    }
38
39    /// Returns the underlying stream clock.
40    pub fn clock(&self) -> &Clock {
41        &self.clock
42    }
43
44    /// Returns the clock's sample rate in hertz.
45    pub fn sample_rate_hz(&self) -> u32 {
46        self.sample_rate_hz
47    }
48
49    /// Builds a transport snapshot at a sample position and play state.
50    pub fn transport_at(&self, sample_pos: u64, playing: bool) -> Transport {
51        Transport {
52            playing,
53            sample_pos,
54            tempo_bpm: 120.0,
55            ppq_pos: 0.0,
56        }
57    }
58}
59
60impl LanBufferedPreviewWindow {
61    /// Creates a window expecting `next_sequence`, tolerating gaps up to
62    /// `reorder_depth` before dropping the missing range.
63    pub fn new(next_sequence: u64, reorder_depth: u64) -> Self {
64        Self {
65            next_sequence,
66            reorder_depth,
67            pending: BTreeMap::new(),
68            diagnostics: Vec::new(),
69        }
70    }
71
72    /// Accepts an envelope and returns any envelopes now in sequence order.
73    ///
74    /// Late, jittered, reordered, or dropped packets are recorded as
75    /// diagnostics retrievable via [`drain_diagnostics`](Self::drain_diagnostics).
76    pub fn push(&mut self, envelope: StreamEnvelope) -> Result<Vec<StreamEnvelope>> {
77        validate_lan_buffered_preview_envelope(&envelope)?;
78        let sequence = envelope.sequence();
79        if sequence < self.next_sequence || self.pending.contains_key(&sequence) {
80            self.record(
81                lan_buffered_preview_late_packet_diagnostic_kind(),
82                format!(
83                    "LAN preview packet {sequence} arrived after sequence {} was expected",
84                    self.next_sequence
85                ),
86            );
87            return Ok(Vec::new());
88        }
89        if sequence > self.next_sequence {
90            let gap = sequence - self.next_sequence;
91            self.record(
92                lan_buffered_preview_jitter_diagnostic_kind(),
93                format!("LAN preview packet {sequence} arrived with a sequence gap of {gap}"),
94            );
95            self.record(
96                lan_buffered_preview_reorder_diagnostic_kind(),
97                format!(
98                    "LAN preview packet {sequence} arrived before expected sequence {}",
99                    self.next_sequence
100                ),
101            );
102            if gap > self.reorder_depth {
103                self.record(
104                    lan_buffered_preview_drop_diagnostic_kind(),
105                    format!(
106                        "LAN preview dropped missing sequence range {}..{sequence}",
107                        self.next_sequence
108                    ),
109                );
110                self.next_sequence = sequence;
111                return Ok(self.accept_ready(envelope));
112            }
113            self.pending.insert(sequence, envelope);
114            return Ok(Vec::new());
115        }
116        Ok(self.accept_ready(envelope))
117    }
118
119    /// Drains and returns the accumulated preview diagnostics.
120    pub fn drain_diagnostics(&mut self) -> Vec<StreamPacket> {
121        std::mem::take(&mut self.diagnostics)
122    }
123
124    fn accept_ready(&mut self, envelope: StreamEnvelope) -> Vec<StreamEnvelope> {
125        let mut ready = vec![envelope];
126        self.next_sequence = self.next_sequence.saturating_add(1);
127        while let Some(envelope) = self.pending.remove(&self.next_sequence) {
128            ready.push(envelope);
129            self.next_sequence = self.next_sequence.saturating_add(1);
130        }
131        ready
132    }
133
134    fn record(&mut self, kind: Symbol, message: String) {
135        self.diagnostics
136            .push(StreamPacket::Diagnostic(StreamDiagnostic::new(
137                kind, message,
138            )));
139    }
140}
141
142/// Validates that an envelope carries PCM media on the LAN buffered audio
143/// preview transport profile.
144pub fn validate_lan_buffered_preview_envelope(envelope: &StreamEnvelope) -> Result<()> {
145    if envelope.media() != StreamMedia::Pcm {
146        return Err(sim_kernel::Error::Eval(
147            "LAN buffered audio preview requires PCM media".to_owned(),
148        ));
149    }
150    if envelope.profile().name()
151        != sim_lib_stream_core::TransportProfile::lan_buffered_audio_preview().name()
152    {
153        return Err(sim_kernel::Error::Eval(
154            "LAN buffered audio preview requires stream/profile/lan-buffered-audio-preview"
155                .to_owned(),
156        ));
157    }
158    Ok(())
159}
160
161/// Returns the diagnostic kind symbol for preview sequence jitter.
162pub fn lan_buffered_preview_jitter_diagnostic_kind() -> Symbol {
163    Symbol::qualified("stream/preview", "Jitter")
164}
165
166/// Returns the diagnostic kind symbol for dropped preview packet ranges.
167pub fn lan_buffered_preview_drop_diagnostic_kind() -> Symbol {
168    Symbol::qualified("stream/preview", "Drop")
169}
170
171/// Returns the diagnostic kind symbol for out-of-order preview packets.
172pub fn lan_buffered_preview_reorder_diagnostic_kind() -> Symbol {
173    Symbol::qualified("stream/preview", "Reorder")
174}
175
176/// Returns the diagnostic kind symbol for late preview packets.
177pub fn lan_buffered_preview_late_packet_diagnostic_kind() -> Symbol {
178    Symbol::qualified("stream/preview", "LatePacket")
179}
sim_lib_audio_graph_live/transport.rs

sim_lib_audio_graph_live/
transport.rs
