Spaces:
Running
Running
import argparse | |
from multiprocessing import Pool, cpu_count | |
import torch | |
import torch.multiprocessing as mp | |
from tqdm import tqdm | |
import utils | |
from config import config | |
from clap_wrapper import get_clap_audio_feature | |
import librosa | |
import os | |
os.environ["OMP_NUM_THREADS"] = "1" | |
os.environ["MKL_NUM_THREADS"] = "1" | |
def process_line(line): | |
device = config.emo_gen_config.device | |
if config.emo_gen_config.use_multi_device: | |
rank = mp.current_process()._identity | |
rank = rank[0] if len(rank) > 0 else 0 | |
if torch.cuda.is_available(): | |
gpu_id = rank % torch.cuda.device_count() | |
device = torch.device(f"cuda:{gpu_id}") | |
else: | |
device = torch.device("cpu") | |
wav_path, _, language_str, text, phones, tone, word2ph = line.strip().split("|") | |
clap_path = wav_path.replace(".WAV", ".wav").replace(".wav", ".emo.pt") | |
if os.path.isfile(clap_path): | |
return | |
audio = librosa.load(wav_path, 48000)[0] | |
# audio = librosa.resample(audio, 44100, 48000) | |
clap = get_clap_audio_feature(audio, device) | |
torch.save(clap, clap_path) | |
if __name__ == "__main__": | |
parser = argparse.ArgumentParser() | |
parser.add_argument( | |
"-c", "--config", type=str, default=config.emo_gen_config.config_path | |
) | |
parser.add_argument( | |
"--num_processes", type=int, default=config.emo_gen_config.num_processes | |
) | |
args, _ = parser.parse_known_args() | |
config_path = args.config | |
hps = utils.get_hparams_from_file(config_path) | |
lines = [] | |
with open(hps.data.training_files, encoding="utf-8") as f: | |
lines.extend(f.readlines()) | |
with open(hps.data.validation_files, encoding="utf-8") as f: | |
lines.extend(f.readlines()) | |
if len(lines) != 0: | |
num_processes = min(args.num_processes, cpu_count()) | |
with Pool(processes=num_processes) as pool: | |
for _ in tqdm(pool.imap_unordered(process_line, lines), total=len(lines)): | |
pass | |
print(f"clap生成完毕!, 共有{len(lines)}个emo.pt生成!") | |