|
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() |
|
|
