| 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() |
|
|
