RMSnow's picture
init and interface
df2accb
raw
history blame
20 kB
# Copyright (c) 2023 Amphion.
#
# This source code is licensed under the MIT license found in the
# LICENSE file in the root directory of this source tree.
import json
import os
import numpy as np
from scipy.interpolate import interp1d
from tqdm import tqdm
from sklearn.preprocessing import StandardScaler
def load_content_feature_path(meta_data, processed_dir, feat_dir):
utt2feat_path = {}
for utt_info in meta_data:
utt = utt_info["Dataset"] + "_" + utt_info["Uid"]
feat_path = os.path.join(
processed_dir, utt_info["Dataset"], feat_dir, f'{utt_info["Uid"]}.npy'
)
utt2feat_path[utt] = feat_path
return utt2feat_path
def load_source_content_feature_path(meta_data, feat_dir):
utt2feat_path = {}
for utt in meta_data:
feat_path = os.path.join(feat_dir, f"{utt}.npy")
utt2feat_path[utt] = feat_path
return utt2feat_path
def get_spk_map(spk2id_path, utt2spk_path):
utt2spk = {}
with open(spk2id_path, "r") as spk2id_file:
spk2id = json.load(spk2id_file)
with open(utt2spk_path, encoding="utf-8") as f:
for line in f.readlines():
utt, spk = line.strip().split("\t")
utt2spk[utt] = spk
return spk2id, utt2spk
def get_target_f0_median(f0_dir):
total_f0 = []
for utt in os.listdir(f0_dir):
if not utt.endswith(".npy"):
continue
f0_feat_path = os.path.join(f0_dir, utt)
f0 = np.load(f0_feat_path)
total_f0 += f0.tolist()
total_f0 = np.array(total_f0)
voiced_position = np.where(total_f0 != 0)
return np.median(total_f0[voiced_position])
def get_conversion_f0_factor(source_f0, target_median, source_median=None):
"""Align the median between source f0 and target f0
Note: Here we use multiplication, whose factor is target_median/source_median
Reference: Frequency and pitch interval
http://blog.ccyg.studio/article/be12c2ee-d47c-4098-9782-ca76da3035e4/
"""
if source_median is None:
voiced_position = np.where(source_f0 != 0)
source_median = np.median(source_f0[voiced_position])
factor = target_median / source_median
return source_median, factor
def transpose_key(frame_pitch, trans_key):
# Transpose by user's argument
print("Transpose key = {} ...\n".format(trans_key))
transed_pitch = frame_pitch * 2 ** (trans_key / 12)
return transed_pitch
def pitch_shift_to_target(frame_pitch, target_pitch_median, source_pitch_median=None):
# Loading F0 Base (median) and shift
source_pitch_median, factor = get_conversion_f0_factor(
frame_pitch, target_pitch_median, source_pitch_median
)
print(
"Auto transposing: source f0 median = {:.1f}, target f0 median = {:.1f}, factor = {:.2f}".format(
source_pitch_median, target_pitch_median, factor
)
)
transed_pitch = frame_pitch * factor
return transed_pitch
def load_frame_pitch(
meta_data,
processed_dir,
pitch_dir,
use_log_scale=False,
return_norm=False,
interoperate=False,
utt2spk=None,
):
utt2pitch = {}
utt2uv = {}
if utt2spk is None:
pitch_scaler = StandardScaler()
for utt_info in meta_data:
utt = utt_info["Dataset"] + "_" + utt_info["Uid"]
pitch_path = os.path.join(
processed_dir, utt_info["Dataset"], pitch_dir, f'{utt_info["Uid"]}.npy'
)
pitch = np.load(pitch_path)
assert len(pitch) > 0
uv = pitch != 0
utt2uv[utt] = uv
if use_log_scale:
nonzero_idxes = np.where(pitch != 0)[0]
pitch[nonzero_idxes] = np.log(pitch[nonzero_idxes])
utt2pitch[utt] = pitch
pitch_scaler.partial_fit(pitch.reshape(-1, 1))
mean, std = pitch_scaler.mean_[0], pitch_scaler.scale_[0]
if return_norm:
for utt_info in meta_data:
utt = utt_info["Dataset"] + "_" + utt_info["Uid"]
pitch = utt2pitch[utt]
normalized_pitch = (pitch - mean) / std
utt2pitch[utt] = normalized_pitch
pitch_statistic = {"mean": mean, "std": std}
else:
spk2utt = {}
pitch_statistic = []
for utt_info in meta_data:
utt = utt_info["Dataset"] + "_" + utt_info["Uid"]
if not utt2spk[utt] in spk2utt:
spk2utt[utt2spk[utt]] = []
spk2utt[utt2spk[utt]].append(utt)
for spk in spk2utt:
pitch_scaler = StandardScaler()
for utt in spk2utt[spk]:
dataset = utt.split("_")[0]
uid = "_".join(utt.split("_")[1:])
pitch_path = os.path.join(
processed_dir, dataset, pitch_dir, f"{uid}.npy"
)
pitch = np.load(pitch_path)
assert len(pitch) > 0
uv = pitch != 0
utt2uv[utt] = uv
if use_log_scale:
nonzero_idxes = np.where(pitch != 0)[0]
pitch[nonzero_idxes] = np.log(pitch[nonzero_idxes])
utt2pitch[utt] = pitch
pitch_scaler.partial_fit(pitch.reshape(-1, 1))
mean, std = pitch_scaler.mean_[0], pitch_scaler.scale_[0]
if return_norm:
for utt in spk2utt[spk]:
pitch = utt2pitch[utt]
normalized_pitch = (pitch - mean) / std
utt2pitch[utt] = normalized_pitch
pitch_statistic.append({"spk": spk, "mean": mean, "std": std})
return utt2pitch, utt2uv, pitch_statistic
# discard
def load_phone_pitch(
meta_data,
processed_dir,
pitch_dir,
utt2dur,
use_log_scale=False,
return_norm=False,
interoperate=True,
utt2spk=None,
):
print("Load Phone Pitch")
utt2pitch = {}
utt2uv = {}
if utt2spk is None:
pitch_scaler = StandardScaler()
for utt_info in tqdm(meta_data):
utt = utt_info["Dataset"] + "_" + utt_info["Uid"]
pitch_path = os.path.join(
processed_dir, utt_info["Dataset"], pitch_dir, f'{utt_info["Uid"]}.npy'
)
frame_pitch = np.load(pitch_path)
assert len(frame_pitch) > 0
uv = frame_pitch != 0
utt2uv[utt] = uv
phone_pitch = phone_average_pitch(frame_pitch, utt2dur[utt], interoperate)
if use_log_scale:
nonzero_idxes = np.where(phone_pitch != 0)[0]
phone_pitch[nonzero_idxes] = np.log(phone_pitch[nonzero_idxes])
utt2pitch[utt] = phone_pitch
pitch_scaler.partial_fit(remove_outlier(phone_pitch).reshape(-1, 1))
mean, std = pitch_scaler.mean_[0], pitch_scaler.scale_[0]
max_value = np.finfo(np.float64).min
min_value = np.finfo(np.float64).max
if return_norm:
for utt_info in meta_data:
utt = utt_info["Dataset"] + "_" + utt_info["Uid"]
pitch = utt2pitch[utt]
normalized_pitch = (pitch - mean) / std
max_value = max(max_value, max(normalized_pitch))
min_value = min(min_value, min(normalized_pitch))
utt2pitch[utt] = normalized_pitch
phone_normalized_pitch_path = os.path.join(
processed_dir,
utt_info["Dataset"],
"phone_level_" + pitch_dir,
f'{utt_info["Uid"]}.npy',
)
pitch_statistic = {
"mean": mean,
"std": std,
"min_value": min_value,
"max_value": max_value,
}
else:
spk2utt = {}
pitch_statistic = []
for utt_info in tqdm(meta_data):
utt = utt_info["Dataset"] + "_" + utt_info["Uid"]
if not utt2spk[utt] in spk2utt:
spk2utt[utt2spk[utt]] = []
spk2utt[utt2spk[utt]].append(utt)
for spk in spk2utt:
pitch_scaler = StandardScaler()
for utt in spk2utt[spk]:
dataset = utt.split("_")[0]
uid = "_".join(utt.split("_")[1:])
pitch_path = os.path.join(
processed_dir, dataset, pitch_dir, f"{uid}.npy"
)
frame_pitch = np.load(pitch_path)
assert len(frame_pitch) > 0
uv = frame_pitch != 0
utt2uv[utt] = uv
phone_pitch = phone_average_pitch(
frame_pitch, utt2dur[utt], interoperate
)
if use_log_scale:
nonzero_idxes = np.where(phone_pitch != 0)[0]
phone_pitch[nonzero_idxes] = np.log(phone_pitch[nonzero_idxes])
utt2pitch[utt] = phone_pitch
pitch_scaler.partial_fit(remove_outlier(phone_pitch).reshape(-1, 1))
mean, std = pitch_scaler.mean_[0], pitch_scaler.scale_[0]
max_value = np.finfo(np.float64).min
min_value = np.finfo(np.float64).max
if return_norm:
for utt in spk2utt[spk]:
pitch = utt2pitch[utt]
normalized_pitch = (pitch - mean) / std
max_value = max(max_value, max(normalized_pitch))
min_value = min(min_value, min(normalized_pitch))
utt2pitch[utt] = normalized_pitch
pitch_statistic.append(
{
"spk": spk,
"mean": mean,
"std": std,
"min_value": min_value,
"max_value": max_value,
}
)
return utt2pitch, utt2uv, pitch_statistic
def phone_average_pitch(pitch, dur, interoperate=False):
pos = 0
if interoperate:
nonzero_ids = np.where(pitch != 0)[0]
interp_fn = interp1d(
nonzero_ids,
pitch[nonzero_ids],
fill_value=(pitch[nonzero_ids[0]], pitch[nonzero_ids[-1]]),
bounds_error=False,
)
pitch = interp_fn(np.arange(0, len(pitch)))
phone_pitch = np.zeros(len(dur))
for i, d in enumerate(dur):
d = int(d)
if d > 0 and pos < len(pitch):
phone_pitch[i] = np.mean(pitch[pos : pos + d])
else:
phone_pitch[i] = 0
pos += d
return phone_pitch
def load_energy(
meta_data,
processed_dir,
energy_dir,
use_log_scale=False,
return_norm=False,
utt2spk=None,
):
utt2energy = {}
if utt2spk is None:
for utt_info in meta_data:
utt = utt_info["Dataset"] + "_" + utt_info["Uid"]
energy_path = os.path.join(
processed_dir, utt_info["Dataset"], energy_dir, f'{utt_info["Uid"]}.npy'
)
if not os.path.exists(energy_path):
continue
energy = np.load(energy_path)
assert len(energy) > 0
if use_log_scale:
nonzero_idxes = np.where(energy != 0)[0]
energy[nonzero_idxes] = np.log(energy[nonzero_idxes])
utt2energy[utt] = energy
if return_norm:
with open(
os.path.join(
processed_dir, utt_info["Dataset"], energy_dir, "statistics.json"
)
) as f:
stats = json.load(f)
mean, std = (
stats[utt_info["Dataset"] + "_" + utt_info["Singer"]][
"voiced_positions"
]["mean"],
stats["LJSpeech_LJSpeech"]["voiced_positions"]["std"],
)
for utt in utt2energy.keys():
energy = utt2energy[utt]
normalized_energy = (energy - mean) / std
utt2energy[utt] = normalized_energy
energy_statistic = {"mean": mean, "std": std}
else:
spk2utt = {}
energy_statistic = []
for utt_info in meta_data:
utt = utt_info["Dataset"] + "_" + utt_info["Uid"]
if not utt2spk[utt] in spk2utt:
spk2utt[utt2spk[utt]] = []
spk2utt[utt2spk[utt]].append(utt)
for spk in spk2utt:
energy_scaler = StandardScaler()
for utt in spk2utt[spk]:
dataset = utt.split("_")[0]
uid = "_".join(utt.split("_")[1:])
energy_path = os.path.join(
processed_dir, dataset, energy_dir, f"{uid}.npy"
)
if not os.path.exists(energy_path):
continue
frame_energy = np.load(energy_path)
assert len(frame_energy) > 0
if use_log_scale:
nonzero_idxes = np.where(frame_energy != 0)[0]
frame_energy[nonzero_idxes] = np.log(frame_energy[nonzero_idxes])
utt2energy[utt] = frame_energy
energy_scaler.partial_fit(frame_energy.reshape(-1, 1))
mean, std = energy_scaler.mean_[0], energy_scaler.scale_[0]
if return_norm:
for utt in spk2utt[spk]:
energy = utt2energy[utt]
normalized_energy = (energy - mean) / std
utt2energy[utt] = normalized_energy
energy_statistic.append({"spk": spk, "mean": mean, "std": std})
return utt2energy, energy_statistic
def load_frame_energy(
meta_data,
processed_dir,
energy_dir,
use_log_scale=False,
return_norm=False,
interoperate=False,
utt2spk=None,
):
utt2energy = {}
if utt2spk is None:
energy_scaler = StandardScaler()
for utt_info in meta_data:
utt = utt_info["Dataset"] + "_" + utt_info["Uid"]
energy_path = os.path.join(
processed_dir, utt_info["Dataset"], energy_dir, f'{utt_info["Uid"]}.npy'
)
frame_energy = np.load(energy_path)
assert len(frame_energy) > 0
if use_log_scale:
nonzero_idxes = np.where(frame_energy != 0)[0]
frame_energy[nonzero_idxes] = np.log(frame_energy[nonzero_idxes])
utt2energy[utt] = frame_energy
energy_scaler.partial_fit(frame_energy.reshape(-1, 1))
mean, std = energy_scaler.mean_[0], energy_scaler.scale_[0]
if return_norm:
for utt_info in meta_data:
utt = utt_info["Dataset"] + "_" + utt_info["Uid"]
energy = utt2energy[utt]
normalized_energy = (energy - mean) / std
utt2energy[utt] = normalized_energy
energy_statistic = {"mean": mean, "std": std}
else:
spk2utt = {}
energy_statistic = []
for utt_info in meta_data:
utt = utt_info["Dataset"] + "_" + utt_info["Uid"]
if not utt2spk[utt] in spk2utt:
spk2utt[utt2spk[utt]] = []
spk2utt[utt2spk[utt]].append(utt)
for spk in spk2utt:
energy_scaler = StandardScaler()
for utt in spk2utt[spk]:
dataset = utt.split("_")[0]
uid = "_".join(utt.split("_")[1:])
energy_path = os.path.join(
processed_dir, dataset, energy_dir, f"{uid}.npy"
)
frame_energy = np.load(energy_path)
assert len(frame_energy) > 0
if use_log_scale:
nonzero_idxes = np.where(frame_energy != 0)[0]
frame_energy[nonzero_idxes] = np.log(frame_energy[nonzero_idxes])
utt2energy[utt] = frame_energy
energy_scaler.partial_fit(frame_energy.reshape(-1, 1))
mean, std = energy_scaler.mean_[0], energy_scaler.scale_[0]
if return_norm:
for utt in spk2utt[spk]:
energy = utt2energy[utt]
normalized_energy = (energy - mean) / std
utt2energy[utt] = normalized_energy
energy_statistic.append({"spk": spk, "mean": mean, "std": std})
return utt2energy, energy_statistic
def align_length(feature, target_len, pad_value=0.0):
feature_len = feature.shape[-1]
dim = len(feature.shape)
# align 1-D data
if dim == 2:
if target_len > feature_len:
feature = np.pad(
feature,
((0, 0), (0, target_len - feature_len)),
constant_values=pad_value,
)
else:
feature = feature[:, :target_len]
# align 2-D data
elif dim == 1:
if target_len > feature_len:
feature = np.pad(
feature, (0, target_len - feature_len), constant_values=pad_value
)
else:
feature = feature[:target_len]
else:
raise NotImplementedError
return feature
def align_whisper_feauture_length(
feature, target_len, fast_mapping=True, source_hop=320, target_hop=256
):
factor = np.gcd(source_hop, target_hop)
source_hop //= factor
target_hop //= factor
# print(
# "Mapping source's {} frames => target's {} frames".format(
# target_hop, source_hop
# )
# )
max_source_len = 1500
target_len = min(target_len, max_source_len * source_hop // target_hop)
width = feature.shape[-1]
if fast_mapping:
source_len = target_len * target_hop // source_hop + 1
feature = feature[:source_len]
else:
source_len = max_source_len
# const ~= target_len * target_hop
const = source_len * source_hop // target_hop * target_hop
# (source_len * source_hop, dim)
up_sampling_feats = np.repeat(feature, source_hop, axis=0)
# (const, dim) -> (const/target_hop, target_hop, dim) -> (const/target_hop, dim)
down_sampling_feats = np.average(
up_sampling_feats[:const].reshape(-1, target_hop, width), axis=1
)
assert len(down_sampling_feats) >= target_len
# (target_len, dim)
feat = down_sampling_feats[:target_len]
return feat
def align_content_feature_length(feature, target_len, source_hop=320, target_hop=256):
factor = np.gcd(source_hop, target_hop)
source_hop //= factor
target_hop //= factor
# print(
# "Mapping source's {} frames => target's {} frames".format(
# target_hop, source_hop
# )
# )
# (source_len, 256)
source_len, width = feature.shape
# const ~= target_len * target_hop
const = source_len * source_hop // target_hop * target_hop
# (source_len * source_hop, dim)
up_sampling_feats = np.repeat(feature, source_hop, axis=0)
# (const, dim) -> (const/target_hop, target_hop, dim) -> (const/target_hop, dim)
down_sampling_feats = np.average(
up_sampling_feats[:const].reshape(-1, target_hop, width), axis=1
)
err = abs(target_len - len(down_sampling_feats))
if err > 4: ## why 4 not 3?
print("target_len:", target_len)
print("raw feature:", feature.shape)
print("up_sampling:", up_sampling_feats.shape)
print("down_sampling_feats:", down_sampling_feats.shape)
exit()
if len(down_sampling_feats) < target_len:
# (1, dim) -> (err, dim)
end = down_sampling_feats[-1][None, :].repeat(err, axis=0)
down_sampling_feats = np.concatenate([down_sampling_feats, end], axis=0)
# (target_len, dim)
feat = down_sampling_feats[:target_len]
return feat
def remove_outlier(values):
values = np.array(values)
p25 = np.percentile(values, 25)
p75 = np.percentile(values, 75)
lower = p25 - 1.5 * (p75 - p25)
upper = p75 + 1.5 * (p75 - p25)
normal_indices = np.logical_and(values > lower, values < upper)
return values[normal_indices]