| import os
|
| import sys
|
| import numpy as np
|
| import pyworld
|
| import torchcrepe
|
| import torch
|
| import parselmouth
|
| import tqdm
|
| from multiprocessing import Process, cpu_count
|
|
|
| current_directory = os.getcwd()
|
| sys.path.append(current_directory)
|
|
|
|
|
| from rvc.lib.utils import load_audio
|
|
|
|
|
| exp_dir = sys.argv[1]
|
| f0_method = sys.argv[2]
|
| num_processes = cpu_count()
|
|
|
| try:
|
| hop_length = int(sys.argv[3])
|
| except ValueError:
|
| hop_length = 128
|
|
|
| DoFormant = False
|
| Quefrency = 1.0
|
| Timbre = 1.0
|
|
|
|
|
| class FeatureInput:
|
| def __init__(self, sample_rate=16000, hop_size=160):
|
| self.fs = sample_rate
|
| self.hop = hop_size
|
|
|
| self.f0_method_dict = self.get_f0_method_dict()
|
|
|
| self.f0_bin = 256
|
| self.f0_max = 1100.0
|
| self.f0_min = 50.0
|
| self.f0_mel_min = 1127 * np.log(1 + self.f0_min / 700)
|
| self.f0_mel_max = 1127 * np.log(1 + self.f0_max / 700)
|
|
|
| def mncrepe(self, method, x, p_len, hop_length):
|
| f0 = None
|
| torch_device_index = 0
|
| torch_device = (
|
| torch.device(f"cuda:{torch_device_index % torch.cuda.device_count()}")
|
| if torch.cuda.is_available()
|
| else (
|
| torch.device("mps")
|
| if torch.backends.mps.is_available()
|
| else torch.device("cpu")
|
| )
|
| )
|
|
|
| audio = torch.from_numpy(x.astype(np.float32)).to(torch_device, copy=True)
|
| audio /= torch.quantile(torch.abs(audio), 0.999)
|
| audio = torch.unsqueeze(audio, dim=0)
|
| if audio.ndim == 2 and audio.shape[0] > 1:
|
| audio = torch.mean(audio, dim=0, keepdim=True).detach()
|
| audio = audio.detach()
|
|
|
| if method == "crepe":
|
| pitch = torchcrepe.predict(
|
| audio,
|
| self.fs,
|
| hop_length,
|
| self.f0_min,
|
| self.f0_max,
|
| "full",
|
| batch_size=hop_length * 2,
|
| device=torch_device,
|
| pad=True,
|
| )
|
| p_len = p_len or x.shape[0] // hop_length
|
| source = np.array(pitch.squeeze(0).cpu().float().numpy())
|
| source[source < 0.001] = np.nan
|
| target = np.interp(
|
| np.arange(0, len(source) * p_len, len(source)) / p_len,
|
| np.arange(0, len(source)),
|
| source,
|
| )
|
| f0 = np.nan_to_num(target)
|
|
|
| return f0
|
|
|
| def get_pm(self, x, p_len):
|
| f0 = (
|
| parselmouth.Sound(x, self.fs)
|
| .to_pitch_ac(
|
| time_step=160 / 16000,
|
| voicing_threshold=0.6,
|
| pitch_floor=self.f0_min,
|
| pitch_ceiling=self.f0_max,
|
| )
|
| .selected_array["frequency"]
|
| )
|
|
|
| return np.pad(
|
| f0,
|
| [
|
| [
|
| max(0, (p_len - len(f0) + 1) // 2),
|
| max(0, p_len - len(f0) - (p_len - len(f0) + 1) // 2),
|
| ]
|
| ],
|
| mode="constant",
|
| )
|
|
|
| def get_harvest(self, x):
|
| f0_spectral = pyworld.harvest(
|
| x.astype(np.double),
|
| fs=self.fs,
|
| f0_ceil=self.f0_max,
|
| f0_floor=self.f0_min,
|
| frame_period=1000 * self.hop / self.fs,
|
| )
|
| return pyworld.stonemask(x.astype(np.double), *f0_spectral, self.fs)
|
|
|
| def get_dio(self, x):
|
| f0_spectral = pyworld.dio(
|
| x.astype(np.double),
|
| fs=self.fs,
|
| f0_ceil=self.f0_max,
|
| f0_floor=self.f0_min,
|
| frame_period=1000 * self.hop / self.fs,
|
| )
|
| return pyworld.stonemask(x.astype(np.double), *f0_spectral, self.fs)
|
|
|
| def get_rmvpe(self, x):
|
| if not hasattr(self, "model_rmvpe"):
|
| from rvc.lib.rmvpe import RMVPE
|
|
|
| self.model_rmvpe = RMVPE("rmvpe.pt", is_half=False, device="cpu")
|
| return self.model_rmvpe.infer_from_audio(x, thred=0.03)
|
|
|
| def get_f0_method_dict(self):
|
| return {
|
| "pm": self.get_pm,
|
| "harvest": self.get_harvest,
|
| "dio": self.get_dio,
|
| "rmvpe": self.get_rmvpe,
|
| }
|
|
|
| def compute_f0(self, path, f0_method, hop_length):
|
| x = load_audio(path, self.fs)
|
| p_len = x.shape[0] // self.hop
|
|
|
| if f0_method in self.f0_method_dict:
|
| f0 = (
|
| self.f0_method_dict[f0_method](x, p_len)
|
| if f0_method == "pm"
|
| else self.f0_method_dict[f0_method](x)
|
| )
|
| elif f0_method == "crepe":
|
| f0 = self.mncrepe(f0_method, x, p_len, hop_length)
|
| return f0
|
|
|
| def coarse_f0(self, f0):
|
| f0_mel = 1127 * np.log(1 + f0 / 700)
|
| f0_mel[f0_mel > 0] = (f0_mel[f0_mel > 0] - self.f0_mel_min) * (
|
| self.f0_bin - 2
|
| ) / (self.f0_mel_max - self.f0_mel_min) + 1
|
|
|
|
|
| f0_mel[f0_mel <= 1] = 1
|
| f0_mel[f0_mel > self.f0_bin - 1] = self.f0_bin - 1
|
| f0_coarse = np.rint(f0_mel).astype(int)
|
| assert f0_coarse.max() <= 255 and f0_coarse.min() >= 1, (
|
| f0_coarse.max(),
|
| f0_coarse.min(),
|
| )
|
| return f0_coarse
|
|
|
| def process_paths(self, paths, f0_method, hop_length, thread_n):
|
| if len(paths) == 0:
|
| print("There are no paths to process.")
|
| return
|
| with tqdm.tqdm(total=len(paths), leave=True, position=thread_n) as pbar:
|
| description = f"Thread {thread_n} | Hop-Length {hop_length}"
|
| pbar.set_description(description)
|
|
|
| for idx, (inp_path, opt_path1, opt_path2) in enumerate(paths):
|
| try:
|
| if os.path.exists(opt_path1 + ".npy") and os.path.exists(
|
| opt_path2 + ".npy"
|
| ):
|
| pbar.update(1)
|
| continue
|
|
|
| feature_pit = self.compute_f0(inp_path, f0_method, hop_length)
|
| np.save(
|
| opt_path2,
|
| feature_pit,
|
| allow_pickle=False,
|
| )
|
| coarse_pit = self.coarse_f0(feature_pit)
|
| np.save(
|
| opt_path1,
|
| coarse_pit,
|
| allow_pickle=False,
|
| )
|
| pbar.update(1)
|
| except Exception as error:
|
| print(f"f0fail-{idx}-{inp_path}-{error}")
|
|
|
|
|
| if __name__ == "__main__":
|
| feature_input = FeatureInput()
|
| paths = []
|
| input_root = f"{exp_dir}/1_16k_wavs"
|
| output_root1 = f"{exp_dir}/2a_f0"
|
| output_root2 = f"{exp_dir}/2b-f0nsf"
|
|
|
| os.makedirs(output_root1, exist_ok=True)
|
| os.makedirs(output_root2, exist_ok=True)
|
| for name in sorted(list(os.listdir(input_root))):
|
| input_path = f"{input_root}/{name}"
|
| if "spec" in input_path:
|
| continue
|
| output_path1 = f"{output_root1}/{name}"
|
| output_path2 = f"{output_root2}/{name}"
|
| paths.append([input_path, output_path1, output_path2])
|
|
|
| processes = []
|
| print("Using f0 method: " + f0_method)
|
| for i in range(num_processes):
|
| p = Process(
|
| target=feature_input.process_paths,
|
| args=(paths[i::num_processes], f0_method, hop_length, i),
|
| )
|
| processes.append(p)
|
| p.start()
|
| for i in range(num_processes):
|
| processes[i].join()
|
|
|
