MCPcopy Create free account
hub / github.com/FireRedTeam/FireRedTTS2 / forward_chunk

Method forward_chunk

fireredtts2/codec/decoder.py:407–468  ·  view source on GitHub ↗

Forward only one frame. Args: spec: shape (B, N, T=chunk_size) cache: previous chunk's last ifft frame, shape (B, N, T=3) last_chunk: if last_chunk, will not trim the last (win-hop) segment Returns: y: shape (B, T=effective_length)

(
        self, spec: torch.Tensor, cache: torch.Tensor = None, last_chunk: bool = False
    )

Source from the content-addressed store, hash-verified

405 return y
406
407 def forward_chunk(
408 self, spec: torch.Tensor, cache: torch.Tensor = None, last_chunk: bool = False
409 ):
410 """Forward only one frame.
411
412 Args:
413 spec: shape (B, N, T=chunk_size)
414 cache: previous chunk's last ifft frame, shape (B, N, T=3)
415 last_chunk: if last_chunk, will not trim the last (win-hop) segment
416 Returns:
417 y: shape (B, T=effective_length)
418 """
419 assert self.padding == "same", "Padding must be same."
420 assert (
421 self.win_length % self.hop_length == 0
422 ), f"{self.win_length} {self.hop_length}"
423 pad = (self.win_length - self.hop_length) // 2
424
425 # Inverse FFT
426 ifft = torch.fft.irfft(spec, self.n_fft, dim=1, norm="backward")
427 ifft = ifft * self.window[None, :, None] # (B, N, T=chunk_size)
428
429 # Append previous cache
430 if cache is not None:
431 ifft = torch.cat([cache, ifft], dim=-1)
432 new_cache_t = self.win_length // self.hop_length - 1
433 new_cache = ifft[..., -new_cache_t:]
434
435 # Overlap and Add
436 output_size = (ifft.shape[-1] - 1) * self.hop_length + self.win_length
437 y = torch.nn.functional.fold(
438 ifft,
439 output_size=(1, output_size),
440 kernel_size=(1, self.win_length),
441 stride=(1, self.hop_length),
442 )[:, 0, 0, :]
443
444 # Window envelope
445 window_sq = (
446 self.window.square().expand(1, ifft.shape[-1], -1).transpose(1, 2)
447 ) # (B=1, N, T)
448 window_envelope = torch.nn.functional.fold(
449 window_sq,
450 output_size=(1, output_size),
451 kernel_size=(1, self.win_length),
452 stride=(1, self.hop_length),
453 ).squeeze()
454
455 # Normalize
456 # assert (window_envelope > 1e-11).all()
457 y = y / window_envelope
458
459 # Only take effective part
460 if cache is None:
461 y = y[:, pad:]
462 else:
463 y = y[:, (self.win_length - self.hop_length) :]
464 if last_chunk:

Callers 5

forward_chunkMethod · 0.45
forward_chunkMethod · 0.45
forward_chunkMethod · 0.45
forward_chunkMethod · 0.45
decode_one_tokenMethod · 0.45

Calls

no outgoing calls

Tested by

no test coverage detected