| |
| |
| |
| |
|
|
| import os |
| import json |
| import torchaudio |
| from tqdm import tqdm |
| from glob import glob |
| from collections import defaultdict |
|
|
| from utils.util import has_existed |
|
|
|
|
| def main(output_path, dataset_path): |
| print("-" * 10) |
| print("Preparing samples for hifitts...\n") |
|
|
| save_dir = os.path.join(output_path, "hifitts") |
| os.makedirs(save_dir, exist_ok=True) |
| print("Saving to ", save_dir) |
|
|
| train_output_file = os.path.join(save_dir, "train.json") |
| test_output_file = os.path.join(save_dir, "test.json") |
| valid_output_file = os.path.join(save_dir, "valid.json") |
| singer_dict_file = os.path.join(save_dir, "singers.json") |
| utt2singer_file = os.path.join(save_dir, "utt2singer") |
| if has_existed(train_output_file): |
| return |
| utt2singer = open(utt2singer_file, "w") |
|
|
| hifitts_path = dataset_path |
|
|
| speakers = [] |
|
|
| train = [] |
| test = [] |
| valid = [] |
|
|
| train_index_count = 0 |
| test_index_count = 0 |
| valid_index_count = 0 |
|
|
| train_total_duration = 0 |
| test_total_duration = 0 |
| valid_total_duration = 0 |
|
|
| distribution_infos = glob(hifitts_path + "/*.json") |
|
|
| for distribution_info in tqdm( |
| distribution_infos, desc="Extracting metadata from distributions" |
| ): |
| distribution = distribution_info.split("/")[-1].split(".")[0] |
| speaker_id = distribution.split("_")[0] |
| speakers.append(speaker_id) |
|
|
| with open(distribution_info, "r", encoding="utf-8") as file: |
| for line in file: |
| entry = json.loads(line) |
| utt_path = entry.get("audio_filepath") |
| chosen_book = utt_path.split("/")[-2] |
| chosen_uid = utt_path.split("/")[-1].split(".")[0] |
| duration = entry.get("duration") |
| text = entry.get("text_normalized") |
| path = os.path.join(hifitts_path, utt_path) |
| assert os.path.exists(path) |
|
|
| res = { |
| "Dataset": "hifitts", |
| "Singer": speaker_id, |
| "Uid": "{}#{}#{}#{}".format( |
| distribution, speaker_id, chosen_book, chosen_uid |
| ), |
| "Text": text, |
| "Path": path, |
| "Duration": duration, |
| } |
|
|
| if "train" in distribution: |
| res["index"] = train_index_count |
| train_total_duration += duration |
| train.append(res) |
| train_index_count += 1 |
|
|
| elif "test" in distribution: |
| res["index"] = test_index_count |
| test_total_duration += duration |
| test.append(res) |
| test_index_count += 1 |
|
|
| elif "dev" in distribution: |
| res["index"] = valid_index_count |
| valid_total_duration += duration |
| valid.append(res) |
| valid_index_count += 1 |
|
|
| utt2singer.write("{}\t{}\n".format(res["Uid"], res["Singer"])) |
|
|
| unique_speakers = list(set(speakers)) |
| unique_speakers.sort() |
|
|
| print("Speakers: \n{}".format("\t".join(unique_speakers))) |
|
|
| print( |
| "#Train = {}, #Test = {}, #Valid = {}".format(len(train), len(test), len(valid)) |
| ) |
| print( |
| "#Train hours= {}, #Test hours= {}, #Valid hours= {}".format( |
| train_total_duration / 3600, |
| test_total_duration / 3600, |
| valid_total_duration / 3600, |
| ) |
| ) |
|
|
| |
| with open(train_output_file, "w") as f: |
| json.dump(train, f, indent=4, ensure_ascii=False) |
| with open(test_output_file, "w") as f: |
| json.dump(test, f, indent=4, ensure_ascii=False) |
| with open(valid_output_file, "w") as f: |
| json.dump(valid, f, indent=4, ensure_ascii=False) |
|
|
| |
| singer_lut = {name: i for i, name in enumerate(unique_speakers)} |
| with open(singer_dict_file, "w") as f: |
| json.dump(singer_lut, f, indent=4, ensure_ascii=False) |
|
|
