preprocessors/vctkfewsinger.py

# 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 os
import json
import pickle
import glob
from collections import defaultdict
from tqdm import tqdm


# Train: male 20 hours, female 10 hours
TRAIN_MALE_MAX_SECONDS = 20 * 3600
TRAIN_FEMALE_MAX_SECONDS = 10 * 3600
TEST_MAX_NUM_EVERY_PERSON = 5


def select_sample_idxs():
    chosen_speakers = get_chosen_speakers()

    with open(os.path.join(vctk_dir, "train.json"), "r") as f:
        raw_train = json.load(f)
    with open(os.path.join(vctk_dir, "test.json"), "r") as f:
        raw_test = json.load(f)

    train_idxs, test_idxs = [], []

    # =========== Test ===========
    test_nums = defaultdict(int)
    for utt in tqdm(raw_train):
        idx = utt["index"]
        singer = utt["Singer"]

        if singer in chosen_speakers and test_nums[singer] < TEST_MAX_NUM_EVERY_PERSON:
            test_nums[singer] += 1
            test_idxs.append("train_{}".format(idx))

    for utt in tqdm(raw_test):
        idx = utt["index"]
        singer = utt["Singer"]

        if singer in chosen_speakers and test_nums[singer] < TEST_MAX_NUM_EVERY_PERSON:
            test_nums[singer] += 1
            test_idxs.append("test_{}".format(idx))

    # =========== Train ===========
    for utt in tqdm(raw_train):
        idx = utt["index"]
        singer = utt["Singer"]

        if singer in chosen_speakers and "train_{}".format(idx) not in test_idxs:
            train_idxs.append("train_{}".format(idx))

    for utt in tqdm(raw_test):
        idx = utt["index"]
        singer = utt["Singer"]

        if singer in chosen_speakers and "test_{}".format(idx) not in test_idxs:
            train_idxs.append("test_{}".format(idx))

    train_idxs.sort()
    test_idxs.sort()
    return train_idxs, test_idxs, raw_train, raw_test


def statistics_of_speakers():
    speaker2time = defaultdict(float)
    sex2time = defaultdict(float)

    with open(os.path.join(vctk_dir, "train.json"), "r") as f:
        train = json.load(f)
    with open(os.path.join(vctk_dir, "test.json"), "r") as f:
        test = json.load(f)

    for utt in train + test:
        # minutes
        speaker2time[utt["Singer"]] += utt["Duration"]
        # hours
        sex2time[utt["Singer"].split("_")[0]] += utt["Duration"]

    print(
        "Female: {:.2f} hours, Male: {:.2f} hours.\n".format(
            sex2time["female"] / 3600, sex2time["male"] / 3600
        )
    )

    speaker2time = sorted(speaker2time.items(), key=lambda x: x[-1], reverse=True)
    for singer, seconds in speaker2time:
        print("{}\t{:.2f} mins".format(singer, seconds / 60))

    return speaker2time


def get_chosen_speakers():
    speaker2time = statistics_of_speakers()

    chosen_time = defaultdict(float)
    chosen_speaker = defaultdict(list)
    train_constrait = {
        "male": TRAIN_MALE_MAX_SECONDS,
        "female": TRAIN_FEMALE_MAX_SECONDS,
    }

    for speaker, seconds in speaker2time:
        sex = speaker.split("_")[0]
        if chosen_time[sex] < train_constrait[sex]:
            chosen_time[sex] += seconds
            chosen_speaker[sex].append(speaker)

    speaker2time = dict(speaker2time)
    chosen_speaker = chosen_speaker["male"] + chosen_speaker["female"]
    print("\n#Chosen speakers = {}".format(len(chosen_speaker)))
    for spk in chosen_speaker:
        print("{}\t{:.2f} mins".format(spk, speaker2time[spk] / 60))

    return chosen_speaker


if __name__ == "__main__":
    root_path = ""
    vctk_dir = os.path.join(root_path, "vctk")
    fewspeaker_dir = os.path.join(root_path, "vctkfewspeaker")
    os.makedirs(fewspeaker_dir, exist_ok=True)

    train_idxs, test_idxs, raw_train, raw_test = select_sample_idxs()
    print("#Train = {}, #Test = {}".format(len(train_idxs), len(test_idxs)))

    # There are no data leakage
    assert len(set(train_idxs).intersection(set(test_idxs))) == 0
    for idx in train_idxs + test_idxs:
        # No test data of raw vctk
        assert "test_" not in idx

    for split, chosen_idxs in zip(["train", "test"], [train_idxs, test_idxs]):
        print("{}: #chosen idx = {}\n".format(split, len(chosen_idxs)))

        # Select features
        feat_files = glob.glob("**/train.pkl", root_dir=vctk_dir, recursive=True)
        for file in tqdm(feat_files):
            raw_file = os.path.join(vctk_dir, file)
            new_file = os.path.join(
                fewspeaker_dir, file.replace("train.pkl", "{}.pkl".format(split))
            )

            new_dir = "/".join(new_file.split("/")[:-1])
            os.makedirs(new_dir, exist_ok=True)

            if "mel_min" in file or "mel_max" in file:
                os.system("cp {} {}".format(raw_file, new_file))
                continue

            with open(raw_file, "rb") as f:
                raw_feats = pickle.load(f)

            print("file: {}, #raw_feats = {}".format(file, len(raw_feats)))
            new_feats = []
            for idx in chosen_idxs:
                chosen_split_is_train, raw_idx = idx.split("_")
                assert chosen_split_is_train == "train"
                new_feats.append(raw_feats[int(raw_idx)])

            with open(new_file, "wb") as f:
                pickle.dump(new_feats, f)
            print("New file: {}, #new_feats = {}".format(new_file, len(new_feats)))

        # Utterance re-index
        news_utts = [raw_train[int(idx.split("_")[-1])] for idx in chosen_idxs]
        for i, utt in enumerate(news_utts):
            utt["Dataset"] = "vctkfewsinger"
            utt["index"] = i

        with open(os.path.join(fewspeaker_dir, "{}.json".format(split)), "w") as f:
            json.dump(news_utts, f, indent=4)