mirror of
https://github.com/storytold/storyteller-ml.git
synced 2026-10-24 08:36:10 +00:00
Delete extract_f0_print.py
This commit is contained in:
Justin John
@@ -1,160 +0,0 @@
|
|||||||
import os, traceback, sys, parselmouth
|
|
||||||
|
|
||||||
now_dir = os.getcwd()
|
|
||||||
sys.path.append(now_dir)
|
|
||||||
from my_utils import load_audio
|
|
||||||
import pyworld
|
|
||||||
from scipy.io import wavfile
|
|
||||||
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()
|
|
||||||
Reference in New Issue
Block a user