maskgct / utils /util.py
Hecheng0625's picture
Upload 61 files
7ee3434 verified
raw
history blame
20.3 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 collections
import glob
import os
import random
import time
import argparse
from collections import OrderedDict
import json5
import numpy as np
import glob
from torch.nn import functional as F
try:
from ruamel.yaml import YAML as yaml
except:
from ruamel_yaml import YAML as yaml
import torch
from utils.hparam import HParams
import logging
from logging import handlers
def str2bool(v):
"""Used in argparse.ArgumentParser.add_argument to indicate
that a type is a bool type and user can enter
- yes, true, t, y, 1, to represent True
- no, false, f, n, 0, to represent False
See https://stackoverflow.com/questions/15008758/parsing-boolean-values-with-argparse # noqa
"""
if isinstance(v, bool):
return v
if v.lower() in ("yes", "true", "t", "y", "1"):
return True
elif v.lower() in ("no", "false", "f", "n", "0"):
return False
else:
raise argparse.ArgumentTypeError("Boolean value expected.")
def find_checkpoint_of_mapper(mapper_ckpt_dir):
mapper_ckpts = glob.glob(os.path.join(mapper_ckpt_dir, "ckpts/*.pt"))
# Select the max steps
mapper_ckpts.sort()
mapper_weights_file = mapper_ckpts[-1]
return mapper_weights_file
def pad_f0_to_tensors(f0s, batched=None):
# Initialize
tensors = []
if batched == None:
# Get the max frame for padding
size = -1
for f0 in f0s:
size = max(size, f0.shape[-1])
tensor = torch.zeros(len(f0s), size)
for i, f0 in enumerate(f0s):
tensor[i, : f0.shape[-1]] = f0[:]
tensors.append(tensor)
else:
start = 0
while start + batched - 1 < len(f0s):
end = start + batched - 1
# Get the max frame for padding
size = -1
for i in range(start, end + 1):
size = max(size, f0s[i].shape[-1])
tensor = torch.zeros(batched, size)
for i in range(start, end + 1):
tensor[i - start, : f0s[i].shape[-1]] = f0s[i][:]
tensors.append(tensor)
start = start + batched
if start != len(f0s):
end = len(f0s)
# Get the max frame for padding
size = -1
for i in range(start, end):
size = max(size, f0s[i].shape[-1])
tensor = torch.zeros(len(f0s) - start, size)
for i in range(start, end):
tensor[i - start, : f0s[i].shape[-1]] = f0s[i][:]
tensors.append(tensor)
return tensors
def pad_mels_to_tensors(mels, batched=None):
"""
Args:
mels: A list of mel-specs
Returns:
tensors: A list of tensors containing the batched mel-specs
mel_frames: A list of tensors containing the frames of the original mel-specs
"""
# Initialize
tensors = []
mel_frames = []
# Split mel-specs into batches to avoid cuda memory exceed
if batched == None:
# Get the max frame for padding
size = -1
for mel in mels:
size = max(size, mel.shape[-1])
tensor = torch.zeros(len(mels), mels[0].shape[0], size)
mel_frame = torch.zeros(len(mels), dtype=torch.int32)
for i, mel in enumerate(mels):
tensor[i, :, : mel.shape[-1]] = mel[:]
mel_frame[i] = mel.shape[-1]
tensors.append(tensor)
mel_frames.append(mel_frame)
else:
start = 0
while start + batched - 1 < len(mels):
end = start + batched - 1
# Get the max frame for padding
size = -1
for i in range(start, end + 1):
size = max(size, mels[i].shape[-1])
tensor = torch.zeros(batched, mels[0].shape[0], size)
mel_frame = torch.zeros(batched, dtype=torch.int32)
for i in range(start, end + 1):
tensor[i - start, :, : mels[i].shape[-1]] = mels[i][:]
mel_frame[i - start] = mels[i].shape[-1]
tensors.append(tensor)
mel_frames.append(mel_frame)
start = start + batched
if start != len(mels):
end = len(mels)
# Get the max frame for padding
size = -1
for i in range(start, end):
size = max(size, mels[i].shape[-1])
tensor = torch.zeros(len(mels) - start, mels[0].shape[0], size)
mel_frame = torch.zeros(len(mels) - start, dtype=torch.int32)
for i in range(start, end):
tensor[i - start, :, : mels[i].shape[-1]] = mels[i][:]
mel_frame[i - start] = mels[i].shape[-1]
tensors.append(tensor)
mel_frames.append(mel_frame)
return tensors, mel_frames
def load_model_config(args):
"""Load model configurations (in args.json under checkpoint directory)
Args:
args (ArgumentParser): arguments to run bins/preprocess.py
Returns:
dict: dictionary that stores model configurations
"""
if args.checkpoint_dir is None:
assert args.checkpoint_file is not None
checkpoint_dir = os.path.split(args.checkpoint_file)[0]
else:
checkpoint_dir = args.checkpoint_dir
config_path = os.path.join(checkpoint_dir, "args.json")
print("config_path: ", config_path)
config = load_config(config_path)
return config
def remove_and_create(dir):
if os.path.exists(dir):
os.system("rm -r {}".format(dir))
os.makedirs(dir, exist_ok=True)
def has_existed(path, warning=False):
if not warning:
return os.path.exists(path)
if os.path.exists(path):
answer = input(
"The path {} has existed. \nInput 'y' (or hit Enter) to skip it, and input 'n' to re-write it [y/n]\n".format(
path
)
)
if not answer == "n":
return True
return False
def remove_older_ckpt(saved_model_name, checkpoint_dir, max_to_keep=5):
if os.path.exists(os.path.join(checkpoint_dir, "checkpoint")):
with open(os.path.join(checkpoint_dir, "checkpoint"), "r") as f:
ckpts = [x.strip() for x in f.readlines()]
else:
ckpts = []
ckpts.append(saved_model_name)
for item in ckpts[:-max_to_keep]:
if os.path.exists(os.path.join(checkpoint_dir, item)):
os.remove(os.path.join(checkpoint_dir, item))
with open(os.path.join(checkpoint_dir, "checkpoint"), "w") as f:
for item in ckpts[-max_to_keep:]:
f.write("{}\n".format(item))
def set_all_random_seed(seed: int):
random.seed(seed)
np.random.seed(seed)
torch.random.manual_seed(seed)
def save_checkpoint(
args,
generator,
g_optimizer,
step,
discriminator=None,
d_optimizer=None,
max_to_keep=5,
):
saved_model_name = "model.ckpt-{}.pt".format(step)
checkpoint_path = os.path.join(args.checkpoint_dir, saved_model_name)
if discriminator and d_optimizer:
torch.save(
{
"generator": generator.state_dict(),
"discriminator": discriminator.state_dict(),
"g_optimizer": g_optimizer.state_dict(),
"d_optimizer": d_optimizer.state_dict(),
"global_step": step,
},
checkpoint_path,
)
else:
torch.save(
{
"generator": generator.state_dict(),
"g_optimizer": g_optimizer.state_dict(),
"global_step": step,
},
checkpoint_path,
)
print("Saved checkpoint: {}".format(checkpoint_path))
if os.path.exists(os.path.join(args.checkpoint_dir, "checkpoint")):
with open(os.path.join(args.checkpoint_dir, "checkpoint"), "r") as f:
ckpts = [x.strip() for x in f.readlines()]
else:
ckpts = []
ckpts.append(saved_model_name)
for item in ckpts[:-max_to_keep]:
if os.path.exists(os.path.join(args.checkpoint_dir, item)):
os.remove(os.path.join(args.checkpoint_dir, item))
with open(os.path.join(args.checkpoint_dir, "checkpoint"), "w") as f:
for item in ckpts[-max_to_keep:]:
f.write("{}\n".format(item))
def attempt_to_restore(
generator, g_optimizer, checkpoint_dir, discriminator=None, d_optimizer=None
):
checkpoint_list = os.path.join(checkpoint_dir, "checkpoint")
if os.path.exists(checkpoint_list):
checkpoint_filename = open(checkpoint_list).readlines()[-1].strip()
checkpoint_path = os.path.join(checkpoint_dir, "{}".format(checkpoint_filename))
print("Restore from {}".format(checkpoint_path))
checkpoint = torch.load(checkpoint_path, map_location="cpu")
if generator:
if not list(generator.state_dict().keys())[0].startswith("module."):
raw_dict = checkpoint["generator"]
clean_dict = OrderedDict()
for k, v in raw_dict.items():
if k.startswith("module."):
clean_dict[k[7:]] = v
else:
clean_dict[k] = v
generator.load_state_dict(clean_dict)
else:
generator.load_state_dict(checkpoint["generator"])
if g_optimizer:
g_optimizer.load_state_dict(checkpoint["g_optimizer"])
global_step = 100000
if discriminator and "discriminator" in checkpoint.keys():
discriminator.load_state_dict(checkpoint["discriminator"])
global_step = checkpoint["global_step"]
print("restore discriminator")
if d_optimizer and "d_optimizer" in checkpoint.keys():
d_optimizer.load_state_dict(checkpoint["d_optimizer"])
print("restore d_optimizer...")
else:
global_step = 0
return global_step
class ExponentialMovingAverage(object):
def __init__(self, decay):
self.decay = decay
self.shadow = {}
def register(self, name, val):
self.shadow[name] = val.clone()
def update(self, name, x):
assert name in self.shadow
update_delta = self.shadow[name] - x
self.shadow[name] -= (1.0 - self.decay) * update_delta
def apply_moving_average(model, ema):
for name, param in model.named_parameters():
if name in ema.shadow:
ema.update(name, param.data)
def register_model_to_ema(model, ema):
for name, param in model.named_parameters():
if param.requires_grad:
ema.register(name, param.data)
class YParams(HParams):
def __init__(self, yaml_file):
if not os.path.exists(yaml_file):
raise IOError("yaml file: {} is not existed".format(yaml_file))
super().__init__()
self.d = collections.OrderedDict()
with open(yaml_file) as fp:
for _, v in yaml().load(fp).items():
for k1, v1 in v.items():
try:
if self.get(k1):
self.set_hparam(k1, v1)
else:
self.add_hparam(k1, v1)
self.d[k1] = v1
except Exception:
import traceback
print(traceback.format_exc())
# @property
def get_elements(self):
return self.d.items()
def override_config(base_config, new_config):
"""Update new configurations in the original dict with the new dict
Args:
base_config (dict): original dict to be overridden
new_config (dict): dict with new configurations
Returns:
dict: updated configuration dict
"""
for k, v in new_config.items():
if type(v) == dict:
if k not in base_config.keys():
base_config[k] = {}
base_config[k] = override_config(base_config[k], v)
else:
base_config[k] = v
return base_config
def get_lowercase_keys_config(cfg):
"""Change all keys in cfg to lower case
Args:
cfg (dict): dictionary that stores configurations
Returns:
dict: dictionary that stores configurations
"""
updated_cfg = dict()
for k, v in cfg.items():
if type(v) == dict:
v = get_lowercase_keys_config(v)
updated_cfg[k.lower()] = v
return updated_cfg
def _load_config(config_fn, lowercase=False):
"""Load configurations into a dictionary
Args:
config_fn (str): path to configuration file
lowercase (bool, optional): whether changing keys to lower case. Defaults to False.
Returns:
dict: dictionary that stores configurations
"""
with open(config_fn, "r") as f:
data = f.read()
config_ = json5.loads(data)
if "base_config" in config_:
# load configurations from new path
p_config_path = os.path.join(os.getenv("WORK_DIR"), config_["base_config"])
p_config_ = _load_config(p_config_path)
config_ = override_config(p_config_, config_)
if lowercase:
# change keys in config_ to lower case
config_ = get_lowercase_keys_config(config_)
return config_
def load_config(config_fn, lowercase=False):
"""Load configurations into a dictionary
Args:
config_fn (str): path to configuration file
lowercase (bool, optional): _description_. Defaults to False.
Returns:
JsonHParams: an object that stores configurations
"""
config_ = _load_config(config_fn, lowercase=lowercase)
# create an JsonHParams object with configuration dict
cfg = JsonHParams(**config_)
return cfg
def save_config(save_path, cfg):
"""Save configurations into a json file
Args:
save_path (str): path to save configurations
cfg (dict): dictionary that stores configurations
"""
with open(save_path, "w") as f:
json5.dump(
cfg, f, ensure_ascii=False, indent=4, quote_keys=True, sort_keys=True
)
class JsonHParams:
def __init__(self, **kwargs):
for k, v in kwargs.items():
if type(v) == dict:
v = JsonHParams(**v)
self[k] = v
def keys(self):
return self.__dict__.keys()
def items(self):
return self.__dict__.items()
def values(self):
return self.__dict__.values()
def __len__(self):
return len(self.__dict__)
def __getitem__(self, key):
return getattr(self, key)
def __setitem__(self, key, value):
return setattr(self, key, value)
def __contains__(self, key):
return key in self.__dict__
def __repr__(self):
return self.__dict__.__repr__()
class ValueWindow:
def __init__(self, window_size=100):
self._window_size = window_size
self._values = []
def append(self, x):
self._values = self._values[-(self._window_size - 1) :] + [x]
@property
def sum(self):
return sum(self._values)
@property
def count(self):
return len(self._values)
@property
def average(self):
return self.sum / max(1, self.count)
def reset(self):
self._values = []
class Logger(object):
def __init__(
self,
filename,
level="info",
when="D",
backCount=10,
fmt="%(asctime)s : %(message)s",
):
self.level_relations = {
"debug": logging.DEBUG,
"info": logging.INFO,
"warning": logging.WARNING,
"error": logging.ERROR,
"crit": logging.CRITICAL,
}
if level == "debug":
fmt = "%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s"
self.logger = logging.getLogger(filename)
format_str = logging.Formatter(fmt)
self.logger.setLevel(self.level_relations.get(level))
sh = logging.StreamHandler()
sh.setFormatter(format_str)
th = handlers.TimedRotatingFileHandler(
filename=filename, when=when, backupCount=backCount, encoding="utf-8"
)
th.setFormatter(format_str)
self.logger.addHandler(sh)
self.logger.addHandler(th)
self.logger.info(
"==========================New Starting Here=============================="
)
def init_weights(m, mean=0.0, std=0.01):
classname = m.__class__.__name__
if classname.find("Conv") != -1:
m.weight.data.normal_(mean, std)
def get_padding(kernel_size, dilation=1):
return int((kernel_size * dilation - dilation) / 2)
def slice_segments(x, ids_str, segment_size=4):
ret = torch.zeros_like(x[:, :, :segment_size])
for i in range(x.size(0)):
idx_str = ids_str[i]
idx_end = idx_str + segment_size
ret[i] = x[i, :, idx_str:idx_end]
return ret
def rand_slice_segments(x, x_lengths=None, segment_size=4):
b, d, t = x.size()
if x_lengths is None:
x_lengths = t
ids_str_max = x_lengths - segment_size + 1
ids_str = (torch.rand([b]).to(device=x.device) * ids_str_max).to(dtype=torch.long)
ret = slice_segments(x, ids_str, segment_size)
return ret, ids_str
def subsequent_mask(length):
mask = torch.tril(torch.ones(length, length)).unsqueeze(0).unsqueeze(0)
return mask
@torch.jit.script
def fused_add_tanh_sigmoid_multiply(input_a, input_b, n_channels):
n_channels_int = n_channels[0]
in_act = input_a + input_b
t_act = torch.tanh(in_act[:, :n_channels_int, :])
s_act = torch.sigmoid(in_act[:, n_channels_int:, :])
acts = t_act * s_act
return acts
def convert_pad_shape(pad_shape):
l = pad_shape[::-1]
pad_shape = [item for sublist in l for item in sublist]
return pad_shape
def sequence_mask(length, max_length=None):
if max_length is None:
max_length = length.max()
x = torch.arange(max_length, dtype=length.dtype, device=length.device)
return x.unsqueeze(0) < length.unsqueeze(1)
def generate_path(duration, mask):
"""
duration: [b, 1, t_x]
mask: [b, 1, t_y, t_x]
"""
device = duration.device
b, _, t_y, t_x = mask.shape
cum_duration = torch.cumsum(duration, -1)
cum_duration_flat = cum_duration.view(b * t_x)
path = sequence_mask(cum_duration_flat, t_y).to(mask.dtype)
path = path.view(b, t_x, t_y)
path = path - F.pad(path, convert_pad_shape([[0, 0], [1, 0], [0, 0]]))[:, :-1]
path = path.unsqueeze(1).transpose(2, 3) * mask
return path
def clip_grad_value_(parameters, clip_value, norm_type=2):
if isinstance(parameters, torch.Tensor):
parameters = [parameters]
parameters = list(filter(lambda p: p.grad is not None, parameters))
norm_type = float(norm_type)
if clip_value is not None:
clip_value = float(clip_value)
total_norm = 0
for p in parameters:
param_norm = p.grad.data.norm(norm_type)
total_norm += param_norm.item() ** norm_type
if clip_value is not None:
p.grad.data.clamp_(min=-clip_value, max=clip_value)
total_norm = total_norm ** (1.0 / norm_type)
return total_norm
def get_current_time():
pass
def make_pad_mask(lengths: torch.Tensor, max_len: int = 0) -> torch.Tensor:
"""
Args:
lengths:
A 1-D tensor containing sentence lengths.
max_len:
The length of masks.
Returns:
Return a 2-D bool tensor, where masked positions
are filled with `True` and non-masked positions are
filled with `False`.
>>> lengths = torch.tensor([1, 3, 2, 5])
>>> make_pad_mask(lengths)
tensor([[False, True, True, True, True],
[False, False, False, True, True],
[False, False, True, True, True],
[False, False, False, False, False]])
"""
assert lengths.ndim == 1, lengths.ndim
max_len = max(max_len, lengths.max())
n = lengths.size(0)
seq_range = torch.arange(0, max_len, device=lengths.device)
expaned_lengths = seq_range.unsqueeze(0).expand(n, max_len)
return expaned_lengths >= lengths.unsqueeze(-1)