Method decode_frames

moss_tts_realtime/fast_api.py:140–155

(self, audio_frames: list[torch.Tensor])

            self.callbacks.on_audio_stream_stop()

    def decode_frames(self, audio_frames: list[torch.Tensor]) -> Iterator[np.ndarray]:
        for frame in audio_frames:
            tokens = frame
            if tokens.dim() == 3:
                tokens = tokens[0]
            if tokens.dim() != 2:
                raise ValueError(f"Expected [T, C] audio tokens, got {tuple(tokens.shape)}")
            tokens, _ = _sanitize_tokens(tokens, self.codebook_size, self.audio_eos_token)
            if tokens.numel() == 0:
                continue
            self.decoder.push_tokens(tokens.detach())
            for wav in self.decoder.audio_chunks():
                if wav.numel() == 0:
                    continue
                self._mark_started()
                yield wav.detach().cpu().numpy().reshape(-1)

    def flush(self) -> Iterator[np.ndarray]:
        final_chunk = self.decoder.flush()