(self, audio_frames: list[torch.Tensor])
| 209 | self.audio_eos_token = audio_eos_token |
| 210 | |
| 211 | def decode_frames(self, audio_frames: list[torch.Tensor]) -> Iterator[np.ndarray]: |
| 212 | for frame in audio_frames: |
| 213 | tokens = frame |
| 214 | if tokens.dim() == 3: |
| 215 | tokens = tokens[0] |
| 216 | if tokens.dim() != 2: |
| 217 | raise ValueError(f"Expected [T, C] audio tokens, got {tuple(tokens.shape)}") |
| 218 | tokens, stop = _sanitize_tokens(tokens, self.codebook_size, self.audio_eos_token) |
| 219 | if tokens.numel() == 0: |
| 220 | if stop: |
| 221 | break |
| 222 | continue |
| 223 | self.decoder.push_tokens(tokens.detach()) |
| 224 | for wav in self.decoder.audio_chunks(): |
| 225 | if wav.numel() == 0: |
| 226 | continue |
| 227 | yield wav.detach().cpu().numpy().reshape(-1) |
| 228 | if stop: |
| 229 | break |
| 230 | |
| 231 | def flush(self) -> Iterator[np.ndarray]: |
| 232 | final_chunk = self.decoder.flush() |
no test coverage detected