| import json
|
| from collections import defaultdict
|
| from random import shuffle
|
| from typing import Optional
|
|
|
| from tqdm import tqdm
|
| import click
|
| from text.cleaner import clean_text_bert
|
| import os
|
| import torch
|
| from text.symbols import symbols, num_languages, num_tones
|
|
|
| @click.command()
|
| @click.option(
|
| "--metadata",
|
| default="data/example/metadata.list",
|
| type=click.Path(exists=True, file_okay=True, dir_okay=False),
|
| )
|
| @click.option("--cleaned-path", default=None)
|
| @click.option("--train-path", default=None)
|
| @click.option("--val-path", default=None)
|
| @click.option(
|
| "--config_path",
|
| default="configs/config.json",
|
| type=click.Path(exists=True, file_okay=True, dir_okay=False),
|
| )
|
| @click.option("--val-per-spk", default=4)
|
| @click.option("--max-val-total", default=8)
|
| @click.option("--clean/--no-clean", default=True)
|
| def main(
|
| metadata: str,
|
| cleaned_path: Optional[str],
|
| train_path: str,
|
| val_path: str,
|
| config_path: str,
|
| val_per_spk: int,
|
| max_val_total: int,
|
| clean: bool,
|
| ):
|
| if train_path is None:
|
| train_path = os.path.join(os.path.dirname(metadata), 'train.list')
|
| if val_path is None:
|
| val_path = os.path.join(os.path.dirname(metadata), 'val.list')
|
| out_config_path = os.path.join(os.path.dirname(metadata), 'config.json')
|
|
|
| if cleaned_path is None:
|
| cleaned_path = metadata + ".cleaned"
|
|
|
| if clean:
|
| out_file = open(cleaned_path, "w", encoding="utf-8")
|
| new_symbols = []
|
| for line in tqdm(open(metadata, encoding="utf-8").readlines()):
|
| try:
|
| utt, spk, language, text = line.strip().split("|")
|
| norm_text, phones, tones, word2ph, bert = clean_text_bert(text, language, device='cuda:0')
|
| for ph in phones:
|
| if ph not in symbols and ph not in new_symbols:
|
| new_symbols.append(ph)
|
| print('update!, now symbols:')
|
| print(new_symbols)
|
| with open(f'{language}_symbol.txt', 'w') as f:
|
| f.write(f'{new_symbols}')
|
|
|
| assert len(phones) == len(tones)
|
| assert len(phones) == sum(word2ph)
|
| out_file.write(
|
| "{}|{}|{}|{}|{}|{}|{}\n".format(
|
| utt,
|
| spk,
|
| language,
|
| norm_text,
|
| " ".join(phones),
|
| " ".join([str(i) for i in tones]),
|
| " ".join([str(i) for i in word2ph]),
|
| )
|
| )
|
| bert_path = utt.replace(".wav", ".bert.pt")
|
| os.makedirs(os.path.dirname(bert_path), exist_ok=True)
|
| torch.save(bert.cpu(), bert_path)
|
| except Exception as error:
|
| print("err!", line, error)
|
|
|
| out_file.close()
|
|
|
| metadata = cleaned_path
|
|
|
| spk_utt_map = defaultdict(list)
|
| spk_id_map = {}
|
| current_sid = 0
|
|
|
| with open(metadata, encoding="utf-8") as f:
|
| for line in f.readlines():
|
| utt, spk, language, text, phones, tones, word2ph = line.strip().split("|")
|
| spk_utt_map[spk].append(line)
|
|
|
| if spk not in spk_id_map.keys():
|
| spk_id_map[spk] = current_sid
|
| current_sid += 1
|
|
|
| train_list = []
|
| val_list = []
|
|
|
| for spk, utts in spk_utt_map.items():
|
| shuffle(utts)
|
| val_list += utts[:val_per_spk]
|
| train_list += utts[val_per_spk:]
|
|
|
| if len(val_list) > max_val_total:
|
| train_list += val_list[max_val_total:]
|
| val_list = val_list[:max_val_total]
|
|
|
| with open(train_path, "w", encoding="utf-8") as f:
|
| for line in train_list:
|
| f.write(line)
|
|
|
| with open(val_path, "w", encoding="utf-8") as f:
|
| for line in val_list:
|
| f.write(line)
|
|
|
| config = json.load(open(config_path, encoding="utf-8"))
|
| config["data"]["spk2id"] = spk_id_map
|
|
|
| config["data"]["training_files"] = train_path
|
| config["data"]["validation_files"] = val_path
|
| config["data"]["n_speakers"] = len(spk_id_map)
|
| config["num_languages"] = num_languages
|
| config["num_tones"] = num_tones
|
| config["symbols"] = symbols
|
|
|
| with open(out_config_path, "w", encoding="utf-8") as f:
|
| json.dump(config, f, indent=2, ensure_ascii=False)
|
|
|
|
|
| if __name__ == "__main__":
|
| main()
|
|
|
