""" RVC + Beatrice v2 Voice Conversion - Single-file app for HuggingFace Spaces RVC-Project + Beatrice v2 (fierce-cats/beatrice-trainer), consolidated into single file - Inference: RVC v2 (.pth) + Beatrice v2 (.pt.gz), CPU or GPU - Training: RVC v2 + Beatrice v2, GPU recommended Usage: CLI: python app.py infer -i input.wav -m model.pth -o output.wav python app.py infer -i input.wav -m beatrice.pt.gz -o output.wav Gradio: python app.py """ import os import sys # MPS fallback for macOS os.environ["PYTORCH_ENABLE_MPS_FALLBACK"] = "1" import argparse import gc import gzip import json as json_module import logging import math import re import secrets import shutil import queue as _queue_module import tempfile import threading import zipfile import time import warnings # Suppress known harmless warnings from HF Spaces / torch internals warnings.filterwarnings("ignore", message=".*torch.distributed.reduce_op.*", category=FutureWarning) warnings.filterwarnings("ignore", message=".*torch.nn.utils.weight_norm.*", category=FutureWarning) from collections import defaultdict from fractions import Fraction from functools import partial from pathlib import Path from random import Random from typing import Optional, List, Tuple, Union, BinaryIO, Literal, Sequence, Iterable, Callable import gradio as gr import numpy as np import torch import torch.nn as nn import torch.nn.functional as F from torch.nn import Conv1d, ConvTranspose1d from torch.nn.utils import weight_norm, remove_weight_norm import librosa import pyworld import soundfile as sf import torchaudio from scipy import signal from huggingface_hub import hf_hub_download from tqdm.auto import tqdm import requests # ── Shared safeguards (training + inference) ── MAX_AUDIO_DURATION = 7200 # 2 hours — max input audio length (seconds) MAX_INFERENCE_TIME = 600 # 10 min — hard inference timeout (seconds) # NOTE: no training time limit is enforced — training runs until it completes # or is explicitly cancelled via the tracking code (see "📡 پیگیری آموزش" tab). MAX_MODEL_DOWNLOAD_BYTES = 2 * 1024 * 1024 * 1024 # 2GB cap for model URL/zip downloads # Cancel flag — checked by training loops at each epoch (concurrency_limit=1 so one flag is fine) _training_cancel = False # 48 Hz high-pass filter to remove low-frequency artifacts (same as Applio) FILTER_ORDER = 5 CUTOFF_FREQUENCY = 48 # Hz SAMPLE_RATE = 16000 # Hz bh, ah = signal.butter(N=FILTER_ORDER, Wn=CUTOFF_FREQUENCY, btype="high", fs=SAMPLE_RATE) def sanitize_model_name(name: str) -> str: """Sanitize model name for safe use in file paths""" name = os.path.basename(name.strip()) name = re.sub(r'[^\w\-.]', '_', name) return name or "unnamed_model" # Default example model DEFAULT_MODEL_REPO = "audo/Benee-RVC" DEFAULT_MODEL_FILE = "BENEE8000.pth" DEFAULT_INDEX_FILE = "added_IVF1054_Flat_nprobe_8.index" # RVC v2 pretrained weights from official repo RVC_PRETRAINED_REPO = "lj1995/VoiceConversionWebUI" RVC_PRETRAINED_V2 = { # Generator with f0 (pitch) "f0G48k": "pretrained_v2/f0G48k.pth", "f0G40k": "pretrained_v2/f0G40k.pth", "f0G32k": "pretrained_v2/f0G32k.pth", # Discriminator with f0 "f0D48k": "pretrained_v2/f0D48k.pth", "f0D40k": "pretrained_v2/f0D40k.pth", "f0D32k": "pretrained_v2/f0D32k.pth", # Generator without f0 "G48k": "pretrained_v2/G48k.pth", "G40k": "pretrained_v2/G40k.pth", "G32k": "pretrained_v2/G32k.pth", # Discriminator without f0 "D48k": "pretrained_v2/D48k.pth", "D40k": "pretrained_v2/D40k.pth", "D32k": "pretrained_v2/D32k.pth", } def download_pretrained_rvc(name: str) -> str: """Download RVC v2 pretrained weights from HuggingFace""" if name not in RVC_PRETRAINED_V2: raise ValueError(f"Unknown pretrained: {name}. Available: {list(RVC_PRETRAINED_V2.keys())}") filepath = RVC_PRETRAINED_V2[name] logger.info(f"Downloading pretrained {name} from {RVC_PRETRAINED_REPO}...") return hf_hub_download(repo_id=RVC_PRETRAINED_REPO, filename=filepath) # Beatrice v2 pretrained assets BEATRICE_REPO = "fierce-cats/beatrice-trainer" BEATRICE_PRETRAINED = { "phone_extractor": "assets/pretrained/122_checkpoint_03000000.pt", "pitch_estimator": "assets/pretrained/104_3_checkpoint_00300000.pt", "pretrained_model": "assets/pretrained/151_checkpoint_libritts_r_200_02750000.pt.gz", } def download_beatrice_asset(name: str) -> str: """Download Beatrice v2 pretrained asset from HuggingFace""" if name not in BEATRICE_PRETRAINED: raise ValueError(f"Unknown asset: {name}. Available: {list(BEATRICE_PRETRAINED.keys())}") filepath = BEATRICE_PRETRAINED[name] logger.info(f"Downloading Beatrice asset {name} from {BEATRICE_REPO}...") return hf_hub_download(repo_id=BEATRICE_REPO, filename=filepath) def download_beatrice_augmentation(): """Download Beatrice augmentation assets (noise + IR) - optional for training""" try: from huggingface_hub import snapshot_download cache_dir = snapshot_download(repo_id=BEATRICE_REPO, allow_patterns=["assets/noise/*", "assets/ir/*"]) noise_dir = os.path.join(cache_dir, "assets", "noise") ir_dir = os.path.join(cache_dir, "assets", "ir") if os.path.isdir(noise_dir) and os.path.isdir(ir_dir): return noise_dir, ir_dir return None, None except Exception as e: logger.warning(f"Could not download augmentation assets: {e}") return None, None def load_pretrained_weights(model: nn.Module, pretrained_path: str) -> None: """Load pretrained weights into model, handling speaker embedding mismatch""" logger.info(f"Loading pretrained weights: {pretrained_path}") state_dict = torch.load(pretrained_path, map_location="cpu", weights_only=True) # Handle different checkpoint formats if "model" in state_dict: state_dict = state_dict["model"] # Filter out mismatched keys, but handle emb_g specially model_state = model.state_dict() filtered_state = {} skipped = [] for k, v in state_dict.items(): if k in model_state: if v.shape == model_state[k].shape: filtered_state[k] = v elif k == "emb_g.weight": # Initialize our speaker embedding with mean of pretrained embeddings # This gives a much better starting point than random initialization mean_emb = v.mean(dim=0, keepdim=True) # [1, 256] num_speakers = model_state[k].shape[0] filtered_state[k] = mean_emb.expand(num_speakers, -1).clone() logger.info(f"Initialized emb_g from pretrained mean ({v.shape[0]} -> {num_speakers} speakers)") else: skipped.append(f"{k}: {v.shape} vs {model_state[k].shape}") else: skipped.append(f"{k}: not in model") if skipped: logger.info(f"Skipped {len(skipped)} mismatched keys") model.load_state_dict(filtered_state, strict=False) logger.info(f"Loaded {len(filtered_state)}/{len(state_dict)} pretrained weights") logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Device selection: # - Inference: Always CPU (HF Spaces free tier, also works everywhere) # - Training: GPU if available for speed, CPU fallback device = torch.device("cpu") # For inference train_device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # For training logger.info(f"Inference device: {device}") logger.info(f"Training device: {train_device}") # ============================================================ # COMMONS - Helper functions from infer/lib/infer_pack/commons.py # ============================================================ def init_weights(m, mean=0.0, std=0.01): classname = m.__class__.__name__ if classname.find("Conv") != -1: m.weight.data.normal_(mean, std) def get_padding(kernel_size, dilation=1): return int((kernel_size * dilation - dilation) / 2) def sequence_mask(length: torch.Tensor, max_length: Optional[int] = None): if max_length is None: max_length = length.max() x = torch.arange(max_length, dtype=length.dtype, device=length.device) return x.unsqueeze(0) < length.unsqueeze(1) @torch.jit.script def fused_add_tanh_sigmoid_multiply(input_a, input_b, n_channels): n_channels_int = n_channels[0] in_act = input_a + input_b t_act = torch.tanh(in_act[:, :n_channels_int, :]) s_act = torch.sigmoid(in_act[:, n_channels_int:, :]) return t_act * s_act def slice_segments(x, ids_str, segment_size=4): """Slice segments from tensor""" ret = torch.zeros_like(x[:, :, :segment_size]) for i in range(x.size(0)): idx_str = ids_str[i] idx_end = idx_str + segment_size ret[i] = x[i, :, idx_str:idx_end] return ret def slice_segments2(x, ids_str, segment_size=4): """Slice segments from 2D tensor""" ret = torch.zeros_like(x[:, :segment_size]) for i in range(x.size(0)): idx_str = ids_str[i] idx_end = idx_str + segment_size ret[i] = x[i, idx_str:idx_end] return ret def rand_slice_segments(x, x_lengths=None, segment_size=4): """Random slice segments""" b, d, t = x.size() if x_lengths is None: x_lengths = t ids_str_max = torch.clamp(x_lengths - segment_size + 1, min=1) ids_str = (torch.rand([b], device=x.device) * ids_str_max.float()).long() ret = slice_segments(x, ids_str, segment_size) return ret, ids_str # ============================================================ # MODULES - From infer/lib/infer_pack/modules.py # ============================================================ LRELU_SLOPE = 0.1 class LayerNorm(nn.Module): def __init__(self, channels, eps=1e-5): super().__init__() self.channels = channels self.eps = eps self.gamma = nn.Parameter(torch.ones(channels)) self.beta = nn.Parameter(torch.zeros(channels)) def forward(self, x): x = x.transpose(1, -1) x = F.layer_norm(x, (self.channels,), self.gamma, self.beta, self.eps) return x.transpose(1, -1) class WN(nn.Module): def __init__(self, hidden_channels, kernel_size, dilation_rate, n_layers, gin_channels=0, p_dropout=0): super().__init__() assert kernel_size % 2 == 1 self.hidden_channels = hidden_channels self.kernel_size = (kernel_size,) self.dilation_rate = dilation_rate self.n_layers = n_layers self.gin_channels = gin_channels self.p_dropout = float(p_dropout) self.in_layers = nn.ModuleList() self.res_skip_layers = nn.ModuleList() self.drop = nn.Dropout(float(p_dropout)) if gin_channels != 0: cond_layer = nn.Conv1d(gin_channels, 2 * hidden_channels * n_layers, 1) self.cond_layer = weight_norm(cond_layer, name="weight") for i in range(n_layers): dilation = dilation_rate ** i padding = int((kernel_size * dilation - dilation) / 2) in_layer = nn.Conv1d(hidden_channels, 2 * hidden_channels, kernel_size, dilation=dilation, padding=padding) in_layer = weight_norm(in_layer, name="weight") self.in_layers.append(in_layer) if i < n_layers - 1: res_skip_channels = 2 * hidden_channels else: res_skip_channels = hidden_channels res_skip_layer = nn.Conv1d(hidden_channels, res_skip_channels, 1) res_skip_layer = weight_norm(res_skip_layer, name="weight") self.res_skip_layers.append(res_skip_layer) def forward(self, x: torch.Tensor, x_mask: torch.Tensor, g: Optional[torch.Tensor] = None): output = torch.zeros_like(x) n_channels_tensor = torch.IntTensor([self.hidden_channels]) if g is not None: g = self.cond_layer(g) for i, (in_layer, res_skip_layer) in enumerate(zip(self.in_layers, self.res_skip_layers)): x_in = in_layer(x) if g is not None: cond_offset = i * 2 * self.hidden_channels g_l = g[:, cond_offset:cond_offset + 2 * self.hidden_channels, :] else: g_l = torch.zeros_like(x_in) acts = fused_add_tanh_sigmoid_multiply(x_in, g_l, n_channels_tensor) acts = self.drop(acts) res_skip_acts = res_skip_layer(acts) if i < self.n_layers - 1: res_acts = res_skip_acts[:, :self.hidden_channels, :] x = (x + res_acts) * x_mask output = output + res_skip_acts[:, self.hidden_channels:, :] else: output = output + res_skip_acts return output * x_mask def remove_weight_norm(self): if self.gin_channels != 0: remove_weight_norm(self.cond_layer) for l in self.in_layers: remove_weight_norm(l) for l in self.res_skip_layers: remove_weight_norm(l) class ResBlock1(nn.Module): def __init__(self, channels, kernel_size=3, dilation=(1, 3, 5)): super().__init__() self.convs1 = nn.ModuleList([ weight_norm(Conv1d(channels, channels, kernel_size, 1, dilation=dilation[0], padding=get_padding(kernel_size, dilation[0]))), weight_norm(Conv1d(channels, channels, kernel_size, 1, dilation=dilation[1], padding=get_padding(kernel_size, dilation[1]))), weight_norm(Conv1d(channels, channels, kernel_size, 1, dilation=dilation[2], padding=get_padding(kernel_size, dilation[2]))), ]) self.convs1.apply(init_weights) self.convs2 = nn.ModuleList([ weight_norm(Conv1d(channels, channels, kernel_size, 1, dilation=1, padding=get_padding(kernel_size, 1))), weight_norm(Conv1d(channels, channels, kernel_size, 1, dilation=1, padding=get_padding(kernel_size, 1))), weight_norm(Conv1d(channels, channels, kernel_size, 1, dilation=1, padding=get_padding(kernel_size, 1))), ]) self.convs2.apply(init_weights) self.lrelu_slope = LRELU_SLOPE def forward(self, x: torch.Tensor, x_mask: Optional[torch.Tensor] = None): for c1, c2 in zip(self.convs1, self.convs2): xt = F.leaky_relu(x, self.lrelu_slope) if x_mask is not None: xt = xt * x_mask xt = c1(xt) xt = F.leaky_relu(xt, self.lrelu_slope) if x_mask is not None: xt = xt * x_mask xt = c2(xt) x = xt + x if x_mask is not None: x = x * x_mask return x def remove_weight_norm(self): for l in self.convs1: remove_weight_norm(l) for l in self.convs2: remove_weight_norm(l) class ResBlock2(nn.Module): def __init__(self, channels, kernel_size=3, dilation=(1, 3)): super().__init__() self.convs = nn.ModuleList([ weight_norm(Conv1d(channels, channels, kernel_size, 1, dilation=dilation[0], padding=get_padding(kernel_size, dilation[0]))), weight_norm(Conv1d(channels, channels, kernel_size, 1, dilation=dilation[1], padding=get_padding(kernel_size, dilation[1]))), ]) self.convs.apply(init_weights) self.lrelu_slope = LRELU_SLOPE def forward(self, x, x_mask: Optional[torch.Tensor] = None): for c in self.convs: xt = F.leaky_relu(x, self.lrelu_slope) if x_mask is not None: xt = xt * x_mask xt = c(xt) x = xt + x if x_mask is not None: x = x * x_mask return x def remove_weight_norm(self): for l in self.convs: remove_weight_norm(l) class Flip(nn.Module): def forward(self, x: torch.Tensor, x_mask: torch.Tensor, g: Optional[torch.Tensor] = None, reverse: bool = False): x = torch.flip(x, [1]) if not reverse: logdet = torch.zeros(x.size(0)).to(dtype=x.dtype, device=x.device) return x, logdet else: return x, torch.zeros([1], device=x.device) class ResidualCouplingLayer(nn.Module): def __init__(self, channels, hidden_channels, kernel_size, dilation_rate, n_layers, p_dropout=0, gin_channels=0, mean_only=False): assert channels % 2 == 0 super().__init__() self.channels = channels self.hidden_channels = hidden_channels self.half_channels = channels // 2 self.mean_only = mean_only self.pre = nn.Conv1d(self.half_channels, hidden_channels, 1) self.enc = WN(hidden_channels, kernel_size, dilation_rate, n_layers, p_dropout=float(p_dropout), gin_channels=gin_channels) self.post = nn.Conv1d(hidden_channels, self.half_channels * (2 - mean_only), 1) self.post.weight.data.zero_() self.post.bias.data.zero_() def forward(self, x: torch.Tensor, x_mask: torch.Tensor, g: Optional[torch.Tensor] = None, reverse: bool = False): x0, x1 = torch.split(x, [self.half_channels] * 2, 1) h = self.pre(x0) * x_mask h = self.enc(h, x_mask, g=g) stats = self.post(h) * x_mask if not self.mean_only: m, logs = torch.split(stats, [self.half_channels] * 2, 1) else: m = stats logs = torch.zeros_like(m) if not reverse: x1 = m + x1 * torch.exp(logs) * x_mask x = torch.cat([x0, x1], 1) logdet = torch.sum(logs, [1, 2]) return x, logdet else: x1 = (x1 - m) * torch.exp(-logs) * x_mask x = torch.cat([x0, x1], 1) return x, torch.zeros([1]) def remove_weight_norm(self): self.enc.remove_weight_norm() # ============================================================ # ATTENTIONS - From infer/lib/infer_pack/attentions.py # ============================================================ class MultiHeadAttention(nn.Module): def __init__(self, channels, out_channels, n_heads, p_dropout=0.0, window_size=None, heads_share=True, proximal_bias=False, proximal_init=False): super().__init__() assert channels % n_heads == 0 self.channels = channels self.out_channels = out_channels self.n_heads = n_heads self.p_dropout = p_dropout self.window_size = window_size self.heads_share = heads_share self.proximal_bias = proximal_bias self.proximal_init = proximal_init self.k_channels = channels // n_heads self.conv_q = nn.Conv1d(channels, channels, 1) self.conv_k = nn.Conv1d(channels, channels, 1) self.conv_v = nn.Conv1d(channels, channels, 1) self.conv_o = nn.Conv1d(channels, out_channels, 1) self.drop = nn.Dropout(p_dropout) if window_size is not None: n_heads_rel = 1 if heads_share else n_heads rel_stddev = self.k_channels ** -0.5 self.emb_rel_k = nn.Parameter(torch.randn(n_heads_rel, window_size * 2 + 1, self.k_channels) * rel_stddev) self.emb_rel_v = nn.Parameter(torch.randn(n_heads_rel, window_size * 2 + 1, self.k_channels) * rel_stddev) nn.init.xavier_uniform_(self.conv_q.weight) nn.init.xavier_uniform_(self.conv_k.weight) nn.init.xavier_uniform_(self.conv_v.weight) if proximal_init: with torch.no_grad(): self.conv_k.weight.copy_(self.conv_q.weight) self.conv_k.bias.copy_(self.conv_q.bias) def forward(self, x: torch.Tensor, c: torch.Tensor, attn_mask: Optional[torch.Tensor] = None): q = self.conv_q(x) k = self.conv_k(c) v = self.conv_v(c) x, _ = self.attention(q, k, v, mask=attn_mask) x = self.conv_o(x) return x def attention(self, query, key, value, mask=None): b, d, t_s = key.size() t_t = query.size(2) query = query.view(b, self.n_heads, self.k_channels, t_t).transpose(2, 3) key = key.view(b, self.n_heads, self.k_channels, t_s).transpose(2, 3) value = value.view(b, self.n_heads, self.k_channels, t_s).transpose(2, 3) scores = torch.matmul(query / math.sqrt(self.k_channels), key.transpose(-2, -1)) if self.window_size is not None: key_relative_embeddings = self._get_relative_embeddings(self.emb_rel_k, t_s) rel_logits = self._matmul_with_relative_keys(query / math.sqrt(self.k_channels), key_relative_embeddings) scores_local = self._relative_position_to_absolute_position(rel_logits) scores = scores + scores_local if self.proximal_bias: scores = scores + self._attention_bias_proximal(t_s).to(device=scores.device, dtype=scores.dtype) if mask is not None: scores = scores.masked_fill(mask == 0, -1e4) p_attn = F.softmax(scores, dim=-1) p_attn = self.drop(p_attn) output = torch.matmul(p_attn, value) if self.window_size is not None: relative_weights = self._absolute_position_to_relative_position(p_attn) value_relative_embeddings = self._get_relative_embeddings(self.emb_rel_v, t_s) output = output + self._matmul_with_relative_values(relative_weights, value_relative_embeddings) output = output.transpose(2, 3).contiguous().view(b, d, t_t) return output, p_attn def _matmul_with_relative_values(self, x, y): return torch.matmul(x, y.unsqueeze(0)) def _matmul_with_relative_keys(self, x, y): return torch.matmul(x, y.unsqueeze(0).transpose(-2, -1)) def _get_relative_embeddings(self, relative_embeddings, length): pad_length = max(length - (self.window_size + 1), 0) slice_start_position = max((self.window_size + 1) - length, 0) slice_end_position = slice_start_position + 2 * length - 1 if pad_length > 0: padded_relative_embeddings = F.pad(relative_embeddings, [0, 0, pad_length, pad_length, 0, 0]) else: padded_relative_embeddings = relative_embeddings return padded_relative_embeddings[:, slice_start_position:slice_end_position] def _relative_position_to_absolute_position(self, x): batch, heads, length, _ = x.size() x = F.pad(x, [0, 1, 0, 0, 0, 0, 0, 0]) x_flat = x.view([batch, heads, length * 2 * length]) x_flat = F.pad(x_flat, [0, int(length) - 1, 0, 0, 0, 0]) x_final = x_flat.view([batch, heads, length + 1, 2 * length - 1])[:, :, :length, length - 1:] return x_final def _absolute_position_to_relative_position(self, x): batch, heads, length, _ = x.size() x = F.pad(x, [0, int(length) - 1, 0, 0, 0, 0, 0, 0]) x_flat = x.view([batch, heads, int(length ** 2) + int(length * (length - 1))]) x_flat = F.pad(x_flat, [length, 0, 0, 0, 0, 0]) x_final = x_flat.view([batch, heads, length, 2 * length])[:, :, :, 1:] return x_final def _attention_bias_proximal(self, length): r = torch.arange(length, dtype=torch.float32) diff = torch.unsqueeze(r, 0) - torch.unsqueeze(r, 1) return torch.unsqueeze(torch.unsqueeze(-torch.log1p(torch.abs(diff)), 0), 0) class FFN(nn.Module): def __init__(self, in_channels, out_channels, filter_channels, kernel_size, p_dropout=0.0, activation=None, causal=False): super().__init__() self.in_channels = in_channels self.out_channels = out_channels self.filter_channels = filter_channels self.kernel_size = kernel_size self.p_dropout = p_dropout self.causal = causal self.is_activation = activation == "gelu" self.conv_1 = nn.Conv1d(in_channels, filter_channels, kernel_size) self.conv_2 = nn.Conv1d(filter_channels, out_channels, kernel_size) self.drop = nn.Dropout(p_dropout) def forward(self, x: torch.Tensor, x_mask: torch.Tensor): x = self.conv_1(self._padding(x, x_mask)) if self.is_activation: x = x * torch.sigmoid(1.702 * x) else: x = torch.relu(x) x = self.drop(x) x = self.conv_2(self._padding(x, x_mask)) return x * x_mask def _padding(self, x, x_mask): if self.causal: if self.kernel_size == 1: return x * x_mask pad_l = self.kernel_size - 1 return F.pad(x * x_mask, [pad_l, 0, 0, 0, 0, 0]) else: if self.kernel_size == 1: return x * x_mask pad_l = (self.kernel_size - 1) // 2 pad_r = self.kernel_size // 2 return F.pad(x * x_mask, [pad_l, pad_r, 0, 0, 0, 0]) class Encoder(nn.Module): def __init__(self, hidden_channels, filter_channels, n_heads, n_layers, kernel_size=1, p_dropout=0.0, window_size=10, **kwargs): super().__init__() self.hidden_channels = hidden_channels self.n_layers = int(n_layers) self.drop = nn.Dropout(p_dropout) self.attn_layers = nn.ModuleList() self.norm_layers_1 = nn.ModuleList() self.ffn_layers = nn.ModuleList() self.norm_layers_2 = nn.ModuleList() for i in range(self.n_layers): self.attn_layers.append(MultiHeadAttention(hidden_channels, hidden_channels, n_heads, p_dropout=p_dropout, window_size=window_size)) self.norm_layers_1.append(LayerNorm(hidden_channels)) self.ffn_layers.append(FFN(hidden_channels, hidden_channels, filter_channels, kernel_size, p_dropout=p_dropout)) self.norm_layers_2.append(LayerNorm(hidden_channels)) def forward(self, x, x_mask): attn_mask = x_mask.unsqueeze(2) * x_mask.unsqueeze(-1) x = x * x_mask for attn, norm1, ffn, norm2 in zip(self.attn_layers, self.norm_layers_1, self.ffn_layers, self.norm_layers_2): y = attn(x, x, attn_mask) y = self.drop(y) x = norm1(x + y) y = ffn(x, x_mask) y = self.drop(y) x = norm2(x + y) return x * x_mask # ============================================================ # MODELS - From infer/lib/infer_pack/models.py # ============================================================ sr2sr = {"32k": 32000, "40k": 40000, "48k": 48000} class TextEncoder256(nn.Module): def __init__(self, out_channels, hidden_channels, filter_channels, n_heads, n_layers, kernel_size, p_dropout, f0=True): super().__init__() self.out_channels = out_channels self.hidden_channels = hidden_channels self.emb_phone = nn.Linear(256, hidden_channels) self.lrelu = nn.LeakyReLU(0.1, inplace=True) if f0: self.emb_pitch = nn.Embedding(256, hidden_channels) self.encoder = Encoder(hidden_channels, filter_channels, n_heads, n_layers, kernel_size, float(p_dropout)) self.proj = nn.Conv1d(hidden_channels, out_channels * 2, 1) def forward(self, phone: torch.Tensor, pitch: Optional[torch.Tensor], lengths: torch.Tensor): if pitch is None: x = self.emb_phone(phone) else: x = self.emb_phone(phone) + self.emb_pitch(pitch) x = x * math.sqrt(self.hidden_channels) x = self.lrelu(x) x = torch.transpose(x, 1, -1) x_mask = torch.unsqueeze(sequence_mask(lengths, x.size(2)), 1).to(x.dtype) x = self.encoder(x * x_mask, x_mask) stats = self.proj(x) * x_mask m, logs = torch.split(stats, self.out_channels, dim=1) return m, logs, x_mask class TextEncoder768(nn.Module): def __init__(self, out_channels, hidden_channels, filter_channels, n_heads, n_layers, kernel_size, p_dropout, f0=True): super().__init__() self.out_channels = out_channels self.hidden_channels = hidden_channels self.emb_phone = nn.Linear(768, hidden_channels) self.lrelu = nn.LeakyReLU(0.1, inplace=True) if f0: self.emb_pitch = nn.Embedding(256, hidden_channels) self.encoder = Encoder(hidden_channels, filter_channels, n_heads, n_layers, kernel_size, float(p_dropout)) self.proj = nn.Conv1d(hidden_channels, out_channels * 2, 1) def forward(self, phone: torch.Tensor, pitch: Optional[torch.Tensor], lengths: torch.Tensor): if pitch is None: x = self.emb_phone(phone) else: x = self.emb_phone(phone) + self.emb_pitch(pitch) x = x * math.sqrt(self.hidden_channels) x = self.lrelu(x) x = torch.transpose(x, 1, -1) x_mask = torch.unsqueeze(sequence_mask(lengths, x.size(2)), 1).to(x.dtype) x = self.encoder(x * x_mask, x_mask) stats = self.proj(x) * x_mask m, logs = torch.split(stats, self.out_channels, dim=1) return m, logs, x_mask class ResidualCouplingBlock(nn.Module): def __init__(self, channels, hidden_channels, kernel_size, dilation_rate, n_layers, n_flows=4, gin_channels=0): super().__init__() self.n_flows = n_flows self.flows = nn.ModuleList() for i in range(n_flows): self.flows.append(ResidualCouplingLayer(channels, hidden_channels, kernel_size, dilation_rate, n_layers, gin_channels=gin_channels, mean_only=True)) self.flows.append(Flip()) def forward(self, x: torch.Tensor, x_mask: torch.Tensor, g: Optional[torch.Tensor] = None, reverse: bool = False): if not reverse: for flow in self.flows: x, _ = flow(x, x_mask, g=g, reverse=reverse) else: for flow in self.flows[::-1]: x, _ = flow.forward(x, x_mask, g=g, reverse=reverse) return x def remove_weight_norm(self): for i in range(self.n_flows): self.flows[i * 2].remove_weight_norm() class PosteriorEncoder(nn.Module): def __init__(self, in_channels, out_channels, hidden_channels, kernel_size, dilation_rate, n_layers, gin_channels=0): super().__init__() self.out_channels = out_channels self.pre = nn.Conv1d(in_channels, hidden_channels, 1) self.enc = WN(hidden_channels, kernel_size, dilation_rate, n_layers, gin_channels=gin_channels) self.proj = nn.Conv1d(hidden_channels, out_channels * 2, 1) def forward(self, x: torch.Tensor, x_lengths: torch.Tensor, g: Optional[torch.Tensor] = None): x_mask = torch.unsqueeze(sequence_mask(x_lengths, x.size(2)), 1).to(x.dtype) x = self.pre(x) * x_mask x = self.enc(x, x_mask, g=g) stats = self.proj(x) * x_mask m, logs = torch.split(stats, self.out_channels, dim=1) z = (m + torch.randn_like(m) * torch.exp(logs)) * x_mask return z, m, logs, x_mask def remove_weight_norm(self): self.enc.remove_weight_norm() class Generator(nn.Module): def __init__(self, initial_channel, resblock, resblock_kernel_sizes, resblock_dilation_sizes, upsample_rates, upsample_initial_channel, upsample_kernel_sizes, gin_channels=0): super().__init__() self.num_kernels = len(resblock_kernel_sizes) self.num_upsamples = len(upsample_rates) self.conv_pre = Conv1d(initial_channel, upsample_initial_channel, 7, 1, padding=3) resblock_class = ResBlock1 if resblock == "1" else ResBlock2 self.ups = nn.ModuleList() for i, (u, k) in enumerate(zip(upsample_rates, upsample_kernel_sizes)): self.ups.append(weight_norm(ConvTranspose1d(upsample_initial_channel // (2 ** i), upsample_initial_channel // (2 ** (i + 1)), k, u, padding=(k - u) // 2))) self.resblocks = nn.ModuleList() for i in range(len(self.ups)): ch = upsample_initial_channel // (2 ** (i + 1)) for j, (k, d) in enumerate(zip(resblock_kernel_sizes, resblock_dilation_sizes)): self.resblocks.append(resblock_class(ch, k, d)) self.conv_post = Conv1d(ch, 1, 7, 1, padding=3, bias=False) self.ups.apply(init_weights) if gin_channels != 0: self.cond = nn.Conv1d(gin_channels, upsample_initial_channel, 1) def forward(self, x: torch.Tensor, g: Optional[torch.Tensor] = None): x = self.conv_pre(x) if g is not None: x = x + self.cond(g) for i in range(self.num_upsamples): x = F.leaky_relu(x, LRELU_SLOPE) x = self.ups[i](x) xs = None for j in range(self.num_kernels): if xs is None: xs = self.resblocks[i * self.num_kernels + j](x) else: xs += self.resblocks[i * self.num_kernels + j](x) x = xs / self.num_kernels x = F.leaky_relu(x) x = self.conv_post(x) x = torch.tanh(x) return x def remove_weight_norm(self): for l in self.ups: remove_weight_norm(l) for l in self.resblocks: l.remove_weight_norm() class SineGen(nn.Module): def __init__(self, samp_rate, harmonic_num=0, sine_amp=0.1, noise_std=0.003, voiced_threshold=0): super().__init__() self.sine_amp = sine_amp self.noise_std = noise_std self.harmonic_num = harmonic_num self.dim = harmonic_num + 1 self.sampling_rate = samp_rate self.voiced_threshold = voiced_threshold def _f02uv(self, f0): uv = torch.ones_like(f0) uv = uv * (f0 > self.voiced_threshold) return uv.float() def forward(self, f0: torch.Tensor, upp: int): with torch.no_grad(): f0 = f0[:, None].transpose(1, 2) f0_buf = torch.zeros(f0.shape[0], f0.shape[1], self.dim, device=f0.device) f0_buf[:, :, 0] = f0[:, :, 0] for idx in range(self.harmonic_num): f0_buf[:, :, idx + 1] = f0_buf[:, :, 0] * (idx + 2) rad_values = (f0_buf / self.sampling_rate) % 1 rand_ini = torch.rand(f0_buf.shape[0], f0_buf.shape[2], device=f0_buf.device) rand_ini[:, 0] = 0 rad_values[:, 0, :] = rad_values[:, 0, :] + rand_ini tmp_over_one = torch.cumsum(rad_values, 1) tmp_over_one *= upp tmp_over_one = F.interpolate(tmp_over_one.transpose(2, 1), scale_factor=float(upp), mode="linear", align_corners=True).transpose(2, 1) rad_values = F.interpolate(rad_values.transpose(2, 1), scale_factor=float(upp), mode="nearest").transpose(2, 1) tmp_over_one %= 1 tmp_over_one_idx = (tmp_over_one[:, 1:, :] - tmp_over_one[:, :-1, :]) < 0 cumsum_shift = torch.zeros_like(rad_values) cumsum_shift[:, 1:, :] = tmp_over_one_idx * -1.0 sine_waves = torch.sin(torch.cumsum(rad_values + cumsum_shift, dim=1) * 2 * torch.pi) sine_waves = sine_waves * self.sine_amp uv = self._f02uv(f0) uv = F.interpolate(uv.transpose(2, 1), scale_factor=float(upp), mode="nearest").transpose(2, 1) noise_amp = uv * self.noise_std + (1 - uv) * self.sine_amp / 3 noise = noise_amp * torch.randn_like(sine_waves) sine_waves = sine_waves * uv + noise return sine_waves, uv, noise class SourceModuleHnNSF(nn.Module): def __init__(self, sampling_rate, harmonic_num=0, sine_amp=0.1, add_noise_std=0.003, voiced_threshod=0, is_half=False): super().__init__() self.l_sin_gen = SineGen(sampling_rate, harmonic_num, sine_amp, add_noise_std, voiced_threshod) self.l_linear = nn.Linear(harmonic_num + 1, 1) self.l_tanh = nn.Tanh() def forward(self, x: torch.Tensor, upp: int = 1): sine_wavs, uv, _ = self.l_sin_gen(x, upp) sine_wavs = sine_wavs.to(dtype=self.l_linear.weight.dtype) sine_merge = self.l_tanh(self.l_linear(sine_wavs)) return sine_merge, None, None class GeneratorNSF(nn.Module): def __init__(self, initial_channel, resblock, resblock_kernel_sizes, resblock_dilation_sizes, upsample_rates, upsample_initial_channel, upsample_kernel_sizes, gin_channels, sr, is_half=False): super().__init__() self.num_kernels = len(resblock_kernel_sizes) self.num_upsamples = len(upsample_rates) self.f0_upsamp = nn.Upsample(scale_factor=math.prod(upsample_rates)) self.m_source = SourceModuleHnNSF(sampling_rate=sr, harmonic_num=0, is_half=is_half) self.noise_convs = nn.ModuleList() self.conv_pre = Conv1d(initial_channel, upsample_initial_channel, 7, 1, padding=3) resblock_class = ResBlock1 if resblock == "1" else ResBlock2 self.ups = nn.ModuleList() for i, (u, k) in enumerate(zip(upsample_rates, upsample_kernel_sizes)): c_cur = upsample_initial_channel // (2 ** (i + 1)) self.ups.append(weight_norm(ConvTranspose1d(upsample_initial_channel // (2 ** i), upsample_initial_channel // (2 ** (i + 1)), k, u, padding=(k - u) // 2))) if i + 1 < len(upsample_rates): stride_f0 = math.prod(upsample_rates[i + 1:]) self.noise_convs.append(Conv1d(1, c_cur, kernel_size=stride_f0 * 2, stride=stride_f0, padding=stride_f0 // 2)) else: self.noise_convs.append(Conv1d(1, c_cur, kernel_size=1)) self.resblocks = nn.ModuleList() for i in range(len(self.ups)): ch = upsample_initial_channel // (2 ** (i + 1)) for j, (k, d) in enumerate(zip(resblock_kernel_sizes, resblock_dilation_sizes)): self.resblocks.append(resblock_class(ch, k, d)) self.conv_post = Conv1d(ch, 1, 7, 1, padding=3, bias=False) self.ups.apply(init_weights) if gin_channels != 0: self.cond = nn.Conv1d(gin_channels, upsample_initial_channel, 1) self.upp = math.prod(upsample_rates) self.lrelu_slope = LRELU_SLOPE def forward(self, x, f0, g: Optional[torch.Tensor] = None): har_source, _, _ = self.m_source(f0, self.upp) har_source = har_source.transpose(1, 2) x = self.conv_pre(x) if g is not None: x = x + self.cond(g) for i, (ups, noise_convs) in enumerate(zip(self.ups, self.noise_convs)): if i < self.num_upsamples: x = F.leaky_relu(x, self.lrelu_slope) x = ups(x) x_source = noise_convs(har_source) x = x + x_source xs = None l = [i * self.num_kernels + j for j in range(self.num_kernels)] for j, resblock in enumerate(self.resblocks): if j in l: if xs is None: xs = resblock(x) else: xs += resblock(x) x = xs / self.num_kernels x = F.leaky_relu(x) x = self.conv_post(x) x = torch.tanh(x) return x def remove_weight_norm(self): for l in self.ups: remove_weight_norm(l) for l in self.resblocks: l.remove_weight_norm() # Synthesizer classes for different model versions class SynthesizerTrnMs256NSFsid(nn.Module): """RVC v1 model with f0""" def __init__(self, spec_channels, segment_size, inter_channels, hidden_channels, filter_channels, n_heads, n_layers, kernel_size, p_dropout, resblock, resblock_kernel_sizes, resblock_dilation_sizes, upsample_rates, upsample_initial_channel, upsample_kernel_sizes, spk_embed_dim, gin_channels, sr, **kwargs): super().__init__() if isinstance(sr, str): sr = sr2sr[sr] self.segment_size = segment_size self.gin_channels = gin_channels self.spk_embed_dim = spk_embed_dim self.enc_p = TextEncoder256(inter_channels, hidden_channels, filter_channels, n_heads, n_layers, kernel_size, float(p_dropout)) self.dec = GeneratorNSF(inter_channels, resblock, resblock_kernel_sizes, resblock_dilation_sizes, upsample_rates, upsample_initial_channel, upsample_kernel_sizes, gin_channels=gin_channels, sr=sr, is_half=kwargs.get("is_half", False)) self.enc_q = PosteriorEncoder(spec_channels, inter_channels, hidden_channels, 5, 1, 16, gin_channels=gin_channels) self.flow = ResidualCouplingBlock(inter_channels, hidden_channels, 5, 1, 3, gin_channels=gin_channels) self.emb_g = nn.Embedding(spk_embed_dim, gin_channels) @torch.jit.export def infer(self, phone: torch.Tensor, phone_lengths: torch.Tensor, pitch: torch.Tensor, nsff0: torch.Tensor, sid: torch.Tensor, rate: Optional[torch.Tensor] = None): g = self.emb_g(sid).unsqueeze(-1) m_p, logs_p, x_mask = self.enc_p(phone, pitch, phone_lengths) z_p = (m_p + torch.exp(logs_p) * torch.randn_like(m_p) * 0.66666) * x_mask if rate is not None: head = int(z_p.shape[2] * (1 - rate.item())) z_p = z_p[:, :, head:] x_mask = x_mask[:, :, head:] nsff0 = nsff0[:, head:] z = self.flow(z_p, x_mask, g=g, reverse=True) o = self.dec(z * x_mask, nsff0, g=g) return o, x_mask, (z, z_p, m_p, logs_p) class SynthesizerTrnMs768NSFsid(nn.Module): """RVC v2 model with f0""" def __init__(self, spec_channels, segment_size, inter_channels, hidden_channels, filter_channels, n_heads, n_layers, kernel_size, p_dropout, resblock, resblock_kernel_sizes, resblock_dilation_sizes, upsample_rates, upsample_initial_channel, upsample_kernel_sizes, spk_embed_dim, gin_channels, sr, **kwargs): super().__init__() if isinstance(sr, str): sr = sr2sr[sr] self.segment_size = segment_size self.gin_channels = gin_channels self.spk_embed_dim = spk_embed_dim self.enc_p = TextEncoder768(inter_channels, hidden_channels, filter_channels, n_heads, n_layers, kernel_size, float(p_dropout)) self.dec = GeneratorNSF(inter_channels, resblock, resblock_kernel_sizes, resblock_dilation_sizes, upsample_rates, upsample_initial_channel, upsample_kernel_sizes, gin_channels=gin_channels, sr=sr, is_half=kwargs.get("is_half", False)) self.enc_q = PosteriorEncoder(spec_channels, inter_channels, hidden_channels, 5, 1, 16, gin_channels=gin_channels) self.flow = ResidualCouplingBlock(inter_channels, hidden_channels, 5, 1, 3, gin_channels=gin_channels) self.emb_g = nn.Embedding(spk_embed_dim, gin_channels) def forward(self, phone, phone_lengths, pitch, pitchf, y, y_lengths, ds): """Training forward pass""" g = self.emb_g(ds).unsqueeze(-1) m_p, logs_p, x_mask = self.enc_p(phone, pitch, phone_lengths) z, m_q, logs_q, y_mask = self.enc_q(y, y_lengths, g=g) z_p = self.flow(z, y_mask, g=g) z_slice, ids_slice = rand_slice_segments(z, y_lengths, self.segment_size) pitchf = slice_segments2(pitchf, ids_slice, self.segment_size) o = self.dec(z_slice, pitchf, g=g) return o, ids_slice, x_mask, y_mask, (z, z_p, m_p, logs_p, m_q, logs_q) @torch.jit.export def infer(self, phone: torch.Tensor, phone_lengths: torch.Tensor, pitch: torch.Tensor, nsff0: torch.Tensor, sid: torch.Tensor, rate: Optional[torch.Tensor] = None): g = self.emb_g(sid).unsqueeze(-1) m_p, logs_p, x_mask = self.enc_p(phone, pitch, phone_lengths) z_p = (m_p + torch.exp(logs_p) * torch.randn_like(m_p) * 0.66666) * x_mask if rate is not None: head = int(z_p.shape[2] * (1.0 - rate.item())) z_p = z_p[:, :, head:] x_mask = x_mask[:, :, head:] nsff0 = nsff0[:, head:] z = self.flow(z_p, x_mask, g=g, reverse=True) o = self.dec(z * x_mask, nsff0, g=g) return o, x_mask, (z, z_p, m_p, logs_p) class SynthesizerTrnMs256NSFsid_nono(nn.Module): """RVC v1 model without f0""" def __init__(self, spec_channels, segment_size, inter_channels, hidden_channels, filter_channels, n_heads, n_layers, kernel_size, p_dropout, resblock, resblock_kernel_sizes, resblock_dilation_sizes, upsample_rates, upsample_initial_channel, upsample_kernel_sizes, spk_embed_dim, gin_channels, sr=None, **kwargs): super().__init__() self.segment_size = segment_size self.gin_channels = gin_channels self.enc_p = TextEncoder256(inter_channels, hidden_channels, filter_channels, n_heads, n_layers, kernel_size, float(p_dropout), f0=False) self.dec = Generator(inter_channels, resblock, resblock_kernel_sizes, resblock_dilation_sizes, upsample_rates, upsample_initial_channel, upsample_kernel_sizes, gin_channels=gin_channels) self.enc_q = PosteriorEncoder(spec_channels, inter_channels, hidden_channels, 5, 1, 16, gin_channels=gin_channels) self.flow = ResidualCouplingBlock(inter_channels, hidden_channels, 5, 1, 3, gin_channels=gin_channels) self.emb_g = nn.Embedding(spk_embed_dim, gin_channels) @torch.jit.export def infer(self, phone: torch.Tensor, phone_lengths: torch.Tensor, sid: torch.Tensor, rate: Optional[torch.Tensor] = None): g = self.emb_g(sid).unsqueeze(-1) m_p, logs_p, x_mask = self.enc_p(phone, None, phone_lengths) z_p = (m_p + torch.exp(logs_p) * torch.randn_like(m_p) * 0.66666) * x_mask if rate is not None: head = int(z_p.shape[2] * (1.0 - rate.item())) z_p = z_p[:, :, head:] x_mask = x_mask[:, :, head:] z = self.flow(z_p, x_mask, g=g, reverse=True) o = self.dec(z * x_mask, g=g) return o, x_mask, (z, z_p, m_p, logs_p) class SynthesizerTrnMs768NSFsid_nono(nn.Module): """RVC v2 model without f0""" def __init__(self, spec_channels, segment_size, inter_channels, hidden_channels, filter_channels, n_heads, n_layers, kernel_size, p_dropout, resblock, resblock_kernel_sizes, resblock_dilation_sizes, upsample_rates, upsample_initial_channel, upsample_kernel_sizes, spk_embed_dim, gin_channels, sr=None, **kwargs): super().__init__() self.segment_size = segment_size self.gin_channels = gin_channels self.enc_p = TextEncoder768(inter_channels, hidden_channels, filter_channels, n_heads, n_layers, kernel_size, float(p_dropout), f0=False) self.dec = Generator(inter_channels, resblock, resblock_kernel_sizes, resblock_dilation_sizes, upsample_rates, upsample_initial_channel, upsample_kernel_sizes, gin_channels=gin_channels) self.enc_q = PosteriorEncoder(spec_channels, inter_channels, hidden_channels, 5, 1, 16, gin_channels=gin_channels) self.flow = ResidualCouplingBlock(inter_channels, hidden_channels, 5, 1, 3, gin_channels=gin_channels) self.emb_g = nn.Embedding(spk_embed_dim, gin_channels) @torch.jit.export def infer(self, phone: torch.Tensor, phone_lengths: torch.Tensor, sid: torch.Tensor, rate: Optional[torch.Tensor] = None): g = self.emb_g(sid).unsqueeze(-1) m_p, logs_p, x_mask = self.enc_p(phone, None, phone_lengths) z_p = (m_p + torch.exp(logs_p) * torch.randn_like(m_p) * 0.66666) * x_mask if rate is not None: head = int(z_p.shape[2] * (1.0 - rate.item())) z_p = z_p[:, :, head:] x_mask = x_mask[:, :, head:] z = self.flow(z_p, x_mask, g=g, reverse=True) o = self.dec(z * x_mask, g=g) return o, x_mask, (z, z_p, m_p, logs_p) # ============================================================ # DISCRIMINATOR - For training # ============================================================ class DiscriminatorS(nn.Module): def __init__(self, use_spectral_norm=False): super().__init__() norm_f = nn.utils.spectral_norm if use_spectral_norm else weight_norm self.convs = nn.ModuleList([ norm_f(Conv1d(1, 16, 15, 1, padding=7)), norm_f(Conv1d(16, 64, 41, 4, groups=4, padding=20)), norm_f(Conv1d(64, 256, 41, 4, groups=16, padding=20)), norm_f(Conv1d(256, 1024, 41, 4, groups=64, padding=20)), norm_f(Conv1d(1024, 1024, 41, 4, groups=256, padding=20)), norm_f(Conv1d(1024, 1024, 5, 1, padding=2)), ]) self.conv_post = norm_f(Conv1d(1024, 1, 3, 1, padding=1)) def forward(self, x): fmap = [] for l in self.convs: x = l(x) x = F.leaky_relu(x, 0.1) fmap.append(x) x = self.conv_post(x) fmap.append(x) x = torch.flatten(x, 1, -1) return x, fmap class DiscriminatorP(nn.Module): def __init__(self, period, use_spectral_norm=False): super().__init__() self.period = period norm_f = nn.utils.spectral_norm if use_spectral_norm else weight_norm self.convs = nn.ModuleList([ norm_f(nn.Conv2d(1, 32, (5, 1), (3, 1), padding=(2, 0))), norm_f(nn.Conv2d(32, 128, (5, 1), (3, 1), padding=(2, 0))), norm_f(nn.Conv2d(128, 512, (5, 1), (3, 1), padding=(2, 0))), norm_f(nn.Conv2d(512, 1024, (5, 1), (3, 1), padding=(2, 0))), norm_f(nn.Conv2d(1024, 1024, (5, 1), 1, padding=(2, 0))), ]) self.conv_post = norm_f(nn.Conv2d(1024, 1, (3, 1), 1, padding=(1, 0))) def forward(self, x): fmap = [] b, c, t = x.shape if t % self.period != 0: n_pad = self.period - (t % self.period) x = F.pad(x, (0, n_pad), "reflect") t = t + n_pad x = x.view(b, c, t // self.period, self.period) for l in self.convs: x = l(x) x = F.leaky_relu(x, 0.1) fmap.append(x) x = self.conv_post(x) fmap.append(x) x = torch.flatten(x, 1, -1) return x, fmap class MultiPeriodDiscriminator(nn.Module): def __init__(self, use_spectral_norm=False): super().__init__() periods = [2, 3, 5, 7, 11, 17, 23, 37] # 8 periods for v2 pretrained (9 total discriminators) self.discriminators = nn.ModuleList( [DiscriminatorS(use_spectral_norm)] + [DiscriminatorP(p, use_spectral_norm) for p in periods] ) def forward(self, y, y_hat): y_d_rs, y_d_gs, fmap_rs, fmap_gs = [], [], [], [] for d in self.discriminators: y_d_r, fmap_r = d(y) y_d_g, fmap_g = d(y_hat) y_d_rs.append(y_d_r) y_d_gs.append(y_d_g) fmap_rs.append(fmap_r) fmap_gs.append(fmap_g) return y_d_rs, y_d_gs, fmap_rs, fmap_gs # ============================================================ # TRAINING LOSSES # ============================================================ def feature_loss(fmap_r, fmap_g): loss = 0 for dr, dg in zip(fmap_r, fmap_g): for rl, gl in zip(dr, dg): loss += torch.mean(torch.abs(rl.float().detach() - gl.float())) return loss * 2 def discriminator_loss(disc_real_outputs, disc_generated_outputs): loss = 0 for dr, dg in zip(disc_real_outputs, disc_generated_outputs): loss += torch.mean((1 - dr.float()) ** 2) + torch.mean(dg.float() ** 2) return loss def generator_loss(disc_outputs): loss = 0 for dg in disc_outputs: loss += torch.mean((1 - dg.float()) ** 2) return loss def kl_loss(z_p, logs_q, m_p, logs_p, z_mask): z_p, logs_q, m_p, logs_p, z_mask = [x.float() for x in [z_p, logs_q, m_p, logs_p, z_mask]] kl = logs_p - logs_q - 0.5 + 0.5 * ((z_p - m_p) ** 2) * torch.exp(-2.0 * logs_p) return torch.sum(kl * z_mask) / torch.sum(z_mask) # ============================================================ # HUBERT EXTRACTION - Using torchaudio bundle # ============================================================ # ContentVec model for v1 (256-dim) and HuBERT for v2 (768-dim) _contentvec_model = None # For v1 models (256-dim output) _hubert_model = None # For v2 models (768-dim output) _hubert_bundle = None CONTENTVEC_REPO = "IAHispano/Applio" CONTENTVEC_MODEL = "Resources/embedders/contentvec/pytorch_model.bin" CONTENTVEC_CONFIG = "Resources/embedders/contentvec/config.json" def load_contentvec(): """Load ContentVec model from HuggingFace for v1 models (256-dim output)""" global _contentvec_model if _contentvec_model is None: try: from transformers import HubertModel, HubertConfig logger.info("Loading ContentVec model from HuggingFace...") # Download model files model_path = hf_hub_download(repo_id=CONTENTVEC_REPO, filename=CONTENTVEC_MODEL) config_path = hf_hub_download(repo_id=CONTENTVEC_REPO, filename=CONTENTVEC_CONFIG) # Create model with final_proj layer class HubertModelWithFinalProj(HubertModel): def __init__(self, config): super().__init__(config) self.final_proj = nn.Linear(config.hidden_size, config.classifier_proj_size) config = HubertConfig.from_pretrained(config_path) _contentvec_model = HubertModelWithFinalProj(config) state_dict = torch.load(model_path, map_location="cpu", weights_only=True) _contentvec_model.load_state_dict(state_dict) _contentvec_model.to(device).eval() logger.info(f"ContentVec loaded: hidden={config.hidden_size}, proj={config.classifier_proj_size}") except Exception as e: logger.warning(f"Failed to load ContentVec: {e}, falling back to torchaudio HuBERT") _contentvec_model = None return _contentvec_model def load_hubert(): """Load HuBERT model via torchaudio for v2 models (768-dim output)""" global _hubert_model, _hubert_bundle if _hubert_model is None: import torchaudio logger.info("Loading HuBERT model via torchaudio...") _hubert_bundle = torchaudio.pipelines.HUBERT_BASE _hubert_model = _hubert_bundle.get_model().to(device) _hubert_model.eval() logger.info("HuBERT model loaded") return _hubert_model, _hubert_bundle def extract_hubert_features(audio: np.ndarray, sr: int = 16000, version: str = "v2") -> torch.Tensor: """Extract ContentVec features from audio (same as Applio) v1 models: Use ContentVec with final_proj (256-dim) v2 models: Use ContentVec without final_proj (768-dim) """ audio = audio.astype(np.float32) if np.abs(audio).max() > 1.0: audio = audio / np.abs(audio).max() inputs = torch.from_numpy(audio).unsqueeze(0).to(device) # Use ContentVec for ALL versions (same as Applio) contentvec = load_contentvec() if contentvec is not None: with torch.no_grad(): output = contentvec(inputs) if version == "v1": # v1: use final_proj for 256-dim feats = contentvec.final_proj(output.last_hidden_state) else: # v2: use raw hidden state (768-dim) feats = output.last_hidden_state return feats # Fallback to torchaudio HuBERT if ContentVec not available logger.warning("ContentVec not available, using torchaudio HuBERT (results may be degraded)") hubert, bundle = load_hubert() with torch.no_grad(): features, _ = hubert.extract_features(inputs) layer_idx = 11 if version == "v2" else 8 feats = features[min(layer_idx, len(features)-1)] if version == "v1": proj = nn.Linear(768, 256, bias=False).to(device) with torch.no_grad(): w = torch.zeros(256, 768) for i in range(256): w[i, i*3:(i+1)*3] = 1/3 proj.weight.copy_(w) feats = proj(feats) return feats # ============================================================ # F0 EXTRACTION # ============================================================ def extract_f0_pm(audio: np.ndarray, sr: int = 16000, f0_up_key: int = 0) -> Tuple[np.ndarray, np.ndarray]: """Extract F0 using parselmouth (pm method)""" import parselmouth p_len = audio.shape[0] // 160 + 1 f0_min = 65 f0_max = 1100 l_pad = int(np.ceil(1.5 / f0_min * 16000)) r_pad = l_pad + 1 s = parselmouth.Sound(np.pad(audio, (l_pad, r_pad)), 16000).to_pitch_ac( time_step=0.01, voicing_threshold=0.6, pitch_floor=f0_min, pitch_ceiling=f0_max, ) f0 = s.selected_array["frequency"] if len(f0) < p_len: f0 = np.pad(f0, (0, p_len - len(f0))) f0 = f0[:p_len] f0 *= pow(2, f0_up_key / 12) return f0_to_coarse(f0) def extract_f0_harvest(audio: np.ndarray, sr: int = 16000, f0_up_key: int = 0) -> Tuple[np.ndarray, np.ndarray]: """Extract F0 using pyworld harvest""" import pyworld from scipy import signal as scipy_signal f0, t = pyworld.harvest(audio.astype(np.double), fs=16000, f0_ceil=1100, f0_floor=50, frame_period=10) f0 = scipy_signal.medfilt(f0, 3) f0 *= pow(2, f0_up_key / 12) return f0_to_coarse(f0) def f0_to_coarse(f0: np.ndarray) -> Tuple[np.ndarray, np.ndarray]: """Convert f0 to coarse representation""" f0_min = 50 f0_max = 1100 f0_mel_min = 1127 * np.log(1 + f0_min / 700) f0_mel_max = 1127 * np.log(1 + f0_max / 700) f0bak = f0.copy() f0_mel = 1127 * np.log(1 + f0 / 700) f0_mel[f0_mel > 0] = (f0_mel[f0_mel > 0] - f0_mel_min) * 254 / (f0_mel_max - f0_mel_min) + 1 f0_mel[f0_mel <= 1] = 1 f0_mel[f0_mel > 255] = 255 f0_coarse = np.rint(f0_mel).astype(np.int32) return f0_coarse, f0bak # ============================================================ # RMVPE F0 EXTRACTION (from Applio - IAHispano/Applio) # ============================================================ class RMVPE_ConvBlockRes(nn.Module): def __init__(self, in_channels, out_channels, momentum=0.01): super().__init__() self.conv = nn.Sequential( nn.Conv2d(in_channels, out_channels, (3, 3), (1, 1), (1, 1), bias=False), nn.BatchNorm2d(out_channels, momentum=momentum), nn.ReLU(), nn.Conv2d(out_channels, out_channels, (3, 3), (1, 1), (1, 1), bias=False), nn.BatchNorm2d(out_channels, momentum=momentum), nn.ReLU(), ) self.shortcut = nn.Conv2d(in_channels, out_channels, (1, 1)) if in_channels != out_channels else None def forward(self, x): r = self.conv(x) return r + self.shortcut(x) if self.shortcut else r + x class RMVPE_ResEncoderBlock(nn.Module): def __init__(self, in_channels, out_channels, kernel_size, n_blocks=1, momentum=0.01): super().__init__() self.conv = nn.ModuleList([RMVPE_ConvBlockRes(in_channels, out_channels, momentum)]) for _ in range(n_blocks - 1): self.conv.append(RMVPE_ConvBlockRes(out_channels, out_channels, momentum)) self.kernel_size = kernel_size if kernel_size is not None: self.pool = nn.AvgPool2d(kernel_size=kernel_size) def forward(self, x): for c in self.conv: x = c(x) return (x, self.pool(x)) if self.kernel_size is not None else x class RMVPE_Encoder(nn.Module): def __init__(self, in_channels, in_size, n_encoders, kernel_size, n_blocks, out_channels=16, momentum=0.01): super().__init__() self.n_encoders = n_encoders self.bn = nn.BatchNorm2d(in_channels, momentum=momentum) self.layers = nn.ModuleList() for _ in range(n_encoders): self.layers.append(RMVPE_ResEncoderBlock(in_channels, out_channels, kernel_size, n_blocks, momentum)) in_channels = out_channels out_channels *= 2 in_size //= 2 self.out_size = in_size self.out_channel = out_channels def forward(self, x): concat_tensors = [] x = self.bn(x) for layer in self.layers: t, x = layer(x) concat_tensors.append(t) return x, concat_tensors class RMVPE_Intermediate(nn.Module): def __init__(self, in_channels, out_channels, n_inters, n_blocks, momentum=0.01): super().__init__() self.layers = nn.ModuleList([RMVPE_ResEncoderBlock(in_channels, out_channels, None, n_blocks, momentum)]) for _ in range(n_inters - 1): self.layers.append(RMVPE_ResEncoderBlock(out_channels, out_channels, None, n_blocks, momentum)) def forward(self, x): for layer in self.layers: x = layer(x) return x class RMVPE_ResDecoderBlock(nn.Module): def __init__(self, in_channels, out_channels, stride, n_blocks=1, momentum=0.01): super().__init__() out_padding = (0, 1) if stride == (1, 2) else (1, 1) self.conv1 = nn.Sequential( nn.ConvTranspose2d(in_channels, out_channels, (3, 3), stride, (1, 1), out_padding, bias=False), nn.BatchNorm2d(out_channels, momentum=momentum), nn.ReLU(), ) self.conv2 = nn.ModuleList([RMVPE_ConvBlockRes(out_channels * 2, out_channels, momentum)]) for _ in range(n_blocks - 1): self.conv2.append(RMVPE_ConvBlockRes(out_channels, out_channels, momentum)) def forward(self, x, concat_tensor): x = self.conv1(x) x = torch.cat((x, concat_tensor), dim=1) for c in self.conv2: x = c(x) return x class RMVPE_Decoder(nn.Module): def __init__(self, in_channels, n_decoders, stride, n_blocks, momentum=0.01): super().__init__() self.layers = nn.ModuleList() for _ in range(n_decoders): out_channels = in_channels // 2 self.layers.append(RMVPE_ResDecoderBlock(in_channels, out_channels, stride, n_blocks, momentum)) in_channels = out_channels self.n_decoders = n_decoders def forward(self, x, concat_tensors): for i in range(self.n_decoders): x = self.layers[i](x, concat_tensors[-1 - i]) return x class RMVPE_DeepUnet(nn.Module): def __init__(self, kernel_size, n_blocks, en_de_layers=5, inter_layers=4, in_channels=1, en_out_channels=16): super().__init__() self.encoder = RMVPE_Encoder(in_channels, 128, en_de_layers, kernel_size, n_blocks, en_out_channels) self.intermediate = RMVPE_Intermediate(self.encoder.out_channel // 2, self.encoder.out_channel, inter_layers, n_blocks) self.decoder = RMVPE_Decoder(self.encoder.out_channel, en_de_layers, kernel_size, n_blocks) def forward(self, x): x, concat_tensors = self.encoder(x) x = self.intermediate(x) x = self.decoder(x, concat_tensors) return x class RMVPE_BiGRU(nn.Module): def __init__(self, input_features, hidden_features, num_layers): super().__init__() self.gru = nn.GRU(input_features, hidden_features, num_layers=num_layers, batch_first=True, bidirectional=True) def forward(self, x): return self.gru(x)[0] RMVPE_N_MELS = 128 RMVPE_N_CLASS = 360 class RMVPE_E2E(nn.Module): def __init__(self, n_blocks, n_gru, kernel_size, en_de_layers=5, inter_layers=4, in_channels=1, en_out_channels=16): super().__init__() self.unet = RMVPE_DeepUnet(kernel_size, n_blocks, en_de_layers, inter_layers, in_channels, en_out_channels) self.cnn = nn.Conv2d(en_out_channels, 3, (3, 3), padding=(1, 1)) if n_gru: self.fc = nn.Sequential( RMVPE_BiGRU(3 * 128, 256, n_gru), nn.Linear(512, RMVPE_N_CLASS), nn.Dropout(0.25), nn.Sigmoid(), ) else: self.fc = nn.Sequential(nn.Linear(3 * RMVPE_N_MELS, RMVPE_N_CLASS), nn.Dropout(0.25), nn.Sigmoid()) def forward(self, mel): mel = mel.transpose(-1, -2).unsqueeze(1) x = self.cnn(self.unet(mel)).transpose(1, 2).flatten(-2) return self.fc(x) class RMVPE_MelSpectrogram(nn.Module): def __init__(self, n_mel_channels=128, sample_rate=16000, win_length=1024, hop_length=160, n_fft=None, mel_fmin=30, mel_fmax=8000, clamp=1e-5): super().__init__() from librosa.filters import mel as librosa_mel n_fft = win_length if n_fft is None else n_fft self.hann_window = {} mel_basis = librosa_mel(sr=sample_rate, n_fft=n_fft, n_mels=n_mel_channels, fmin=mel_fmin, fmax=mel_fmax, htk=True) self.register_buffer("mel_basis", torch.from_numpy(mel_basis).float()) self.n_fft = n_fft self.hop_length = hop_length self.win_length = win_length self.clamp = clamp def forward(self, audio, keyshift=0, speed=1, center=True): factor = 2 ** (keyshift / 12) n_fft_new = int(np.round(self.n_fft * factor)) win_length_new = int(np.round(self.win_length * factor)) hop_length_new = int(np.round(self.hop_length * speed)) key = f"{keyshift}_{audio.device}" if key not in self.hann_window: self.hann_window[key] = torch.hann_window(win_length_new).to(audio.device) fft = torch.stft(audio, n_fft=n_fft_new, hop_length=hop_length_new, win_length=win_length_new, window=self.hann_window[key], center=center, return_complex=True) magnitude = torch.sqrt(fft.real.pow(2) + fft.imag.pow(2)) if keyshift != 0: size = self.n_fft // 2 + 1 resize = magnitude.size(1) if resize < size: magnitude = F.pad(magnitude, (0, 0, 0, size - resize)) magnitude = magnitude[:, :size, :] * self.win_length / win_length_new mel_output = torch.matmul(self.mel_basis, magnitude) return torch.log(torch.clamp(mel_output, min=self.clamp)) _rmvpe_model = None def load_rmvpe(): """Download and load RMVPE model for f0 extraction""" global _rmvpe_model if _rmvpe_model is None: logger.info("Downloading RMVPE model...") rmvpe_path = hf_hub_download(repo_id="IAHispano/Applio", filename="Resources/predictors/rmvpe.pt") model = RMVPE_E2E(4, 1, (2, 2)) ckpt = torch.load(rmvpe_path, map_location="cpu", weights_only=True) model.load_state_dict(ckpt) model.eval().to(device) mel_extractor = RMVPE_MelSpectrogram().to(device) cents_mapping = 20 * np.arange(RMVPE_N_CLASS) + 1997.3794084376191 _rmvpe_model = (model, mel_extractor, np.pad(cents_mapping, (4, 4))) logger.info("RMVPE model loaded") return _rmvpe_model def extract_f0_rmvpe(audio: np.ndarray, sr: int = 16000, f0_up_key: int = 0, thred: float = 0.03) -> Tuple[np.ndarray, np.ndarray]: """Extract F0 using RMVPE (best quality, neural network based)""" model, mel_extractor, cents_mapping = load_rmvpe() audio_t = torch.from_numpy(audio).float().to(device).unsqueeze(0) mel = mel_extractor(audio_t, center=True) del audio_t # mel2hidden with chunking with torch.no_grad(): n_frames = mel.shape[-1] mel_padded = F.pad(mel, (0, 32 * ((n_frames - 1) // 32 + 1) - n_frames), mode="reflect") chunks = [] for start in range(0, mel_padded.shape[-1], 32000): end = min(start + 32000, mel_padded.shape[-1]) chunks.append(model(mel_padded[..., start:end])) hidden = torch.cat(chunks, dim=1)[:, :n_frames].squeeze(0).cpu().numpy() # Decode hidden to f0 center = np.argmax(hidden, axis=1) salience = np.pad(hidden, ((0, 0), (4, 4))) center += 4 todo_salience = [] todo_cents = [] for idx in range(salience.shape[0]): s, e = center[idx] - 4, center[idx] + 5 todo_salience.append(salience[idx, s:e]) todo_cents.append(cents_mapping[s:e]) todo_salience = np.array(todo_salience) todo_cents = np.array(todo_cents) cents_pred = np.sum(todo_salience * todo_cents, 1) / np.sum(todo_salience, 1) cents_pred[np.max(salience, axis=1) <= thred] = 0 f0 = 10 * (2 ** (cents_pred / 1200)) f0[f0 == 10] = 0 f0 *= pow(2, f0_up_key / 12) return f0_to_coarse(f0) # ============================================================ # MODEL LOADING # ============================================================ _model_cache = {} def load_rvc_model(model_path: str): """Load RVC model and auto-detect version""" if model_path in _model_cache: return _model_cache[model_path] logger.info(f"Loading RVC model: {model_path}") try: cpt = torch.load(model_path, map_location="cpu", weights_only=True) except Exception: logger.warning("Model requires unsafe loading - may be an older format") cpt = torch.load(model_path, map_location="cpu", weights_only=False) weight_key = None for key in ["weight", "model", "state_dict", "net_g"]: if key in cpt: weight_key = key break if weight_key is None: raise ValueError(f"Cannot find model weights. Keys: {list(cpt.keys())}") config = cpt.get("config", None) if config is None: config = [1025, 32, 192, 192, 768, 2, 6, 3, 0, "1", [3, 7, 11], [[1, 3, 5], [1, 3, 5], [1, 3, 5]], [10, 10, 2, 2], 512, [16, 16, 4, 4], 109, 256, 40000] logger.warning("No config found, using v2 defaults") version = cpt.get("version", "v1") if_f0 = cpt.get("f0", 1) if weight_key in cpt: emb_weight = cpt[weight_key].get("emb_g.weight") if emb_weight is not None: config[-3] = emb_weight.shape[0] sr = config[-1] if isinstance(config[-1], int) else 40000 if version == "v1": model_class = SynthesizerTrnMs256NSFsid if if_f0 == 1 else SynthesizerTrnMs256NSFsid_nono else: model_class = SynthesizerTrnMs768NSFsid if if_f0 == 1 else SynthesizerTrnMs768NSFsid_nono model = model_class( spec_channels=config[0], segment_size=config[1], inter_channels=config[2], hidden_channels=config[3], filter_channels=config[4], n_heads=config[5], n_layers=config[6], kernel_size=config[7], p_dropout=config[8], resblock=config[9], resblock_kernel_sizes=config[10], resblock_dilation_sizes=config[11], upsample_rates=config[12], upsample_initial_channel=config[13], upsample_kernel_sizes=config[14], spk_embed_dim=config[15], gin_channels=config[16], sr=sr, is_half=False ) model.load_state_dict(cpt[weight_key], strict=False) model.eval().to(device) _model_cache[model_path] = (model, sr, version, if_f0) logger.info(f"Model loaded: version={version}, f0={if_f0}, sr={sr}") return model, sr, version, if_f0 # ============================================================ # TRAINING - Simplified for CPU testing # ============================================================ def spectrogram_torch(y, n_fft, hop_size, win_size, center=False): """Compute spectrogram""" hann_window = torch.hann_window(win_size).to(y.device) y = F.pad(y.unsqueeze(1), (int((n_fft-hop_size)/2), int((n_fft-hop_size)/2)), mode='reflect').squeeze(1) spec = torch.stft(y, n_fft, hop_length=hop_size, win_length=win_size, window=hann_window, center=center, pad_mode='reflect', normalized=False, onesided=True, return_complex=True) spec = torch.sqrt(spec.real.pow(2) + spec.imag.pow(2) + 1e-6) return spec # Mel spectrogram for training loss _mel_basis_cache = {} def spec_to_mel_torch(spec, n_fft=2048, num_mels=125, sampling_rate=40000, fmin=0, fmax=None): """Convert spectrogram to mel spectrogram""" from librosa.filters import mel as librosa_mel_fn global _mel_basis_cache if fmax is None: fmax = sampling_rate // 2 key = f"{n_fft}_{num_mels}_{sampling_rate}_{fmin}_{fmax}_{spec.dtype}_{spec.device}" if key not in _mel_basis_cache: mel = librosa_mel_fn(sr=sampling_rate, n_fft=n_fft, n_mels=num_mels, fmin=fmin, fmax=fmax) _mel_basis_cache[key] = torch.from_numpy(mel).to(dtype=spec.dtype, device=spec.device) melspec = torch.matmul(_mel_basis_cache[key], spec) melspec = torch.log(torch.clamp(melspec, min=1e-5)) # Log-amplitude return melspec def preprocess_audio_for_training(audio_path: str, output_dir: str, target_sr: int = 40000, f0_method: str = "rmvpe"): """Preprocess audio file for training - slice and extract features""" import scipy.signal as signal os.makedirs(output_dir, exist_ok=True) os.makedirs(f"{output_dir}/wavs", exist_ok=True) os.makedirs(f"{output_dir}/hubert", exist_ok=True) os.makedirs(f"{output_dir}/f0", exist_ok=True) logger.info(f"Preprocessing: {audio_path}") # Load and resample audio audio, sr = librosa.load(audio_path, sr=target_sr, mono=True) # High-pass filter bh, ah = signal.butter(N=5, Wn=48, btype="high", fs=target_sr) audio = signal.lfilter(bh, ah, audio) # Slice into chunks (3.7 seconds with 0.3 overlap) chunk_size = int(3.7 * target_sr) hop = int(3.4 * target_sr) chunks = [] for i, start in enumerate(range(0, len(audio) - chunk_size, hop)): chunk = audio[start:start + chunk_size] # Normalize max_val = np.abs(chunk).max() if max_val > 0.01: # Skip silence chunk = chunk / max_val * 0.9 chunks.append((i, chunk)) if not chunks: logger.warning("No valid audio chunks found") return None logger.info(f"Created {len(chunks)} chunks") # Save chunks and extract features manifest = [] for idx, chunk in chunks: # Save wav wav_path = f"{output_dir}/wavs/{idx:04d}.wav" sf.write(wav_path, chunk, target_sr) # Resample to 16k for HuBERT chunk_16k = librosa.resample(chunk, orig_sr=target_sr, target_sr=16000) # Extract HuBERT features feats = extract_hubert_features(chunk_16k, sr=16000, version="v2") hubert_path = f"{output_dir}/hubert/{idx:04d}.npy" np.save(hubert_path, feats.squeeze(0).cpu().numpy()) # Extract F0 if f0_method == "rmvpe": f0_coarse, f0 = extract_f0_rmvpe(chunk_16k, 16000, 0) elif f0_method == "harvest": f0_coarse, f0 = extract_f0_harvest(chunk_16k, 16000, 0) else: f0_coarse, f0 = extract_f0_pm(chunk_16k, 16000, 0) f0_path = f"{output_dir}/f0/{idx:04d}.npy" np.save(f0_path, np.stack([f0_coarse, f0], axis=0)) manifest.append(f"{idx:04d}") # Save manifest with open(f"{output_dir}/manifest.txt", "w") as f: f.write("\n".join(manifest)) logger.info(f"Preprocessing complete: {len(manifest)} samples") return output_dir def train_rvc_generator( data_dir: str, output_dir: str, epochs: int = 10, batch_size: int = 2, lr: float = 1e-5, # Lower LR prevents overfitting on small data target_sr: int = 40000, progress_callback=None ): """Generator version of train_rvc - yields (epoch_msg, ckpt_path) tuples""" logger.info(f"Starting training: {data_dir} -> {output_dir}") os.makedirs(output_dir, exist_ok=True) # Load manifest with open(f"{data_dir}/manifest.txt") as f: samples = [l.strip() for l in f if l.strip()] if len(samples) < 1: logger.error("No training samples found") return None logger.info(f"Training with {len(samples)} samples") # Model config (v2 40k defaults) config = [1025, 32, 192, 192, 768, 2, 6, 3, 0, "1", [3, 7, 11], [[1, 3, 5], [1, 3, 5], [1, 3, 5]], [10, 10, 2, 2], 512, [16, 16, 4, 4], 1, 256, target_sr] # Create models (v2 only - 768-dim HuBERT features) net_g = SynthesizerTrnMs768NSFsid( spec_channels=config[0], segment_size=config[1], inter_channels=config[2], hidden_channels=config[3], filter_channels=config[4], n_heads=config[5], n_layers=config[6], kernel_size=config[7], p_dropout=config[8], resblock=config[9], resblock_kernel_sizes=config[10], resblock_dilation_sizes=config[11], upsample_rates=config[12], upsample_initial_channel=config[13], upsample_kernel_sizes=config[14], spk_embed_dim=config[15], gin_channels=config[16], sr=target_sr ).to(train_device) net_d = MultiPeriodDiscriminator().to(train_device) logger.info(f"Training on device: {train_device}") # Download and load pretrained weights (essential for good results) sr_key = f"{target_sr // 1000}k" # e.g., "40k" try: pretrain_g_path = download_pretrained_rvc(f"f0G{sr_key}") pretrain_d_path = download_pretrained_rvc(f"f0D{sr_key}") load_pretrained_weights(net_g, pretrain_g_path) load_pretrained_weights(net_d, pretrain_d_path) except Exception as e: logger.warning(f"Failed to load pretrained weights: {e}") logger.warning("Training from scratch (results may be poor)") # Optimizers (after loading pretrained weights) optim_g = torch.optim.AdamW(net_g.parameters(), lr=lr, betas=(0.8, 0.99)) optim_d = torch.optim.AdamW(net_d.parameters(), lr=lr, betas=(0.8, 0.99)) # LR scheduler (matches Applio - exponential decay) lr_decay = 0.999875 scheduler_g = torch.optim.lr_scheduler.ExponentialLR(optim_g, gamma=lr_decay) scheduler_d = torch.optim.lr_scheduler.ExponentialLR(optim_d, gamma=lr_decay) net_g.train() net_d.train() ckpt_path = f"{output_dir}/model.pth" def _save_checkpoint(): torch.save({ "weight": net_g.state_dict(), "config": config, "version": "v2", "f0": 1, }, ckpt_path) logger.info(f"Saved checkpoint: {ckpt_path}") # Training loop for epoch in range(epochs): total_loss_g, total_loss_d = 0, 0 np.random.shuffle(samples) for i in range(0, len(samples), batch_size): batch_samples = samples[i:i+batch_size] # Load batch data wavs, huberts, f0s = [], [], [] for s in batch_samples: wav, _ = librosa.load(f"{data_dir}/wavs/{s}.wav", sr=target_sr, mono=True) hubert = np.load(f"{data_dir}/hubert/{s}.npy") # Upsample 50Hz -> 100Hz using interpolation (same as inference) hubert_t = torch.from_numpy(hubert).unsqueeze(0).permute(0, 2, 1) # (1, 768, seq) hubert_t = F.interpolate(hubert_t, scale_factor=2, mode='linear', align_corners=False) hubert = hubert_t.permute(0, 2, 1).squeeze(0).numpy() # (seq*2, 768) f0_data = np.load(f"{data_dir}/f0/{s}.npy") wavs.append(wav) huberts.append(hubert) f0s.append(f0_data) # Compute spectrogram first to get target length max_wav_len = max(len(w) for w in wavs) wav_batch = np.zeros((len(wavs), max_wav_len)) for j, w in enumerate(wavs): wav_batch[j, :len(w)] = w wav_t = torch.FloatTensor(wav_batch).unsqueeze(1).to(train_device) spec = spectrogram_torch(wav_t.squeeze(1), 2048, 400, 2048) spec_len = spec.shape[2] # Target length for all features # Pad/truncate features to match spec length exactly hubert_batch = np.zeros((len(huberts), spec_len, huberts[0].shape[1])) f0_batch = np.zeros((len(f0s), spec_len)) f0f_batch = np.zeros((len(f0s), spec_len)) for j, (h, f) in enumerate(zip(huberts, f0s)): # Truncate or pad HuBERT to spec_len h_len = min(h.shape[0], spec_len) hubert_batch[j, :h_len] = h[:h_len] # Truncate or pad F0 to spec_len f0_len = min(f.shape[1], spec_len) f0_batch[j, :f0_len] = f[0, :f0_len] f0f_batch[j, :f0_len] = f[1, :f0_len] # To tensors - all features now have spec_len hubert_t = torch.FloatTensor(hubert_batch).to(train_device) f0_t = torch.LongTensor(f0_batch.astype(np.int64)).to(train_device) f0f_t = torch.FloatTensor(f0f_batch).to(train_device) lengths_t = torch.LongTensor([spec_len] * len(batch_samples)).to(train_device) sid_t = torch.LongTensor([0] * len(batch_samples)).to(train_device) spec_lengths = torch.LongTensor([spec_len] * len(batch_samples)).to(train_device) # Forward pass generator # Args: phone, phone_lengths, pitch, pitchf, y (spec), y_lengths, ds try: y_hat, ids_slice, x_mask, z_mask, (z, z_p, m_p, logs_p, m_q, logs_q) = net_g( hubert_t, lengths_t, f0_t, f0f_t, spec, spec_lengths, sid_t ) except Exception as e: logger.warning(f"Generator forward failed: {e}") continue # Slice wav at same position model generated (CRITICAL for proper loss) # ids_slice is in latent space, multiply by hop_length to get waveform position hop_length = 400 segment_size_wav = 32 * hop_length # segment_size in latent * hop_length y = slice_segments(wav_t, ids_slice * hop_length, segment_size_wav) # Discriminator forward y_d_rs, y_d_gs, fmap_rs, fmap_gs = net_d(y, y_hat.detach()) # Discriminator loss loss_d = discriminator_loss(y_d_rs, y_d_gs) optim_d.zero_grad() loss_d.backward() optim_d.step() # Generator loss y_d_rs, y_d_gs, fmap_rs, fmap_gs = net_d(y, y_hat) loss_gen = generator_loss(y_d_gs) loss_fm = feature_loss(fmap_rs, fmap_gs) loss_kl = kl_loss(z_p, logs_q, m_p, logs_p, z_mask) # Mel spectrogram loss (crucial for quality) # Config: n_fft=2048, hop=400, win=2048, n_mels=125, fmin=0, fmax=None y_mel = spec_to_mel_torch(spectrogram_torch(y.squeeze(1), 2048, 400, 2048), n_fft=2048, num_mels=125, sampling_rate=target_sr, fmin=0, fmax=None) y_hat_mel = spec_to_mel_torch(spectrogram_torch(y_hat.squeeze(1), 2048, 400, 2048), n_fft=2048, num_mels=125, sampling_rate=target_sr, fmin=0, fmax=None) # Align lengths if needed min_len = min(y_mel.shape[2], y_hat_mel.shape[2]) loss_mel = F.l1_loss(y_mel[:, :, :min_len], y_hat_mel[:, :, :min_len]) * 45 # c_mel = 45 loss_g = loss_gen + loss_fm + loss_mel + loss_kl optim_g.zero_grad() loss_g.backward() optim_g.step() total_loss_g += loss_g.item() total_loss_d += loss_d.item() avg_loss_g = total_loss_g / max(1, len(samples) // batch_size) avg_loss_d = total_loss_d / max(1, len(samples) // batch_size) epoch_msg = f"Epoch {epoch+1}/{epochs} - G: {avg_loss_g:.2f}, D: {avg_loss_d:.2f}" logger.info(epoch_msg) # Update progress callback if provided if progress_callback: progress_pct = 0.30 + (0.65 * (epoch + 1) / epochs) progress_callback(progress_pct, epoch_msg) # Check cancel flag global _training_cancel if _training_cancel: _training_cancel = False _save_checkpoint() logger.info(f"Cancelled at epoch {epoch+1}/{epochs}, checkpoint saved") del net_g, net_d, optim_g, optim_d, scheduler_g, scheduler_d gc.collect() if torch.cuda.is_available(): torch.cuda.empty_cache() yield f"🛑 Cancelled at epoch {epoch+1}. Checkpoint saved.", ckpt_path, None return # Periodic mid-training checkpoint save, so the in-progress model can # be downloaded/tested (via tracking code) before training finishes — # e.g. grabbing the epoch-600 checkpoint of a 1000-epoch run. checkpoint_interval = max(1, epochs // 10) is_checkpoint_epoch = ((epoch + 1) % checkpoint_interval == 0) or (epoch == epochs - 1) if is_checkpoint_epoch: _save_checkpoint() yield f"{epoch_msg} | 💾 checkpoint saved ({epoch+1}/{epochs} epochs)", ckpt_path, None else: # Yield epoch message for live UI updates yield epoch_msg, None, None # Step LR schedulers scheduler_g.step() scheduler_d.step() # Normal completion — save final checkpoint + index _save_checkpoint() index_path = None try: import faiss hubert_dir = f"{data_dir}/hubert" npys = [] for name in sorted(os.listdir(hubert_dir)): if name.endswith('.npy'): phone = np.load(os.path.join(hubert_dir, name)) npys.append(phone) if npys: big_npy = np.concatenate(npys, axis=0) n_ivf = min(int(16 * np.sqrt(big_npy.shape[0])), big_npy.shape[0] // 39) n_ivf = max(1, n_ivf) # Ensure at least 1 index = faiss.index_factory(big_npy.shape[1], f"IVF{n_ivf},Flat") index.train(big_npy) index.add(big_npy) index_path = f"{output_dir}/model.index" faiss.write_index(index, index_path) logger.info(f"Saved index: {index_path}") except Exception as e: logger.warning(f"Failed to generate index: {e}") del net_g, net_d, optim_g, optim_d, scheduler_g, scheduler_d gc.collect() if torch.cuda.is_available(): torch.cuda.empty_cache() yield "Training complete!", ckpt_path, index_path def train_rvc( data_dir: str, output_dir: str, epochs: int = 10, batch_size: int = 2, lr: float = 1e-5, # Lower LR prevents overfitting on small data target_sr: int = 40000, progress_callback=None ): """Non-generator wrapper for CLI use - returns (checkpoint_path, index_path)""" ckpt = None idx = None for msg, path, index in train_rvc_generator(data_dir, output_dir, epochs, batch_size, lr, target_sr, progress_callback): if path: ckpt = path if index: idx = index return ckpt, idx # ============================================================ # INFERENCE # ============================================================ def convert_voice( source_audio: str, model_file, index_file=None, pitch_shift: int = 0, f0_method: str = "pm", index_rate: float = 0.5, protect: float = 0.33, volume_envelope: float = 1.0, progress=gr.Progress() ) -> Tuple[str, str]: """Convert voice using RVC model (Applio-compatible pipeline).""" try: if source_audio is None: return None, "Please upload source audio" if model_file is None: return None, "Please upload RVC model (.pth)" model_path = model_file.name if hasattr(model_file, 'name') else model_file progress(0.1, "Loading model...") model, tgt_sr, version, if_f0 = load_rvc_model(model_path) progress(0.2, "Loading audio...") audio, sr = librosa.load(source_audio, sr=16000, mono=True) # Apply 48Hz high-pass filter (critical - removes low-frequency artifacts) audio = signal.filtfilt(bh, ah, audio) # Normalize audio audio_max = np.abs(audio).max() / 0.95 if audio_max > 1: audio /= audio_max # Pipeline constants (same as Applio) window = 160 # Critical for feature/pitch alignment x_pad = 1 # Padding in seconds t_pad = 16000 * x_pad # Padding in samples # Pad audio audio_pad = np.pad(audio, (t_pad, t_pad), mode="reflect") p_len = audio_pad.shape[0] // window progress(0.3, "Extracting features...") feats = extract_hubert_features(audio_pad, sr=16000, version=version) # Save original features for protect mechanism feats0 = feats.clone() if if_f0 == 1 and protect < 0.5 else None # Index retrieval (speaker similarity) if index_file is not None and index_rate > 0: try: import faiss index_path = index_file.name if hasattr(index_file, 'name') else index_file progress(0.4, "Loading index...") index = faiss.read_index(index_path) big_npy = index.reconstruct_n(0, index.ntotal) npy = feats[0].cpu().numpy().astype("float32") score, ix = index.search(npy, k=8) weight = np.square(1 / score) weight /= weight.sum(axis=1, keepdims=True) npy = np.sum(big_npy[ix] * np.expand_dims(weight, axis=2), axis=1) feats = torch.from_numpy(npy).unsqueeze(0).to(device) * index_rate + (1 - index_rate) * feats except Exception as e: logger.warning(f"Index retrieval failed: {e}") # Feature upsampling by 2x feats = F.interpolate(feats.permute(0, 2, 1), scale_factor=2).permute(0, 2, 1) # Adjust length based on audio p_len = min(audio_pad.shape[0] // window, feats.shape[1]) pitch, pitchf = None, None if if_f0 == 1: progress(0.5, f"Extracting F0 ({f0_method})...") if f0_method == "rmvpe": pitch, pitchf = extract_f0_rmvpe(audio_pad, 16000, pitch_shift) elif f0_method == "harvest": pitch, pitchf = extract_f0_harvest(audio_pad, 16000, pitch_shift) else: pitch, pitchf = extract_f0_pm(audio_pad, 16000, pitch_shift) pitch = pitch[:p_len] pitchf = pitchf[:p_len] # Upsample feats0 for protect if feats0 is not None: feats0 = F.interpolate(feats0.permute(0, 2, 1), scale_factor=2).permute(0, 2, 1) # Apply protect mechanism (preserve original features for unvoiced segments) if protect < 0.5 and feats0 is not None: pitchf_tensor = torch.from_numpy(pitchf).float().to(device) pitchff = pitchf_tensor.clone() pitchff[pitchf_tensor > 0] = 1 pitchff[pitchf_tensor < 1] = protect pitchff = pitchff.unsqueeze(0).unsqueeze(-1) feats = feats[:, :p_len, :] * pitchff + feats0[:, :p_len, :] * (1 - pitchff) if len(pitch) < p_len: pitch = np.pad(pitch, (0, p_len - len(pitch))) pitchf = np.pad(pitchf, (0, p_len - len(pitchf))) pitch = torch.LongTensor(pitch).unsqueeze(0).to(device) pitchf = torch.FloatTensor(pitchf).unsqueeze(0).to(device) p_len_tensor = torch.LongTensor([p_len]).to(device) sid = torch.LongTensor([0]).to(device) progress(0.7, "Running inference...") with torch.no_grad(): if if_f0 == 1: audio_out = model.infer(feats[:, :p_len, :], p_len_tensor, pitch, pitchf, sid)[0][0, 0].data.cpu().float().numpy() else: audio_out = model.infer(feats[:, :p_len, :], p_len_tensor, sid)[0][0, 0].data.cpu().float().numpy() # Remove padding from output t_pad_tgt = int(t_pad * tgt_sr / 16000) if len(audio_out) > 2 * t_pad_tgt: audio_out = audio_out[t_pad_tgt:-t_pad_tgt] # RMS mixing - match volume dynamics of source audio if volume_envelope != 1.0: try: source_at_tgt_sr = librosa.resample(audio, orig_sr=16000, target_sr=tgt_sr) frame_len = tgt_sr // 2 * 2 hop_len = tgt_sr // 2 rms_source = librosa.feature.rms(y=source_at_tgt_sr, frame_length=frame_len, hop_length=hop_len) rms_output = librosa.feature.rms(y=audio_out, frame_length=frame_len, hop_length=hop_len) rms_source = F.interpolate( torch.from_numpy(rms_source).float().unsqueeze(0), size=audio_out.shape[0], mode="linear" ).squeeze() rms_output = F.interpolate( torch.from_numpy(rms_output).float().unsqueeze(0), size=audio_out.shape[0], mode="linear" ).squeeze() rms_output = torch.maximum(rms_output, torch.zeros_like(rms_output) + 1e-6) # Applio formula: target * (source^(1-rate) * output^(rate-1)) audio_out = audio_out * (torch.pow(rms_source, 1 - volume_envelope) * torch.pow(rms_output, volume_envelope - 1)).numpy() except Exception as e: logger.warning(f"RMS mixing failed: {e}") # Final normalization audio_max = np.abs(audio_out).max() / 0.99 if audio_max > 1: audio_out /= audio_max progress(0.9, "Saving output...") fd, output_path = tempfile.mkstemp(suffix=".wav") os.close(fd) sf.write(output_path, audio_out, tgt_sr) # Cleanup tensors and model cache to prevent OOM on HF Spaces del model, feats, audio_out, audio, audio_pad if feats0 is not None: del feats0 _model_cache.clear() gc.collect() return output_path, f"Converted: {version}, sr={tgt_sr}, pitch={pitch_shift:+d}" except Exception as e: logger.exception("Conversion failed") _model_cache.clear() gc.collect() return None, f"Error: {str(e)}" # ============================================================ # DEFAULT MODEL DOWNLOAD # ============================================================ def load_example_model(): """Download and load the default example model from HuggingFace""" import shutil try: logger.info(f"Downloading example model from {DEFAULT_MODEL_REPO}...") model_path = hf_hub_download(repo_id=DEFAULT_MODEL_REPO, filename=DEFAULT_MODEL_FILE) index_path = hf_hub_download(repo_id=DEFAULT_MODEL_REPO, filename=DEFAULT_INDEX_FILE) # Gradio 6 requires files to be in allowed directories (cwd or /tmp) # Copy from HF cache to temp directory temp_dir = tempfile.mkdtemp() temp_model = os.path.join(temp_dir, DEFAULT_MODEL_FILE) temp_index = os.path.join(temp_dir, DEFAULT_INDEX_FILE) shutil.copy2(model_path, temp_model) shutil.copy2(index_path, temp_index) return temp_model, temp_index, f"Loaded: {DEFAULT_MODEL_REPO}" except Exception as e: logger.exception("Failed to download example model") return None, None, f"Error: {str(e)}" # ============================================================ # BEATRICE V2 MODEL # ============================================================ def beatrice_load_audio(file, **kwargs): """Load audio using soundfile directly (for Beatrice dataset)""" data, sr = sf.read(file, dtype='float32') # soundfile returns (samples, channels), convert to torch (channels, samples) wav = torch.from_numpy(data) if wav.ndim == 1: wav = wav.unsqueeze(0) # mono -> (1, samples) else: wav = wav.T # (samples, channels) -> (channels, samples) return wav, sr class AttrDict(dict): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.__dict__ = self def dump_params(params: torch.Tensor, f: BinaryIO): if params is None: return if params.dtype == torch.bfloat16: f.write( params.detach() .clone() .float() .view(torch.short) .numpy() .ravel()[1::2] .tobytes() ) else: f.write(params.detach().numpy().ravel().tobytes()) f.flush() def dump_layer(layer: nn.Module, f: BinaryIO): dump = partial(dump_params, f=f) if hasattr(layer, "dump"): layer.dump(f) elif isinstance(layer, (nn.Linear, nn.Conv1d, nn.LayerNorm)): dump(layer.weight) dump(layer.bias) elif isinstance(layer, nn.MultiheadAttention): embed_dim = layer.embed_dim num_heads = layer.num_heads # [3 * embed_dim, embed_dim] in_proj_weight = layer.in_proj_weight.data.clone() in_proj_weight[: 2 * embed_dim] *= 1.0 / math.sqrt( math.sqrt(embed_dim // num_heads) ) in_proj_weight = in_proj_weight.view( 3, num_heads, embed_dim // num_heads, embed_dim ) # [num_heads, 3, embed_dim / num_heads, embed_dim] in_proj_weight = in_proj_weight.transpose(0, 1) # [3 * embed_dim] in_proj_bias = layer.in_proj_bias.data.clone() in_proj_bias[: 2 * embed_dim] *= 1.0 / math.sqrt( math.sqrt(embed_dim // num_heads) ) in_proj_bias = in_proj_bias.view(3, num_heads, embed_dim // num_heads) # [num_heads, 3, embed_dim / num_heads] in_proj_bias = in_proj_bias.transpose(0, 1) dump(in_proj_weight) dump(in_proj_bias) dump(layer.out_proj.weight) dump(layer.out_proj.bias) elif isinstance(layer, nn.Embedding): dump(layer.weight) elif isinstance(layer, nn.Parameter): dump(layer) elif isinstance(layer, nn.ModuleList): for layer_i in layer: dump_layer(layer_i, f) else: assert False, layer class CausalConv1d(nn.Conv1d): def __init__( self, in_channels: int, out_channels: int, kernel_size: int, stride: int = 1, dilation: int = 1, groups: int = 1, bias: bool = True, delay: int = 0, ): padding = (kernel_size - 1) * dilation - delay self.trim = (kernel_size - 1) * dilation - 2 * delay if self.trim < 0: raise ValueError super().__init__( in_channels, out_channels, kernel_size=kernel_size, stride=stride, padding=padding, dilation=dilation, groups=groups, bias=bias, ) def forward(self, input: torch.Tensor) -> torch.Tensor: result = super().forward(input) if self.trim == 0: return result else: return result[:, :, : -self.trim] class WSConv1d(CausalConv1d): def __init__( self, in_channels: int, out_channels: int, kernel_size: int, stride: int = 1, dilation: int = 1, groups: int = 1, bias: bool = True, delay: int = 0, ): super().__init__( in_channels, out_channels, kernel_size=kernel_size, stride=stride, dilation=dilation, groups=groups, bias=bias, delay=delay, ) self.weight.data.normal_( 0.0, math.sqrt(1.0 / (in_channels * kernel_size // groups)) ) if bias: self.bias.data.zero_() self.gain = nn.Parameter(torch.ones((out_channels, 1, 1))) def standardized_weight(self) -> torch.Tensor: var, mean = torch.var_mean(self.weight, [1, 2], keepdim=True) scale = ( self.gain * ( self.in_channels * self.kernel_size[0] // self.groups * var + 1e-8 ).rsqrt() ) return scale * (self.weight - mean) def forward(self, input: torch.Tensor) -> torch.Tensor: result = F.conv1d( input, self.standardized_weight(), self.bias, self.stride, self.padding, self.dilation, self.groups, ) if self.trim == 0: return result else: return result[:, :, : -self.trim] def merge_weights(self): self.weight.data[:] = self.standardized_weight().detach() self.gain.data.fill_(1.0) class WSLinear(nn.Linear): def __init__(self, in_features: int, out_features: int, bias: bool = True): super().__init__(in_features, out_features, bias) self.weight.data.normal_(0.0, math.sqrt(1.0 / in_features)) self.bias.data.zero_() self.gain = nn.Parameter(torch.ones((out_features, 1))) def standardized_weight(self) -> torch.Tensor: var, mean = torch.var_mean(self.weight, 1, keepdim=True) scale = self.gain * (self.in_features * var + 1e-8).rsqrt() return scale * (self.weight - mean) def forward(self, input: torch.Tensor) -> torch.Tensor: return F.linear(input, self.standardized_weight(), self.bias) def merge_weights(self): self.weight.data[:] = self.standardized_weight().detach() self.gain.data.fill_(1.0) class CrossAttention(nn.Module): def __init__( self, qk_channels: int, vo_channels: int, num_heads: int, in_q_channels: int, in_kv_channels: int, out_channels: int, dropout: float = 0.0, ): super().__init__() assert qk_channels % num_heads == 0 self.qk_channels = qk_channels self.vo_channels = vo_channels self.num_heads = num_heads self.in_q_channels = in_q_channels self.in_kv_channels = in_kv_channels self.out_channels = out_channels self.dropout = dropout self.head_qk_channels = qk_channels // num_heads self.head_vo_channels = vo_channels // num_heads self.q_projection = nn.Linear(in_q_channels, qk_channels) self.q_projection.weight.data.normal_(0.0, math.sqrt(1.0 / in_q_channels)) self.q_projection.bias.data.zero_() self.kv_projection = nn.Linear(in_kv_channels, qk_channels + vo_channels) self.kv_projection.weight.data.normal_(0.0, math.sqrt(1.0 / in_kv_channels)) self.kv_projection.bias.data.zero_() self.out_projection = nn.Linear(vo_channels, out_channels) self.out_projection.weight.data.normal_(0.0, math.sqrt(1.0 / vo_channels)) self.out_projection.bias.data.zero_() def forward( self, q: torch.Tensor, kv: torch.Tensor, ) -> torch.Tensor: # q: [batch_size, q_length, in_q_channels] # kv: [batch_size, kv_length, in_kv_channels] batch_size, q_length, _ = q.size() _, kv_length, _ = kv.size() # [batch_size, q_length, qk_channels] q = self.q_projection(q) # [batch_size, kv_length, qk_channels + vo_channels] kv = self.kv_projection(kv) # [batch_size, kv_length, qk_channels], [batch_size, kv_length, vo_channels] k, v = kv.split([self.qk_channels, self.vo_channels], dim=2) q = q.view( batch_size, q_length, self.num_heads, self.head_qk_channels ).transpose(1, 2) k = k.view( batch_size, kv_length, self.num_heads, self.head_qk_channels ).transpose(1, 2) v = v.view( batch_size, kv_length, self.num_heads, self.head_vo_channels ).transpose(1, 2) # [batch_size, num_heads, q_length, head_vo_channels] attn_out = F.scaled_dot_product_attention(q, k, v, dropout_p=self.dropout) # [batch_size, q_length, vo_channels] attn_out = ( attn_out.transpose(1, 2) .contiguous() .view(batch_size, q_length, self.vo_channels) ) # [batch_size, q_length, out_channels] attn_out = self.out_projection(attn_out) return attn_out def dump(self, f: Union[BinaryIO, str, bytes, os.PathLike]): if isinstance(f, (str, bytes, os.PathLike)): with open(f, "wb") as f: self.dump(f) return if not hasattr(f, "write"): raise TypeError q_projection_weight = self.q_projection.weight.data.clone() q_projection_bias = self.q_projection.bias.data.clone() q_projection_weight *= 1.0 / math.sqrt(math.sqrt(self.head_qk_channels)) q_projection_bias *= 1.0 / math.sqrt(math.sqrt(self.head_qk_channels)) dump_params(q_projection_weight, f) dump_params(q_projection_bias, f) dump_layer(self.out_projection, f) def dump_kv(self, f: Union[BinaryIO, str, bytes, os.PathLike]): if isinstance(f, (str, bytes, os.PathLike)): with open(f, "wb") as f: self.dump_kv(f) return if not hasattr(f, "write"): raise TypeError kv_projection_weight = self.kv_projection.weight.data.clone() kv_projection_bias = self.kv_projection.bias.data.clone() k_projection_weight, v_projection_weight = kv_projection_weight.split( [self.qk_channels, self.vo_channels] ) k_projection_bias, v_projection_bias = kv_projection_bias.split( [self.qk_channels, self.vo_channels] ) k_projection_weight *= 1.0 / math.sqrt(math.sqrt(self.head_qk_channels)) k_projection_bias *= 1.0 / math.sqrt(math.sqrt(self.head_qk_channels)) # [qk_channels, in_kv_channels] -> [num_heads, head_qk_channels, in_kv_channels] k_projection_weight = k_projection_weight.view( self.num_heads, self.head_qk_channels, self.in_kv_channels ) # [qk_channels] -> [num_heads, head_qk_channels] k_projection_bias = k_projection_bias.view( self.num_heads, self.head_qk_channels ) # [vo_channels, in_kv_channels] -> [num_heads, head_vo_channels, in_kv_channels] v_projection_weight = v_projection_weight.view( self.num_heads, self.head_vo_channels, self.in_kv_channels ) # [vo_channels] -> [num_heads, head_vo_channels] v_projection_bias = v_projection_bias.view( self.num_heads, self.head_vo_channels ) for i in range(self.num_heads): # [head_qk_channels, in_kv_channels] dump_params(k_projection_weight[i], f) # [head_vo_channels, in_kv_channels] dump_params(v_projection_weight[i], f) for i in range(self.num_heads): # [head_qk_channels] dump_params(k_projection_bias[i], f) # [head_vo_channels] dump_params(v_projection_bias[i], f) class ConvNeXtBlock(nn.Module): def __init__( self, channels: int, intermediate_channels: int, layer_scale_init_value: float, kernel_size: int = 7, use_weight_standardization: bool = False, enable_scaling: bool = False, pre_scale: float = 1.0, post_scale: float = 1.0, use_mha: bool = False, cross_attention: bool = False, num_heads: int = 4, attention_dropout: float = 0.1, attention_channels: Optional[int] = None, kv_channels: Optional[int] = None, ): super().__init__() self.use_weight_standardization = use_weight_standardization self.enable_scaling = enable_scaling self.use_mha = use_mha self.cross_attention = cross_attention if use_mha: self.attn_norm = nn.LayerNorm(channels) if cross_attention: self.mha = CrossAttention( qk_channels=attention_channels, vo_channels=attention_channels, num_heads=num_heads, in_q_channels=channels, in_kv_channels=kv_channels, out_channels=channels, dropout=attention_dropout, ) else: # self-attention assert attention_channels is None assert kv_channels is None self.mha = nn.MultiheadAttention( embed_dim=channels, num_heads=num_heads, dropout=attention_dropout, batch_first=True, ) self.dwconv = CausalConv1d( channels, channels, kernel_size=kernel_size, groups=channels ) self.norm = nn.LayerNorm(channels) self.pwconv1 = nn.Linear(channels, intermediate_channels) self.pwconv2 = nn.Linear(intermediate_channels, channels) self.gamma = nn.Parameter(torch.full((channels,), layer_scale_init_value)) self.dwconv.weight.data.normal_(0.0, math.sqrt(1.0 / kernel_size)) self.dwconv.bias.data.zero_() self.pwconv1.weight.data.normal_(0.0, math.sqrt(2.0 / channels)) self.pwconv1.bias.data.zero_() self.pwconv2.weight.data.normal_(0.0, math.sqrt(1.0 / intermediate_channels)) self.pwconv2.bias.data.zero_() if use_weight_standardization: self.norm = nn.Identity() self.dwconv = WSConv1d(channels, channels, kernel_size, groups=channels) self.pwconv1 = WSLinear(channels, intermediate_channels) self.pwconv2 = WSLinear(intermediate_channels, channels) del self.gamma if enable_scaling: self.register_buffer("pre_scale", torch.tensor(pre_scale)) self.register_buffer("post_scale", torch.tensor(post_scale)) self.post_scale_weight = nn.Parameter(torch.ones(())) def forward( self, x: torch.Tensor, attn_mask: Optional[torch.Tensor] = None, kv: Optional[torch.Tensor] = None, ) -> torch.Tensor: if self.use_mha: batch_size, channels, length = x.size() if self.cross_attention: assert kv is not None else: assert kv is None assert length % 4 == 0 identity = x if self.cross_attention: # kv: [batch_size, kv_length, kv_channels] x = x.transpose(1, 2) x = self.attn_norm(x) x = self.mha(x, kv) x = x.transpose(1, 2) else: x = x.view(batch_size, channels, length // 4, 4) x = x.permute(0, 3, 2, 1) x = x.reshape(batch_size * 4, length // 4, channels) x = self.attn_norm(x) x, _ = self.mha( x, x, x, attn_mask=attn_mask, is_causal=True, need_weights=False ) x = x.view(batch_size, 4, length // 4, channels) x = x.permute(0, 3, 2, 1) x = x.reshape(batch_size, channels, length) x += identity identity = x if self.enable_scaling: x = x * self.pre_scale x = self.dwconv(x) x = x.transpose(1, 2) x = self.norm(x) x = self.pwconv1(x) x = F.gelu(x, approximate="tanh") x = self.pwconv2(x) if not self.use_weight_standardization: x *= self.gamma if self.enable_scaling: x *= self.post_scale * self.post_scale_weight x = x.transpose(1, 2) x += identity return x def merge_weights(self): if self.use_mha: if self.cross_attention: assert isinstance(self.mha, CrossAttention) self.mha.q_projection.bias.data += torch.mv( self.mha.q_projection.weight.data, self.attn_norm.bias.data ) self.mha.q_projection.weight.data *= self.attn_norm.weight.data[None, :] self.attn_norm.bias.data[:] = 0.0 self.attn_norm.weight.data[:] = 1.0 else: # self-attention assert isinstance(self.mha, nn.MultiheadAttention) self.mha.in_proj_bias.data += torch.mv( self.mha.in_proj_weight.data, self.attn_norm.bias.data ) self.mha.in_proj_weight.data *= self.attn_norm.weight.data[None, :] self.attn_norm.bias.data[:] = 0.0 self.attn_norm.weight.data[:] = 1.0 if self.use_weight_standardization: self.dwconv.merge_weights() self.pwconv1.merge_weights() self.pwconv2.merge_weights() else: self.pwconv1.bias.data += torch.mv( self.pwconv1.weight.data, self.norm.bias.data ) self.pwconv1.weight.data *= self.norm.weight.data[None, :] self.norm.bias.data[:] = 0.0 self.norm.weight.data[:] = 1.0 self.pwconv2.weight.data *= self.gamma.data[:, None] self.pwconv2.bias.data *= self.gamma.data self.gamma.data[:] = 1.0 if self.enable_scaling: self.dwconv.weight.data *= self.pre_scale.data self.pre_scale.data.fill_(1.0) self.pwconv2.weight.data *= ( self.post_scale.data * self.post_scale_weight.data ) self.pwconv2.bias.data *= self.post_scale.data * self.post_scale_weight.data self.post_scale.data.fill_(1.0) self.post_scale_weight.data.fill_(1.0) def dump(self, f: Union[BinaryIO, str, bytes, os.PathLike]): if isinstance(f, (str, bytes, os.PathLike)): with open(f, "wb") as f: self.dump(f) return if not hasattr(f, "write"): raise TypeError if self.use_mha: dump_layer(self.mha, f) dump_layer(self.dwconv, f) dump_layer(self.pwconv1, f) dump_layer(self.pwconv2, f) class ConvNeXtStack(nn.Module): def __init__( self, in_channels: int, channels: int, intermediate_channels: int, n_blocks: int, delay: int, embed_kernel_size: int, kernel_size: int, use_weight_standardization: bool = False, enable_scaling: bool = False, use_mha: bool = False, cross_attention: bool = False, kv_channels: Optional[int] = None, ): super().__init__() assert delay * 2 + 1 <= embed_kernel_size assert not (use_weight_standardization and use_mha) # 未対応 self.use_weight_standardization = use_weight_standardization self.use_mha = use_mha self.cross_attention = cross_attention self.embed = CausalConv1d(in_channels, channels, embed_kernel_size, delay=delay) self.norm = nn.LayerNorm(channels) self.convnext = nn.ModuleList() for i in range(n_blocks): pre_scale = 1.0 / math.sqrt(1.0 + i / n_blocks) if enable_scaling else 1.0 post_scale = 1.0 / math.sqrt(n_blocks) if enable_scaling else 1.0 block = ConvNeXtBlock( channels=channels, intermediate_channels=intermediate_channels, layer_scale_init_value=1.0 / n_blocks, kernel_size=kernel_size, use_weight_standardization=use_weight_standardization, enable_scaling=enable_scaling, pre_scale=pre_scale, post_scale=post_scale, use_mha=use_mha, cross_attention=cross_attention, num_heads=4, attention_dropout=0.1, attention_channels=kv_channels, kv_channels=kv_channels, ) self.convnext.append(block) self.final_layer_norm = nn.LayerNorm(channels) self.embed.weight.data.normal_( 0.0, math.sqrt(0.5 / (embed_kernel_size * in_channels)) ) self.embed.bias.data.zero_() if use_weight_standardization: self.embed = WSConv1d(in_channels, channels, embed_kernel_size, delay=delay) self.norm = nn.Identity() self.final_layer_norm = nn.Identity() def forward( self, x: torch.Tensor, kv: Optional[torch.Tensor] = None ) -> torch.Tensor: x = self.embed(x) x = self.norm(x.transpose(1, 2)).transpose(1, 2) if self.use_mha and not self.cross_attention: pad_length = -x.size(2) % 4 if pad_length: x = F.pad(x, (0, pad_length)) t40 = x.size(2) // 4 attn_mask = torch.ones((t40, t40), dtype=torch.bool, device=x.device).triu( 1 ) else: attn_mask = None for conv_block in self.convnext: x = conv_block(x, attn_mask=attn_mask, kv=kv) if self.use_mha and not self.cross_attention and pad_length: x = x[:, :, :-pad_length] x = self.final_layer_norm(x.transpose(1, 2)).transpose(1, 2) return x def merge_weights(self): if self.use_weight_standardization: self.embed.merge_weights() for conv_block in self.convnext: conv_block.merge_weights() def dump(self, f: Union[BinaryIO, str, bytes, os.PathLike]): if isinstance(f, (str, bytes, os.PathLike)): with open(f, "wb") as f: self.dump(f) return if not hasattr(f, "write"): raise TypeError dump_layer(self.embed, f) if not self.use_weight_standardization: dump_layer(self.norm, f) dump_layer(self.convnext, f) if not self.use_weight_standardization: dump_layer(self.final_layer_norm, f) def dump_kv(self, f: Union[BinaryIO, str, bytes, os.PathLike]): if isinstance(f, (str, bytes, os.PathLike)): with open(f, "wb") as f: self.dump_kv(f) return if not hasattr(f, "write"): raise TypeError assert self.use_mha and self.cross_attention for conv_block in self.convnext: if not conv_block.use_mha or not conv_block.cross_attention: continue assert isinstance(conv_block, ConvNeXtBlock) assert hasattr(conv_block, "mha") assert isinstance(conv_block.mha, CrossAttention) conv_block.mha.dump_kv(f) class FeatureExtractor(nn.Module): def __init__(self, hidden_channels: int): super().__init__() # fmt: off self.conv0 = weight_norm(nn.Conv1d(1, hidden_channels // 8, 10, 5, bias=False)) self.conv1 = weight_norm(nn.Conv1d(hidden_channels // 8, hidden_channels // 4, 3, 2, bias=False)) self.conv2 = weight_norm(nn.Conv1d(hidden_channels // 4, hidden_channels // 2, 3, 2, bias=False)) self.conv3 = weight_norm(nn.Conv1d(hidden_channels // 2, hidden_channels, 3, 2, bias=False)) self.conv4 = weight_norm(nn.Conv1d(hidden_channels, hidden_channels, 3, 2, bias=False)) self.conv5 = weight_norm(nn.Conv1d(hidden_channels, hidden_channels, 2, 2, bias=False)) # fmt: on def forward(self, x: torch.Tensor) -> torch.Tensor: # x: [batch_size, 1, wav_length] wav_length = x.size(2) if wav_length % 160 != 0: warnings.warn("wav_length % 160 != 0") x = F.pad(x, (40, 40)) x = F.gelu(self.conv0(x), approximate="tanh") x = F.gelu(self.conv1(x), approximate="tanh") x = F.gelu(self.conv2(x), approximate="tanh") x = F.gelu(self.conv3(x), approximate="tanh") x = F.gelu(self.conv4(x), approximate="tanh") x = F.gelu(self.conv5(x), approximate="tanh") # [batch_size, hidden_channels, wav_length / 160] return x def remove_weight_norm(self): remove_weight_norm(self.conv0) remove_weight_norm(self.conv1) remove_weight_norm(self.conv2) remove_weight_norm(self.conv3) remove_weight_norm(self.conv4) remove_weight_norm(self.conv5) def dump(self, f: Union[BinaryIO, str, bytes, os.PathLike]): if isinstance(f, (str, bytes, os.PathLike)): with open(f, "wb") as f: self.dump(f) return if not hasattr(f, "write"): raise TypeError dump_layer(self.conv0, f) dump_layer(self.conv1, f) dump_layer(self.conv2, f) dump_layer(self.conv3, f) dump_layer(self.conv4, f) dump_layer(self.conv5, f) class FeatureProjection(nn.Module): def __init__(self, channels: int): super().__init__() self.norm = nn.LayerNorm(channels) self.dropout = nn.Dropout(0.1) def forward(self, x: torch.Tensor) -> torch.Tensor: # [batch_size, channels, length] x = self.norm(x.transpose(1, 2)).transpose(1, 2) x = self.dropout(x) return x class PhoneExtractor(nn.Module): def __init__( self, phone_channels: int = 128, hidden_channels: int = 128, backbone_embed_kernel_size: int = 9, kernel_size: int = 17, n_blocks: int = 20, ): super().__init__() self.feature_extractor = FeatureExtractor(hidden_channels) self.feature_projection = FeatureProjection(hidden_channels) self.backbone = ConvNeXtStack( in_channels=hidden_channels, channels=hidden_channels, intermediate_channels=hidden_channels * 3, n_blocks=n_blocks, delay=0, embed_kernel_size=backbone_embed_kernel_size, kernel_size=kernel_size, use_mha=True, ) self.head = weight_norm(nn.Conv1d(hidden_channels, phone_channels, 1)) def forward( self, x: torch.Tensor, return_stats: bool = True ) -> Union[torch.Tensor, tuple[torch.Tensor, dict[str, float]]]: # x: [batch_size, 1, wav_length] stats = {} # [batch_size, 1, wav_length] -> [batch_size, feature_extractor_hidden_channels, length] x = self.feature_extractor(x) if return_stats: stats["feature_norm"] = x.detach().norm(dim=1).mean() # [batch_size, feature_extractor_hidden_channels, length] -> [batch_size, hidden_channels, length] x = self.feature_projection(x) # [batch_size, hidden_channels, length] x = self.backbone(x) # [batch_size, hidden_channels, length] -> [batch_size, phone_channels, length] phone = self.head(F.gelu(x, approximate="tanh")) results = [phone] if return_stats: stats["code_norm"] = phone.detach().norm(dim=1).mean() results.append(stats) if len(results) == 1: return results[0] return tuple(results) @torch.inference_mode() def units(self, x: torch.Tensor) -> torch.Tensor: # x: [batch_size, 1, wav_length] # [batch_size, 1, wav_length] -> [batch_size, phone_channels, length] phone = self.forward(x, return_stats=False) # [batch_size, phone_channels, length] -> [batch_size, length, phone_channels] phone = phone.transpose(1, 2) # [batch_size, length, phone_channels] return phone def remove_weight_norm(self): self.feature_extractor.remove_weight_norm() remove_weight_norm(self.head) def merge_weights(self): self.backbone.merge_weights() self.backbone.embed.bias.data += ( ( self.feature_projection.norm.bias.data[None, :, None] * self.backbone.embed.weight.data # [o, i, k] ) .sum(1) .sum(1) ) self.backbone.embed.weight.data *= self.feature_projection.norm.weight.data[ None, :, None ] self.feature_projection.norm.bias.data[:] = 0.0 self.feature_projection.norm.weight.data[:] = 1.0 def dump(self, f: Union[BinaryIO, str, bytes, os.PathLike]): if isinstance(f, (str, bytes, os.PathLike)): with open(f, "wb") as f: self.dump(f) return if not hasattr(f, "write"): raise TypeError dump_layer(self.feature_extractor, f) dump_layer(self.backbone, f) dump_layer(self.head, f) class VectorQuantizer(nn.Module): def __init__( self, n_speakers: int, codebook_size: int, channels: int, topk: int = 4, training_time_vq: Literal["none", "self", "random"] = "none", ): super().__init__() assert 1 <= topk <= codebook_size self.n_speakers = n_speakers self.codebook_size = codebook_size self.channels = channels self.topk = topk self.training_time_vq = training_time_vq self.register_buffer( "codebooks", torch.empty(n_speakers, codebook_size, channels, dtype=torch.half), ) self.codebooks: torch.Tensor # VQ の適用箇所を変更しやすいように hook にしている self._hook_handle: Optional[torch.utils.hooks.RemovableHandle] = None self.target_speaker_ids: Optional[torch.Tensor] = None def _hook(_, __, output): return self(output, self.target_speaker_ids) self._hook_fn = _hook @torch.no_grad() def build_codebooks( self, collector_func: Callable, target_layer: nn.Module, inputs: Sequence[Iterable[torch.Tensor]], kmeans_n_iters: int = 50, ): assert len(inputs) == self.n_speakers assert self._hook_handle is None, "hook already installed" device = next(self.buffers()).device for spk_id, inps in enumerate(tqdm(inputs, desc="Building codebooks")): activations: list[torch.Tensor] = [] # TODO: データ多すぎる場合に間引く処理をする def _collect(_, __, output): # output: [batch_size, channels, length] activations.append(output.detach()) handle = target_layer.register_forward_hook(_collect) for x in inps: collector_func(x.to(device)) handle.remove() if not activations: raise RuntimeError(f"No activation collected for speaker {spk_id}") # [n_data, channels] activations: torch.Tensor = torch.cat( [ a.transpose(1, 2).reshape(a.size(0) * a.size(2), self.channels) for a in activations ] ) activations = activations.float() activations = F.normalize(activations, dim=1, eps=1e-6) # [codebook_size, channels] centers = ( self._kmeans_plus_plus(activations, self.codebook_size, kmeans_n_iters) if activations.size(0) >= self.codebook_size else self._pad_replicate(activations, self.codebook_size) ) self.codebooks[spk_id] = centers.to(self.codebooks.dtype) def forward( self, x: torch.Tensor, speaker_ids: Optional[torch.Tensor] = None ) -> torch.Tensor: batch_size, channels, length = x.size() assert channels == self.channels device = x.device dtype = x.dtype if self.training: if self.training_time_vq == "none": return x elif self.training_time_vq == "self": if self.target_speaker_ids is None: raise ValueError("target_speaker_ids is not set") elif self.training_time_vq == "random": speaker_ids = torch.randint( 0, self.n_speakers, (batch_size,), device=device ) else: raise ValueError(f"Unknown training_time_vq: {self.training_time_vq}") else: if speaker_ids is None: return x speaker_ids = speaker_ids.to(device) # [batch_size, channels, length] → [batch_size, length, channels] q = F.normalize(x, dim=1, eps=1e-6) codes = self.codebooks[speaker_ids].to(q.dtype) # [batch_size, length, codebook_size] sim = torch.einsum("bcl,bkc->blk", q, codes) # [batch_size, length, topk] _, topk_idx = sim.topk(self.topk, dim=-1) # [batch_size, length, codebook_size, channels] expanded_codes = codes[:, None, :, :].expand(-1, length, -1, -1) # [batch_size, length, topk, channels] expanded_topk_idx = topk_idx[:, :, :, None].expand(-1, -1, -1, channels) # [batch_size, length, topk, channels] gathered = expanded_codes.gather(2, expanded_topk_idx) # [batch_size, length, channels] gathered = gathered.mean(2) # [batch_size, channels, length] return gathered.transpose(1, 2).to(dtype) def enable_hook(self, target_layer: nn.Module): if self._hook_handle is not None: raise RuntimeError("hook already installed") self._hook_handle = target_layer.register_forward_hook(self._hook_fn) def disable_hook(self): if self._hook_handle is None: raise RuntimeError("hook not installed") self._hook_handle.remove() self._hook_handle = None def set_target_speaker_ids(self, speaker_ids: Optional[torch.Tensor]): # この話者が使われる条件は forward() を参照 self.target_speaker_ids = speaker_ids @staticmethod def _pad_replicate(x: torch.Tensor, n: int) -> torch.Tensor: # データ数が n に満たないとき適当に複製して埋める idx = torch.arange(n, device=x.device) % x.size(0) return x[idx] @staticmethod def _kmeans_plus_plus( x: torch.Tensor, n_clusters: int, n_iters: int = 50 ) -> torch.Tensor: n_data, _ = x.size() center_indices = [torch.randint(0, n_data, ()).item()] min_distances = torch.full((n_data,), math.inf, device=x.device) for _ in range(1, n_clusters): last_center_index = center_indices[-1] min_distances = min_distances.minimum( torch.cdist(x, x[last_center_index : last_center_index + 1]) .float() .square_() .squeeze_(1) ) probs = min_distances / (min_distances.sum() + 1e-12) center_indices.append(torch.multinomial(probs, 1).item()) centers = x[center_indices] del min_distances, probs for _ in range(n_iters): distances = torch.cdist(x, centers) # [n_data, n_clusters] labels = distances.argmin(1) # [n_data] # [n_clusters, dim] new_centers = torch.zeros_like(centers).index_add_(0, labels, x) # [n_clusters] counts = labels.bincount(minlength=n_clusters) if (counts == 0).sum().item() != 0: # TODO: 割り当てがないクラスタの処理 warnings.warn("Some clusters have no assigned data points.") new_centers /= counts[:, None].clamp_(min=1).float() centers = new_centers return centers def extract_pitch_features( y: torch.Tensor, # [..., wav_length] hop_length: int = 160, # 10ms win_length: int = 560, # 35ms max_corr_period: int = 256, # 16ms, 62.5Hz (16000 / 256) corr_win_length: int = 304, # 19ms instfreq_features_cutoff_bin: int = 64, # 1828Hz (16000 * 64 / 560) ) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]: assert max_corr_period + corr_win_length == win_length # パディングする padding_length = (win_length - hop_length) // 2 y = F.pad(y, (padding_length, padding_length)) # フレームにする # [..., win_length, n_frames] y_frames = y.unfold(-1, win_length, hop_length).transpose_(-2, -1) # 複素スペクトログラム # Complex[..., (win_length // 2 + 1), n_frames] spec: torch.Tensor = torch.fft.rfft(y_frames, n=win_length, dim=-2) # Complex[..., instfreq_features_cutoff_bin, n_frames] spec = spec[..., :instfreq_features_cutoff_bin, :] # 対数パワースペクトログラム log_power_spec = spec.abs().add_(1e-5).log10_() # 瞬時位相の時間差分 # 時刻 0 の値は 0 delta_spec = spec[..., :, 1:] * spec[..., :, :-1].conj() delta_spec /= delta_spec.abs().add_(1e-5) delta_spec = torch.cat( [torch.zeros_like(delta_spec[..., :, :1]), delta_spec], dim=-1 ) # [..., instfreq_features_cutoff_bin * 3, n_frames] instfreq_features = torch.cat( [log_power_spec, delta_spec.real, delta_spec.imag], dim=-2 ) # 自己相関 # 元々これに 2.0 / corr_win_length を掛けて使おうと思っていたが、 # この値は振幅の 2 乗に比例していて、NN に入力するために良い感じに分散を # 標準化する方法が思いつかなかったのでやめた flipped_y_frames = y_frames.flip((-2,)) a = torch.fft.rfft(flipped_y_frames, n=win_length, dim=-2) b = torch.fft.rfft(y_frames[..., -corr_win_length:, :], n=win_length, dim=-2) # [..., max_corr_period, n_frames] corr = torch.fft.irfft(a * b, n=win_length, dim=-2)[..., corr_win_length:, :] # エネルギー項 energy = flipped_y_frames.square_().cumsum_(-2) energy0 = energy[..., corr_win_length - 1 : corr_win_length, :] energy = energy[..., corr_win_length:, :] - energy[..., :-corr_win_length, :] # Difference function corr_diff = (energy0 + energy).sub_(corr.mul_(2.0)) assert corr_diff.min() >= -1e-3, corr_diff.min() corr_diff.clamp_(min=0.0) # 計算誤差対策 # 標準化 corr_diff *= 2.0 / corr_win_length corr_diff.sqrt_() # 変換モデルへの入力用のエネルギー energy = ( (y_frames * torch.signal.windows.cosine(win_length, device=y.device)[..., None]) .square_() .sum(-2, keepdim=True) ) energy.clamp_(min=1e-3).log10_() # >= -3, 振幅 1 の正弦波なら大体 2.15 energy *= 0.5 # >= -1.5, 振幅 1 の正弦波なら大体 1.07, 1 の差は振幅で 20dB の差 return ( instfreq_features, # [..., instfreq_features_cutoff_bin * 3, n_frames] corr_diff, # [..., max_corr_period, n_frames] energy, # [..., 1, n_frames] ) class PitchEstimator(nn.Module): def __init__( self, input_instfreq_channels: int = 192, input_corr_channels: int = 256, pitch_bins: int = 448, channels: int = 192, intermediate_channels: int = 192 * 2, n_blocks: int = 9, delay: int = 1, # 10ms, 特徴抽出と合わせると 22.5ms embed_kernel_size: int = 3, kernel_size: int = 33, pitch_bins_per_octave: int = 96, ): super().__init__() self.pitch_bins_per_octave = pitch_bins_per_octave self.instfreq_embed_0 = nn.Conv1d(input_instfreq_channels, channels, 1) self.instfreq_embed_1 = nn.Conv1d(channels, channels, 1) self.corr_embed_0 = nn.Conv1d(input_corr_channels, channels, 1) self.corr_embed_1 = nn.Conv1d(channels, channels, 1) self.backbone = ConvNeXtStack( channels, channels, intermediate_channels, n_blocks, delay, embed_kernel_size, kernel_size, enable_scaling=True, ) self.head = nn.Conv1d(channels, pitch_bins, 1) def forward(self, wav: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor]: # wav: [batch_size, 1, wav_length] # [batch_size, input_instfreq_channels, length], # [batch_size, input_corr_channels, length] with torch.amp.autocast("cuda" if torch.cuda.is_available() else "cpu", enabled=False): instfreq_features, corr_diff, energy = extract_pitch_features( wav.squeeze(1), hop_length=160, win_length=560, max_corr_period=256, corr_win_length=304, instfreq_features_cutoff_bin=64, ) instfreq_features = F.gelu( self.instfreq_embed_0(instfreq_features), approximate="tanh" ) instfreq_features = self.instfreq_embed_1(instfreq_features) corr_diff = F.gelu(self.corr_embed_0(corr_diff), approximate="tanh") corr_diff = self.corr_embed_1(corr_diff) # [batch_size, channels, length] x = F.gelu(instfreq_features + corr_diff, approximate="tanh") x = self.backbone(x) # [batch_size, pitch_bins, length] x = self.head(x) return x, energy def sample_pitch( self, pitch: torch.Tensor, band_width: int = 4, return_features: bool = False ) -> Union[torch.Tensor, tuple[torch.Tensor, torch.Tensor]]: # pitch: [batch_size, pitch_bins, length] # 返されるピッチの値には 0 は含まれない batch_size, pitch_bins, length = pitch.size() pitch = pitch.softmax(1) if return_features: unvoiced_proba = pitch[:, :1, :].clone() pitch[:, 0, :] = -100.0 pitch = ( pitch.transpose(1, 2).contiguous().view(batch_size * length, 1, pitch_bins) ) band_pitch = F.conv1d( pitch, torch.ones((1, 1, 1), device=pitch.device).expand(1, 1, band_width), ) # [batch_size * length, 1, pitch_bins - band_width + 1] -> Long[batch_size * length, 1] quantized_band_pitch = band_pitch.argmax(2) if return_features: # [batch_size * length, 1] band_proba = band_pitch.gather(2, quantized_band_pitch[:, :, None]) # [batch_size * length, 1] half_pitch_band_proba = band_pitch.gather( 2, (quantized_band_pitch - self.pitch_bins_per_octave).clamp_(min=1)[ :, :, None ], ) half_pitch_band_proba[ quantized_band_pitch <= self.pitch_bins_per_octave ] = 0.0 half_pitch_proba = (half_pitch_band_proba / (band_proba + 1e-6)).view( batch_size, 1, length ) # [batch_size * length, 1] double_pitch_band_proba = band_pitch.gather( 2, (quantized_band_pitch + self.pitch_bins_per_octave).clamp_( max=pitch_bins - band_width )[:, :, None], ) double_pitch_band_proba[ quantized_band_pitch > pitch_bins - band_width - self.pitch_bins_per_octave ] = 0.0 double_pitch_proba = (double_pitch_band_proba / (band_proba + 1e-6)).view( batch_size, 1, length ) # Long[1, pitch_bins] mask = torch.arange(pitch_bins, device=pitch.device)[None, :] # bool[batch_size * length, pitch_bins] mask = (quantized_band_pitch <= mask) & ( mask < quantized_band_pitch + band_width ) # Long[batch_size, length] quantized_pitch = (pitch.squeeze(1) * mask).argmax(1).view(batch_size, length) if return_features: features = torch.cat( [unvoiced_proba, half_pitch_proba, double_pitch_proba], dim=1 ) # Long[batch_size, length], [batch_size, 3, length] return quantized_pitch, features else: return quantized_pitch def merge_weights(self): self.backbone.merge_weights() def dump(self, f: Union[BinaryIO, str, bytes, os.PathLike]): if isinstance(f, (str, bytes, os.PathLike)): with open(f, "wb") as f: self.dump(f) return if not hasattr(f, "write"): raise TypeError dump_layer(self.instfreq_embed_0, f) dump_layer(self.instfreq_embed_1, f) dump_layer(self.corr_embed_0, f) dump_layer(self.corr_embed_1, f) dump_layer(self.backbone, f) dump_layer(self.head, f) def overlap_add( ir_amp: torch.Tensor, ir_phase: torch.Tensor, window: torch.Tensor, pitch: torch.Tensor, hop_length: int = 240, delay: int = 0, sr: float = 24000.0, ) -> torch.Tensor: batch_size, ir_length, length = ir_amp.size() ir_length = (ir_length - 1) * 2 assert ir_phase.size() == ir_amp.size() assert window.size() == (ir_length,), (window.size(), ir_amp.size()) assert pitch.size() == (batch_size, length * hop_length) assert 0 <= delay < ir_length, (delay, ir_length) # 正規化角周波数 [2π rad] normalized_freq = pitch / sr # 初期位相 [2π rad] をランダムに設定 normalized_freq[:, 0] = torch.rand(batch_size, device=pitch.device) with torch.amp.autocast("cuda" if torch.cuda.is_available() else "cpu", enabled=False): phase = (normalized_freq.double().cumsum_(1) % 1.0).float() # 重ねる箇所を求める # [n_pitchmarks], [n_pitchmarks] indices0, indices1 = torch.nonzero(phase[:, :-1] > phase[:, 1:], as_tuple=True) # 重ねる箇所の小数部分 (位相の遅れ) を求める numer = 1.0 - phase[indices0, indices1] # [n_pitchmarks] fractional_part = numer / (numer + phase[indices0, indices1 + 1]) # 重ねる値を求める # Complex[n_pitchmarks, ir_length / 2 + 1] ir_amp = ir_amp[indices0, :, indices1 // hop_length] ir_phase = ir_phase[indices0, :, indices1 // hop_length] # 位相遅れの量 [rad] # [n_pitchmarks, ir_length / 2 + 1] delay_phase = ( torch.arange(ir_length // 2 + 1, device=pitch.device, dtype=torch.float32)[ None, : ] * (-math.tau / ir_length) * fractional_part[:, None] ) # Complex[n_pitchmarks, ir_length / 2 + 1] spec = torch.polar(ir_amp, ir_phase + delay_phase) # [n_pitchmarks, ir_length] ir = torch.fft.irfft(spec, n=ir_length, dim=1) ir *= window # 加算する値をサンプル単位にばらす # [n_pitchmarks * ir_length] ir = ir.ravel() # Long[n_pitchmarks * ir_length] indices0 = indices0[:, None].expand(-1, ir_length).ravel() # Long[n_pitchmarks * ir_length] indices1 = ( indices1[:, None] + torch.arange(ir_length, device=pitch.device) ).ravel() # overlap-add する overlap_added_signal = torch.zeros( (batch_size, length * hop_length + ir_length), device=pitch.device ) overlap_added_signal.index_put_((indices0, indices1), ir, accumulate=True) overlap_added_signal = overlap_added_signal[:, delay : -ir_length + delay] return overlap_added_signal def generate_noise( aperiodicity: torch.Tensor, delay: int = 0 ) -> tuple[torch.Tensor, torch.Tensor]: # aperiodicity: [batch_size, hop_length, length] batch_size, hop_length, length = aperiodicity.size() excitation = torch.rand( batch_size, (length + 1) * hop_length, device=aperiodicity.device ) excitation -= 0.5 n_fft = 2 * hop_length # 矩形窓で分析 # Complex[batch_size, hop_length + 1, length] noise = torch.stft( excitation, n_fft=n_fft, hop_length=hop_length, window=torch.ones(n_fft, device=excitation.device), center=False, return_complex=True, ) assert noise.size(2) == aperiodicity.size(2) noise[:, 0, :] = 0.0 noise[:, 1:, :] *= aperiodicity # ハン窓で合成 # torch.istft は最適合成窓が使われるので使えないことに注意 # [batch_size, 2 * hop_length, length] noise = torch.fft.irfft(noise, n=2 * hop_length, dim=1) noise *= torch.hann_window(2 * hop_length, device=noise.device)[None, :, None] # [batch_size, (length + 1) * hop_length] noise = F.fold( noise, (1, (length + 1) * hop_length), (1, 2 * hop_length), stride=(1, hop_length), ).squeeze_((1, 2)) assert delay < hop_length noise = noise[:, delay : -hop_length + delay] excitation = excitation[:, delay : -hop_length + delay] return noise, excitation # [batch_size, length * hop_length] D4C_PREVENT_ZERO_DIVISION = True # False にすると本家の処理 def interp(x: torch.Tensor, y: torch.Tensor, xi: torch.Tensor) -> torch.Tensor: # x が単調増加で等間隔と仮定 # 外挿は起こらないと仮定 x = torch.as_tensor(x) y = torch.as_tensor(y) xi = torch.as_tensor(xi) if xi.ndim < y.ndim: diff_ndim = y.ndim - xi.ndim xi = xi.view(tuple([1] * diff_ndim) + xi.size()) if xi.size()[:-1] != y.size()[:-1]: xi = xi.expand(y.size()[:-1] + (xi.size(-1),)) assert (x.min(-1).values == x[..., 0]).all() assert (x.max(-1).values == x[..., -1]).all() assert (xi.min(-1).values >= x[..., 0]).all() assert (xi.max(-1).values <= x[..., -1]).all() delta_x = (x[..., -1].double() - x[..., 0].double()) / (x.size(-1) - 1.0) delta_x = delta_x.to(x.dtype) xi = (xi - x[..., :1]).div_(delta_x[..., None]) xi_base = xi.floor() xi_fraction = xi.sub_(xi_base) xi_base = xi_base.long() delta_y = y.diff(dim=-1, append=y[..., -1:]) yi = y.gather(-1, xi_base) + delta_y.gather(-1, xi_base) * xi_fraction return yi def linear_smoothing( group_delay: torch.Tensor, sr: int, n_fft: int, width: torch.Tensor ) -> torch.Tensor: group_delay = torch.as_tensor(group_delay) assert group_delay.size(-1) == n_fft // 2 + 1 width = torch.as_tensor(width) boundary = (width.max() * n_fft / sr).long() + 1 dtype = group_delay.dtype device = group_delay.device fft_resolution = sr / n_fft mirroring_freq_axis = ( torch.arange(-boundary, n_fft // 2 + 1 + boundary, dtype=dtype, device=device) .add(0.5) .mul(fft_resolution) ) if group_delay.ndim == 1: mirroring_spec = F.pad( group_delay[None], (boundary, boundary), mode="reflect" ).squeeze_(0) elif group_delay.ndim >= 4: shape = group_delay.size() mirroring_spec = F.pad( group_delay.view(math.prod(shape[:-1]), group_delay.size(-1)), (boundary, boundary), mode="reflect", ).view(shape[:-1] + (shape[-1] + 2 * boundary,)) else: mirroring_spec = F.pad(group_delay, (boundary, boundary), mode="reflect") mirroring_segment = mirroring_spec.mul(fft_resolution).cumsum_(-1) center_freq = torch.arange(n_fft // 2 + 1, dtype=dtype, device=device).mul_( fft_resolution ) low_freq = center_freq - width[..., None] * 0.5 high_freq = center_freq + width[..., None] * 0.5 levels = interp( mirroring_freq_axis, mirroring_segment, torch.cat([low_freq, high_freq], dim=-1) ) low_levels, high_levels = levels.split([n_fft // 2 + 1] * 2, dim=-1) smoothed = (high_levels - low_levels).div_(width[..., None]) return smoothed def dc_correction( spec: torch.Tensor, sr: int, n_fft: int, f0: torch.Tensor ) -> torch.Tensor: spec = torch.as_tensor(spec) f0 = torch.as_tensor(f0) dtype = spec.dtype device = spec.device upper_limit = 2 + (f0 * (n_fft / sr)).long() max_upper_limit = upper_limit.max() upper_limit_mask = ( torch.arange(max_upper_limit - 1, device=device) < (upper_limit - 1)[..., None] ) low_freq_axis = torch.arange(max_upper_limit + 1, dtype=dtype, device=device) * ( sr / n_fft ) low_freq_replica = interp( f0[..., None] - low_freq_axis.flip(-1), spec[..., : max_upper_limit + 1].flip(-1), low_freq_axis[..., : max_upper_limit - 1] * upper_limit_mask, ) output = spec.clone() output[..., : max_upper_limit - 1] += low_freq_replica * upper_limit_mask return output def nuttall(n: int, device: torch.types.Device) -> torch.Tensor: t = torch.linspace(0, math.tau, n, device=device) coefs = torch.tensor([0.355768, -0.487396, 0.144232, -0.012604], device=device) terms = torch.tensor([0.0, 1.0, 2.0, 3.0], device=device) cos_matrix = (terms[:, None] * t).cos_() # [4, n] window = coefs.matmul(cos_matrix) return window def get_windowed_waveform( x: torch.Tensor, sr: int, f0: torch.Tensor, position: torch.Tensor, half_window_length_ratio: float, window_type: Literal["hann", "blackman"], n_fft: int, ) -> tuple[torch.Tensor, torch.Tensor]: x = torch.as_tensor(x) f0 = torch.as_tensor(f0) position = torch.as_tensor(position) current_sample = position * sr # [...] half_window_length = (half_window_length_ratio * sr / f0).add_(0.5).long() # [..., fft_size] base_index = -half_window_length[..., None] + torch.arange(n_fft, device=x.device) base_index_mask = base_index <= half_window_length[..., None] # [..., fft_size] safe_index = ((current_sample + 0.501).long()[..., None] + base_index).clamp_( 0, x.size(-1) - 1 ) # [..., fft_size] time_axis = base_index.to(x.dtype).div_(half_window_length_ratio) # [...] normalized_f0 = math.pi / sr * f0 # [..., fft_size] phase = time_axis.mul_(normalized_f0[..., None]) if window_type == "hann": window = phase.cos_().mul_(0.5).add_(0.5) elif window_type == "blackman": window = phase.mul(2.0).cos_().mul_(0.08).add_(phase.cos().mul_(0.5)).add_(0.42) else: assert False window *= base_index_mask prefix_shape = tuple( max(x_size, i_size) for x_size, i_size in zip(x.size(), safe_index.size()) )[:-1] waveform = ( x.expand(prefix_shape + (-1,)) .gather(-1, safe_index.expand(prefix_shape + (-1,))) .mul_(window) ) if not D4C_PREVENT_ZERO_DIVISION: waveform += torch.randn_like(window).mul_(1e-12) waveform *= base_index_mask waveform -= window * waveform.sum(-1, keepdim=True).div_( window.sum(-1, keepdim=True) ) return waveform, window def get_centroid(x: torch.Tensor, n_fft: int) -> torch.Tensor: x = torch.as_tensor(x) if D4C_PREVENT_ZERO_DIVISION: x = x / x.norm(dim=-1, keepdim=True).clamp(min=6e-8) else: x = x / x.norm(dim=-1, keepdim=True) spec0 = torch.fft.rfft(x, n_fft) spec1 = torch.fft.rfft( x * torch.arange(1, x.size(-1) + 1, dtype=x.dtype, device=x.device).div_(n_fft), n_fft, ) centroid = spec0.real * spec1.real + spec0.imag * spec1.imag return centroid def get_static_centroid( x: torch.Tensor, sr: int, f0: torch.Tensor, position: torch.Tensor, n_fft: int ) -> torch.Tensor: """First step: calculation of temporally static parameters on basis of group delay""" x1, _ = get_windowed_waveform( x, sr, f0, position + 0.25 / f0, 2.0, "blackman", n_fft ) x2, _ = get_windowed_waveform( x, sr, f0, position - 0.25 / f0, 2.0, "blackman", n_fft ) centroid1 = get_centroid(x1, n_fft) centroid2 = get_centroid(x2, n_fft) return dc_correction(centroid1 + centroid2, sr, n_fft, f0) def get_smoothed_power_spec( x: torch.Tensor, sr: int, f0: torch.Tensor, position: torch.Tensor, n_fft: int ) -> tuple[torch.Tensor, torch.Tensor]: x = torch.as_tensor(x) f0 = torch.as_tensor(f0) x, window = get_windowed_waveform(x, sr, f0, position, 2.0, "hann", n_fft) window_weight = window.square().sum(-1, keepdim=True) rms = x.square().sum(-1, keepdim=True).div_(window_weight).sqrt_() if D4C_PREVENT_ZERO_DIVISION: x = x / (rms * math.sqrt(n_fft)).clamp_(min=6e-8) smoothed_power_spec = torch.fft.rfft(x, n_fft).abs().square_() smoothed_power_spec = dc_correction(smoothed_power_spec, sr, n_fft, f0) smoothed_power_spec = linear_smoothing(smoothed_power_spec, sr, n_fft, f0) return smoothed_power_spec, rms.detach().squeeze(-1) def get_static_group_delay( static_centroid: torch.Tensor, smoothed_power_spec: torch.Tensor, sr: int, f0: torch.Tensor, n_fft: int, ) -> torch.Tensor: """Second step: calculation of parameter shaping""" if D4C_PREVENT_ZERO_DIVISION: smoothed_power_spec = smoothed_power_spec.clamp(min=6e-8) static_group_delay = static_centroid / smoothed_power_spec # t_g static_group_delay = linear_smoothing( static_group_delay, sr, n_fft, f0 * 0.5 ) # t_gs smoothed_group_delay = linear_smoothing(static_group_delay, sr, n_fft, f0) # t_gb static_group_delay = static_group_delay - smoothed_group_delay # t_D return static_group_delay def get_coarse_aperiodicity( group_delay: torch.Tensor, sr: int, n_fft: int, freq_interval: int, n_aperiodicities: int, window: torch.Tensor, ) -> torch.Tensor: """Third step: estimation of band-aperiodicity""" group_delay = torch.as_tensor(group_delay) window = torch.as_tensor(window) boundary = int(round(n_fft * 8 / window.size(-1))) half_window_length = window.size(-1) // 2 coarse_aperiodicity = torch.empty( group_delay.size()[:-1] + (n_aperiodicities,), dtype=group_delay.dtype, device=group_delay.device, ) for i in range(n_aperiodicities): center = freq_interval * (i + 1) * n_fft // sr segment = ( group_delay[ ..., center - half_window_length : center + half_window_length + 1 ] * window ) power_spec: torch.Tensor = torch.fft.rfft(segment, n_fft).abs().square_() cumulative_power_spec = power_spec.sort(-1).values.cumsum_(-1) if D4C_PREVENT_ZERO_DIVISION: cumulative_power_spec.clamp_(min=6e-8) coarse_aperiodicity[..., i] = ( cumulative_power_spec[..., n_fft // 2 - boundary - 1] / cumulative_power_spec[..., -1] ) coarse_aperiodicity.log10_().mul_(10.0) return coarse_aperiodicity def d4c_love_train( x: torch.Tensor, sr: int, f0: torch.Tensor, position: torch.Tensor, threshold: float ) -> int: x = torch.as_tensor(x) position = torch.as_tensor(position) f0: torch.Tensor = torch.as_tensor(f0) vuv = f0 != 0 lowest_f0 = 40 f0 = f0.clamp(min=lowest_f0) n_fft = 1 << (3 * sr // lowest_f0).bit_length() boundary0 = (100 * n_fft - 1) // sr + 1 boundary1 = (4000 * n_fft - 1) // sr + 1 boundary2 = (7900 * n_fft - 1) // sr + 1 waveform, _ = get_windowed_waveform(x, sr, f0, position, 1.5, "blackman", n_fft) power_spec = torch.fft.rfft(waveform, n_fft).abs().square_() power_spec[..., : boundary0 + 1] = 0.0 cumulative_spec = power_spec.cumsum_(-1) vuv = vuv & ( cumulative_spec[..., boundary1] > threshold * cumulative_spec[..., boundary2] ) return vuv def d4c_general_body( x: torch.Tensor, sr: int, f0: torch.Tensor, freq_interval: int, position: torch.Tensor, n_fft: int, n_aperiodicities: int, window: torch.Tensor, ) -> tuple[torch.Tensor, torch.Tensor]: static_centroid = get_static_centroid(x, sr, f0, position, n_fft) smoothed_power_spec, rms = get_smoothed_power_spec(x, sr, f0, position, n_fft) static_group_delay = get_static_group_delay( static_centroid, smoothed_power_spec, sr, f0, n_fft ) coarse_aperiodicity = get_coarse_aperiodicity( static_group_delay, sr, n_fft, freq_interval, n_aperiodicities, window ) coarse_aperiodicity.add_((f0[..., None] - 100.0).div_(50.0)).clamp_(max=0.0) return coarse_aperiodicity, rms def d4c( x: torch.Tensor, f0: torch.Tensor, t: torch.Tensor, sr: int, threshold: float = 0.85, n_fft_spec: Optional[int] = None, coarse_only: bool = False, ) -> tuple[torch.Tensor, torch.Tensor]: """Adapted from https://github.com/tuanad121/Python-WORLD/blob/master/world/d4c.py""" FLOOR_F0 = 71 FLOOR_F0_D4C = 47 UPPER_LIMIT = 15000 FREQ_INTERVAL = 3000 assert sr == int(sr) sr = int(sr) assert sr % 2 == 0 x = torch.as_tensor(x) f0 = torch.as_tensor(f0) temporal_positions = torch.as_tensor(t) n_fft_d4c = 1 << (4 * sr // FLOOR_F0_D4C).bit_length() if n_fft_spec is None: n_fft_spec = 1 << (3 * sr // FLOOR_F0).bit_length() n_aperiodicities = min(UPPER_LIMIT, sr // 2 - FREQ_INTERVAL) // FREQ_INTERVAL assert n_aperiodicities >= 1 window_length = FREQ_INTERVAL * n_fft_d4c // sr * 2 + 1 window = nuttall(window_length, device=x.device) freq_axis = torch.arange(n_fft_spec // 2 + 1, device=x.device) * (sr / n_fft_spec) coarse_aperiodicity, rms = d4c_general_body( x[..., None, :], sr, f0.clamp(min=FLOOR_F0_D4C), FREQ_INTERVAL, temporal_positions, n_fft_d4c, n_aperiodicities, window, ) if coarse_only: return coarse_aperiodicity, rms even_coarse_axis = ( torch.arange(n_aperiodicities + 3, device=x.device) * FREQ_INTERVAL ) assert even_coarse_axis[-2] <= sr // 2 < even_coarse_axis[-1], sr coarse_axis_low = ( torch.arange(n_aperiodicities + 1, dtype=torch.float, device=x.device) * FREQ_INTERVAL ) aperiodicity_low = interp( coarse_axis_low, F.pad(coarse_aperiodicity, (1, 0), value=-60.0), freq_axis[freq_axis < n_aperiodicities * FREQ_INTERVAL], ) coarse_axis_high = torch.tensor( [n_aperiodicities * FREQ_INTERVAL, sr * 0.5], device=x.device ) aperiodicity_high = interp( coarse_axis_high, F.pad(coarse_aperiodicity[..., -1:], (0, 1), value=-1e-12), freq_axis[freq_axis >= n_aperiodicities * FREQ_INTERVAL], ) aperiodicity = torch.cat([aperiodicity_low, aperiodicity_high], -1) aperiodicity = 10.0 ** (aperiodicity / 20.0) vuv = d4c_love_train(x[..., None, :], sr, f0, temporal_positions, threshold) aperiodicity = torch.where(vuv[..., None], aperiodicity, 1 - 1e-12) return aperiodicity, coarse_aperiodicity class Vocoder(nn.Module): def __init__( self, channels: int, speaker_embedding_channels: int = 128, hop_length: int = 240, n_pre_blocks: int = 4, out_sample_rate: float = 24000.0, ): super().__init__() self.hop_length = hop_length self.out_sample_rate = out_sample_rate self.prenet = ConvNeXtStack( in_channels=channels, channels=channels, intermediate_channels=channels * 2, n_blocks=n_pre_blocks, delay=2, # 20ms 遅延 embed_kernel_size=7, kernel_size=33, enable_scaling=True, use_mha=True, cross_attention=True, kv_channels=speaker_embedding_channels, ) self.ir_generator = ConvNeXtStack( in_channels=channels, channels=channels, intermediate_channels=channels * 2, n_blocks=2, delay=0, embed_kernel_size=3, kernel_size=33, use_weight_standardization=True, enable_scaling=True, ) self.ir_generator_post = WSConv1d(channels, 512, 1) self.register_buffer("ir_scale", torch.tensor(1.0)) self.ir_window = nn.Parameter(torch.ones(512)) self.aperiodicity_generator = ConvNeXtStack( in_channels=channels, channels=channels, intermediate_channels=channels * 2, n_blocks=1, delay=0, embed_kernel_size=3, kernel_size=33, use_weight_standardization=True, enable_scaling=True, ) self.aperiodicity_generator_post = WSConv1d(channels, hop_length, 1, bias=False) self.register_buffer("aperiodicity_scale", torch.tensor(0.005)) self.post_filter_generator = ConvNeXtStack( in_channels=channels, channels=channels, intermediate_channels=channels * 2, n_blocks=1, delay=0, embed_kernel_size=3, kernel_size=33, use_weight_standardization=True, enable_scaling=True, ) self.post_filter_generator_post = WSConv1d(channels, 512, 1, bias=False) self.register_buffer("post_filter_scale", torch.tensor(0.01)) def forward( self, x: torch.Tensor, pitch: torch.Tensor, speaker_embedding: torch.Tensor ) -> tuple[torch.Tensor, dict[str, torch.Tensor]]: # x: [batch_size, channels, length] # pitch: [batch_size, length] # speaker_embedding: [batch_size, speaker_embedding_length, speaker_embedding_channels] batch_size, _, length = x.size() x = self.prenet(x, speaker_embedding) ir = self.ir_generator(x) ir = F.silu(ir, inplace=True) # [batch_size, 512, length] ir = self.ir_generator_post(ir) ir *= self.ir_scale ir_amp = ir[:, : ir.size(1) // 2 + 1, :].exp() ir_phase = F.pad(ir[:, ir.size(1) // 2 + 1 :, :], (0, 0, 1, 1)) ir_phase[:, 1::2, :] += math.pi # TODO: 直流成分が正の値しか取れないのを修正する # 最近傍補間 # [batch_size, length * hop_length] pitch = torch.repeat_interleave(pitch, self.hop_length, dim=1) # [batch_size, length * hop_length] periodic_signal = overlap_add( ir_amp, ir_phase, self.ir_window, pitch, self.hop_length, delay=0, sr=self.out_sample_rate, ) aperiodicity = self.aperiodicity_generator(x) aperiodicity = F.silu(aperiodicity, inplace=True) # [batch_size, hop_length, length] aperiodicity = self.aperiodicity_generator_post(aperiodicity) aperiodicity *= self.aperiodicity_scale # [batch_size, length * hop_length], [batch_size, length * hop_length] aperiodic_signal, noise_excitation = generate_noise(aperiodicity, delay=0) post_filter = self.post_filter_generator(x) post_filter = F.silu(post_filter, inplace=True) # [batch_size, 512, length] post_filter = self.post_filter_generator_post(post_filter) post_filter *= self.post_filter_scale post_filter[:, 0, :] += 1.0 # [batch_size, length, 512] post_filter = post_filter.transpose(1, 2) with torch.amp.autocast("cuda" if torch.cuda.is_available() else "cpu", enabled=False): periodic_signal = periodic_signal.float() aperiodic_signal = aperiodic_signal.float() post_filter = post_filter.float() post_filter = torch.fft.rfft(post_filter, n=768) # [batch_size, length, 768] periodic_signal = torch.fft.irfft( torch.fft.rfft( periodic_signal.view(batch_size, length, self.hop_length), n=768 ) * post_filter, n=768, ) aperiodic_signal = torch.fft.irfft( torch.fft.rfft( aperiodic_signal.view(batch_size, length, self.hop_length), n=768 ) * post_filter, n=768, ) periodic_signal = F.fold( periodic_signal.transpose(1, 2), (1, (length - 1) * self.hop_length + 768), (1, 768), stride=(1, self.hop_length), ).squeeze_((1, 2)) aperiodic_signal = F.fold( aperiodic_signal.transpose(1, 2), (1, (length - 1) * self.hop_length + 768), (1, 768), stride=(1, self.hop_length), ).squeeze_((1, 2)) periodic_signal = periodic_signal[:, 120 : 120 + length * self.hop_length] aperiodic_signal = aperiodic_signal[:, 120 : 120 + length * self.hop_length] noise_excitation = noise_excitation[:, 120:] # TODO: compensation の正確さが怪しくなってくる。今も本当に必要なのか? # [batch_size, 1, length * hop_length] y_g_hat = (periodic_signal + aperiodic_signal)[:, None, :] return y_g_hat, { "periodic_signal": periodic_signal.detach(), "aperiodic_signal": aperiodic_signal.detach(), "noise_excitation": noise_excitation.detach(), } def merge_weights(self): self.prenet.merge_weights() self.ir_generator.merge_weights() self.ir_generator_post.merge_weights() self.aperiodicity_generator.merge_weights() self.aperiodicity_generator_post.merge_weights() self.ir_generator_post.weight.data *= self.ir_scale self.ir_generator_post.bias.data *= self.ir_scale self.ir_scale.fill_(1.0) self.aperiodicity_generator_post.weight.data *= self.aperiodicity_scale self.aperiodicity_scale.fill_(1.0) self.post_filter_generator.merge_weights() self.post_filter_generator_post.merge_weights() self.post_filter_generator_post.weight.data *= self.post_filter_scale self.post_filter_scale.fill_(1.0) def dump(self, f: Union[BinaryIO, str, bytes, os.PathLike]): if isinstance(f, (str, bytes, os.PathLike)): with open(f, "wb") as f: self.dump(f) return if not hasattr(f, "write"): raise TypeError dump_layer(self.prenet, f) dump_layer(self.ir_generator, f) dump_layer(self.ir_generator_post, f) dump_layer(self.ir_window, f) dump_layer(self.aperiodicity_generator, f) dump_layer(self.aperiodicity_generator_post, f) dump_layer(self.post_filter_generator, f) dump_layer(self.post_filter_generator_post, f) def compute_loudness( x: torch.Tensor, sr: int, win_lengths: list[int] ) -> list[torch.Tensor]: # x: [batch_size, wav_length] assert x.ndim == 2 n_fft = 2048 chunk_length = n_fft // 2 n_taps = chunk_length + 1 results = [] with torch.amp.autocast("cuda" if torch.cuda.is_available() else "cpu", enabled=False): if not hasattr(compute_loudness, "filter"): compute_loudness.filter = {} if sr not in compute_loudness.filter: ir = torch.zeros(n_taps, device=x.device, dtype=torch.double) ir[0] = 0.5 ir = torchaudio.functional.treble_biquad( ir, sr, 4.0, 1500.0, 1.0 / math.sqrt(2) ) ir = torchaudio.functional.highpass_biquad(ir, sr, 38.0, 0.5) ir *= 2.0 compute_loudness.filter[sr] = torch.fft.rfft(ir, n=n_fft).to( torch.complex64 ) x = x.float() wav_length = x.size(-1) if wav_length % chunk_length != 0: x = F.pad(x, (0, chunk_length - wav_length % chunk_length)) padded_wav_length = x.size(-1) x = x.view(x.size()[:-1] + (padded_wav_length // chunk_length, chunk_length)) x = torch.fft.irfft( torch.fft.rfft(x, n=n_fft) * compute_loudness.filter[sr], n=n_fft, ) x = F.fold( x.transpose(-2, -1), (1, padded_wav_length + chunk_length), (1, n_fft), stride=(1, chunk_length), ).squeeze_((-3, -2))[..., :wav_length] x.square_() for win_length in win_lengths: hop_length = win_length // 4 # [..., n_frames] energy = ( x.unfold(-1, win_length, hop_length) .matmul(torch.hann_window(win_length, device=x.device)) .add_(win_length / 4.0 * 1e-5) .log10_() ) # フィルタリング後の波形が振幅 1 の正弦波なら大体 log10(win_length/4), 1 の差は 10dB の差 results.append(energy) return results def beatrice_slice_segments( x: torch.Tensor, start_indices: torch.Tensor, segment_length: int ) -> torch.Tensor: batch_size, channels, _ = x.size() # [batch_size, 1, segment_size] indices = start_indices[:, None, None] + torch.arange( segment_length, device=start_indices.device ) # [batch_size, channels, segment_size] indices = indices.expand(batch_size, channels, segment_length) return x.gather(2, indices) class ConverterNetwork(nn.Module): def __init__( self, phone_extractor: PhoneExtractor, pitch_estimator: PitchEstimator, n_speakers: int, pitch_bins: int, hidden_channels: int, vq_topk: int = 4, training_time_vq: Literal["none", "self", "random"] = "none", phone_noise_ratio: int = 0.5, floor_noise_level: float = 1e-3, ): super().__init__() self.frozen_modules = { "phone_extractor": phone_extractor.eval().requires_grad_(False), "pitch_estimator": pitch_estimator.eval().requires_grad_(False), } self.pitch_bins = pitch_bins self.phone_noise_ratio = phone_noise_ratio self.floor_noise_level = floor_noise_level self.out_sample_rate = out_sample_rate = 24000 phone_channels = 128 self.vq = VectorQuantizer( n_speakers=n_speakers, codebook_size=512, channels=phone_channels, topk=vq_topk, training_time_vq=training_time_vq, ) self.embed_phone = nn.Conv1d(phone_channels, hidden_channels, 1) self.embed_phone.weight.data.normal_(0.0, math.sqrt(2.0 / (256 * 5))) self.embed_phone.bias.data.zero_() self.embed_quantized_pitch = nn.Embedding(pitch_bins, hidden_channels) phase = ( torch.arange(pitch_bins, dtype=torch.float)[:, None] * ( torch.arange(0, hidden_channels, 2, dtype=torch.float) * (-math.log(10000.0) / hidden_channels) ).exp_() ) self.embed_quantized_pitch.weight.data[:, 0::2] = phase.sin() self.embed_quantized_pitch.weight.data[:, 1::2] = phase.cos_() self.embed_quantized_pitch.weight.data *= math.sqrt(4.0 / 5.0) self.embed_quantized_pitch.weight.requires_grad_(False) self.embed_pitch_features = nn.Conv1d(4, hidden_channels, 1) self.embed_pitch_features.weight.data.normal_(0.0, math.sqrt(2.0 / (4 * 5))) self.embed_pitch_features.bias.data.zero_() self.embed_speaker = nn.Embedding(n_speakers, hidden_channels) self.embed_speaker.weight.data.normal_(0.0, math.sqrt(2.0 / 5.0)) self.embed_formant_shift = nn.Embedding(9, hidden_channels) self.embed_formant_shift.weight.data.normal_(0.0, math.sqrt(2.0 / 5.0)) self.key_value_speaker_embedding_length = 384 self.key_value_speaker_embedding_channels = 128 self.key_value_speaker_embedding = nn.Embedding( n_speakers, self.key_value_speaker_embedding_length * self.key_value_speaker_embedding_channels, ) self.key_value_speaker_embedding.weight.data[0].normal_() self.key_value_speaker_embedding.weight.data[1:] = ( self.key_value_speaker_embedding.weight.data[0] ) self.vocoder = Vocoder( channels=hidden_channels, speaker_embedding_channels=self.key_value_speaker_embedding_channels, hop_length=out_sample_rate // 100, n_pre_blocks=4, out_sample_rate=out_sample_rate, ) self.melspectrograms = nn.ModuleList() for win_length, n_mels in [ (32, 5), (64, 10), (128, 20), (256, 40), (512, 80), (1024, 160), (2048, 320), ]: self.melspectrograms.append( torchaudio.transforms.MelSpectrogram( sample_rate=out_sample_rate, n_fft=win_length, win_length=win_length, hop_length=win_length // 4, n_mels=n_mels, power=2, norm="slaney", mel_scale="slaney", ) ) def initialize_vq(self, inputs: Sequence[Iterable[torch.Tensor]]): collector_func = self.frozen_modules["phone_extractor"].units target_layer = self.frozen_modules["phone_extractor"].head self.vq.build_codebooks( collector_func, target_layer, inputs, ) self.vq.enable_hook(target_layer) def enable_hook(self): target_layer = self.frozen_modules["phone_extractor"].head self.vq.enable_hook(target_layer) def _get_resampler( self, orig_freq, new_freq, device, cache={} ) -> torchaudio.transforms.Resample: key = orig_freq, new_freq if key in cache: return cache[key] resampler = torchaudio.transforms.Resample(orig_freq, new_freq).to( device, non_blocking=True ) cache[key] = resampler return resampler def forward( self, x: torch.Tensor, target_speaker_id: torch.Tensor, formant_shift_semitone: torch.Tensor, pitch_shift_semitone: Optional[torch.Tensor] = None, slice_start_indices: Optional[torch.Tensor] = None, slice_segment_length: Optional[int] = None, return_stats: bool = False, ) -> Union[torch.Tensor, tuple[torch.Tensor, dict[str, float]]]: # x: [batch_size, 1, wav_length] # target_speaker_id: Long[batch_size] # formant_shift_semitone: [batch_size] # pitch_shift_semitone: [batch_size] # slice_start_indices: [batch_size] batch_size, _, _ = x.size() self.vq.set_target_speaker_ids(target_speaker_id) with torch.inference_mode(): phone_extractor: PhoneExtractor = self.frozen_modules["phone_extractor"] pitch_estimator: PitchEstimator = self.frozen_modules["pitch_estimator"] # [batch_size, 1, wav_length] -> [batch_size, phone_channels, length] phone = phone_extractor.units(x).transpose(1, 2) if self.training and self.phone_noise_ratio != 0.0: phone *= (1.0 - self.phone_noise_ratio) / phone.square().mean( 1, keepdim=True ).sqrt_() noise = torch.randn_like(phone) noise *= ( self.phone_noise_ratio / noise.square().mean(1, keepdim=True).sqrt_() ) phone += noise # F.rms_norm は PyTorch >= 2.4 が必要 phone *= ( 1.0 / phone.square() .mean(1, keepdim=True) .add_(torch.finfo(torch.float).eps) .sqrt_() ) # [batch_size, 1, wav_length] -> [batch_size, pitch_bins, length], [batch_size, 1, length] pitch, energy = pitch_estimator(x) # augmentation if self.training: # [batch_size, pitch_bins - 1] weights = pitch.softmax(1)[:, 1:, :].mean(2) # [batch_size] mean_pitch = ( weights * torch.arange( 1, self.embed_quantized_pitch.num_embeddings, device=weights.device, ) ).sum(1) / weights.sum(1) mean_pitch = mean_pitch.round_().long() target_pitch = torch.randint_like(mean_pitch, 64, 257) shift = target_pitch - mean_pitch shift_ratio = ( 2.0 ** (shift.float() / pitch_estimator.pitch_bins_per_octave) ).tolist() shift = [] interval_length = 100 # 1s interval_zeros = torch.zeros( (1, 1, interval_length * 160), device=x.device ) concatenated_shifted_x = [] offsets = [0] torch.backends.cudnn.benchmark = False for i in range(batch_size): shift_ratio_i = shift_ratio[i] shift_ratio_fraction_i = Fraction.from_float( shift_ratio_i ).limit_denominator(30) shift_numer_i = shift_ratio_fraction_i.numerator shift_denom_i = shift_ratio_fraction_i.denominator shift_ratio_i = shift_numer_i / shift_denom_i shift_i = int( round( math.log2(shift_ratio_i) * pitch_estimator.pitch_bins_per_octave ) ) shift.append(shift_i) shift_ratio[i] = shift_ratio_i # [1, 1, wav_length / shift_ratio] with torch.amp.autocast("cuda" if torch.cuda.is_available() else "cpu", enabled=False): shifted_x_i = self._get_resampler( shift_numer_i, shift_denom_i, x.device )(x[i])[None] if shifted_x_i.size(2) % 160 != 0: shifted_x_i = F.pad( shifted_x_i, (0, 160 - shifted_x_i.size(2) % 160), mode="reflect", ) assert shifted_x_i.size(2) % 160 == 0 offsets.append( offsets[-1] + interval_length + shifted_x_i.size(2) // 160 ) concatenated_shifted_x.extend([interval_zeros, shifted_x_i]) if offsets[-1] % 256 != 0: # 長さが同じ方が何かのキャッシュが効いて早くなるようなので # 適当に 256 の倍数になるようにパディングして長さのパターン数を減らす concatenated_shifted_x.append( torch.zeros( (1, 1, (256 - offsets[-1] % 256) * 160), device=x.device ) ) # [batch_size, 1, sum(wav_length) + batch_size * 16000] concatenated_shifted_x = torch.cat(concatenated_shifted_x, dim=2) assert concatenated_shifted_x.size(2) % (256 * 160) == 0 # [1, pitch_bins, length / shift_ratio], [1, 1, length / shift_ratio] concatenated_pitch, concatenated_energy = pitch_estimator( concatenated_shifted_x ) for i in range(batch_size): shift_i = shift[i] shift_ratio_i = shift_ratio[i] left = offsets[i] + interval_length right = offsets[i + 1] pitch_i = concatenated_pitch[:, :, left:right] energy_i = concatenated_energy[:, :, left:right] pitch_i = F.interpolate( pitch_i, scale_factor=shift_ratio_i, mode="linear", align_corners=False, ) energy_i = F.interpolate( energy_i, scale_factor=shift_ratio_i, mode="linear", align_corners=False, ) assert pitch_i.size(2) == energy_i.size(2) assert abs(pitch_i.size(2) - pitch.size(2)) <= 10 length = min(pitch_i.size(2), pitch.size(2)) if shift_i > 0: pitch[i : i + 1, :1, :length] = pitch_i[:, :1, :length] pitch[i : i + 1, 1:-shift_i, :length] = pitch_i[ :, 1 + shift_i :, :length ] pitch[i : i + 1, -shift_i:, :length] = -10.0 elif shift_i < 0: pitch[i : i + 1, :1, :length] = pitch_i[:, :1, :length] pitch[i : i + 1, 1 : 1 - shift_i, :length] = -10.0 pitch[i : i + 1, 1 - shift_i :, :length] = pitch_i[ :, 1:shift_i, :length ] energy[i : i + 1, :, :length] = energy_i[:, :, :length] torch.backends.cudnn.benchmark = True # [batch_size, pitch_bins, length] -> Long[batch_size, length], [batch_size, 3, length] quantized_pitch, pitch_features = pitch_estimator.sample_pitch( pitch, return_features=True ) if pitch_shift_semitone is not None: quantized_pitch = torch.where( quantized_pitch == 0, quantized_pitch, ( quantized_pitch + ( pitch_shift_semitone[:, None] * (pitch_estimator.pitch_bins_per_octave / 12.0) ) .round_() .long() ).clamp_(1, self.pitch_bins - 1), ) pitch = 55.0 * 2.0 ** ( quantized_pitch.float() / pitch_estimator.pitch_bins_per_octave ) # phone が 2.5ms 先読みしているのに対して、 # energy は 12.5ms, pitch_features は 22.5ms 先読みしているので、 # ずらして phone に合わせる energy = F.pad(energy[:, :, :-1], (1, 0), mode="reflect") quantized_pitch = F.pad(quantized_pitch[:, :-2], (2, 0), mode="reflect") pitch_features = F.pad(pitch_features[:, :, :-2], (2, 0), mode="reflect") # [batch_size, 1, length], [batch_size, 3, length] -> [batch_size, 4, length] pitch_features = torch.cat([energy, pitch_features], dim=1) formant_shift_indices = ( ((formant_shift_semitone + 2.0) * 2.0).round_().long() ) phone = phone.clone() quantized_pitch = quantized_pitch.clone() pitch_features = pitch_features.clone() formant_shift_indices = formant_shift_indices.clone() pitch = pitch.clone() # [batch_sise, hidden_channels, length] x = ( self.embed_phone(phone) + self.embed_quantized_pitch(quantized_pitch).transpose(1, 2) + self.embed_pitch_features(pitch_features) + ( self.embed_speaker(target_speaker_id)[:, :, None] + self.embed_formant_shift(formant_shift_indices)[:, :, None] ) ) if slice_start_indices is not None: assert slice_segment_length is not None # [batch_size, hidden_channels, length] -> [batch_size, hidden_channels, segment_length] x = beatrice_slice_segments(x, slice_start_indices, slice_segment_length) x = F.silu(x, inplace=True) speaker_embedding = self.key_value_speaker_embedding(target_speaker_id).view( batch_size, self.key_value_speaker_embedding_length, self.key_value_speaker_embedding_channels, ) # [batch_size, hidden_channels, segment_length] -> [batch_size, 1, segment_length * 240] y_g_hat, stats = self.vocoder(x, pitch, speaker_embedding) stats["pitch"] = pitch if return_stats: return y_g_hat, stats else: return y_g_hat def _normalize_melsp(self, x): return x.clamp(min=1e-10).log_() def forward_and_compute_loss( self, noisy_wavs_16k: torch.Tensor, target_speaker_id: torch.Tensor, formant_shift_semitone: torch.Tensor, slice_start_indices: torch.Tensor, slice_segment_length: int, y_all: torch.Tensor, enable_loss_ap: bool = False, ) -> tuple[ torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor, dict[str, float], ]: # noisy_wavs_16k: [batch_size, 1, wav_length] # target_speaker_id: Long[batch_size] # formant_shift_semitone: [batch_size] # slice_start_indices: [batch_size] # slice_segment_length: int # y_all: [batch_size, 1, wav_length] stats = {} loss_mel = 0.0 loss_loudness = 0.0 loudness_win_lengths = [512, 1024, 2048, 4096] # [batch_size, 1, wav_length] -> [batch_size, 1, wav_length * 240] y_hat_all, intermediates = self( noisy_wavs_16k, target_speaker_id, formant_shift_semitone, return_stats=True, ) y_hat_all = y_hat_all.detach().where(y_all == 0.0, y_hat_all) with torch.amp.autocast("cuda" if torch.cuda.is_available() else "cpu", enabled=False): periodic_signal = intermediates["periodic_signal"].float() aperiodic_signal = intermediates["aperiodic_signal"].float() noise_excitation = intermediates["noise_excitation"].float() periodic_signal = periodic_signal[:, : noise_excitation.size(1)] aperiodic_signal = aperiodic_signal[:, : noise_excitation.size(1)] y_hat_all = y_hat_all.float() floor_noise = torch.randn_like(y_all) * self.floor_noise_level y_all = y_all + floor_noise y_hat_all += floor_noise y_hat_all_truncated = y_hat_all.squeeze(1)[:, : periodic_signal.size(1)] y_all_truncated = y_all.squeeze(1)[:, : periodic_signal.size(1)] y_loudness = compute_loudness( y_all_truncated, self.out_sample_rate, loudness_win_lengths ) y_hat_loudness = compute_loudness( y_hat_all_truncated, self.out_sample_rate, loudness_win_lengths ) for win_length, y_loudness_i, y_hat_loudness_i in zip( loudness_win_lengths, y_loudness, y_hat_loudness ): loss_loudness_i = F.mse_loss(y_hat_loudness_i, y_loudness_i) loss_loudness += loss_loudness_i * math.sqrt(win_length) stats[f"loss_loudness_{win_length}"] = loss_loudness_i.item() for melspectrogram in self.melspectrograms: melsp_periodic_signal = melspectrogram(periodic_signal) melsp_aperiodic_signal = melspectrogram(aperiodic_signal) melsp_noise_excitation = melspectrogram(noise_excitation) # [1, n_mels, 1] # 1/6 ... [-0.5, 0.5] の一様乱数の平均パワー # 3/8 ... ハン窓をかけた時のパワー減衰 # 0.5 ... 謎 reference_melsp = melspectrogram.mel_scale( torch.full( (1, melspectrogram.n_fft // 2 + 1, 1), (1 / 6) * (3 / 8) * 0.5 * melspectrogram.win_length, device=noisy_wavs_16k.device, ) ) aperiodic_ratio = melsp_aperiodic_signal / ( melsp_periodic_signal + melsp_aperiodic_signal + 1e-5 ) compensation_ratio = reference_melsp / (melsp_noise_excitation + 1e-5) melsp_y_hat = melspectrogram(y_hat_all_truncated) melsp_y_hat = melsp_y_hat * ( (1.0 - aperiodic_ratio) + aperiodic_ratio * compensation_ratio ) y_hat_mel = self._normalize_melsp(melsp_y_hat) y_mel = self._normalize_melsp(melspectrogram(y_all_truncated)) loss_mel_i = F.l1_loss(y_hat_mel, y_mel) loss_mel += loss_mel_i stats[ f"loss_mel_{melspectrogram.win_length}_{melspectrogram.n_mels}" ] = loss_mel_i.item() loss_mel /= len(self.melspectrograms) if enable_loss_ap: t = ( torch.arange(intermediates["pitch"].size(1), device=y_all.device) * 0.01 + 0.005 ) y_coarse_aperiodicity, y_rms = d4c( y_all.squeeze(1), intermediates["pitch"], t, self.vocoder.out_sample_rate, coarse_only=True, ) y_coarse_aperiodicity = 10.0 ** (y_coarse_aperiodicity / 10.0) y_hat_coarse_aperiodicity, y_hat_rms = d4c( y_hat_all.squeeze(1), intermediates["pitch"], t, self.vocoder.out_sample_rate, coarse_only=True, ) y_hat_coarse_aperiodicity = 10.0 ** (y_hat_coarse_aperiodicity / 10.0) rms = torch.maximum(y_rms, y_hat_rms) loss_ap = F.mse_loss( y_hat_coarse_aperiodicity, y_coarse_aperiodicity, reduction="none" ) loss_ap *= (rms / (rms + 1e-3) * (rms > 1e-5))[:, :, None] loss_ap = loss_ap.mean() else: loss_ap = torch.tensor(0.0) # [batch_size, 1, wav_length] -> [batch_size, 1, slice_segment_length * 240] y_hat = beatrice_slice_segments( y_hat_all, slice_start_indices * 240, slice_segment_length * 240 ) # [batch_size, 1, wav_length] -> [batch_size, 1, slice_segment_length * 240] y = beatrice_slice_segments(y_all, slice_start_indices * 240, slice_segment_length * 240) return y, y_hat, y_hat_all, loss_loudness, loss_mel, loss_ap, stats def merge_weights(self): self.vocoder.merge_weights() def dump(self, f: Union[BinaryIO, str, bytes, os.PathLike]): if isinstance(f, (str, bytes, os.PathLike)): with open(f, "wb") as f: self.dump(f) return if not hasattr(f, "write"): raise TypeError dump_layer(self.embed_phone, f) dump_layer(self.embed_quantized_pitch, f) dump_layer(self.embed_pitch_features, f) dump_layer(self.vocoder, f) def dump_speaker_embeddings(self, f: Union[BinaryIO, str, bytes, os.PathLike]): if isinstance(f, (str, bytes, os.PathLike)): with open(f, "wb") as f: self.dump_speaker_embeddings(f) return if not hasattr(f, "write"): raise TypeError dump_params(self.vq.codebooks, f) dump_layer(self.embed_speaker, f) dump_layer(self.embed_formant_shift, f) dump_layer(self.key_value_speaker_embedding, f) def dump_embedding_setter(self, f: Union[BinaryIO, str, bytes, os.PathLike]): if isinstance(f, (str, bytes, os.PathLike)): with open(f, "wb") as f: self.dump_embedding_setter(f) return if not hasattr(f, "write"): raise TypeError self.vocoder.prenet.dump_kv(f) # Discriminator def _normalize(tensor: torch.Tensor, dim: int) -> torch.Tensor: denom = tensor.norm(p=2.0, dim=dim, keepdim=True).clamp_min(1e-6) return tensor / denom class SANConv2d(nn.Conv2d): def __init__( self, in_channels: int, out_channels: int, kernel_size: int, stride: int = 1, padding: int = 0, dilation: int = 1, bias: bool = True, padding_mode="zeros", device=None, dtype=None, ): super().__init__( in_channels, out_channels, kernel_size, stride, padding=padding, dilation=dilation, groups=1, bias=bias, padding_mode=padding_mode, device=device, dtype=dtype, ) scale = self.weight.norm(p=2.0, dim=[1, 2, 3], keepdim=True).clamp_min(1e-6) self.weight = nn.parameter.Parameter(self.weight / scale.expand_as(self.weight)) self.scale = nn.parameter.Parameter(scale.view(out_channels)) if bias: self.bias = nn.parameter.Parameter( torch.zeros(in_channels, device=device, dtype=dtype) ) else: self.register_parameter("bias", None) def forward( self, input: torch.Tensor, flg_san_train: bool = False ) -> Union[torch.Tensor, tuple[torch.Tensor, torch.Tensor]]: if self.bias is not None: input = input + self.bias.view(self.in_channels, 1, 1) normalized_weight = self._get_normalized_weight() scale = self.scale.view(self.out_channels, 1, 1) if flg_san_train: out_fun = F.conv2d( input, normalized_weight.detach(), None, self.stride, self.padding, self.dilation, self.groups, ) out_dir = F.conv2d( input.detach(), normalized_weight, None, self.stride, self.padding, self.dilation, self.groups, ) out = out_fun * scale, out_dir * scale.detach() else: out = F.conv2d( input, normalized_weight, None, self.stride, self.padding, self.dilation, self.groups, ) out = out * scale return out @torch.no_grad() def normalize_weight(self): self.weight.data = self._get_normalized_weight() def _get_normalized_weight(self) -> torch.Tensor: return _normalize(self.weight, dim=[1, 2, 3]) def get_padding(kernel_size: int, dilation: int = 1) -> int: return (kernel_size * dilation - dilation) // 2 class BeatriceDiscriminatorP(nn.Module): def __init__( self, period: int, kernel_size: int = 5, stride: int = 3, san: bool = False ): super().__init__() self.period = period self.san = san # fmt: off self.convs = nn.ModuleList([ weight_norm(nn.Conv2d(1, 32, (kernel_size, 1), (stride, 1), (get_padding(kernel_size, 1), 0))), weight_norm(nn.Conv2d(32, 128, (kernel_size, 1), (stride, 1), (get_padding(kernel_size, 1), 0))), weight_norm(nn.Conv2d(128, 512, (kernel_size, 1), (stride, 1), (get_padding(kernel_size, 1), 0))), weight_norm(nn.Conv2d(512, 1024, (kernel_size, 1), (stride, 1), (get_padding(kernel_size, 1), 0))), weight_norm(nn.Conv2d(1024, 1024, (kernel_size, 1), 1, (get_padding(kernel_size, 1), 0))), ]) # fmt: on if san: self.conv_post = SANConv2d(1024, 1, (3, 1), 1, (1, 0)) else: self.conv_post = weight_norm(nn.Conv2d(1024, 1, (3, 1), 1, (1, 0))) def forward( self, x: torch.Tensor, flg_san_train: bool = False ) -> tuple[ Union[torch.Tensor, tuple[torch.Tensor, torch.Tensor]], list[torch.Tensor] ]: fmap = [] b, c, t = x.shape if t % self.period != 0: n_pad = self.period - (t % self.period) x = F.pad(x, (0, n_pad), "reflect") t = t + n_pad x = x.view(b, c, t // self.period, self.period) for conv in self.convs: x = conv(x) x = F.silu(x, inplace=True) fmap.append(x) if self.san: x = self.conv_post(x, flg_san_train=flg_san_train) else: x = self.conv_post(x) if flg_san_train: x_fun, x_dir = x fmap.append(x_fun) x_fun = torch.flatten(x_fun, 1, -1) x_dir = torch.flatten(x_dir, 1, -1) x = x_fun, x_dir else: fmap.append(x) x = torch.flatten(x, 1, -1) return x, fmap class BeatriceDiscriminatorR(nn.Module): def __init__(self, resolution: int, san: bool = False): super().__init__() self.resolution = resolution self.san = san assert len(self.resolution) == 3 self.convs = nn.ModuleList( [ weight_norm(nn.Conv2d(1, 32, (3, 9), padding=(1, 4))), weight_norm(nn.Conv2d(32, 32, (3, 9), (1, 2), (1, 4))), weight_norm(nn.Conv2d(32, 32, (3, 9), (1, 2), (1, 4))), weight_norm(nn.Conv2d(32, 32, (3, 9), (1, 2), (1, 4))), weight_norm(nn.Conv2d(32, 32, (3, 3), padding=(1, 1))), ] ) if san: self.conv_post = SANConv2d(32, 1, (3, 3), padding=(1, 1)) else: self.conv_post = weight_norm(nn.Conv2d(32, 1, (3, 3), padding=(1, 1))) def forward( self, x: torch.Tensor, flg_san_train: bool = False ) -> tuple[ Union[torch.Tensor, tuple[torch.Tensor, torch.Tensor]], list[torch.Tensor] ]: fmap = [] x = self._spectrogram(x).unsqueeze(1) for conv in self.convs: x = conv(x) x = F.silu(x, inplace=True) fmap.append(x) if self.san: x = self.conv_post(x, flg_san_train=flg_san_train) else: x = self.conv_post(x) if flg_san_train: x_fun, x_dir = x fmap.append(x_fun) x_fun = torch.flatten(x_fun, 1, -1) x_dir = torch.flatten(x_dir, 1, -1) x = x_fun, x_dir else: fmap.append(x) x = torch.flatten(x, 1, -1) return x, fmap def _spectrogram(self, x: torch.Tensor) -> torch.Tensor: n_fft, hop_length, win_length = self.resolution x = F.pad( x, ((n_fft - hop_length) // 2, (n_fft - hop_length) // 2), mode="reflect" ).squeeze(1) with torch.amp.autocast("cuda" if torch.cuda.is_available() else "cpu", enabled=False): mag = torch.stft( x.float(), n_fft=n_fft, hop_length=hop_length, win_length=win_length, window=torch.ones(win_length, device=x.device), center=False, return_complex=True, ).abs() return mag class BeatriceMultiPeriodDiscriminator(nn.Module): def __init__(self, san: bool = False): super().__init__() resolutions = [[1024, 120, 600], [2048, 240, 1200], [512, 50, 240]] periods = [2, 3, 5, 7, 11] self.discriminators = nn.ModuleList( [BeatriceDiscriminatorR(r, san=san) for r in resolutions] + [BeatriceDiscriminatorP(p, san=san) for p in periods] ) self.discriminator_names = [f"R_{n}_{h}_{w}" for n, h, w in resolutions] + [ f"P_{p}" for p in periods ] self.san = san def forward( self, y: torch.Tensor, y_hat: torch.Tensor, flg_san_train: bool = False ) -> tuple[ list[Union[torch.Tensor, tuple[torch.Tensor, torch.Tensor]]], list[Union[torch.Tensor, tuple[torch.Tensor, torch.Tensor]]], list[list[torch.Tensor]], list[list[torch.Tensor]], ]: batch_size = y.size(0) concatenated_y_y_hat = torch.cat([y, y_hat]) y_d_rs = [] y_d_gs = [] fmap_rs = [] fmap_gs = [] for d in self.discriminators: if flg_san_train: (y_d_fun, y_d_dir), fmap = d( concatenated_y_y_hat, flg_san_train=flg_san_train ) y_d_r_fun, y_d_g_fun = torch.split(y_d_fun, batch_size) y_d_r_dir, y_d_g_dir = torch.split(y_d_dir, batch_size) y_d_r = y_d_r_fun, y_d_r_dir y_d_g = y_d_g_fun, y_d_g_dir else: y_d, fmap = d(concatenated_y_y_hat, flg_san_train=flg_san_train) y_d_r, y_d_g = torch.split(y_d, batch_size) fmap_r = [] fmap_g = [] for fm in fmap: fm_r, fm_g = torch.split(fm, batch_size) fmap_r.append(fm_r) fmap_g.append(fm_g) y_d_rs.append(y_d_r) y_d_gs.append(y_d_g) fmap_rs.append(fmap_r) fmap_gs.append(fmap_g) return y_d_rs, y_d_gs, fmap_rs, fmap_gs def forward_and_compute_loss( self, y: torch.Tensor, y_hat: torch.Tensor ) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, dict[str, float]]: y_d_rs, y_d_gs, fmap_rs, fmap_gs = self(y, y_hat, flg_san_train=self.san) stats = {} assert len(y_d_gs) == len(y_d_rs) == len(self.discriminators) with torch.amp.autocast("cuda" if torch.cuda.is_available() else "cpu", enabled=False): # discriminator loss d_loss = 0.0 for dr, dg, name in zip(y_d_rs, y_d_gs, self.discriminator_names): if self.san: dr_fun, dr_dir = map(lambda x: x.float(), dr) dg_fun, dg_dir = map(lambda x: x.float(), dg) r_loss_fun = F.softplus(1.0 - dr_fun).square().mean() g_loss_fun = F.softplus(dg_fun).square().mean() r_loss_dir = F.softplus(1.0 - dr_dir).square().mean() g_loss_dir = -F.softplus(1.0 - dg_dir).square().mean() r_loss = r_loss_fun + r_loss_dir g_loss = g_loss_fun + g_loss_dir else: dr = dr.float() dg = dg.float() r_loss = (1.0 - dr).square().mean() g_loss = dg.square().mean() stats[f"{name}_dr_loss"] = r_loss.item() stats[f"{name}_dg_loss"] = g_loss.item() d_loss += r_loss + g_loss # adversarial loss adv_loss = 0.0 for dg, name in zip(y_d_gs, self.discriminator_names): if self.san: dg_fun = dg[0].float() g_loss = F.softplus(1.0 - dg_fun).square().mean() else: dg = dg.float() g_loss = (1.0 - dg).square().mean() stats[f"{name}_gg_loss"] = g_loss.item() adv_loss += g_loss # feature mathcing loss fm_loss = 0.0 for fr, fg, name in zip(fmap_rs, fmap_gs, self.discriminator_names): fm_loss_i = 0.0 for j, (r, g) in enumerate(zip(fr, fg)): fm_loss_ij = (r.detach().float() - g.float()).abs().mean() stats[f"~{name}_fm_loss_{j}"] = fm_loss_ij.item() fm_loss_i += fm_loss_ij stats[f"{name}_fm_loss"] = fm_loss_i.item() fm_loss += fm_loss_i return d_loss, adv_loss, fm_loss, stats class GradBalancer: """Adapted from https://github.com/facebookresearch/encodec/blob/main/encodec/balancer.py""" def __init__( self, weights: dict[str, float], rescale_grads: bool = True, total_norm: float = 1.0, ema_decay: float = 0.999, per_batch_item: bool = True, ): self.weights = weights self.per_batch_item = per_batch_item self.total_norm = total_norm self.ema_decay = ema_decay self.rescale_grads = rescale_grads self.ema_total: dict[str, float] = defaultdict(float) self.ema_fix: dict[str, float] = defaultdict(float) def backward( self, losses: dict[str, torch.Tensor], input: torch.Tensor, scaler: Optional[torch.amp.GradScaler] = None, skip_update_ema: bool = False, ) -> dict[str, float]: stats = {} if skip_update_ema: assert len(losses) == len(self.ema_total) ema_norms = {k: tot / self.ema_fix[k] for k, tot in self.ema_total.items()} else: # 各 loss に対して d loss / d input とそのノルムを計算する norms = {} grads = {} for name, loss in losses.items(): if scaler is not None: loss = scaler.scale(loss) (grad,) = torch.autograd.grad(loss, [input], retain_graph=True) if not grad.isfinite().all(): input.backward(grad) return {} grad = grad.detach() / (1.0 if scaler is None else scaler.get_scale()) if self.per_batch_item: dims = tuple(range(1, grad.dim())) ema_norm = grad.norm(dim=dims).mean() else: ema_norm = grad.norm() norms[name] = float(ema_norm) grads[name] = grad # ノルムの移動平均を計算する for key, value in norms.items(): self.ema_total[key] = self.ema_total[key] * self.ema_decay + value self.ema_fix[key] = self.ema_fix[key] * self.ema_decay + 1.0 ema_norms = {k: tot / self.ema_fix[k] for k, tot in self.ema_total.items()} # ログを取る total_ema_norm = sum(ema_norms.values()) for k, ema_norm in ema_norms.items(): stats[f"grad_norm_value_{k}"] = ema_norm stats[f"grad_norm_ratio_{k}"] = ema_norm / (total_ema_norm + 1e-12) # loss の係数の比率を計算する if self.rescale_grads: total_weights = sum([self.weights[k] for k in ema_norms]) ratios = {k: w / total_weights for k, w in self.weights.items()} # 勾配を修正する loss = 0.0 for name, ema_norm in ema_norms.items(): if self.rescale_grads: scale = ratios[name] * self.total_norm / (ema_norm + 1e-12) else: scale = self.weights[name] loss += (losses if skip_update_ema else grads)[name] * scale if scaler is not None: loss = scaler.scale(loss) if skip_update_ema: (loss,) = torch.autograd.grad(loss, [input]) input.backward(loss) return stats def state_dict(self) -> dict[str, dict[str, float]]: return { "ema_total": dict(self.ema_total), "ema_fix": dict(self.ema_fix), } def load_state_dict(self, state_dict): self.ema_total = defaultdict(float, state_dict["ema_total"]) self.ema_fix = defaultdict(float, state_dict["ema_fix"]) class QualityTester(nn.Module): def __init__(self): super().__init__() self.utmos = torch.hub.load( "tarepan/SpeechMOS:v1.0.0", "utmos22_strong", trust_repo=True ).eval() @torch.inference_mode() def compute_mos(self, wav: torch.Tensor) -> dict[str, list[float]]: res = {"utmos": self.utmos(wav, sr=16000).tolist()} return res def test( self, converted_wav: torch.Tensor, source_wav: torch.Tensor ) -> dict[str, list[float]]: # [batch_size, wav_length] res = {} res.update(self.compute_mos(converted_wav)) return res def test_many( self, converted_wavs: list[torch.Tensor], source_wavs: list[torch.Tensor] ) -> tuple[dict[str, float], dict[str, list[float]]]: # list[batch_size, wav_length] results = defaultdict(list) assert len(converted_wavs) == len(source_wavs) for converted_wav, source_wav in zip(converted_wavs, source_wavs): res = self.test(converted_wav, source_wav) for metric_name, value in res.items(): results[metric_name].extend(value) return { metric_name: sum(values) / len(values) for metric_name, values in results.items() }, results def compute_grad_norm( model: nn.Module, return_stats: bool = False ) -> Union[float, dict[str, float]]: total_norm = 0.0 stats = {} for name, p in model.named_parameters(): if p.grad is None: continue param_norm = p.grad.data.norm().item() if not math.isfinite(param_norm): param_norm = p.grad.data.float().norm().item() total_norm += param_norm * param_norm if return_stats: stats[f"grad_norm_{name}"] = param_norm total_norm = math.sqrt(total_norm) if return_stats: return total_norm, stats else: return total_norm def compute_mean_f0( files: list[Path], method: Literal["dio", "harvest"] = "dio" ) -> float: sum_log_f0 = 0.0 n_frames = 0 for file in files: wav, sr = beatrice_load_audio(file) if method == "dio": f0, _ = pyworld.dio(wav.ravel().numpy().astype(np.float64), sr) elif method == "harvest": f0, _ = pyworld.harvest(wav.ravel().numpy().astype(np.float64), sr) else: raise ValueError(f"Invalid method: {method}") f0 = f0[f0 > 0] sum_log_f0 += float(np.log(f0).sum()) n_frames += len(f0) if n_frames == 0: return math.nan mean_log_f0 = sum_log_f0 / n_frames return math.exp(mean_log_f0) def get_resampler( sr_before: int, sr_after: int, device="cpu", cache={} ) -> torchaudio.transforms.Resample: if not isinstance(device, str): device = str(device) if (sr_before, sr_after, device) not in cache: cache[(sr_before, sr_after, device)] = torchaudio.transforms.Resample( sr_before, sr_after ).to(device) return cache[(sr_before, sr_after, device)] def convolve(signal: torch.Tensor, ir: torch.Tensor) -> torch.Tensor: n = 1 << (signal.size(-1) + ir.size(-1) - 2).bit_length() res = torch.fft.irfft(torch.fft.rfft(signal, n=n) * torch.fft.rfft(ir, n=n), n=n) return res[..., : signal.size(-1)] def random_formant_shift( wav: torch.Tensor, sample_rate: int, formant_shift_semitone_min: float = -3.0, formant_shift_semitone_max: float = 3.0, ) -> torch.Tensor: assert wav.ndim == 2 assert wav.size(0) == 1 device = wav.device hop_length = 256 # [wav_length] wav_np = wav.ravel().double().cpu().numpy() f0, t = pyworld.dio( wav_np, sample_rate, f0_floor=55, f0_ceil=1400, frame_period=hop_length * 1000 / sample_rate, ) f0 = pyworld.stonemask(wav_np, f0, t, sample_rate) world_sp = pyworld.cheaptrick(wav_np, f0, t, sample_rate) world_sp = ( torch.from_numpy(world_sp).float().to(device).sqrt_()[None] ) # [1, length, n_fft // 2 + 1] n_fft = win_length = (world_sp.size(2) - 1) * 2 window = torch.hann_window(win_length, device=device) # [1, n_fft // 2 + 1, length] stft_sp = torch.stft( wav, n_fft=n_fft, hop_length=hop_length, win_length=win_length, window=window, return_complex=True, ) assert world_sp.size(1) == stft_sp.size(2), (world_sp.size(), stft_sp.size()) assert world_sp.size(2) == stft_sp.size(1), (world_sp.size(), stft_sp.size()) shift_semitones = ( torch.rand(()).item() * (formant_shift_semitone_max - formant_shift_semitone_min) + formant_shift_semitone_min ) shift_ratio = 2.0 ** (shift_semitones / 12.0) shifted_world_sp = F.interpolate( world_sp, scale_factor=shift_ratio, mode="linear", align_corners=True ) if shifted_world_sp.size(2) > n_fft // 2 + 1: shifted_world_sp = shifted_world_sp[:, :, : n_fft // 2 + 1] elif shifted_world_sp.size(2) < n_fft // 2 + 1: shifted_world_sp = F.pad( shifted_world_sp, (0, n_fft // 2 + 1 - shifted_world_sp.size(2)) ) ratio = ((shifted_world_sp + 1e-5) / (world_sp + 1e-5)).clamp(0.1, 10.0) stft_sp *= ratio.transpose(-2, -1) # [1, n_fft // 2 + 1, length] out = torch.istft( stft_sp, n_fft=n_fft, hop_length=hop_length, win_length=win_length, window=window, length=wav.size(-1), ) return out def random_filter(audio: torch.Tensor) -> torch.Tensor: assert audio.ndim == 2 ab = torch.rand(audio.size(0), 6) * 0.75 - 0.375 a, b = ab[:, :3], ab[:, 3:] a[:, 0] = 1.0 b[:, 0] = 1.0 audio = torchaudio.functional.lfilter(audio, a, b, clamp=False) return audio def get_noise( n_samples: int, sample_rate: float, files: list[Union[str, bytes, os.PathLike]] ) -> torch.Tensor: resample_augmentation_candidates = [0.9, 0.95, 1.0, 1.05, 1.1] wavs = [] current_length = 0 while current_length < n_samples: idx_files = torch.randint(0, len(files), ()) file = files[idx_files] wav, sr = beatrice_load_audio(file) assert wav.size(0) == 1 augmented_sample_rate = int( round( sample_rate * resample_augmentation_candidates[ torch.randint(0, len(resample_augmentation_candidates), ()) ] ) ) resampler = get_resampler(sr, augmented_sample_rate) wav = resampler(wav) wav = random_filter(wav) wav *= 0.99 / (wav.abs().max() + 1e-5) wavs.append(wav) current_length += wav.size(1) start = torch.randint(0, current_length - n_samples + 1, ()) wav = torch.cat(wavs, dim=1)[:, start : start + n_samples] assert wav.size() == (1, n_samples), wav.size() return wav def get_butterworth_lpf( cutoff_freq: float, sample_rate: int, cache={} ) -> tuple[torch.Tensor, torch.Tensor]: if (cutoff_freq, sample_rate) not in cache: q = math.sqrt(0.5) omega = math.tau * cutoff_freq / sample_rate cos_omega = math.cos(omega) alpha = math.sin(omega) / (2.0 * q) b1 = (1.0 - cos_omega) / (1.0 + alpha) b0 = b1 * 0.5 a1 = -2.0 * cos_omega / (1.0 + alpha) a2 = (1.0 - alpha) / (1.0 + alpha) cache[(cutoff_freq, sample_rate)] = ( torch.tensor([b0, b1, b0]), torch.tensor([1.0, a1, a2]), ) return cache[(cutoff_freq, sample_rate)] def augment_audio( clean: torch.Tensor, sample_rate: int, noise_files: list[Union[str, bytes, os.PathLike]], ir_files: list[Union[str, bytes, os.PathLike]], snr_candidates: list[float] = [20.0, 25.0, 30.0, 35.0, 40.0, 45.0], formant_shift_probability: float = 0.5, formant_shift_semitone_min: float = -3.0, formant_shift_semitone_max: float = 3.0, reverb_probability: float = 0.5, lpf_probability: float = 0.2, lpf_cutoff_freq_candidates: list[float] = [2000.0, 3000.0, 4000.0, 6000.0], ) -> torch.Tensor: # [1, wav_length] assert clean.size(0) == 1 n_samples = clean.size(1) original_clean_rms = clean.square().mean().sqrt_() # clean をフォルマントシフトする if torch.rand(()) < formant_shift_probability: clean = random_formant_shift( clean, sample_rate, formant_shift_semitone_min, formant_shift_semitone_max ) # noise を取得して clean と concat する noise = get_noise(n_samples, sample_rate, noise_files) signals = torch.cat([clean, noise]) # clean, noise に異なるランダムフィルタをかける signals = random_filter(signals) # clean, noise にリバーブをかける if torch.rand(()) < reverb_probability: ir_file = ir_files[torch.randint(0, len(ir_files), ())] ir, sr = beatrice_load_audio(ir_file) assert ir.size() == (2, sr), ir.size() assert sr == sample_rate, (sr, sample_rate) signals = convolve(signals, ir) # clean, noise に同じ LPF をかける if torch.rand(()) < lpf_probability: if signals.abs().max() > 0.8: signals /= signals.abs().max() * 1.25 cutoff_freq = lpf_cutoff_freq_candidates[ torch.randint(0, len(lpf_cutoff_freq_candidates), ()) ] b, a = get_butterworth_lpf(cutoff_freq, sample_rate) signals = torchaudio.functional.lfilter(signals, a, b, clamp=False) # clean の音量を合わせる clean, noise = signals clean_rms = clean.square().mean().sqrt_() clean *= original_clean_rms / clean_rms if len(snr_candidates) >= 1: # clean, noise の音量をピークを重視して取る clean_level = clean.square().square_().mean().sqrt_().sqrt_() noise_level = noise.square().square_().mean().sqrt_().sqrt_() # SNR snr = snr_candidates[torch.randint(0, len(snr_candidates), ())] # noisy を生成 noisy = clean + noise * ( 0.1 ** (snr / 20.0) * clean_level / (noise_level + 1e-5) ) return noisy class WavDataset(torch.utils.data.Dataset): def __init__( self, audio_files: list[tuple[Path, int]], in_sample_rate: int = 16000, out_sample_rate: int = 24000, wav_length: int = 4 * 24000, # 4s segment_length: int = 100, # 1s noise_files: Optional[list[Union[str, bytes, os.PathLike]]] = None, ir_files: Optional[list[Union[str, bytes, os.PathLike]]] = None, augmentation_snr_candidates: list[float] = [20.0, 25.0, 30.0, 35.0, 40.0, 45.0], augmentation_formant_shift_probability: float = 0.5, augmentation_formant_shift_semitone_min: float = -3.0, augmentation_formant_shift_semitone_max: float = 3.0, augmentation_reverb_probability: float = 0.5, augmentation_lpf_probability: float = 0.2, augmentation_lpf_cutoff_freq_candidates: list[float] = [ 2000.0, 3000.0, 4000.0, 6000.0, ], ): self.audio_files = audio_files self.in_sample_rate = in_sample_rate self.out_sample_rate = out_sample_rate self.wav_length = wav_length self.segment_length = segment_length self.noise_files = noise_files self.ir_files = ir_files self.augmentation_snr_candidates = augmentation_snr_candidates self.augmentation_formant_shift_probability = ( augmentation_formant_shift_probability ) self.augmentation_formant_shift_semitone_min = ( augmentation_formant_shift_semitone_min ) self.augmentation_formant_shift_semitone_max = ( augmentation_formant_shift_semitone_max ) self.augmentation_reverb_probability = augmentation_reverb_probability self.augmentation_lpf_probability = augmentation_lpf_probability self.augmentation_lpf_cutoff_freq_candidates = ( augmentation_lpf_cutoff_freq_candidates ) if (noise_files is None) is not (ir_files is None): raise ValueError("noise_files and ir_files must be both None or not None") self.in_hop_length = in_sample_rate // 100 self.out_hop_length = out_sample_rate // 100 # 10ms 刻み def __getitem__(self, index: int) -> tuple[torch.Tensor, torch.Tensor, int, int]: file, speaker_id = self.audio_files[index] clean_wav, sample_rate = beatrice_load_audio(file) if clean_wav.size(0) != 1: ch = torch.randint(0, clean_wav.size(0), ()) clean_wav = clean_wav[ch : ch + 1] formant_shift_candidates = [-2.0, -1.5, -1.0, -0.5, 0.0, 0.5, 1.0, 1.5, 2.0] formant_shift = formant_shift_candidates[ torch.randint(0, len(formant_shift_candidates), ()).item() ] resampler_fraction = Fraction( sample_rate / self.out_sample_rate * 2.0 ** (formant_shift / 12.0) ).limit_denominator(300) clean_wav = get_resampler( resampler_fraction.numerator, resampler_fraction.denominator )(clean_wav) assert clean_wav.size(0) == 1 assert clean_wav.size(1) != 0 clean_wav = F.pad(clean_wav, (self.wav_length, self.wav_length)) if self.noise_files is None: noisy_wav_16k = get_resampler(self.out_sample_rate, self.in_sample_rate)( clean_wav ) else: clean_wav_16k = get_resampler(self.out_sample_rate, self.in_sample_rate)( clean_wav ) noisy_wav_16k = augment_audio( clean_wav_16k, self.in_sample_rate, self.noise_files, self.ir_files, self.augmentation_snr_candidates, self.augmentation_formant_shift_probability, self.augmentation_formant_shift_semitone_min, self.augmentation_formant_shift_semitone_max, self.augmentation_reverb_probability, self.augmentation_lpf_probability, self.augmentation_lpf_cutoff_freq_candidates, ) clean_wav = clean_wav.squeeze_(0) noisy_wav_16k = noisy_wav_16k.squeeze_(0) # 音量をランダマイズする amplitude = torch.rand(()).item() * 0.899 + 0.1 factor = amplitude / clean_wav.abs().max() clean_wav *= factor noisy_wav_16k *= factor while noisy_wav_16k.abs().max() >= 1.0: clean_wav *= 0.5 noisy_wav_16k *= 0.5 return clean_wav, noisy_wav_16k, speaker_id, formant_shift def __len__(self) -> int: return len(self.audio_files) def collate( self, batch: list[tuple[torch.Tensor, torch.Tensor, int, int]] ) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor]: assert self.wav_length % self.out_hop_length == 0 length = self.wav_length // self.out_hop_length clean_wavs = [] noisy_wavs = [] slice_starts = [] speaker_ids = [] formant_shifts = [] for clean_wav, noisy_wav, speaker_id, formant_shift in batch: # 発声部分をランダムに 1 箇所選ぶ (voiced,) = clean_wav.nonzero(as_tuple=True) assert voiced.numel() != 0 center = voiced[torch.randint(0, voiced.numel(), ()).item()].item() # 発声部分が中央にくるように、スライス区間を選ぶ slice_start = center - self.segment_length * self.out_hop_length // 2 assert slice_start >= 0 # スライス区間が含まれるように、ランダムに wav_length の長さを切り出す r = torch.randint(0, length - self.segment_length + 1, ()).item() offset = slice_start - r * self.out_hop_length clean_wavs.append(clean_wav[offset : offset + self.wav_length]) offset_in_sample_rate = int( round(offset * self.in_sample_rate / self.out_sample_rate) ) noisy_wavs.append( noisy_wav[ offset_in_sample_rate : offset_in_sample_rate + length * self.in_hop_length ] ) slice_start = r slice_starts.append(slice_start) speaker_ids.append(speaker_id) formant_shifts.append(formant_shift) clean_wavs = torch.stack(clean_wavs) noisy_wavs = torch.stack(noisy_wavs) slice_starts = torch.tensor(slice_starts) speaker_ids = torch.tensor(speaker_ids) formant_shifts = torch.tensor(formant_shifts) return ( clean_wavs, # [batch_size, wav_length] noisy_wavs, # [batch_size, wav_length] slice_starts, # Long[batch_size] speaker_ids, # Long[batch_size] formant_shifts, # Long[batch_size] ) AUDIO_FILE_SUFFIXES = { ".wav", ".aif", ".aiff", ".fla", ".flac", ".oga", ".ogg", ".opus", ".mp3", } def get_compressed_optimizer_state_dict( optimizer: torch.optim.Optimizer, ) -> dict: state_dict = {} for k0, v0 in optimizer.state_dict().items(): if k0 != "state": state_dict[k0] = v0 continue state_dict[k0] = {} for k1, v1 in v0.items(): state_dict[k0][k1] = {} for k2, v2 in v1.items(): if isinstance(v2, torch.Tensor): state_dict[k0][k1][k2] = v2.bfloat16() assert state_dict[k0][k1][k2].isfinite().all() else: state_dict[k0][k1][k2] = v2 return state_dict def get_decompressed_optimizer_state_dict(compressed_state_dict: dict) -> dict: state_dict = {} for k0, v0 in compressed_state_dict.items(): if k0 != "state": state_dict[k0] = v0 continue state_dict[k0] = {} for k1, v1 in v0.items(): state_dict[k0][k1] = {} for k2, v2 in v1.items(): if isinstance(v2, torch.Tensor): state_dict[k0][k1][k2] = v2.float() assert state_dict[k0][k1][k2].isfinite().all() else: state_dict[k0][k1][k2] = v2 return state_dict # ============================================================ # BEATRICE V2 TRAINING - Embedded (downloads assets from HuggingFace) # ============================================================ BEATRICE_AUDIO_FILE_SUFFIXES = {".wav", ".aif", ".aiff", ".fla", ".flac", ".oga", ".ogg", ".opus", ".mp3"} def preprocess_audio_for_beatrice(audio_path: str, output_dir: str, speaker_name: str = "speaker"): """Preprocess audio for Beatrice training using silence-based splitting""" # Create speaker directory structure required by Beatrice speaker_dir = os.path.join(output_dir, speaker_name) os.makedirs(speaker_dir, exist_ok=True) # Load audio at 16kHz (Beatrice input sample rate) audio, sr = librosa.load(audio_path, sr=16000, mono=True) # Simple silence-based splitting (RMS threshold) chunk_size = int(4.0 * sr) # 4 second chunks hop = int(3.5 * sr) # 0.5s overlap threshold = 0.01 # RMS threshold chunks_saved = 0 for i, start in enumerate(range(0, len(audio) - chunk_size, hop)): chunk = audio[start:start + chunk_size] rms = np.sqrt(np.mean(chunk ** 2)) if rms > threshold: # Skip silence # Normalize max_val = np.abs(chunk).max() if max_val > 0: chunk = chunk / max_val * 0.9 chunk_path = os.path.join(speaker_dir, f"{speaker_name}_{chunks_saved:04d}.wav") sf.write(chunk_path, chunk, sr) chunks_saved += 1 logger.info(f"Beatrice preprocessing: {chunks_saved} chunks saved to {speaker_dir}") return chunks_saved, output_dir def train_beatrice_generator( data_dir: str, output_dir: str, epochs: int = 30, batch_size: int = 8, lr_g: float = 5e-5, lr_d: float = 5e-5, use_augmentation: bool = False, resume: bool = False, progress_callback=None, ): """Train Beatrice v2 model - generator yielding (message, model_path) tuples""" device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # Download pretrained yield "Downloading pretrained models...", None phone_extractor_path = download_beatrice_asset("phone_extractor") pitch_estimator_path = download_beatrice_asset("pitch_estimator") pretrained_model_path = download_beatrice_asset("pretrained_model") # Discover speakers from directory structure # Expected: data_dir/speaker_name/*.wav speakers = [] training_filelist = [] speaker_audio_files = [] for speaker_dir in sorted(Path(data_dir).iterdir()): if not speaker_dir.is_dir(): continue candidates = [f for f in sorted(speaker_dir.rglob("*")) if f.is_file() and f.suffix.lower() in BEATRICE_AUDIO_FILE_SUFFIXES] if candidates: speaker_id = len(speakers) speakers.append(speaker_dir.name) training_filelist.extend([(f, speaker_id) for f in candidates]) speaker_audio_files.append(candidates) n_speakers = len(speakers) if n_speakers == 0: yield "Error: No speakers found in data directory", None return yield f"Found {n_speakers} speaker(s), {len(training_filelist)} files", None # Augmentation assets (optional) noise_files = None ir_files = None if use_augmentation: try: noise_dir, ir_dir = download_beatrice_augmentation() if noise_dir and ir_dir: noise_files = sorted(list(Path(noise_dir).rglob("*.wav")) + list(Path(noise_dir).rglob("*.flac"))) ir_files = sorted(list(Path(ir_dir).rglob("*.wav")) + list(Path(ir_dir).rglob("*.flac"))) if noise_files and ir_files: yield f"Loaded augmentation: {len(noise_files)} noise, {len(ir_files)} IR files", None else: noise_files = None ir_files = None except Exception as e: yield f"Warning: Could not load augmentation assets: {e}", None # Build models yield "Building models...", None phone_extractor = PhoneExtractor().to(device).eval().requires_grad_(False) pe_ckpt = torch.load(phone_extractor_path, map_location="cpu", weights_only=True) phone_extractor.load_state_dict(pe_ckpt["phone_extractor"], strict=False) del pe_ckpt pitch_estimator = PitchEstimator().to(device).eval().requires_grad_(False) pi_ckpt = torch.load(pitch_estimator_path, map_location="cpu", weights_only=True) pitch_estimator.load_state_dict(pi_ckpt["pitch_estimator"]) del pi_ckpt hidden_channels = 256 pitch_bins = 448 net_g = ConverterNetwork( phone_extractor, pitch_estimator, n_speakers=n_speakers, pitch_bins=pitch_bins, hidden_channels=hidden_channels, vq_topk=4, training_time_vq="none", phone_noise_ratio=0.5, floor_noise_level=1e-3, ).to(device) net_d = BeatriceMultiPeriodDiscriminator(san=True).to(device) # Optimizers optim_g = torch.optim.AdamW(net_g.parameters(), lr_g, betas=(0.8, 0.99), eps=1e-6) optim_d = torch.optim.AdamW(net_d.parameters(), lr_d, betas=(0.8, 0.99), eps=1e-6) grad_scaler = torch.amp.GradScaler(device.type, enabled=device.type == "cuda") grad_balancer = GradBalancer( weights={ "loss_loudness": 1.0, "loss_mel": 45.0, "loss_adv": 1.0, "loss_fm": 2.0, }, ema_decay=0.999, ) initial_iteration = 0 os.makedirs(output_dir, exist_ok=True) # Load pretrained or resume if resume: latest_ckpt = os.path.join(output_dir, "checkpoint_latest.pt.gz") if os.path.isfile(latest_ckpt): yield "Resuming from checkpoint...", None with gzip.open(latest_ckpt, "rb") as f: ckpt = torch.load(f, map_location="cpu", weights_only=True) net_g.load_state_dict(ckpt["net_g"], strict=False) # Filter discriminator for shape mismatches net_d_state = net_d.state_dict() filtered_d = {k: v for k, v in ckpt["net_d"].items() if k in net_d_state and v.shape == net_d_state[k].shape} net_d.load_state_dict(filtered_d, strict=False) optim_g.load_state_dict(get_decompressed_optimizer_state_dict(ckpt["optim_g"])) optim_d.load_state_dict(get_decompressed_optimizer_state_dict(ckpt["optim_d"])) if "grad_balancer" in ckpt: grad_balancer.load_state_dict(ckpt["grad_balancer"]) if "grad_scaler" in ckpt: grad_scaler.load_state_dict(ckpt["grad_scaler"]) initial_iteration = ckpt.get("iteration", 0) del ckpt else: yield "No checkpoint found, starting fresh with pretrained", None resume = False if not resume: yield "Loading pretrained weights...", None with gzip.open(pretrained_model_path, "rb") as f: pretrained_ckpt = torch.load(f, map_location="cpu", weights_only=True) # Adapt pretrained for our n_speakers initial_speaker_emb = pretrained_ckpt["net_g"]["embed_speaker.weight"][:1] pretrained_ckpt["net_g"]["embed_speaker.weight"] = initial_speaker_emb[[0] * n_speakers] initial_kv_emb = pretrained_ckpt["net_g"]["key_value_speaker_embedding.weight"][:1] pretrained_ckpt["net_g"]["key_value_speaker_embedding.weight"] = initial_kv_emb[[0] * n_speakers] pretrained_ckpt["net_g"]["vq.codebooks"] = pretrained_ckpt["net_g"]["vq.codebooks"][[0] * n_speakers] net_g.load_state_dict(pretrained_ckpt["net_g"], strict=False) # Filter discriminator state dict for shape mismatches (pretrained may use san=False) net_d_state = net_d.state_dict() filtered_d = {k: v for k, v in pretrained_ckpt["net_d"].items() if k in net_d_state and v.shape == net_d_state[k].shape} net_d.load_state_dict(filtered_d, strict=False) logger.info(f"Loaded {len(filtered_d)}/{len(pretrained_ckpt['net_d'])} discriminator weights") # Don't load grad_balancer/grad_scaler from pretrained - our loss weights may differ # These will be re-initialized fresh for fine-tuning del pretrained_ckpt # Build VQ codebooks yield "Building VQ codebooks...", None def wav_iterator(files): for file in files: wav, sr = beatrice_load_audio(file) wav = wav.to(device) if sr != 16000: wav = get_resampler(sr, 16000, str(device))(wav) yield wav[:, None, :] if resume: net_g.enable_hook() else: net_g.initialize_vq([wav_iterator(files) for files in speaker_audio_files]) # Dataset dataset = WavDataset( training_filelist, in_sample_rate=16000, out_sample_rate=24000, wav_length=96000, segment_length=100, noise_files=noise_files, ir_files=ir_files, ) _num_workers = min(4, os.cpu_count() or 1) effective_batch = min(batch_size, len(training_filelist)) dataloader = torch.utils.data.DataLoader( dataset, num_workers=_num_workers, collate_fn=dataset.collate, shuffle=True, batch_size=effective_batch, pin_memory=True, drop_last=len(training_filelist) > effective_batch, persistent_workers=_num_workers > 0, ) # Calculate steps steps_per_epoch = max(1, len(training_filelist) // batch_size) total_steps = epochs * steps_per_epoch warmup_steps = min(total_steps // 4, 5000) # LR scheduler with warmup def lr_lambda(step): if step < warmup_steps: return step / max(1, warmup_steps) return 0.999 ** (step - warmup_steps) scheduler_g = torch.optim.lr_scheduler.LambdaLR(optim_g, lr_lambda) scheduler_d = torch.optim.lr_scheduler.LambdaLR(optim_d, lr_lambda) # Advance schedulers if resuming with warnings.catch_warnings(): warnings.filterwarnings("ignore", message=r"Detected call of `lr_scheduler\.step\(\)") for _ in range(initial_iteration + 1): scheduler_g.step() scheduler_d.step() net_g.train() net_d.train() yield f"Training {total_steps} steps ({epochs} epochs x {steps_per_epoch} steps/epoch)", None # Training loop step = initial_iteration data_iter = None ckpt_path = None for epoch in range(epochs): epoch_loss_g = 0.0 epoch_loss_d = 0.0 epoch_steps = 0 for batch_idx in range(steps_per_epoch): if data_iter is None: data_iter = iter(dataloader) batch = next(data_iter, None) if batch is None: data_iter = iter(dataloader) batch = next(data_iter, None) if batch is None: break clean_wavs, noisy_wavs_16k, slice_starts, speaker_ids, formant_shifts = \ [x.to(device, non_blocking=True) for x in batch] with torch.amp.autocast(device.type, enabled=device.type == "cuda"): # Generator forward y, y_hat, y_hat_for_backward, loss_loudness, loss_mel, loss_ap, gen_stats = \ net_g.forward_and_compute_loss( noisy_wavs_16k[:, None, :], speaker_ids, formant_shifts, slice_start_indices=slice_starts, slice_segment_length=100, y_all=clean_wavs[:, None, :], ) # Discriminator forward loss_disc, loss_adv, loss_fm, disc_stats = \ net_d.forward_and_compute_loss(y, y_hat) # Discriminator backward optim_d.zero_grad(set_to_none=True) grad_scaler.scale(loss_disc).backward(retain_graph=True, inputs=list(net_d.parameters())) grad_scaler.unscale_(optim_d) # Generator backward optim_g.zero_grad(set_to_none=True) grad_balancer.backward( {"loss_loudness": loss_loudness, "loss_mel": loss_mel, "loss_adv": loss_adv, "loss_fm": loss_fm}, y_hat_for_backward, grad_scaler, skip_update_ema=step > 10 and step % 5 != 0, ) grad_scaler.unscale_(optim_g) # Update grad_scaler.step(optim_g) grad_scaler.step(optim_d) grad_scaler.update() optim_g.zero_grad(set_to_none=True) optim_d.zero_grad(set_to_none=True) scheduler_g.step() scheduler_d.step() epoch_loss_g += loss_mel.item() epoch_loss_d += loss_disc.item() epoch_steps += 1 step += 1 if progress_callback: progress_callback(step / total_steps) avg_loss_g = epoch_loss_g / max(1, epoch_steps) avg_loss_d = epoch_loss_d / max(1, epoch_steps) yield f"Epoch {epoch+1}/{epochs} | G loss: {avg_loss_g:.4f} | D loss: {avg_loss_d:.4f} | LR: {scheduler_g.get_last_lr()[0]:.2e}", None # Save checkpoint periodically if (epoch + 1) % max(1, epochs // 5) == 0 or epoch == epochs - 1: ckpt_path = os.path.join(output_dir, f"checkpoint_{step:08d}.pt.gz") with gzip.open(ckpt_path, "wb") as f: torch.save({ "iteration": step, "net_g": net_g.state_dict(), "phone_extractor": phone_extractor.state_dict(), "pitch_estimator": pitch_estimator.state_dict(), "net_d": {k: v.half() for k, v in net_d.state_dict().items()}, "optim_g": get_compressed_optimizer_state_dict(optim_g), "optim_d": get_compressed_optimizer_state_dict(optim_d), "grad_balancer": grad_balancer.state_dict(), "grad_scaler": grad_scaler.state_dict(), "h": { "hidden_channels": hidden_channels, "pitch_bins": pitch_bins, "vq_topk": 4, "training_time_vq": "none", "phone_noise_ratio": 0.5, "floor_noise_level": 1e-3, "san": True, }, "speakers": speakers, }, f) shutil.copy(ckpt_path, os.path.join(output_dir, "checkpoint_latest.pt.gz")) yield f"Saved checkpoint: {ckpt_path}", ckpt_path # Cleanup del net_g, net_d, optim_g, optim_d, phone_extractor, pitch_estimator gc.collect() if torch.cuda.is_available(): torch.cuda.empty_cache() yield "Training complete!", ckpt_path def convert_voice_beatrice( source_audio, model_file, target_speaker: int = 0, pitch_shift: int = 0, formant_shift: float = 0.0, progress=None, ): """Convert voice using Beatrice v2 model Args: source_audio: Path to source audio file model_file: Path to Beatrice checkpoint (.pt.gz) or file object with .name target_speaker: Target speaker index pitch_shift: Pitch shift in semitones formant_shift: Formant shift in semitones (-2 to 2) progress: Gradio progress callback Returns: (output_path, status_message) tuple """ device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # Get model path if hasattr(model_file, 'name'): model_path = model_file.name elif isinstance(model_file, str): model_path = model_file else: return None, "Invalid model file" if not model_path or not os.path.exists(model_path): return None, f"Model file not found: {model_path}" try: if progress: progress(0.1, "Loading models...") # Download pretrained assets (phone extractor + pitch estimator) phone_extractor_path = download_beatrice_asset("phone_extractor") pitch_estimator_path = download_beatrice_asset("pitch_estimator") # Build phone extractor phone_extractor = PhoneExtractor().to(device).eval().requires_grad_(False) pe_ckpt = torch.load(phone_extractor_path, map_location="cpu", weights_only=True) phone_extractor.load_state_dict(pe_ckpt["phone_extractor"], strict=False) del pe_ckpt # Build pitch estimator pitch_estimator = PitchEstimator().to(device).eval().requires_grad_(False) pi_ckpt = torch.load(pitch_estimator_path, map_location="cpu", weights_only=True) pitch_estimator.load_state_dict(pi_ckpt["pitch_estimator"]) del pi_ckpt if progress: progress(0.3, "Loading trained model...") # Load trained checkpoint with gzip.open(model_path, "rb") as f: checkpoint = torch.load(f, map_location="cpu", weights_only=True) # Determine model params from checkpoint n_speakers = checkpoint["net_g"]["embed_speaker.weight"].shape[0] h = checkpoint.get("h", {}) hidden_channels = h.get("hidden_channels", 256) pitch_bins = h.get("pitch_bins", 448) speakers = checkpoint.get("speakers", [f"Speaker {i}" for i in range(n_speakers)]) if target_speaker >= n_speakers: target_speaker = 0 net_g = ConverterNetwork( phone_extractor, pitch_estimator, n_speakers=n_speakers, pitch_bins=pitch_bins, hidden_channels=hidden_channels, vq_topk=h.get("vq_topk", 4), training_time_vq=h.get("training_time_vq", "none"), phone_noise_ratio=h.get("phone_noise_ratio", 0.5), floor_noise_level=h.get("floor_noise_level", 1e-3), ).to(device).eval() net_g.load_state_dict(checkpoint["net_g"], strict=False) net_g.enable_hook() del checkpoint if progress: progress(0.5, "Converting voice...") # Load audio at 16kHz audio_path = source_audio if isinstance(source_audio, str) else source_audio audio, sr = librosa.load(audio_path, sr=16000, mono=True) audio_tensor = torch.from_numpy(audio).float().unsqueeze(0).unsqueeze(0).to(device) # Pad to multiple of 160 (phone extractor stride) original_length = audio_tensor.shape[-1] if original_length % 160 != 0: pad_len = 160 - original_length % 160 audio_tensor = F.pad(audio_tensor, (0, pad_len)) # Convert with torch.inference_mode(): y_hat = net_g( audio_tensor, torch.tensor([target_speaker], device=device), torch.tensor([formant_shift], device=device), torch.tensor([float(pitch_shift)], device=device), ) # Output is 24kHz, trim to match input duration output_length = original_length // 160 * 240 # 16kHz→24kHz frame ratio output = y_hat.squeeze().cpu().numpy()[:output_length] # Save fd, output_path = tempfile.mkstemp(suffix=".wav") os.close(fd) sf.write(output_path, output, 24000) # Cleanup del net_g, phone_extractor, pitch_estimator gc.collect() if torch.cuda.is_available(): torch.cuda.empty_cache() speaker_name = speakers[target_speaker] if target_speaker < len(speakers) else f"Speaker {target_speaker}" return output_path, f"Converted using Beatrice v2 | 24kHz | Speaker: {speaker_name} | Pitch: {pitch_shift:+d} | Formant: {formant_shift:+.1f}" except Exception as e: logger.exception("Beatrice inference error") return None, f"Error: {str(e)}" # ============================================================ # BACKGROUND TRAINING JOBS + TRACKING CODES # ============================================================ # Lets a user submit a training request, receive a short tracking code, # close/refresh the browser, and come back later to check status and # download the result by entering that code. Training keeps running in a # background thread on the server regardless of the browser connection. JOBS_DIR = "trained_models/_jobs" JOBS_FILE = f"{JOBS_DIR}/jobs.json" MAX_TRAIN_AUDIO_FILES = 100 # how many "add another file" upload slots the UI offers # Audio extensions recognized when auto-extracting a ZIP of training files _ZIP_AUDIO_EXTS = { ".wav", ".mp3", ".flac", ".ogg", ".m4a", ".aac", ".wma", ".opus", ".aiff", ".aif", ".mp4", } def _extract_audio_from_zip(zip_path): """Extract every recognized audio file from an uploaded ZIP (including files inside sub-folders) into a fresh temp directory. Returns (sorted_file_paths, error_message_or_None). Skips non-audio files, hidden/junk entries (e.g. __MACOSX, .DS_Store), and guards against zip-slip path traversal.""" extract_dir = tempfile.mkdtemp(prefix="train_zip_") found = [] try: with zipfile.ZipFile(zip_path, "r") as zf: for idx, info in enumerate(zf.infolist()): if info.is_dir(): continue name = info.filename base = os.path.basename(name) if not base or base.startswith(".") or "__MACOSX" in name: continue ext = os.path.splitext(base)[1].lower() if ext not in _ZIP_AUDIO_EXTS: continue # Safe destination path (avoid zip-slip); flatten folders and # prefix with an index so same-named files never collide. safe_base = re.sub(r"[^\w.\-]+", "_", base) dest_path = os.path.join(extract_dir, f"{idx:04d}_{safe_base}") with zf.open(info) as src, open(dest_path, "wb") as dst: shutil.copyfileobj(src, dst) found.append(dest_path) except zipfile.BadZipFile: return [], "❌ فایل ZIP معتبر نیست یا خراب است" except Exception as e: logger.exception("Failed extracting training-audio zip") return [], f"❌ خطا در استخراج فایل ZIP: {str(e)}" found.sort() return found, None def handle_train_zip_upload(zip_file): """Gradio handler for the training-audio ZIP uploader: extracts every audio file inside the ZIP and auto-fills the training audio slots (auto-detects the file count, no manual selection needed), up to MAX_TRAIN_AUDIO_FILES. Extra files beyond that cap are reported, not silently dropped.""" n = MAX_TRAIN_AUDIO_FILES if zip_file is None: return [gr.update()] * n + [gr.update(value="")] zip_path = zip_file if isinstance(zip_file, str) else getattr(zip_file, "name", zip_file) files, err = _extract_audio_from_zip(zip_path) if err: return [gr.update()] * n + [gr.update(value=err)] if not files: return [gr.update()] * n + [ gr.update(value="⚠️ هیچ فایل صوتی معتبری داخل ZIP پیدا نشد (فرمت‌های پشتیبانی‌شده: wav, mp3, flac, ogg, m4a, aac, wma, opus, aiff)") ] note = "" if len(files) > n: extra = len(files) - n files = files[:n] note = f" ⚠️ سقف {n} فایل است؛ {extra} فایل اضافی نادیده گرفته شد." updates = [] for i in range(n): if i < len(files): updates.append(gr.update(value=files[i], visible=True)) else: updates.append(gr.update()) updates.append(gr.update(value=f"✅ {len(files)} فایل صوتی از ZIP استخراج و به‌صورت خودکار اضافه شد.{note}")) return updates def _combine_audio_files(paths): """Concatenate multiple uploaded training audio files (in the order they were added) into a single temp wav file, so the existing single-file training/preprocessing pipeline can be reused without changes. A short silence gap is inserted between files to avoid an abrupt splice.""" sr_out = 44100 pieces = [] for p in paths: if not p: continue y, _ = librosa.load(p, sr=sr_out, mono=True) if len(pieces) > 0: pieces.append(np.zeros(int(0.2 * sr_out), dtype=y.dtype)) pieces.append(y) if not pieces: return None audio = np.concatenate(pieces) fd, out_path = tempfile.mkstemp(suffix=".wav") os.close(fd) sf.write(out_path, audio, sr_out) return out_path _jobs = {} _jobs_lock = threading.Lock() _job_queue = _queue_module.Queue() _worker_thread = None _worker_lock = threading.Lock() _current_job_code = None def _load_jobs(): global _jobs try: if os.path.exists(JOBS_FILE): with open(JOBS_FILE, "r", encoding="utf-8") as f: _jobs = json_module.load(f) except Exception: logger.exception("Failed to load jobs.json - starting fresh") _jobs = {} def _save_jobs(): try: os.makedirs(JOBS_DIR, exist_ok=True) with open(JOBS_FILE, "w", encoding="utf-8") as f: json_module.dump(_jobs, f, ensure_ascii=False, indent=2) except Exception: logger.exception("Failed to save jobs.json") def _new_tracking_code() -> str: """Generate a short, human-typeable tracking code, e.g. 4F9A2C.""" while True: code = secrets.token_hex(3).upper() with _jobs_lock: if code not in _jobs: return code def _update_job(code: str, **kwargs): with _jobs_lock: job = _jobs.setdefault(code, {}) job.update(kwargs) job["updated_at"] = time.time() _save_jobs() def _get_job(code: str) -> dict: with _jobs_lock: return dict(_jobs.get(code, {})) _load_jobs() class _JobProgress: """A gr.Progress-compatible callable that writes into the job store instead of pushing updates over a live browser connection.""" def __init__(self, code): self.code = code def __call__(self, frac, desc=""): try: _update_job(self.code, progress=float(frac)) except Exception: pass def tqdm(self, iterable, **kwargs): return iterable def _execute_training_job(code: str, params: dict): """Runs on the background worker thread. Consumes the same generator the UI used to stream live, but writes each update into the job store.""" _update_job(code, status="running", is_partial=True) try: gen = train_unified( params["trainer"], params["audio_path"], params["name"], params["rvc_epochs"], params["rvc_batch"], params["rvc_sr"], params["rvc_f0"], params["beat_epochs"], params["beat_batch"], params["beat_resume"], progress=_JobProgress(code), ) ckpt, idx = None, None for ck, ix, log_text in gen: updates = {"log": log_text} if ck: ckpt = ck # Expose the freshly-saved checkpoint right away (job stays # "running") so it can already be retrieved/tested by tracking # code before the whole training run finishes. updates["model_path"] = ck updates["is_partial"] = True if ix: idx = ix updates["index_path"] = ix _update_job(code, **updates) job = _get_job(code) if job.get("status") == "cancelled": pass # cancel handler already finalized the status elif ckpt: _update_job(code, status="done", model_path=ckpt, index_path=idx, progress=1.0, is_partial=False) else: _update_job(code, status="error") except Exception as e: logger.exception("Background training job failed") job = _get_job(code) _update_job(code, status="error", log=(job.get("log", "") + f"\n❌ خطای غیرمنتظره: {str(e)}")) def _job_worker_loop(): global _current_job_code while True: code = _job_queue.get() try: job = _get_job(code) if job.get("cancel_requested"): _update_job(code, status="cancelled", log=job.get("log", "") + "\n🛑 لغو شد (قبل از شروع)") continue _current_job_code = code _execute_training_job(code, job.get("params", {})) except Exception: logger.exception("Unhandled error in job worker loop") _update_job(code, status="error") finally: _current_job_code = None _job_queue.task_done() def _ensure_worker(): global _worker_thread with _worker_lock: if _worker_thread is None or not _worker_thread.is_alive(): _worker_thread = threading.Thread(target=_job_worker_loop, daemon=True) _worker_thread.start() def submit_training_request( trainer, name, *rest, ): """Validates the request, combines every uploaded training-audio slot into one file, copies it to a stable path, registers a background job, and immediately returns a tracking code - the browser does not need to stay open for training to proceed. `rest` = (audio_slot_1, ..., audio_slot_N, rvc_epochs, rvc_batch, rvc_sr, rvc_f0, beat_epochs, beat_batch, beat_resume) """ audio_slots = list(rest[:MAX_TRAIN_AUDIO_FILES]) (rvc_epochs, rvc_batch, rvc_sr, rvc_f0, beat_epochs, beat_batch, beat_resume) = rest[MAX_TRAIN_AUDIO_FILES:] audio_paths = [a for a in audio_slots if a] if not audio_paths: return "", "❌ لطفاً حداقل یک فایل صوتی برای آموزش بارگذاری کنید" if not name or not name.strip(): return "", "❌ لطفاً یک نام برای مدل وارد کنید" try: combined_path = _combine_audio_files(audio_paths) except Exception as e: logger.exception("Failed to combine training audio files") return "", f"❌ خطا در ترکیب فایل‌های صوتی: {str(e)}" if combined_path is None: return "", "❌ فایل صوتی معتبر پیدا نشد" duration, err = _check_audio_duration(combined_path) if err: try: os.remove(combined_path) except Exception: pass return "", err code = _new_tracking_code() job_dir = f"{JOBS_DIR}/{code}" try: os.makedirs(job_dir, exist_ok=True) stable_audio_path = os.path.abspath(f"{job_dir}/input.wav") shutil.copyfile(combined_path, stable_audio_path) except Exception as e: logger.exception("Failed to stage training audio") return "", f"❌ خطا در ذخیره فایل صوتی: {str(e)}" finally: try: os.remove(combined_path) except Exception: pass params = dict( trainer=trainer, audio_path=stable_audio_path, name=name, rvc_epochs=int(rvc_epochs), rvc_batch=int(rvc_batch), rvc_sr=int(rvc_sr), rvc_f0=rvc_f0, beat_epochs=int(beat_epochs), beat_batch=int(beat_batch), beat_resume=bool(beat_resume), ) _update_job( code, status="queued", log=f"⏳ درخواست شما ثبت شد و در صف پردازش قرار گرفت... ({len(audio_paths)} فایل صوتی ترکیب شد)", trainer=trainer, model_name=name, progress=0.0, cancel_requested=False, created_at=time.time(), model_path=None, index_path=None, params=params, ) _ensure_worker() _job_queue.put(code) msg = ( "✅ درخواست آموزش با موفقیت ثبت شد!\n\n" f"📎 تعداد فایل‌های صوتی ترکیب‌شده: {len(audio_paths)}\n" f"🔑 کد پیگیری شما: {code}\n\n" "این کد را حتماً جایی یادداشت یا کپی کنید.\n" "می‌توانید همین حالا این صفحه را ببندید یا رفرش کنید — آموزش در پس‌زمینه‌ی سرور ادامه پیدا می‌کند.\n" "برای مشاهده وضعیت و دانلود مدل بعداً، به تب «📡 پیگیری آموزش» بروید و این کد را وارد کنید." ) return code, msg _STATUS_LABELS_FA = { "queued": "⏳ در صف انتظار", "running": "🏃 در حال آموزش", "done": "✅ آموزش کامل شد", "error": "❌ خطا در آموزش", "cancelled": "🛑 لغو شده", } def check_tracking_status(code): code = (code or "").strip().upper() if not code: return "❌ لطفاً کد پیگیری را وارد کنید", 0.0, None, None job = _get_job(code) if not job: return "❌ کد پیگیری پیدا نشد. لطفاً کد را دوباره بررسی کنید", 0.0, None, None status = job.get("status", "unknown") status_fa = _STATUS_LABELS_FA.get(status, status) log = job.get("log", "") progress = float(job.get("progress") or 0.0) model_path = job.get("model_path") index_path = job.get("index_path") is_partial = bool(job.get("is_partial")) model_out = model_path if (model_path and os.path.exists(model_path)) else None index_out = index_path if (index_path and os.path.exists(index_path)) else None header = f"وضعیت: {status_fa}" if job.get("model_name"): header += f" | مدل: {job['model_name']}" if model_out and is_partial and status == "running": header += f"\n🔶 یک checkpoint میان‌کاری (تا همین لحظه، {progress*100:.0f}٪ پیشرفت) برای دانلود آماده است — می‌توانید همین حالا دانلودش کنید یا با کد پیگیری در تب Voice Conversion امتحانش کنید؛ آموزش هم‌زمان ادامه دارد." full_log = f"{header}\n{'─'*40}\n{log}" return full_log, progress, model_out, index_out def cancel_tracking_job(code): code = (code or "").strip().upper() if not code: return "❌ لطفاً کد پیگیری را وارد کنید" job = _get_job(code) if not job: return "❌ کد پیگیری پیدا نشد" status = job.get("status") if status == "queued": _update_job(code, cancel_requested=True, status="cancelled") return "🛑 درخواست از صف حذف شد" elif status == "running": global _training_cancel if _current_job_code == code: _training_cancel = True return "🛑 درخواست لغو ثبت شد؛ پس از پایان اپوک جاری، آموزش متوقف شده و چک‌پوینت ذخیره می‌شود" return "⚠️ این آموزش هم‌اکنون در حال اجرا نیست" else: return f"ℹ️ این آموزش در وضعیت «{_STATUS_LABELS_FA.get(status, status)}» است و قابل لغو نیست" # ============================================================ # VOICE-CONVERSION TAB: LOAD A TRAINED MODEL WITHOUT RE-UPLOADING # (by tracking code, or by a direct URL to a model/zip) — avoids # spending the user's upload bandwidth for models already on the server. # ============================================================ def retrieve_model_by_tracking_code(code): """Looks up a training job by its tracking code and loads its model (and index, for RVC) straight into the Voice Conversion inputs — no re-upload needed, since the file already lives on the server from training. Works both for a finished job and for one still in progress: if training is still running, the latest mid-training checkpoint (e.g. epoch 600 of a 1000-epoch run) is retrieved instead, so it can be downloaded or test-converted before the run finishes.""" no_change = gr.update() code = (code or "").strip().upper() if not code: return None, None, None, no_change, "❌ لطفاً کد پیگیری را وارد کنید" job = _get_job(code) if not job: return None, None, None, no_change, "❌ کد پیگیری پیدا نشد" status = job.get("status") model_path = job.get("model_path") if status not in ("done", "running"): status_fa = _STATUS_LABELS_FA.get(status, status or "نامشخص") return None, None, None, no_change, f"⚠️ این آموزش مدل قابل بازیابی ندارد (وضعیت: {status_fa})" if status == "running" and not model_path: progress = float(job.get("progress") or 0.0) return None, None, None, no_change, ( f"⏳ آموزش هنوز در جریان است ({progress*100:.0f}٪) و اولین checkpoint هنوز ذخیره نشده — " "کمی بعد دوباره امتحان کنید." ) index_path = job.get("index_path") trainer = job.get("trainer") or "RVC v2" if not model_path or not os.path.exists(model_path): return None, None, None, no_change, "❌ فایل مدل این کد پیگیری دیگر روی سرور پیدا نشد (شاید سرور ری‌استارت شده باشد)" try: # Gradio needs the file inside an allowed dir (cwd/tmp) — copy it there. temp_dir = tempfile.mkdtemp(prefix="retrieve_") temp_model = os.path.join(temp_dir, os.path.basename(model_path)) shutil.copy2(model_path, temp_model) temp_index = None if index_path and os.path.exists(index_path): temp_index = os.path.join(temp_dir, os.path.basename(index_path)) shutil.copy2(index_path, temp_index) except Exception as e: logger.exception("Failed to stage retrieved model for Voice Conversion tab") return None, None, None, no_change, f"❌ خطا در بازیابی فایل مدل: {str(e)}" is_rvc = (trainer == "RVC v2") model_name = job.get("model_name", "") if status == "running": progress = float(job.get("progress") or 0.0) msg = ( f"🔶 مدل «{model_name}» ({trainer}) هنوز در حال آموزش است — یک checkpoint میان‌کاری " f"(تا همین لحظه، حدود {progress*100:.0f}٪ پیشرفت) بازیابی و بارگذاری شد. " "می‌توانید همین حالا برای تست تبدیل صدا استفاده کنید؛ آموزش در پس‌زمینه ادامه دارد و بعداً " "می‌توانید دوباره همین کد را وارد کنید تا آخرین نسخه را بگیرید." ) else: msg = f"✅ مدل «{model_name}» ({trainer}) کامل شده با موفقیت از کد پیگیری بازیابی و بارگذاری شد" if is_rvc: return temp_model, temp_index, None, gr.update(value="RVC v2"), msg else: return None, None, temp_model, gr.update(value="Beatrice v2"), msg def load_model_from_url(url): """Downloads a model directly on the server from a URL (a .zip containing the model +/- index, or a raw .pth/.pt.gz file) and loads it into the Voice Conversion inputs. The file never passes through the user's upload bandwidth — only the server fetches it.""" no_change = gr.update() url = (url or "").strip() if not url: return None, None, None, no_change, "❌ لطفاً یک لینک وارد کنید" if not (url.startswith("http://") or url.startswith("https://")): return None, None, None, no_change, "❌ لینک نامعتبر است؛ باید با http:// یا https:// شروع شود" temp_dir = tempfile.mkdtemp(prefix="url_model_") raw_path = os.path.join(temp_dir, "downloaded.bin") try: with requests.get(url, stream=True, timeout=60, headers={"User-Agent": "Mozilla/5.0"}) as r: r.raise_for_status() declared_len = int(r.headers.get("Content-Length") or 0) if declared_len and declared_len > MAX_MODEL_DOWNLOAD_BYTES: return None, None, None, no_change, ( f"❌ حجم فایل ({declared_len / (1024*1024):.0f}MB) بیشتر از سقف مجاز " f"({MAX_MODEL_DOWNLOAD_BYTES // (1024*1024)}MB) است" ) downloaded = 0 with open(raw_path, "wb") as f: for chunk in r.iter_content(chunk_size=262144): if not chunk: continue downloaded += len(chunk) if downloaded > MAX_MODEL_DOWNLOAD_BYTES: return None, None, None, no_change, ( f"❌ حجم فایل بیشتر از سقف مجاز ({MAX_MODEL_DOWNLOAD_BYTES // (1024*1024)}MB) است" ) f.write(chunk) except requests.exceptions.RequestException as e: return None, None, None, no_change, f"❌ خطا در دانلود از لینک: {str(e)}" model_out, index_out, beatrice_out, trainer_guess = None, None, None, None if zipfile.is_zipfile(raw_path): try: with zipfile.ZipFile(raw_path) as zf: names = [n for n in zf.namelist() if "__MACOSX" not in n and not n.endswith("/")] pth_name = next((n for n in names if n.lower().endswith(".pth")), None) idx_name = next((n for n in names if n.lower().endswith(".index")), None) beat_name = next((n for n in names if n.lower().endswith(".pt.gz")), None) if pth_name: dest = os.path.join(temp_dir, os.path.basename(pth_name)) with zf.open(pth_name) as src, open(dest, "wb") as dst: shutil.copyfileobj(src, dst) model_out, trainer_guess = dest, "RVC v2" if idx_name: dest = os.path.join(temp_dir, os.path.basename(idx_name)) with zf.open(idx_name) as src, open(dest, "wb") as dst: shutil.copyfileobj(src, dst) index_out = dest if beat_name: dest = os.path.join(temp_dir, os.path.basename(beat_name)) with zf.open(beat_name) as src, open(dest, "wb") as dst: shutil.copyfileobj(src, dst) beatrice_out = dest trainer_guess = trainer_guess or "Beatrice v2" except zipfile.BadZipFile: return None, None, None, no_change, "❌ فایل ZIP معتبر نیست یا خراب است" if not (model_out or beatrice_out): return None, None, None, no_change, "❌ هیچ فایل مدل (.pth یا .pt.gz) داخل فایل ZIP پیدا نشد" else: lower_url = url.lower().split("?")[0] if lower_url.endswith(".pth"): final_path = os.path.join(temp_dir, os.path.basename(lower_url)) os.replace(raw_path, final_path) model_out, trainer_guess = final_path, "RVC v2" elif lower_url.endswith(".gz"): final_path = os.path.join(temp_dir, os.path.basename(lower_url)) os.replace(raw_path, final_path) beatrice_out, trainer_guess = final_path, "Beatrice v2" else: return None, None, None, no_change, ( "❌ فایل دانلودشده نه ZIP است و نه فرمت مدل شناخته‌شده (.pth یا .pt.gz)؛ " "لینک باید مستقیم به یک فایل ZIP یا مدل باشد" ) parts = ["✅ مدل با موفقیت از لینک دانلود و بارگذاری شد (بدون مصرف اینترنت شما — سرور دانلود کرد)"] if model_out: parts.append(f"📦 مدل RVC: {os.path.basename(model_out)}") if index_out: parts.append(f"📎 Index: {os.path.basename(index_out)}") if beatrice_out: parts.append(f"📦 مدل Beatrice: {os.path.basename(beatrice_out)}") type_update = gr.update(value=trainer_guess) if trainer_guess else no_change return model_out, index_out, beatrice_out, type_update, "\n".join(parts) # ============================================================ # GRADIO UI - Gradio 6 Compatible # ============================================================ def train_ui( audio_file, model_name: str, epochs: int, batch_size: int, sample_rate: int, f0_method: str = "rmvpe", progress=gr.Progress() ): """Training function for Gradio UI - Generator for live log updates""" if audio_file is None: yield None, None, "❌ Please upload training audio" return if not model_name or model_name.strip() == "": yield None, None, "❌ Please enter a model name" return # Check if CUDA available has_cuda = torch.cuda.is_available() device_info = "GPU (CUDA)" if has_cuda else "CPU" # Log accumulator for live updates logs = [] try: model_name = sanitize_model_name(model_name) output_dir = f"trained_models/{model_name}" data_dir = f"{output_dir}/data" # Preprocessing phase logs.append(f"🚀 Starting on {device_info}") logs.append(f"📂 Output: {output_dir}") yield None, None, "\n".join(logs) progress(0.1, "Preprocessing...") logs.append("🔄 Preprocessing audio...") yield None, None, "\n".join(logs) audio_path = audio_file if isinstance(audio_file, str) else audio_file.name result = preprocess_audio_for_training(audio_path, data_dir, target_sr=sample_rate, f0_method=f0_method) if result is None: logs.append("❌ Preprocessing failed - no valid audio chunks") yield None, None, "\n".join(logs) return logs.append("✅ Preprocessing complete") logs.append(f"🏋️ Training {epochs} epochs...") logs.append("─" * 40) yield None, None, "\n".join(logs) # Training phase - iterate over generator for live updates ckpt = None idx = None for msg, path, index in train_rvc_generator( data_dir=data_dir, output_dir=output_dir, epochs=epochs, batch_size=batch_size, lr=1e-5, target_sr=sample_rate, progress_callback=progress ): logs.append(msg) if path: ckpt = path if index: idx = index # Propagate any freshly-saved mid-training checkpoint immediately # (not just at the very end), so a tracking-code lookup can pick # up e.g. the epoch-600 checkpoint of a 1000-epoch run while it's # still training. yield (path or None), (index or None), "\n".join(logs) if ckpt: logs.append("─" * 40) logs.append(f"✅ Training complete!") logs.append(f"📦 Model: {ckpt}") if idx: logs.append(f"📦 Index: {idx}") progress(1.0, "Done!") yield ckpt, idx, "\n".join(logs) else: logs.append("❌ Training failed") yield None, None, "\n".join(logs) except Exception as e: logger.exception("Training error") logs.append(f"❌ Error: {str(e)}") yield None, None, "\n".join(logs) with gr.Blocks() as demo: gr.Markdown(f"# 🎤 Voice Conversion (RVC + Beatrice)\nInference: CPU • Training: {'GPU (CUDA)' if torch.cuda.is_available() else 'CPU'}") with gr.Tabs(): # ==================== TAB 1: VOICE CONVERSION ==================== with gr.Tab("🎵 Voice Conversion"): with gr.Row(): with gr.Column(): source_audio = gr.Audio(label="Source Audio", type="filepath") gr.Markdown("### Model") with gr.Accordion("📥 بازیابی خودکار مدل (بدون آپلود دوباره)", open=False): gr.Markdown( "به‌جای آپلود فایل مدل از روی سیستم خودتان، می‌توانید مدل را از یکی از دو راه زیر " "مستقیماً روی سرور بارگذاری کنید — اینترنت شما مصرف نمی‌شود.\n\n" "💡 کد پیگیری حتی برای یک آموزش **هنوز ناتمام** هم کار می‌کند: مثلاً اگر هدف ۱۰۰۰ " "اپوک است و تا الان ۶۰۰ اپوک انجام شده، همین کد را وارد کنید تا آخرین checkpoint " "ذخیره‌شده (اپوک ۶۰۰) بازیابی و همین‌جا برای تست تبدیل صدا بارگذاری شود — بدون اینکه " "صبر کنید آموزش تمام شود. آموزش هم‌زمان در پس‌زمینه ادامه پیدا می‌کند." ) with gr.Row(): retrieve_code_input = gr.Textbox( label="کد پیگیری آموزش (کامل یا در حال انجام)", placeholder="مثال: 4F9A2C", scale=3 ) retrieve_code_btn = gr.Button("🔄 بازیابی مدل", variant="secondary", scale=1) with gr.Row(): model_url_input = gr.Textbox( label="لینک مستقیم فایل ZIP یا مدل (.pth / .pt.gz)", placeholder="https://example.com/my-model.zip", scale=3, ) model_url_btn = gr.Button("⬇️ دانلود از لینک", variant="secondary", scale=1) retrieve_status = gr.Textbox(label="وضعیت بازیابی", lines=3, interactive=False) model_type = gr.Radio( ["RVC v2", "Beatrice v2"], value="RVC v2", label="Model Type", info="RVC: .pth files | Beatrice: .pt.gz files" ) # RVC model inputs with gr.Group(visible=True) as rvc_model_group: with gr.Row(): model_file = gr.File(label="RVC Model (.pth)", file_types=[".pth"]) load_example_btn = gr.Button("Load Example (Benee)", size="sm") index_file = gr.File(label="Index File (.index) - Optional", file_types=[".index"]) # Beatrice model inputs with gr.Group(visible=False) as beatrice_model_group: beatrice_model_file = gr.File(label="Beatrice Model (.pt.gz)", file_types=[".gz"]) with gr.Row(): beatrice_target_speaker = gr.Number(value=0, label="Target Speaker", precision=0) beatrice_formant_shift = gr.Slider(-2, 2, value=0.0, step=0.5, label="Formant Shift") with gr.Row(): pitch_shift = gr.Slider(-12, 12, value=0, step=1, label="Pitch (semitones)") f0_method = gr.Radio(["rmvpe", "pm", "harvest"], value="rmvpe", label="F0 Method", visible=True) with gr.Row(visible=True) as rvc_extra_options: index_rate = gr.Slider(0, 1, value=0.75, step=0.05, label="Index Rate") protect = gr.Slider(0, 0.5, value=0.33, step=0.01, label="Protect (voiceless consonants)") convert_btn = gr.Button("Convert", variant="primary") with gr.Column(): output_audio = gr.Audio(label="Converted Audio", type="filepath") output_info = gr.Textbox(label="Status", lines=2) def update_model_type(model_type_val): is_rvc = model_type_val == "RVC v2" return ( gr.update(visible=is_rvc), # rvc_model_group gr.update(visible=not is_rvc), # beatrice_model_group gr.update(visible=is_rvc), # f0_method gr.update(visible=is_rvc), # rvc_extra_options ) model_type.change( update_model_type, [model_type], [rvc_model_group, beatrice_model_group, f0_method, rvc_extra_options] ) load_example_btn.click( load_example_model, [], [model_file, index_file, output_info] ) # Retrieve a model by training tracking code, or fetch it from a # direct URL — both run server-side, so the user's own upload # bandwidth is never spent moving the model file. retrieve_code_btn.click( retrieve_model_by_tracking_code, [retrieve_code_input], [model_file, index_file, beatrice_model_file, model_type, retrieve_status], ) model_url_btn.click( load_model_from_url, [model_url_input], [model_file, index_file, beatrice_model_file, model_type, retrieve_status], ) def convert_unified(source, m_type, rvc_model, rvc_index, beat_model, beat_speaker, beat_formant, pitch, f0, idx_rate, prot, progress=gr.Progress()): # Audio duration check for inference if source is not None: src_path = source if isinstance(source, str) else source.name try: dur = librosa.get_duration(path=src_path) except Exception: dur = librosa.get_duration(filename=src_path) if dur > MAX_AUDIO_DURATION: return None, f"❌ Audio too long ({dur/3600:.1f}h). Max {MAX_AUDIO_DURATION//3600}h." if m_type == "RVC v2": return convert_voice(source, rvc_model, rvc_index, pitch, f0, idx_rate, prot, progress) else: return convert_voice_beatrice( source, beat_model, target_speaker=int(beat_speaker), pitch_shift=int(pitch), formant_shift=float(beat_formant), progress=progress ) convert_btn.click( convert_unified, [source_audio, model_type, model_file, index_file, beatrice_model_file, beatrice_target_speaker, beatrice_formant_shift, pitch_shift, f0_method, index_rate, protect], [output_audio, output_info], api_name="convert", concurrency_limit=1, ) gr.Markdown("**Models:** [HuggingFace](https://huggingface.co/models?search=rvc) | [Weights.gg](https://weights.gg)") # ==================== TAB 2: TRAINING ==================== with gr.Tab("🏋️ Training"): # GPU Warning gpu_status = "🟢 GPU Available (CUDA)" if torch.cuda.is_available() else "🟡 CPU Only (Training will be slow)" gr.Markdown(f""" ### Training Status: {gpu_status} {'**GPU detected!** Training will use CUDA acceleration.' if torch.cuda.is_available() else '**No GPU detected.** Training will run on CPU (~30 sec/epoch). For faster training, run locally with CUDA GPU.'} """) # Trainer selector - Beatrice always available (downloads assets from HF) trainer_selector = gr.Dropdown( choices=["RVC v2", "Beatrice v2"], value="RVC v2", label="Trainer", info="RVC v2: general purpose | Beatrice v2: low latency (~50ms)" ) with gr.Row(): with gr.Column(): gr.Markdown( f"**Training Audio** — فایل اول را آپلود کنید؛ بعد از هر آپلود، دکمه افزودن فایل بعدی زیرش " f"ظاهر می‌شود. تا **{MAX_TRAIN_AUDIO_FILES} فایل صوتی جداگانه** می‌توانید اضافه کنید — " f"محدودیت خاصی جز همین سقف نیست.\n\n" f"💡 راه سریع‌تر: به‌جای آپلود تک‌تک، یک **فایل ZIP** حاوی همه فایل‌های صوتی‌تان بسازید و " f"همان را در کادر زیر آپلود کنید — همه فایل‌های صوتی داخلش (حتی داخل پوشه‌ها) به‌طور " f"خودکار استخراج، تعدادشان شناسایی و در جای درست اضافه می‌شوند." ) with gr.Group(): train_zip_upload = gr.File( label="📦 آپلود فایل ZIP حاوی همه فایل‌های صوتی آموزش (استخراج خودکار)", file_types=[".zip"], ) train_zip_status = gr.Markdown("") gr.Markdown("— یا به‌صورت جداگانه اضافه کنید —") train_audio_slots = [] add_audio_btns = [] for _i in range(MAX_TRAIN_AUDIO_FILES): _slot = gr.Audio( label=f"Training Audio #{_i + 1}", type="filepath", visible=(_i == 0), ) train_audio_slots.append(_slot) if _i < MAX_TRAIN_AUDIO_FILES - 1: _add_btn = gr.Button( "➕ افزودن فایل صوتی دیگر", size="sm", visible=False ) add_audio_btns.append(_add_btn) # Uploading a file reveals the "add another" button right below it; # clearing it hides that button again. for _i, _slot in enumerate(train_audio_slots[:-1]): _slot.change( lambda v: gr.update(visible=(v is not None)), inputs=[_slot], outputs=[add_audio_btns[_i]], ) # Clicking "add another" reveals the next empty upload slot. for _i, _btn in enumerate(add_audio_btns): _btn.click( lambda: gr.update(visible=True), outputs=[train_audio_slots[_i + 1]], ) # Uploading a ZIP auto-extracts every audio file inside it # (any format, any folder depth) and fills the slots above # automatically — no manual per-file selection needed. train_zip_upload.change( handle_train_zip_upload, inputs=[train_zip_upload], outputs=[*train_audio_slots, train_zip_status], ) train_model_name = gr.Textbox(label="Model Name", placeholder="my_voice", value="my_voice") # RVC-specific options with gr.Group(visible=True) as rvc_options: with gr.Row(): train_epochs = gr.Slider(1, 5000, value=50, step=1, label="Epochs") train_batch = gr.Slider(1, 8, value=2, step=1, label="Batch Size") with gr.Row(): train_sr = gr.Radio([32000, 40000, 48000], value=40000, label="Sample Rate") train_f0 = gr.Radio(["rmvpe", "pm", "harvest"], value="rmvpe", label="F0 Method") # Beatrice-specific options with gr.Group(visible=False) as beatrice_options: with gr.Row(): beatrice_epochs = gr.Slider(1, 100, value=30, step=1, label="Epochs (30 recommended)") beatrice_batch = gr.Slider(1, 64, value=8, step=1, label="Batch Size") beatrice_resume = gr.Checkbox(label="Resume from checkpoint", value=False) train_btn = gr.Button("Start Training", variant="primary") with gr.Column(): train_tracking_code = gr.Textbox( label="🔑 کد پیگیری (بعد از ثبت درخواست اینجا نمایش داده می‌شود)", interactive=False, ) train_status = gr.Textbox(label="وضعیت ثبت درخواست", lines=8) gr.Markdown( "آموزش در پس‌زمینه‌ی سرور اجرا می‌شود، پس نیازی به باز نگه‌داشتن این صفحه نیست. " "برای مشاهده پیشرفت، لغو، یا دانلود مدل نهایی به تب **📡 پیگیری آموزش** بروید و کد بالا را وارد کنید." ) # Toggle visibility based on trainer selection def update_trainer_options(trainer): is_rvc = trainer == "RVC v2" return gr.update(visible=is_rvc), gr.update(visible=not is_rvc) trainer_selector.change( update_trainer_options, [trainer_selector], [rvc_options, beatrice_options] ) def _check_audio_duration(audio): """Check audio duration, return (seconds, error_msg).""" if audio is None: return 0, "❌ Please upload training audio" audio_path = audio if isinstance(audio, str) else audio.name try: duration = librosa.get_duration(path=audio_path) except Exception: duration = librosa.get_duration(filename=audio_path) if duration > MAX_AUDIO_DURATION: hours = duration / 3600 return duration, f"❌ Audio too long ({hours:.1f}h). Max 2 hours to keep the Space usable for everyone." return duration, None # Unified training function def train_unified(trainer, audio, name, rvc_epochs, rvc_batch, rvc_sr, rvc_f0, beat_epochs, beat_batch, beat_resume, progress=gr.Progress()): import time as _time _train_start = _time.time() # Reset cancel flag at start global _training_cancel _training_cancel = False # Audio duration check duration, err = _check_audio_duration(audio) if err: yield None, None, err return if trainer == "RVC v2": for result in train_ui(audio, name, rvc_epochs, rvc_batch, rvc_sr, rvc_f0, progress): yield result else: # Beatrice training if not name or name.strip() == "": yield None, None, "❌ Please enter a model name" return try: name = sanitize_model_name(name) output_dir = f"trained_models/beatrice_{name}" data_dir = f"{output_dir}/training_data" logs = [] logs.append(f"🚀 Starting Beatrice training on {'GPU (CUDA)' if torch.cuda.is_available() else 'CPU'}") logs.append(f"📎 Audio: {duration/60:.0f} min") yield None, None, "\n".join(logs) progress(0.05, "Preprocessing audio...") audio_path = audio if isinstance(audio, str) else audio.name chunks, _ = preprocess_audio_for_beatrice(audio_path, data_dir, name) if chunks == 0: yield None, None, "❌ Preprocessing failed - no valid audio chunks" return logs.append(f"✅ Preprocessed {chunks} audio chunks") logs.append("─" * 40) yield None, None, "\n".join(logs) ckpt = None for msg, path in train_beatrice_generator( data_dir=data_dir, output_dir=output_dir, epochs=beat_epochs, batch_size=beat_batch, resume=beat_resume, progress_callback=progress ): if _training_cancel: logs.append("🛑 Cancelled by user") if path: ckpt = path yield ckpt, None, "\n".join(logs) return logs.append(msg) if path: ckpt = path # Propagate any freshly-saved mid-training checkpoint # immediately, so a tracking-code lookup can retrieve # e.g. epoch 600/1000 while training is still running. yield (path or None), None, "\n".join(logs) if ckpt: logs.append("─" * 40) logs.append(f"✅ Training complete!") logs.append(f"📦 Model: {ckpt}") progress(1.0, "Done!") yield ckpt, None, "\n".join(logs) else: logs.append("❌ Training failed") yield None, None, "\n".join(logs) except Exception as e: logger.exception("Beatrice training error") yield None, None, f"❌ Error: {str(e)}" # Submitting a training request only registers a background job and # returns a tracking code immediately — it does not block on training # itself, so the browser tab can be closed/refreshed right away. train_btn.click( submit_training_request, [trainer_selector, train_model_name, *train_audio_slots, train_epochs, train_batch, train_sr, train_f0, beatrice_epochs, beatrice_batch, beatrice_resume], [train_tracking_code, train_status], api_name="train", ) gr.Markdown(""" --- ### Training Tips **RVC v2:** - 50-100 epochs for quick test, 200-500 for quality - CPU training: ~30 sec/epoch (100 epochs ≈ 50 min) **Beatrice v2:** - 20-50 epochs recommended, GPU recommended (CPU works but slow) - Pretrained assets downloaded automatically from HuggingFace - Output: .pt.gz checkpoint (use in Voice Conversion tab) - Lower latency (~50ms vs ~100ms) ### CLI ```bash python app.py train -a voice.mp3 -o ./model --epochs 100 python app.py train-beatrice -a voice.mp3 -o ./beatrice_model --epochs 30 python app.py infer -i input.wav -m beatrice_model.pt.gz -o output.wav ``` """) # ==================== TAB 3: TRACK TRAINING ==================== with gr.Tab("📡 پیگیری آموزش"): gr.Markdown(""" ### بررسی وضعیت آموزش با کد پیگیری وقتی یک درخواست آموزش در تب «🏋️ Training» ثبت می‌کنید، یک **کد پیگیری** کوتاه دریافت می‌کنید. می‌توانید صفحه را ببندید یا رفرش کنید — آموزش در پس‌زمینه‌ی سرور ادامه می‌یابد. هر وقت خواستید، با وارد کردن همان کد در پایین، وضعیت را بررسی کنید. 🔶 **نیازی نیست منتظر اتمام کامل آموزش بمانید:** به‌صورت دوره‌ای در حین آموزش هم یک checkpoint ذخیره می‌شود (مثلاً اگر هدف ۱۰۰۰ اپوک است و ۶۰۰ اپوک انجام شده، همان checkpoint ۶۰۰ اپوکی). همین‌جا هم می‌توانید آن را دانلود کنید، هم با همین کد پیگیری در تب **🎵 Voice Conversion** (بخش «📥 بازیابی خودکار مدل») مستقیماً امتحانش کنید — بدون آپلود دوباره. """) with gr.Row(): track_code_input = gr.Textbox( label="کد پیگیری", placeholder="مثال: 4F9A2C", scale=3 ) track_check_btn = gr.Button("🔍 بررسی وضعیت", variant="primary", scale=1) track_cancel_btn = gr.Button("🛑 لغو این آموزش", variant="stop", scale=1) track_status = gr.Textbox(label="وضعیت / گزارش آموزش", lines=12) track_progress = gr.Slider( 0, 1, value=0, step=0.01, label="پیشرفت", interactive=False ) with gr.Row(): track_model_file = gr.File(label="📦 دانلود مدل آموزش‌دیده") track_index_file = gr.File(label="📦 دانلود فایل Index (در صورت وجود)") track_check_btn.click( check_tracking_status, [track_code_input], [track_status, track_progress, track_model_file, track_index_file], api_name="check_training_status", ) track_cancel_btn.click( cancel_tracking_job, [track_code_input], [track_status], api_name="cancel_training", ) gr.Markdown( "💡 نکته: کد پیگیری را حتماً یادداشت کنید. این کد تا زمانی که سرور اپلیکیشن ری‌استارت نشود معتبر می‌ماند." ) def cli_convert(args): """CLI mode conversion - supports both RVC and Beatrice models""" print(f"Converting: {args.input}") print(f"Model: {args.model}") # Auto-detect model type from extension or --type flag model_type = getattr(args, 'type', None) model_path = args.model index_path = getattr(args, 'index', None) if args.example: print("Downloading example model...") model_path = hf_hub_download(repo_id=DEFAULT_MODEL_REPO, filename=DEFAULT_MODEL_FILE) index_path = hf_hub_download(repo_id=DEFAULT_MODEL_REPO, filename=DEFAULT_INDEX_FILE) model_type = "rvc" print(f"Model: {model_path}") if not model_path: print("Error: No model specified. Use -m MODEL.pth or --example") sys.exit(1) # Auto-detect type from extension if not specified if model_type is None: if model_path.endswith('.pt.gz') or model_path.endswith('.gz'): model_type = "beatrice" else: model_type = "rvc" if model_type == "beatrice": # Beatrice inference print(f"Using Beatrice v2 inference") output_path, status = convert_voice_beatrice( source_audio=args.input, model_file=model_path, target_speaker=getattr(args, 'speaker', 0), pitch_shift=args.pitch, formant_shift=getattr(args, 'formant_shift', 0.0), ) else: # RVC inference class FileObj: def __init__(self, path): self.name = path model_file = FileObj(model_path) index_file = FileObj(index_path) if index_path else None output_path, status = convert_voice( source_audio=args.input, model_file=model_file, index_file=index_file, pitch_shift=args.pitch, f0_method=args.f0, index_rate=args.index_rate, progress=lambda *a, **k: None ) if output_path: shutil.copy(output_path, args.output) print(f"Output: {args.output}") print(status) else: print(f"Failed: {status}") sys.exit(1) def cli_train_beatrice(args): """CLI Beatrice training mode (embedded - downloads assets from HF)""" print(f"=== Beatrice v2 Training (Embedded) ===") print(f"Input audio: {args.audio}") print(f"Output dir: {args.output}") # Preprocess audio - Beatrice expects: data_dir/speaker_name/*.wav data_dir = os.path.join(args.output, "training_data") speaker_name = os.path.splitext(os.path.basename(args.audio))[0] print(f"\n[1/2] Preprocessing audio for Beatrice...") chunks, _ = preprocess_audio_for_beatrice(args.audio, data_dir, speaker_name) if chunks == 0: print("Preprocessing failed - no valid audio chunks") sys.exit(1) print(f"Created {chunks} audio chunks") # Train using embedded code print(f"\n[2/2] Training Beatrice model ({args.epochs} epochs)...") ckpt = None for msg, path in train_beatrice_generator( data_dir=data_dir, output_dir=args.output, epochs=args.epochs, batch_size=args.batch, resume=args.resume, ): print(msg) if path: ckpt = path if ckpt: print(f"\nTraining complete!") print(f"Model: {ckpt}") else: print("Training failed!") sys.exit(1) def cli_train(args): """CLI training mode""" print(f"=== RVC Training ===") print(f"Input audio: {args.audio}") print(f"Output dir: {args.output}") # Create temp dir for preprocessing data_dir = f"{args.output}/data" # Preprocess print("\n[1/2] Preprocessing audio...") result = preprocess_audio_for_training(args.audio, data_dir, target_sr=args.sr, f0_method=args.f0) if result is None: print("Preprocessing failed!") sys.exit(1) # Train print(f"\n[2/2] Training for {args.epochs} epochs...") ckpt, idx = train_rvc( data_dir=data_dir, output_dir=args.output, epochs=args.epochs, batch_size=args.batch, lr=args.lr, target_sr=args.sr ) if ckpt: print(f"\nTraining complete!") print(f"Model saved: {ckpt}") if idx: print(f"Index saved: {idx}") else: print("Training failed!") sys.exit(1) if __name__ == "__main__": # Check if any CLI args (besides script name) if len(sys.argv) > 1: parser = argparse.ArgumentParser( description="RVC Voice Conversion - Inference and Training", formatter_class=argparse.RawDescriptionHelpFormatter, ) subparsers = parser.add_subparsers(dest="command", help="Commands") # Inference subcommand infer_parser = subparsers.add_parser("infer", help="Voice conversion inference (RVC + Beatrice)", epilog=""" Examples: python app.py infer -i voice.wav -m model.pth -o output.wav python app.py infer -i voice.wav -m beatrice_model.pt.gz -o output.wav --type beatrice python app.py infer -i voice.wav --example -o output.wav """) infer_parser.add_argument("-i", "--input", required=True, help="Input audio file") infer_parser.add_argument("-o", "--output", required=True, help="Output audio file") infer_parser.add_argument("-m", "--model", help="Model file (.pth for RVC, .pt.gz for Beatrice)") infer_parser.add_argument("--type", choices=["rvc", "beatrice"], default=None, help="Model type (auto-detected from extension)") infer_parser.add_argument("--index", help="Index file (.index) - RVC only") infer_parser.add_argument("--example", action="store_true", help="Use example model (Benee-RVC)") infer_parser.add_argument("-p", "--pitch", type=int, default=0, help="Pitch shift (-12 to 12)") infer_parser.add_argument("--f0", choices=["rmvpe", "pm", "harvest"], default="rmvpe", help="F0 method (RVC only)") infer_parser.add_argument("--index-rate", type=float, default=0.75, help="Index rate 0-1 (RVC only)") infer_parser.add_argument("--speaker", type=int, default=0, help="Target speaker index (Beatrice only)") infer_parser.add_argument("--formant-shift", type=float, default=0.0, help="Formant shift -2 to 2 (Beatrice only)") # Training subcommand (RVC) train_parser = subparsers.add_parser("train", help="Train RVC model", epilog=""" Examples: python app.py train -a voice.mp3 -o ./my_model --epochs 5 python app.py train -a dataset.wav -o ./trained --epochs 10 --batch 4 """) train_parser.add_argument("-a", "--audio", required=True, help="Training audio file") train_parser.add_argument("-o", "--output", required=True, help="Output directory for model") train_parser.add_argument("--epochs", type=int, default=5, help="Number of epochs (default: 5)") train_parser.add_argument("--batch", type=int, default=2, help="Batch size (default: 2)") train_parser.add_argument("--lr", type=float, default=1e-5, help="Learning rate (default: 1e-5)") train_parser.add_argument("--sr", type=int, default=40000, help="Sample rate (default: 40000)") train_parser.add_argument("--f0", choices=["rmvpe", "pm", "harvest"], default="rmvpe", help="F0 method (default: rmvpe)") # Beatrice v2 Training subcommand beatrice_parser = subparsers.add_parser("train-beatrice", help="Train Beatrice v2 model (downloads assets from HF)", epilog=""" Examples: python app.py train-beatrice -a voice.mp3 -o ./beatrice_model --epochs 20 python app.py train-beatrice -a dataset.wav -o ./trained --epochs 50 --batch 8 --resume Pretrained assets are downloaded automatically from HuggingFace. """) beatrice_parser.add_argument("-a", "--audio", required=True, help="Training audio file") beatrice_parser.add_argument("-o", "--output", required=True, help="Output directory for model") beatrice_parser.add_argument("--epochs", type=int, default=20, help="Number of epochs (default: 20)") beatrice_parser.add_argument("--batch", type=int, default=24, help="Batch size (default: 24, reduce for less VRAM)") beatrice_parser.add_argument("--resume", action="store_true", help="Resume from checkpoint") args = parser.parse_args() if args.command == "infer": cli_convert(args) elif args.command == "train": cli_train(args) elif args.command == "train-beatrice": cli_train_beatrice(args) else: parser.print_help() else: # Cleanup when user closes browser tab (Gradio heartbeat detects disconnect ~15s) # Running generators (training) auto-cancel via GeneratorExit on disconnect def _on_unload(req: gr.Request): gc.collect() demo.unload(_on_unload) # No args = Gradio mode (Gradio 6 syntax) demo.launch( mcp_server=True, show_error=True, ssr_mode=False, )