(self, pcm_audio: bytes)
| 87 | self._buffer = bytearray() |
| 88 | |
| 89 | def encode(self, pcm_audio: bytes) -> list[bytes]: |
| 90 | packets: list[bytes] = [] |
| 91 | self._buffer.extend(pcm_audio) |
| 92 | |
| 93 | while len(self._buffer) >= self._bytes_per_frame: |
| 94 | chunk = bytes(self._buffer[: self._bytes_per_frame]) |
| 95 | del self._buffer[: self._bytes_per_frame] |
| 96 | |
| 97 | samples = np.frombuffer(chunk, dtype=np.int16).reshape(self.channels, -1) |
| 98 | frame = av.AudioFrame.from_ndarray(samples, format="s16", layout=self._codec.layout.name) |
| 99 | frame.sample_rate = self.sample_rate |
| 100 | packets.extend(bytes(packet) for packet in self._codec.encode(frame)) |
| 101 | |
| 102 | return packets |
| 103 | |
| 104 | def flush(self, pad_final_frame: bool = False) -> list[bytes]: |
| 105 | if not self._buffer: |
no outgoing calls
no test coverage detected