From caf28c459319f59da6fd394a15336ef5bddb9d64 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Tue, 14 Jul 2026 12:09:28 +0200 Subject: [PATCH 1/6] Update AudioStreamTrack to use stateful av.AudioResampler for better quality --- getstream/video/rtc/audio_track.py | 327 +++++++++++----------------- getstream/video/rtc/track_util.py | 81 +++++++ tests/rtc/test_track_util.py | 105 +++++++++ tests/test_audio_stream_track.py | 328 +++++++++++++++++++---------- uv.lock | 4 - 5 files changed, 527 insertions(+), 318 deletions(-) diff --git a/getstream/video/rtc/audio_track.py b/getstream/video/rtc/audio_track.py index 975e07c5..f4066400 100644 --- a/getstream/video/rtc/audio_track.py +++ b/getstream/video/rtc/audio_track.py @@ -1,257 +1,178 @@ -import time import asyncio +import fractions import logging +import time +from collections import deque import aiortc -from av import AudioFrame -from av.frame import Frame -import fractions +import av +import numpy as np -from getstream.video.rtc.track_util import PcmData +from .track_util import AudioFormat, AudioFormatType, FrameResampler, PcmData logger = logging.getLogger(__name__) class AudioStreamTrack(aiortc.mediastreams.MediaStreamTrack): - """ - Audio stream track that accepts PcmData objects and buffers them as bytes. - - Works with PcmData objects instead of raw bytes, avoiding format conversion issues. - Internally buffers as bytes for efficient memory usage. + """aiortc audio track that streams buffered PCM into a WebRTC call. - Usage: - track = AudioStreamTrack(sample_rate=48000, channels=2) + `write()` accepts PcmData at any sample rate, channel layout, or format, and converts it to fixed 20ms packed-s16 frames at the track's output + rate/layout and queues them (dropping the oldest once the queue exceeds + audio_buffer_size_ms). - # Write PcmData objects (any format, any sample rate, any channels) - await track.write(pcm_data) + `recv()` paces the queue out in real time via _FramePacer, + handing back one frame per call with its pts stamped, and synthesizes silence + when the queue is empty so the RTP timeline never stalls. - # The track will automatically resample/convert to the configured format + `flush()` clears the queue to support interruption/barge-in. """ kind = "audio" def __init__( self, - sample_rate: int = 48000, - channels: int = 1, - format: str = "s16", - audio_buffer_size_ms: int = 30000, # 30 seconds default + sample_rate: int = 48000, # rate of emitted frames; 48kHz matches Opus (avoids a re-resample) + channels: int = 1, # output channel count (1=mono, 2=stereo) + format: AudioFormatType = AudioFormat.S16, # output sample format; must be s16 (aiortc's Opus encoder requirement) + audio_buffer_size_ms: int = 30000, # max audio to hold before dropping oldest ): - """ - Initialize an AudioStreamTrack that accepts PcmData objects. - - Args: - sample_rate: Target sample rate in Hz (default: 48000) - channels: Number of channels - 1=mono, 2=stereo (default: 1) - format: Audio format - "s16" or "f32" (default: "s16") - audio_buffer_size_ms: Maximum buffer size in milliseconds (default: 30000ms = 30s) - """ super().__init__() + if format != AudioFormat.S16: + raise ValueError( + f"AudioStreamTrack output format must be 's16', got {format!r}; " + "aiortc's Opus encoder only accepts s16 frames." + ) self.sample_rate = sample_rate self.channels = channels self.format = format self.audio_buffer_size_ms = audio_buffer_size_ms - logger.debug( - "Initialized AudioStreamTrack", - extra={ - "sample_rate": sample_rate, - "channels": channels, - "format": format, - "audio_buffer_size_ms": audio_buffer_size_ms, - }, + self._frame_buffer: deque[av.AudioFrame] = deque() + # Running per-channel sample total, to enforce the size cap cheaply. + self._buffered_samples = 0 + # Serializes buffer mutation between write() and recv(). + self._frame_lock = asyncio.Lock() + + self._layout = "stereo" if channels == 2 else "mono" + # Samples-per-channel in one 20ms frame (e.g. 0.02 * 48000 = 960). + self._samples_per_frame = int(aiortc.mediastreams.AUDIO_PTIME * sample_rate) + # Cap in per-channel samples; beyond this, the oldest frames are dropped. + self._max_samples = int((audio_buffer_size_ms / 1000) * sample_rate) + # Pre-built zeros reused to synthesize a silence frame on starvation. + self._silence = np.zeros( + (1, self._samples_per_frame * channels), dtype=np.int16 ) - # Internal bytearray buffer for audio data - self._buffer = bytearray() - self._buffer_lock = asyncio.Lock() - - # Timing for frame pacing - self._start = None - self._timestamp = None - self._last_frame_time = None - - # Calculate bytes per sample based on format - self._bytes_per_sample = 2 if format == "s16" else 4 # s16=2 bytes, f32=4 bytes - self._bytes_per_frame = int( - aiortc.mediastreams.AUDIO_PTIME - * self.sample_rate - * self.channels - * self._bytes_per_sample + # Resampling and pacing state live in their own helpers; the track only + # calls their methods. + self._resampler = FrameResampler( + rate=sample_rate, + layout=self._layout, + format=format, + frame_size=self._samples_per_frame, ) + self._pacer = _FramePacer(sample_rate, self._samples_per_frame) - async def write(self, pcm: PcmData): - """ - Add PcmData to the buffer. + async def write(self, pcm: PcmData, final: bool = False) -> None: + """Write PCM data to the track. - The PcmData will be automatically resampled/converted to match - the track's configured sample_rate, channels, and format, - then converted to bytes and stored in the buffer. + Under the hood, it resamples to fixed 20ms frames and buffers them. + When final is True, drain the resampler's tail so the utterance plays out. Args: - pcm: PcmData object with audio data + pcm: PCM data to write + final: if True, drain the resampler's tail and add it to the buffer + (e.g. on end-of-utterance event). """ - # Normalize the PCM data to target format immediately - pcm_normalized = self._normalize_pcm(pcm) - - # Convert to bytes - audio_bytes = pcm_normalized.to_bytes() - - async with self._buffer_lock: - # Check buffer size before adding - max_buffer_bytes = int( - (self.audio_buffer_size_ms / 1000) - * self.sample_rate - * self.channels - * self._bytes_per_sample - ) + # Zero or more finished 20ms frames (empty while swr is still buffering). + frames = self._resampler.resample(pcm, flush=final) + if not frames: + return - # Add new data to buffer first - self._buffer.extend(audio_bytes) - - # Check if we exceeded the limit - if len(self._buffer) > max_buffer_bytes: - # Calculate how many bytes to drop from the beginning - bytes_to_drop = len(self._buffer) - max_buffer_bytes - dropped_ms = ( - bytes_to_drop - / (self.sample_rate * self.channels * self._bytes_per_sample) - ) * 1000 - - logger.debug( - "Audio buffer overflow, dropping %.1fms of audio. Buffer max is %dms", - dropped_ms, - self.audio_buffer_size_ms, - extra={ - "buffer_size_bytes": len(self._buffer), - "incoming_bytes": len(audio_bytes), - "dropped_bytes": bytes_to_drop, - }, - ) + async with self._frame_lock: + for frame in frames: + self._frame_buffer.append(frame) + self._buffered_samples += frame.samples - # Drop from the beginning of the buffer to keep latest data - del self._buffer[:bytes_to_drop] - - buffer_duration_ms = ( - len(self._buffer) - / (self.sample_rate * self.channels * self._bytes_per_sample) - ) * 1000 - - logger.debug( - "Added audio to buffer", - extra={ - "pcm_duration_ms": pcm.duration_ms, - "buffer_duration_ms": buffer_duration_ms, - "buffer_size_bytes": len(self._buffer), - }, - ) + # Bound latency/memory: if the producer outran the consumer, drop the + # oldest frames until back under the cap. + while self._buffered_samples > self._max_samples and self._frame_buffer: + self._buffered_samples -= self._frame_buffer.popleft().samples async def flush(self) -> None: - """ - Clear any pending audio from the buffer. - Playback stops immediately. - """ - async with self._buffer_lock: - bytes_cleared = len(self._buffer) - self._buffer.clear() - - logger.debug("Flushed audio buffer", extra={"cleared_bytes": bytes_cleared}) - - async def recv(self) -> Frame: - """ - Receive the next 20ms audio frame. - - Returns: - AudioFrame with the configured sample_rate, channels, and format - """ + """Drop pending frames and reset the resampler (interruption).""" + async with self._frame_lock: + # Cut off queued playback (barge-in) and reset the resampler so no + # samples bleed into the next utterance. + self._frame_buffer.clear() + self._buffered_samples = 0 + self._resampler.flush() + + async def recv(self) -> av.AudioFrame: + """Return the next 20ms frame, synthesizing silence when starved.""" + # aiortc calls recv() in a loop; once the track is stopped, signal EOF. if self.readyState != "live": raise aiortc.mediastreams.MediaStreamError - # Calculate samples needed for 20ms frame - samples_per_frame = int(aiortc.mediastreams.AUDIO_PTIME * self.sample_rate) + # Block until this frame is due, and get the pts to stamp on it. + pts = await self._pacer.next_pts() - # Initialize timestamp if not already done - if self._timestamp is None: - self._start = time.time() - self._timestamp = 0 - self._last_frame_time = time.time() - else: - # Use timestamp-based pacing to avoid drift over time - # This ensures we stay synchronized with the expected audio rate - # even if individual frames have slight timing variations - self._timestamp += samples_per_frame - start_ts = self._start or time.time() - wait = start_ts + (self._timestamp / self.sample_rate) - time.time() - if wait > 0: - await asyncio.sleep(wait) - - self._last_frame_time = time.time() - - # Get 20ms of audio data from buffer - async with self._buffer_lock: - if len(self._buffer) >= self._bytes_per_frame: - # We have enough data - audio_bytes = bytes(self._buffer[: self._bytes_per_frame]) - del self._buffer[: self._bytes_per_frame] - elif len(self._buffer) > 0: - # We have some data but not enough - pad with silence - audio_bytes = bytes(self._buffer) - padding_needed = self._bytes_per_frame - len(audio_bytes) - audio_bytes += bytes(padding_needed) # Pad with zeros (silence) - self._buffer.clear() - - logger.debug( - "Padded audio frame with silence", - extra={ - "available_bytes": len(audio_bytes) - padding_needed, - "required_bytes": self._bytes_per_frame, - "padding_bytes": padding_needed, - }, - ) + async with self._frame_lock: + if self._frame_buffer: + # Normal path: hand out the next ready frame. + frame = self._frame_buffer.popleft() + self._buffered_samples -= frame.samples else: - # No data at all - emit silence - audio_bytes = bytes(self._bytes_per_frame) - - # Create AudioFrame - layout = "stereo" if self.channels == 2 else "mono" - - # Convert format name: "s16" -> "s16", "f32" -> "flt" - if self.format == "s16": - av_format = "s16" # Packed int16 - elif self.format == "f32": - av_format = "flt" # Packed float32 - else: - av_format = "s16" # Default to s16 - - frame = AudioFrame(format=av_format, layout=layout, samples=samples_per_frame) + # Starved (gap between utterances): emit a silence frame so the RTP + # timeline stays continuous instead of stalling. + frame = av.AudioFrame.from_ndarray( + self._silence, format="s16", layout=self._layout + ) - # Fill frame with data - frame.planes[0].update(audio_bytes) + # A flushed tail can be shorter than a full frame; pad it with trailing + # silence so every emitted frame fills its fixed-rate pts slot. + if frame.samples < self._samples_per_frame: + frame = self._pad_to_full_frame(frame) - # Set frame properties - frame.pts = self._timestamp + # Stamp the frame's timing so the encoder/RTP sender place it correctly. + frame.pts = pts frame.sample_rate = self.sample_rate frame.time_base = fractions.Fraction(1, self.sample_rate) - return frame - def _normalize_pcm(self, pcm: PcmData) -> PcmData: - """ - Normalize PcmData to match the track's target format. + def _pad_to_full_frame(self, frame: av.AudioFrame) -> av.AudioFrame: + """Pad a short (flushed-tail) frame up to a full frame with trailing silence.""" + data = frame.to_ndarray() + pad = self._samples_per_frame * self.channels - data.shape[1] + padded = np.pad(data, ((0, 0), (0, pad))) + return av.AudioFrame.from_ndarray(padded, format="s16", layout=self._layout) - Args: - pcm: Input PcmData - Returns: - PcmData resampled/converted to target sample_rate, channels, and format - """ +class _FramePacer: + """Real-time clock for fixed-size frames. - pcm = pcm.resample(self.sample_rate, target_channels=self.channels) + aiortc sends whatever recv() returns immediately, so recv() must block until each + frame is due. next_pts() sleeps the right amount and returns the pts to stamp. + """ - # Convert format if needed - if self.format == "s16" and pcm.format != "s16": - pcm = pcm.to_int16() - elif self.format == "f32" and pcm.format != "f32": - pcm = pcm.to_float32() + def __init__(self, sample_rate: int, samples_per_frame: int): + self._sample_rate = sample_rate + self._samples_per_frame = samples_per_frame + # Wall-clock anchor and monotonic sample cursor (= next pts); None until start. + self._start: float | None = None + self._ts: int | None = None - return pcm + async def next_pts(self) -> int: + if self._ts is None: + # First frame: anchor the clock and emit without waiting. + self._start = time.time() + self._ts = 0 + else: + # Advance one frame and sleep until that wall-clock time. Anchoring to + # _start (not the previous frame) keeps drift from accumulating. + self._ts += self._samples_per_frame + start = self._start or time.time() + wait = start + self._ts / self._sample_rate - time.time() + if wait > 0: + await asyncio.sleep(wait) + return self._ts diff --git a/getstream/video/rtc/track_util.py b/getstream/video/rtc/track_util.py index 3c7c9f48..3eae1f9f 100644 --- a/getstream/video/rtc/track_util.py +++ b/getstream/video/rtc/track_util.py @@ -2597,3 +2597,84 @@ async def _run_track(self): time_base=time_base, ) ) + + +class FrameResampler: + """Wraps av.AudioResampler to emit fixed-size packed av.AudioFrames from PcmData. + + A single av.AudioResampler locks onto its first input frame's rate/layout/format, + so this rebuilds the underlying resampler whenever the input signature changes. + """ + + def __init__( + self, rate: int, layout: str, format: AudioFormatType, frame_size: int + ): + """ + Args: + rate: Target output sample rate in Hz. + layout: Target channel layout, e.g. "mono" or "stereo". + format: Output sample format passed to av.AudioResampler (packed, e.g. "s16"). + frame_size: Samples per channel in each emitted frame; input is buffered + until a full frame_size can be emitted. 0 emits variable-size frames. + The tail returned by flush() may be shorter than frame_size. + """ + self._rate = rate + self._layout = layout + self._format = format + self._frame_size = frame_size + self._resampler: Optional[av.AudioResampler] = None + # Input signature the current resampler was built for. + self._in_rate: Optional[int] = None + self._in_channels: Optional[int] = None + self._in_format: Optional[str] = None + + def resample(self, pcm: PcmData, flush: bool = False) -> list[av.AudioFrame]: + """Resample pcm data and return finished av.AudioFrames. + The resampler keeps some samples in the buffer for smoothing unless `flush` is True. + + Args: + pcm: PcmData audio + flush: if True, also flush the tail and reset.""" + frames: list[av.AudioFrame] = [] + # Empty input (e.g. the final marker) yields no frames but can still flush. + if pcm.samples.size: + resampler = self._ensure_av_resampler(pcm) + frames = resampler.resample(pcm.to_av_frame()) + + if flush: + frames.extend(self.flush()) + + return frames + + def flush(self) -> list[av.AudioFrame]: + """ + Flush the resampler's buffered tail and reset it. + """ + frames: list[av.AudioFrame] = [] + if self._resampler is not None: + # resample(None) flushes swr to EOF: it raises on any further input, so + # drop it here and _ensure_av_resampler rebuilds on the next write. + frames = self._resampler.resample(None) + self._resampler = None + return frames + + def _ensure_av_resampler(self, pcm: PcmData) -> av.AudioResampler: + resampler = self._resampler + if ( + resampler is None + or self._in_rate != pcm.sample_rate + or self._in_channels != pcm.channels + or self._in_format != pcm.format + ): + resampler = av.AudioResampler( + format=self._format, + layout=self._layout, + rate=self._rate, + # frame_size makes swr emit exactly-20ms frames, ready to send. + frame_size=self._frame_size, + ) + self._resampler = resampler + self._in_rate = pcm.sample_rate + self._in_channels = pcm.channels + self._in_format = pcm.format + return resampler diff --git a/tests/rtc/test_track_util.py b/tests/rtc/test_track_util.py index 006085d4..aef1e75b 100644 --- a/tests/rtc/test_track_util.py +++ b/tests/rtc/test_track_util.py @@ -8,6 +8,7 @@ AudioTrackHandler, PcmData, AudioFormat, + FrameResampler, ) import getstream.video.rtc.track_util as track_util @@ -946,3 +947,107 @@ def truncate_audio_to_last_n_seconds( result_new = pcm_exact.tail(duration_s=8.0, pad=True, pad_at="start") np.testing.assert_array_equal(result_new.samples, result_original) + + +SINE_FREQ = 1000.0 + + +def _sine_chunks( + freq: float, + sample_rate: int, + total_ms: int, + chunk_ms: int = 20, + amplitude: int = 10000, +) -> list[PcmData]: + n_total = int(sample_rate * total_ms / 1000) + t = np.arange(n_total) / sample_rate + wave = (amplitude * np.sin(2 * np.pi * freq * t)).astype(np.int16) + chunk = int(sample_rate * chunk_ms / 1000) + return [ + PcmData( + samples=wave[i : i + chunk], + sample_rate=sample_rate, + format="s16", + channels=1, + ) + for i in range(0, n_total, chunk) + if len(wave[i : i + chunk]) == chunk + ] + + +def _silence_chunk(sample_rate: int, chunk_ms: int = 20) -> PcmData: + n = int(sample_rate * chunk_ms / 1000) + return PcmData( + samples=np.zeros(n, dtype=np.int16), + sample_rate=sample_rate, + format="s16", + channels=1, + ) + + +def _frames_samples(frames: list) -> np.ndarray: + if not frames: + return np.array([], dtype=np.int16) + return np.concatenate([f.to_ndarray().reshape(-1) for f in frames]) + + +class TestFrameResampler: + @pytest.fixture + def resampler(self) -> FrameResampler: + return FrameResampler(rate=48000, layout="mono", format="s16", frame_size=0) + + def test_resample_preserves_tone_at_target_rate(self, resampler): + out = [ + _frames_samples(resampler.resample(chunk)) + for chunk in _sine_chunks(SINE_FREQ, 24000, total_ms=200) + ] + signal = np.concatenate(out).astype(np.float64) + signal = signal[: np.nonzero(np.abs(signal) > 1)[0][-1] + 1] + + spectrum = np.abs(np.fft.rfft(signal * np.hanning(len(signal)))) + freqs = np.fft.rfftfreq(len(signal), 1 / 48000) + assert abs(freqs[np.argmax(spectrum)] - SINE_FREQ) < 20 + + def test_matched_rate_is_passed_through(self, resampler): + # The resampler targets 48000 mono s16; a chunk already at that rate needs no + # resampling and comes back unchanged (same-rate swr is a lossless passthrough). + chunk = _sine_chunks(SINE_FREQ, 48000, total_ms=20)[0] + out = _frames_samples(resampler.resample(chunk)) + assert np.array_equal(out, chunk.samples.reshape(-1)) + + def test_flush_returns_tail_then_resets(self, resampler): + for chunk in _sine_chunks(SINE_FREQ, 24000, total_ms=200): + resampler.resample(chunk) + + assert len(resampler.flush()) > 0 + assert resampler.flush() == [] + + def test_resample_is_usable_after_flush(self, resampler): + chunks = _sine_chunks(SINE_FREQ, 24000, total_ms=100) + for chunk in chunks: + resampler.resample(chunk) + resampler.flush() + + # swr is at EOF after a flush; the wrapper must rebuild, not raise EOFError. + produced = _frames_samples( + [frame for chunk in chunks for frame in resampler.resample(chunk)] + ) + assert produced.size > 0 + + def test_rebuilds_on_input_rate_change(self, resampler): + at_24k = _frames_samples( + [ + frame + for chunk in _sine_chunks(SINE_FREQ, 24000, total_ms=100) + for frame in resampler.resample(chunk) + ] + ) + at_16k = _frames_samples( + [ + frame + for chunk in _sine_chunks(SINE_FREQ, 16000, total_ms=100) + for frame in resampler.resample(chunk) + ] + ) + assert at_24k.size > 0 + assert at_16k.size > 0 diff --git a/tests/test_audio_stream_track.py b/tests/test_audio_stream_track.py index a418b3be..dc5209c7 100644 --- a/tests/test_audio_stream_track.py +++ b/tests/test_audio_stream_track.py @@ -1,11 +1,37 @@ import asyncio import time -import pytest + +import aiortc import numpy as np +import pytest from getstream.video.rtc.audio_track import AudioStreamTrack -from getstream.video.rtc.track_util import PcmData, AudioFormat -import aiortc +from getstream.video.rtc.track_util import AudioFormat, PcmData + +SINE_FREQ = 1000.0 + + +def _sine_chunks( + freq: float, + sample_rate: int, + total_ms: int, + chunk_ms: int = 20, + amplitude: int = 10000, +) -> list[PcmData]: + n_total = int(sample_rate * total_ms / 1000) + t = np.arange(n_total) / sample_rate + wave = (amplitude * np.sin(2 * np.pi * freq * t)).astype(np.int16) + chunk = int(sample_rate * chunk_ms / 1000) + return [ + PcmData( + samples=wave[i : i + chunk], + sample_rate=sample_rate, + format="s16", + channels=1, + ) + for i in range(0, n_total, chunk) + if len(wave[i : i + chunk]) == chunk + ] class TestAudioStreamTrack: @@ -27,12 +53,12 @@ async def test_initialization(self): track = AudioStreamTrack( sample_rate=16000, channels=2, - format="f32", + format="s16", audio_buffer_size_ms=10000, ) assert track.sample_rate == 16000 assert track.channels == 2 - assert track.format == "f32" + assert track.format == "s16" assert track.audio_buffer_size_ms == 10000 @pytest.mark.asyncio @@ -65,32 +91,38 @@ async def test_write_and_recv_basic(self): assert frame2.sample_rate == 48000 assert frame2.samples == 960 + def test_rejects_f32_output(self): + """Output format must be s16; f32 is rejected at construction.""" + with pytest.raises(ValueError): + AudioStreamTrack(format="f32") + @pytest.mark.asyncio async def test_format_conversion(self): - """Test that write converts formats correctly.""" - track = AudioStreamTrack(sample_rate=48000, channels=1, format="f32") + """Test that write converts input to the track's s16 output format.""" + track = AudioStreamTrack(sample_rate=48000, channels=1, format="s16") - # Write s16 data to f32 track - samples = np.array([100, 200, 300], dtype=np.int16) + # Write a full 20ms of f32 input at half scale. pcm = PcmData( - samples=samples, + samples=np.full(960, 0.5, dtype=np.float32), sample_rate=48000, - format=AudioFormat.S16, + format=AudioFormat.F32, channels=1, ) await track.write(pcm) - # Check that buffer contains f32 data - assert track._bytes_per_sample == 4 # f32 = 4 bytes + # The emitted frame is s16 with the f32 value scaled into the int16 range. + frame = await track.recv() + assert frame.format.name == "s16" + assert 15000 < int(np.max(frame.to_ndarray())) < 17000 # ~0.5 * 32767 @pytest.mark.asyncio async def test_sample_rate_conversion(self): - """Test that write resamples audio correctly.""" + """Test that write resamples audio to the track's output rate.""" track = AudioStreamTrack(sample_rate=48000, channels=1, format="s16") - # Write 16kHz data to 48kHz track - samples_16k = np.zeros(320, dtype=np.int16) # 20ms at 16kHz + # 20ms at 16kHz = 320 samples; upsampled to 48kHz that is one 960-sample frame. + samples_16k = np.zeros(320, dtype=np.int16) pcm = PcmData( samples=samples_16k, sample_rate=16000, @@ -100,14 +132,14 @@ async def test_sample_rate_conversion(self): await track.write(pcm) - # After resampling, we should have 3x more samples (960 samples) - expected_bytes = 960 * 2 # 960 samples * 2 bytes per s16 sample - assert len(track._buffer) == expected_bytes + frame = await track.recv() + assert frame.sample_rate == 48000 + assert frame.samples == 960 # 320 samples @16kHz -> 960 @48kHz @pytest.mark.asyncio async def test_channel_conversion(self): """Test mono to stereo and stereo to mono conversion.""" - # Test mono to stereo + # Mono input to a stereo track -> stereo output frame. track_stereo = AudioStreamTrack(sample_rate=48000, channels=2, format="s16") samples_mono = np.zeros(960, dtype=np.int16) # 20ms mono @@ -120,11 +152,11 @@ async def test_channel_conversion(self): await track_stereo.write(pcm_mono) - # Should have doubled the data for stereo - expected_bytes = 960 * 2 * 2 # samples * bytes_per_sample * channels - assert len(track_stereo._buffer) == expected_bytes + frame = await track_stereo.recv() + assert len(frame.layout.channels) == 2 + assert frame.samples == 960 - # Test stereo to mono + # Stereo input to a mono track -> mono output frame. track_mono = AudioStreamTrack(sample_rate=48000, channels=1, format="s16") samples_stereo = np.zeros((2, 960), dtype=np.int16) # 20ms stereo @@ -137,21 +169,21 @@ async def test_channel_conversion(self): await track_mono.write(pcm_stereo) - # Should have halved the data for mono - expected_bytes = 960 * 2 # samples * bytes_per_sample - assert len(track_mono._buffer) == expected_bytes + frame = await track_mono.recv() + assert len(frame.layout.channels) == 1 + assert frame.samples == 960 @pytest.mark.asyncio async def test_buffer_overflow(self): - """Test that buffer drops old data when it exceeds max size.""" - # Set small buffer size for testing (100ms) + """Test that the buffer drops old data when it exceeds max size.""" + # 100ms cap holds at most five 20ms frames. track = AudioStreamTrack( sample_rate=48000, channels=1, format="s16", audio_buffer_size_ms=100 ) - # Try to write 200ms of data + # Write 200ms (ten 20ms frames) of non-zero audio; the cap must drop the oldest. samples_200ms = int(0.2 * 48000) # 9600 samples - audio_data = np.zeros(samples_200ms, dtype=np.int16) + audio_data = np.full(samples_200ms, 100, dtype=np.int16) pcm = PcmData( samples=audio_data, sample_rate=48000, @@ -161,10 +193,13 @@ async def test_buffer_overflow(self): await track.write(pcm) - # Buffer should only contain 100ms worth of data - max_samples = int(0.1 * 48000) # 4800 samples - max_bytes = max_samples * 2 # * 2 bytes per s16 sample - assert len(track._buffer) == max_bytes + # Only the last 100ms survives: five frames carry data, the rest are silence. + data_frames = 0 + for _ in range(8): + frame = await track.recv() + if np.any(frame.to_ndarray() != 0): + data_frames += 1 + assert data_frames == 5 @pytest.mark.asyncio async def test_silence_emission(self): @@ -180,38 +215,48 @@ async def test_silence_emission(self): assert np.all(audio_data == 0) @pytest.mark.asyncio - async def test_partial_frame_padding(self): - """Test that partial frames are padded with silence.""" - track = AudioStreamTrack(sample_rate=48000, channels=1, format="s16") - - # Write only 10ms of data (480 samples) - samples_10ms = int(0.01 * 48000) - audio_data = np.ones(samples_10ms, dtype=np.int16) * 100 # Non-zero data - pcm = PcmData( - samples=audio_data, - sample_rate=48000, - format=AudioFormat.S16, - channels=1, + async def test_partial_frame_buffered_until_flush(self): + """Sub-frame writes are held; a final flush emits them, padded by recv to a full frame.""" + # Without a final flush, a sub-frame write (10ms) emits nothing. + held = AudioStreamTrack(sample_rate=48000, channels=1, format="s16") + await held.write( + PcmData( + samples=np.full(480, 100, dtype=np.int16), + sample_rate=48000, + format=AudioFormat.S16, + channels=1, + ) ) - - await track.write(pcm) - - # Receive 20ms frame - frame = await track.recv() - assert frame.samples == 960 # Full 20ms frame - - # Check that first half has data and second half is silence - audio_array = frame.to_ndarray() - assert np.any(audio_array[:480] != 0) # First 10ms has data - assert np.all(audio_array[480:] == 0) # Last 10ms is silence + starved = await held.recv() + assert np.all( + starved.to_ndarray() == 0 + ) # partial held -> recv starves to silence + + # With final=True the buffered partial is flushed; recv pads it to a full 20ms + # frame: the first 10ms carries the data, the last 10ms is silence. + flushed = AudioStreamTrack(sample_rate=48000, channels=1, format="s16") + await flushed.write( + PcmData( + samples=np.full(480, 100, dtype=np.int16), + sample_rate=48000, + format=AudioFormat.S16, + channels=1, + ), + final=True, + ) + frame = await flushed.recv() + assert frame.samples == 960 + audio = frame.to_ndarray().flatten() + assert np.all(audio[:480] != 0) # flushed data + assert np.all(audio[480:] == 0) # silence padding @pytest.mark.asyncio async def test_flush(self): - """Test that flush clears the buffer.""" + """Test that flush drops pending audio so recv falls back to silence.""" track = AudioStreamTrack(sample_rate=48000, channels=1, format="s16") - # Write some data - samples = np.zeros(960, dtype=np.int16) + # Buffer 40ms of non-zero audio (two frames). + samples = np.full(1920, 100, dtype=np.int16) pcm = PcmData( samples=samples, sample_rate=48000, @@ -220,14 +265,10 @@ async def test_flush(self): ) await track.write(pcm) - # Verify data is in buffer - assert len(track._buffer) > 0 - - # Flush + # Flush drops the buffered frames; the next recv is synthesized silence. await track.flush() - - # Verify buffer is empty - assert len(track._buffer) == 0 + frame = await track.recv() + assert np.all(frame.to_ndarray() == 0) @pytest.mark.asyncio async def test_frame_timing(self, monkeypatch): @@ -258,6 +299,57 @@ async def test_frame_timing(self, monkeypatch): # Allow reasonable tolerance for asyncio.sleep() accuracy assert 15.0 <= interval_ms <= 25.0 + @pytest.mark.asyncio + async def test_first_frame_is_immediate_and_pts_zero(self): + """The first recv anchors the clock: it returns without waiting, pts=0.""" + track = AudioStreamTrack(sample_rate=48000, channels=1, format="s16") + + start = time.time() + frame = await track.recv() + elapsed_ms = (time.time() - start) * 1000 + + assert frame.pts == 0 + assert elapsed_ms < 10 # no pacing wait on the first frame + + @pytest.mark.asyncio + async def test_pts_advances_by_one_frame_per_recv(self): + """Successive frames carry pts stepping by samples_per_frame (960 @48kHz).""" + track = AudioStreamTrack(sample_rate=48000, channels=1, format="s16") + + pts = [(await track.recv()).pts for _ in range(4)] + + assert pts == [0, 960, 1920, 2880] + + @pytest.mark.asyncio + async def test_slow_consumer_does_not_accumulate_drift(self): + """A consumer that falls behind gets overdue frames immediately, not delayed. + + Pacing anchors each pts to the start time, not the previous frame, so a stalled + consumer catches up rather than stretching the timeline. + """ + track = AudioStreamTrack(sample_rate=48000, channels=1, format="s16") + + await track.recv() # anchor the clock at pts=0 + await asyncio.sleep(0.1) # fall ~5 frames behind + + start = time.time() + frame = await track.recv() + elapsed_ms = (time.time() - start) * 1000 + + assert frame.pts == 960 + assert elapsed_ms < 10 # overdue frame is due already -> no wait + + @pytest.mark.asyncio + async def test_pacing_is_independent_of_buffer_content(self): + """Pacing advances even while starved: silence frames still step pts by a frame.""" + track = AudioStreamTrack(sample_rate=48000, channels=1, format="s16") + + # No write(): every recv starves to silence, but the clock keeps ticking. + frames = [await track.recv() for _ in range(3)] + + assert [f.pts for f in frames] == [0, 960, 1920] + assert all(np.all(f.to_ndarray() == 0) for f in frames) + @pytest.mark.asyncio async def test_continuous_streaming(self): """Test continuous audio streaming scenario.""" @@ -303,12 +395,14 @@ async def _continuous_reader(self, track): assert frame.samples == 960 @pytest.mark.asyncio - async def test_recv_does_not_reallocate_buffer(self): - """Test that recv consumes data in-place without creating a new buffer object.""" + async def test_recv_consumes_frames_in_order(self): + """recv hands out buffered frames one at a time, in FIFO order, then starves.""" track = AudioStreamTrack(sample_rate=48000, channels=1, format="s16") - # Write 40ms of data (enough for 2 frames) - samples = np.zeros(1920, dtype=np.int16) + # Two distinguishable 20ms frames: first all 100, then all 200. + samples = np.concatenate( + [np.full(960, 100, dtype=np.int16), np.full(960, 200, dtype=np.int16)] + ) pcm = PcmData( samples=samples, sample_rate=48000, @@ -317,55 +411,46 @@ async def test_recv_does_not_reallocate_buffer(self): ) await track.write(pcm) - buffer_ref = track._buffer # save reference before recv - - # Receive a frame (consumes 20ms from buffer) - await track.recv() + first = await track.recv() + second = await track.recv() + third = await track.recv() - assert track._buffer is buffer_ref, ( - "recv should modify buffer in-place, not create a new one" - ) - assert len(track._buffer) == 960 * 2, ( - "should have 20ms of data remaining (960 samples * 2 bytes)" - ) + assert np.all(first.to_ndarray() == 100) + assert np.all(second.to_ndarray() == 200) + assert np.all(third.to_ndarray() == 0) # buffer drained -> silence @pytest.mark.asyncio - async def test_buffer_overflow_does_not_reallocate(self): - """Test that buffer overflow trims in-place without creating a new buffer object.""" + async def test_buffer_overflow_drops_oldest_across_writes(self): + """Overflow across multiple writes drops the oldest frames, keeping the newest.""" track = AudioStreamTrack( sample_rate=48000, channels=1, format="s16", audio_buffer_size_ms=100 ) - # Write 50ms of data first to get a buffer reference - samples_50ms = np.zeros(2400, dtype=np.int16) - pcm = PcmData( - samples=samples_50ms, - sample_rate=48000, - format=AudioFormat.S16, - channels=1, + # First 40ms (value 100), then 200ms (value 200); total far exceeds the 100ms cap. + await track.write( + PcmData( + samples=np.full(1920, 100, dtype=np.int16), + sample_rate=48000, + format=AudioFormat.S16, + channels=1, + ) ) - await track.write(pcm) - buffer_ref = track._buffer # save reference before overflow - - # Write 200ms of data (exceeds 100ms limit, triggers overflow trim) - samples_200ms = np.zeros(9600, dtype=np.int16) - pcm_large = PcmData( - samples=samples_200ms, - sample_rate=48000, - format=AudioFormat.S16, - channels=1, + await track.write( + PcmData( + samples=np.full(9600, 200, dtype=np.int16), + sample_rate=48000, + format=AudioFormat.S16, + channels=1, + ) ) - await track.write(pcm_large) - assert track._buffer is buffer_ref, ( - "overflow trim should modify buffer in-place, not create a new one" - ) - max_buffer_seconds = 100 / 1000 - bytes_per_sample = 2 - expected_max_bytes = int(max_buffer_seconds * 48000) * bytes_per_sample - assert len(track._buffer) == expected_max_bytes, ( - "buffer should be trimmed to configured max size" - ) + # Only the newest five frames survive: all value 200, the older 100s were dropped. + values = [] + for _ in range(6): + frame = await track.recv() + arr = frame.to_ndarray() + values.append(0 if np.all(arr == 0) else int(arr.flatten()[0])) + assert values == [200, 200, 200, 200, 200, 0] @pytest.mark.asyncio async def test_media_stream_error(self): @@ -378,3 +463,24 @@ async def test_media_stream_error(self): # Try to receive - should raise error with pytest.raises(aiortc.mediastreams.MediaStreamError): await track.recv() + + @pytest.mark.asyncio + @pytest.mark.parametrize("src_rate, dst_rate", [(24000, 48000), (48000, 24000)]) + async def test_resampling_is_high_quality(self, src_rate, dst_rate): + # A clean tone resampled up or down must stay clean: the fundamental should + # dominate, i.e. high SINAD. Linear interpolation lands around ~28 dB here. + track = AudioStreamTrack(sample_rate=dst_rate, channels=1) + for chunk in _sine_chunks(SINE_FREQ, src_rate, total_ms=500): + await track.write(chunk) + drained = np.concatenate( + [(await track.recv()).to_ndarray().reshape(-1) for _ in range(28)] + ).astype(np.float64) + drained = drained[: np.nonzero(np.abs(drained) > 1)[0][-1] + 1] + + spectrum = np.abs(np.fft.rfft(drained * np.hanning(len(drained)))) ** 2 + freqs = np.fft.rfftfreq(len(drained), 1 / dst_rate) + df = freqs[1] - freqs[0] + fundamental = spectrum[np.abs(freqs - SINE_FREQ) <= 4 * df].sum() + noise = spectrum.sum() - fundamental - spectrum[freqs < 30].sum() + + assert 10 * np.log10(fundamental / noise) > 60 diff --git a/uv.lock b/uv.lock index 4b2671db..d55c7c1d 100644 --- a/uv.lock +++ b/uv.lock @@ -9,10 +9,6 @@ resolution-markers = [ "python_full_version < '3.11'", ] -[options] -exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for backwards compatibility when using relative exclude-newer values. -exclude-newer-span = "P7D" - [[package]] name = "aiohappyeyeballs" version = "2.6.1" From 0f6907b71c2b8f4526094ba4f83fed3d40ee1edd Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Tue, 14 Jul 2026 13:23:09 +0200 Subject: [PATCH 2/6] Use monotonic clock instead of time.time --- getstream/video/rtc/audio_track.py | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/getstream/video/rtc/audio_track.py b/getstream/video/rtc/audio_track.py index f4066400..15034e13 100644 --- a/getstream/video/rtc/audio_track.py +++ b/getstream/video/rtc/audio_track.py @@ -158,21 +158,22 @@ class _FramePacer: def __init__(self, sample_rate: int, samples_per_frame: int): self._sample_rate = sample_rate self._samples_per_frame = samples_per_frame - # Wall-clock anchor and monotonic sample cursor (= next pts); None until start. + # Monotonic clock anchor and sample cursor (= next pts); None until start. self._start: float | None = None self._ts: int | None = None async def next_pts(self) -> int: if self._ts is None: # First frame: anchor the clock and emit without waiting. - self._start = time.time() + self._start = time.monotonic() self._ts = 0 else: - # Advance one frame and sleep until that wall-clock time. Anchoring to - # _start (not the previous frame) keeps drift from accumulating. + # Advance one frame and sleep until that time. Anchoring to _start + # (not the previous frame) keeps drift from accumulating. monotonic() + # is used so a wall-clock step (NTP, VM migration) can't stall pacing. self._ts += self._samples_per_frame - start = self._start or time.time() - wait = start + self._ts / self._sample_rate - time.time() + start = self._start or time.monotonic() + wait = start + self._ts / self._sample_rate - time.monotonic() if wait > 0: await asyncio.sleep(wait) return self._ts From 80ff276af1bf9c0dfdd0150bf6c2334ff96d4146 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Tue, 14 Jul 2026 13:24:49 +0200 Subject: [PATCH 3/6] Wrap insides of AudioStreamTrack.write() into asyncio.Lock() to avoid dirty concurrent flush --- getstream/video/rtc/audio_track.py | 12 ++++---- tests/test_audio_stream_track.py | 44 ++++++++++++++++++++++++++++++ 2 files changed, 51 insertions(+), 5 deletions(-) diff --git a/getstream/video/rtc/audio_track.py b/getstream/video/rtc/audio_track.py index 15034e13..08f4d1e6 100644 --- a/getstream/video/rtc/audio_track.py +++ b/getstream/video/rtc/audio_track.py @@ -84,12 +84,14 @@ async def write(self, pcm: PcmData, final: bool = False) -> None: final: if True, drain the resampler's tail and add it to the buffer (e.g. on end-of-utterance event). """ - # Zero or more finished 20ms frames (empty while swr is still buffering). - frames = self._resampler.resample(pcm, flush=final) - if not frames: - return - + # Resample under the lock so an interleaved flush() (barge-in) can't reset + # the resampler between resampling and enqueuing, leaking pre-flush audio. async with self._frame_lock: + # Zero or more finished frames (empty while the resampler is buffering). + frames = self._resampler.resample(pcm, flush=final) + if not frames: + return + for frame in frames: self._frame_buffer.append(frame) self._buffered_samples += frame.samples diff --git a/tests/test_audio_stream_track.py b/tests/test_audio_stream_track.py index dc5209c7..3f6cd103 100644 --- a/tests/test_audio_stream_track.py +++ b/tests/test_audio_stream_track.py @@ -350,6 +350,50 @@ async def test_pacing_is_independent_of_buffer_content(self): assert [f.pts for f in frames] == [0, 960, 1920] assert all(np.all(f.to_ndarray() == 0) for f in frames) + @pytest.mark.asyncio + async def test_flush_during_inflight_write_discards_pre_flush_audio(self): + """A barge-in racing an in-flight write must not leak pre-flush audio. + + write() must resample and enqueue inside the same lock flush() uses, so a + flush that lands mid-write can't be followed by frames the write computed + before it. + """ + track = AudioStreamTrack(sample_rate=48000, channels=1, format="s16") + + # Prime the resampler with a sub-frame chunk of a distinctive "old" value: + # held in the resampler, nothing enqueued yet. + await track.write( + PcmData( + samples=np.full(480, 999, dtype=np.int16), + sample_rate=48000, + format=AudioFormat.S16, + channels=1, + ) + ) + + # Hold the buffer lock to force the adversarial interleaving: a concurrent + # flush() and write() both queue on it, flush ahead of write. + await track._frame_lock.acquire() + flush_task = asyncio.create_task(track.flush()) + write_task = asyncio.create_task( + track.write( + PcmData( + samples=np.full(480, 111, dtype=np.int16), + sample_rate=48000, + format=AudioFormat.S16, + channels=1, + ) + ) + ) + await asyncio.sleep(0) # let both tasks run up to the lock + track._frame_lock.release() + await flush_task + await write_task + + # Barge-in wins: the pre-flush "old" audio (999) must not survive. + frame = await track.recv() + assert not np.any(frame.to_ndarray() == 999) + @pytest.mark.asyncio async def test_continuous_streaming(self): """Test continuous audio streaming scenario.""" From 47a6320a8c46ca8e9c3de19543476442dc854697 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Tue, 14 Jul 2026 13:25:41 +0200 Subject: [PATCH 4/6] Increase timeouts in tests --- tests/test_audio_stream_track.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/tests/test_audio_stream_track.py b/tests/test_audio_stream_track.py index 3f6cd103..134249c8 100644 --- a/tests/test_audio_stream_track.py +++ b/tests/test_audio_stream_track.py @@ -309,7 +309,7 @@ async def test_first_frame_is_immediate_and_pts_zero(self): elapsed_ms = (time.time() - start) * 1000 assert frame.pts == 0 - assert elapsed_ms < 10 # no pacing wait on the first frame + assert elapsed_ms < 15 # no pacing wait on the first frame (< one 20ms frame) @pytest.mark.asyncio async def test_pts_advances_by_one_frame_per_recv(self): @@ -337,7 +337,9 @@ async def test_slow_consumer_does_not_accumulate_drift(self): elapsed_ms = (time.time() - start) * 1000 assert frame.pts == 960 - assert elapsed_ms < 10 # overdue frame is due already -> no wait + assert ( + elapsed_ms < 15 + ) # overdue frame is due already -> no wait (< one 20ms frame) @pytest.mark.asyncio async def test_pacing_is_independent_of_buffer_content(self): From 271074bb98e57de1e0b764936a9e588c693bd8f9 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Tue, 14 Jul 2026 13:26:46 +0200 Subject: [PATCH 5/6] Drain the internal resampler before resetting it --- getstream/video/rtc/track_util.py | 26 ++++++++++++++++++++------ tests/rtc/test_track_util.py | 31 +++++++++++++++++++++++++++++++ 2 files changed, 51 insertions(+), 6 deletions(-) diff --git a/getstream/video/rtc/track_util.py b/getstream/video/rtc/track_util.py index 3eae1f9f..72fe8860 100644 --- a/getstream/video/rtc/track_util.py +++ b/getstream/video/rtc/track_util.py @@ -2638,8 +2638,10 @@ def resample(self, pcm: PcmData, flush: bool = False) -> list[av.AudioFrame]: frames: list[av.AudioFrame] = [] # Empty input (e.g. the final marker) yields no frames but can still flush. if pcm.samples.size: - resampler = self._ensure_av_resampler(pcm) - frames = resampler.resample(pcm.to_av_frame()) + resampler, tail = self._reset_resampler(pcm) + # `tail` holds the old resampler's drained buffer when the input + # signature changed; emit it before this input's frames. + frames = tail + resampler.resample(pcm.to_av_frame()) if flush: frames.extend(self.flush()) @@ -2653,12 +2655,19 @@ def flush(self) -> list[av.AudioFrame]: frames: list[av.AudioFrame] = [] if self._resampler is not None: # resample(None) flushes swr to EOF: it raises on any further input, so - # drop it here and _ensure_av_resampler rebuilds on the next write. + # drop it here and _reset_resampler rebuilds on the next write. frames = self._resampler.resample(None) self._resampler = None return frames - def _ensure_av_resampler(self, pcm: PcmData) -> av.AudioResampler: + def _reset_resampler( + self, pcm: PcmData + ) -> tuple[av.AudioResampler, list[av.AudioFrame]]: + """Return the resampler for pcm's signature, rebuilding it on a change. + + On a rebuild the old resampler's buffered tail is drained and returned so it + isn't dropped; the output signature is fixed, so the tail stays compatible. + """ resampler = self._resampler if ( resampler is None @@ -2666,15 +2675,20 @@ def _ensure_av_resampler(self, pcm: PcmData) -> av.AudioResampler: or self._in_channels != pcm.channels or self._in_format != pcm.format ): + tail: list[av.AudioFrame] = [] + if resampler is not None: + # Drain the outgoing resampler before swapping it out. + tail = resampler.resample(None) resampler = av.AudioResampler( format=self._format, layout=self._layout, rate=self._rate, - # frame_size makes swr emit exactly-20ms frames, ready to send. + # frame_size makes the resampler emit fixed frame_size-sample frames. frame_size=self._frame_size, ) self._resampler = resampler self._in_rate = pcm.sample_rate self._in_channels = pcm.channels self._in_format = pcm.format - return resampler + return resampler, tail + return resampler, [] diff --git a/tests/rtc/test_track_util.py b/tests/rtc/test_track_util.py index aef1e75b..9458c7fd 100644 --- a/tests/rtc/test_track_util.py +++ b/tests/rtc/test_track_util.py @@ -1051,3 +1051,34 @@ def test_rebuilds_on_input_rate_change(self, resampler): ) assert at_24k.size > 0 assert at_16k.size > 0 + + def test_flushes_buffered_tail_before_rebuild_on_signature_change(self): + # frame_size buffers a sub-frame chunk instead of emitting it. + r = FrameResampler(rate=48000, layout="mono", format="s16", frame_size=960) + + # 10ms at the target rate (passthrough); held toward a full 960 frame, so + # nothing is emitted yet. Distinctive value marks this "old" audio. + held = r.resample( + PcmData( + samples=np.full(480, 100, dtype=np.int16), + sample_rate=48000, + format=AudioFormat.S16, + channels=1, + ) + ) + assert held == [] + + # Input rate changes -> the resampler rebuilds. The buffered tail must be + # drained first, not silently dropped. + frames = r.resample( + PcmData( + samples=np.full(1600, 50, dtype=np.int16), + sample_rate=16000, + format=AudioFormat.S16, + channels=1, + ) + ) + frames += r.flush() + + samples = np.concatenate([f.to_ndarray().flatten() for f in frames]) + assert np.any(samples == 100) # the held tail survived the rebuild From 7be9b75c52bc56cc7ed0d4b4509887f79878a33d Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Tue, 14 Jul 2026 17:04:09 +0200 Subject: [PATCH 6/6] Handle non-contiguous arrays --- getstream/video/rtc/track_util.py | 6 +++++- tests/rtc/test_track_util.py | 20 ++++++++++++++++++++ 2 files changed, 25 insertions(+), 1 deletion(-) diff --git a/getstream/video/rtc/track_util.py b/getstream/video/rtc/track_util.py index 72fe8860..5bc0c523 100644 --- a/getstream/video/rtc/track_util.py +++ b/getstream/video/rtc/track_util.py @@ -811,7 +811,11 @@ def to_av_frame(self) -> "av.AudioFrame": # Create PyAV AudioFrame layout = "mono" if pcm_formatted.channels == 1 else "stereo" - frame = av.AudioFrame.from_ndarray(samples, format=av_format, layout=layout) + # from_ndarray requires C-contiguous input; .T above and strided chunk + # views (PcmData.chunks on stereo) can make samples non-contiguous. + frame = av.AudioFrame.from_ndarray( + np.ascontiguousarray(samples), format=av_format, layout=layout + ) frame.sample_rate = pcm_formatted.sample_rate frame.pts = pcm_formatted.pts return frame diff --git a/tests/rtc/test_track_util.py b/tests/rtc/test_track_util.py index 9458c7fd..40854a54 100644 --- a/tests/rtc/test_track_util.py +++ b/tests/rtc/test_track_util.py @@ -1082,3 +1082,23 @@ def test_flushes_buffered_tail_before_rebuild_on_signature_change(self): samples = np.concatenate([f.to_ndarray().flatten() for f in frames]) assert np.any(samples == 100) # the held tail survived the rebuild + + +class TestToAvFrameNonContiguous: + """to_av_frame must accept non-contiguous sample arrays.""" + + def test_stereo_transposed_samples(self): + """(num_samples, channels) input hits the internal .T branch, which + produces a non-contiguous (channels, num_samples) view; it must convert.""" + # Caller supplies interleaved (N, 2) samples; to_av_frame transposes. + samples = np.arange(8, dtype=np.int16).reshape(4, 2) + pcm = PcmData( + samples=samples, sample_rate=48000, format=AudioFormat.S16, channels=2 + ) + + frame = pcm.to_av_frame() + + assert frame.sample_rate == 48000 + assert frame.layout.name == "stereo" + # Frame holds the transposed (channels, num_samples) data. + np.testing.assert_array_equal(frame.to_ndarray(), samples.T)