43 lines
1.2 KiB
Python
43 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
|
|
class AudioPacer:
|
|
def __init__(self, *, frame_bytes: int) -> None:
|
|
self._frame_bytes = max(int(frame_bytes), 1)
|
|
self._buffer = bytearray()
|
|
|
|
@property
|
|
def buffered_bytes(self) -> int:
|
|
return len(self._buffer)
|
|
|
|
def push(self, audio_chunk: bytes) -> list[bytes]:
|
|
if not audio_chunk:
|
|
return []
|
|
self._buffer.extend(audio_chunk)
|
|
return self._pop_frames()
|
|
|
|
def flush(self, *, pad_final_frame: bool = True) -> list[bytes]:
|
|
frames = self._pop_frames()
|
|
if not self._buffer:
|
|
return frames
|
|
if pad_final_frame:
|
|
padded = bytes(self._buffer) + (b"\x00" * (self._frame_bytes - len(self._buffer)))
|
|
frames.append(padded)
|
|
else:
|
|
frames.append(bytes(self._buffer))
|
|
self._buffer.clear()
|
|
return frames
|
|
|
|
def clear(self) -> None:
|
|
self._buffer.clear()
|
|
|
|
def reset(self) -> None:
|
|
self.clear()
|
|
|
|
def _pop_frames(self) -> list[bytes]:
|
|
frames: list[bytes] = []
|
|
while len(self._buffer) >= self._frame_bytes:
|
|
frames.append(bytes(self._buffer[: self._frame_bytes]))
|
|
del self._buffer[: self._frame_bytes]
|
|
return frames
|