Spaces:
Running
Running
| import os, traceback, sys, parselmouth | |
| now_dir = os.getcwd() | |
| sys.path.append(now_dir) | |
| from my_utils import load_audio | |
| import pyworld | |
| import numpy as np, logging | |
| logging.getLogger("numba").setLevel(logging.WARNING) | |
| from multiprocessing import Process | |
| exp_dir = sys.argv[1] | |
| f = open("%s/extract_f0_feature.log" % exp_dir, "a+") | |
| def printt(strr): | |
| print(strr) | |
| f.write("%s\n" % strr) | |
| f.flush() | |
| n_p = int(sys.argv[2]) | |
| f0method = sys.argv[3] | |
| class FeatureInput(object): | |
| def __init__(self, samplerate=16000, hop_size=160): | |
| self.fs = samplerate | |
| self.hop = hop_size | |
| 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 compute_f0(self, path, f0_method): | |
| x = load_audio(path, self.fs) | |
| p_len = x.shape[0] // self.hop | |
| if f0_method == "pm": | |
| time_step = 160 / 16000 * 1000 | |
| f0_min = 50 | |
| f0_max = 1100 | |
| f0 = ( | |
| parselmouth.Sound(x, self.fs) | |
| .to_pitch_ac( | |
| time_step=time_step / 1000, | |
| voicing_threshold=0.6, | |
| pitch_floor=f0_min, | |
| pitch_ceiling=f0_max, | |
| ) | |
| .selected_array["frequency"] | |
| ) | |
| pad_size = (p_len - len(f0) + 1) // 2 | |
| if pad_size > 0 or p_len - len(f0) - pad_size > 0: | |
| f0 = np.pad( | |
| f0, [[pad_size, p_len - len(f0) - pad_size]], mode="constant" | |
| ) | |
| elif f0_method == "harvest": | |
| f0, t = 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, | |
| ) | |
| f0 = pyworld.stonemask(x.astype(np.double), f0, t, self.fs) | |
| elif f0_method == "dio": | |
| f0, t = 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, | |
| ) | |
| f0 = pyworld.stonemask(x.astype(np.double), f0, t, self.fs) | |
| 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 | |
| # use 0 or 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 go(self, paths, f0_method): | |
| if len(paths) == 0: | |
| printt("no-f0-todo") | |
| else: | |
| printt("todo-f0-%s" % len(paths)) | |
| n = max(len(paths) // 5, 1) # 每个进程最多打印5条 | |
| for idx, (inp_path, opt_path1, opt_path2) in enumerate(paths): | |
| try: | |
| if idx % n == 0: | |
| printt("f0ing,now-%s,all-%s,-%s" % (idx, len(paths), inp_path)) | |
| if ( | |
| os.path.exists(opt_path1 + ".npy") == True | |
| and os.path.exists(opt_path2 + ".npy") == True | |
| ): | |
| continue | |
| featur_pit = self.compute_f0(inp_path, f0_method) | |
| np.save( | |
| opt_path2, | |
| featur_pit, | |
| allow_pickle=False, | |
| ) # nsf | |
| coarse_pit = self.coarse_f0(featur_pit) | |
| np.save( | |
| opt_path1, | |
| coarse_pit, | |
| allow_pickle=False, | |
| ) # ori | |
| except: | |
| printt("f0fail-%s-%s-%s" % (idx, inp_path, traceback.format_exc())) | |
| if __name__ == "__main__": | |
| # exp_dir=r"E:\codes\py39\dataset\mi-test" | |
| # n_p=16 | |
| # f = open("%s/log_extract_f0.log"%exp_dir, "w") | |
| printt(sys.argv) | |
| featureInput = FeatureInput() | |
| paths = [] | |
| inp_root = "%s/1_16k_wavs" % (exp_dir) | |
| opt_root1 = "%s/2a_f0" % (exp_dir) | |
| opt_root2 = "%s/2b-f0nsf" % (exp_dir) | |
| os.makedirs(opt_root1, exist_ok=True) | |
| os.makedirs(opt_root2, exist_ok=True) | |
| for name in sorted(list(os.listdir(inp_root))): | |
| inp_path = "%s/%s" % (inp_root, name) | |
| if "spec" in inp_path: | |
| continue | |
| opt_path1 = "%s/%s" % (opt_root1, name) | |
| opt_path2 = "%s/%s" % (opt_root2, name) | |
| paths.append([inp_path, opt_path1, opt_path2]) | |
| ps = [] | |
| for i in range(n_p): | |
| p = Process( | |
| target=featureInput.go, | |
| args=( | |
| paths[i::n_p], | |
| f0method, | |
| ), | |
| ) | |
| ps.append(p) | |
| p.start() | |
| for i in range(n_p): | |
| ps[i].join() | |