fairseq/examples/speech_text_joint_to_text/scripts/g2p_encode.py

# Copyright (c) Facebook, Inc. and its affiliates.
#
# This source code is licensed under the MIT license found in the
# LICENSE file in the root directory of this source tree.

import argparse
import itertools
import logging
import re
import time

from g2p_en import G2p

logger = logging.getLogger(__name__)

FAIL_SENT = "FAILED_SENTENCE"


def parse():
    parser = argparse.ArgumentParser()
    parser.add_argument("--data-path", type=str, required=True)
    parser.add_argument("--out-path", type=str, required=True)
    parser.add_argument("--lower-case", action="store_true")
    parser.add_argument("--do-filter", action="store_true")
    parser.add_argument("--use-word-start", action="store_true")
    parser.add_argument("--dup-vowel", default=1, type=int)
    parser.add_argument("--dup-consonant", default=1, type=int)
    parser.add_argument("--no-punc", action="store_true")
    parser.add_argument("--reserve-word", type=str, default="")
    parser.add_argument(
        "--reserve-first-column",
        action="store_true",
        help="first column is sentence id",
    )
    ###
    parser.add_argument("--parallel-process-num", default=1, type=int)
    parser.add_argument("--logdir", default="")
    args = parser.parse_args()
    return args


def process_sent(sent, g2p, res_wrds, args):
    sents = pre_process_sent(sent, args.do_filter, args.lower_case, res_wrds)
    pho_seqs = [do_g2p(g2p, s, res_wrds, i == 0) for i, s in enumerate(sents)]
    pho_seq = (
        [FAIL_SENT]
        if [FAIL_SENT] in pho_seqs
        else list(itertools.chain.from_iterable(pho_seqs))
    )
    if args.no_punc:
        pho_seq = remove_punc(pho_seq)
    if args.dup_vowel > 1 or args.dup_consonant > 1:
        pho_seq = dup_pho(pho_seq, args.dup_vowel, args.dup_consonant)
    if args.use_word_start:
        pho_seq = add_word_start(pho_seq)
    return " ".join(pho_seq)


def remove_punc(sent):
    ns = []
    regex = re.compile("[^a-zA-Z0-9 ]")
    for p in sent:
        if (not regex.search(p)) or p == FAIL_SENT:
            if p == " " and (len(ns) == 0 or ns[-1] == " "):
                continue
            ns.append(p)
    return ns


def do_g2p(g2p, sent, res_wrds, is_first_sent):
    if sent in res_wrds:
        pho_seq = [res_wrds[sent]]
    else:
        pho_seq = g2p(sent)
    if not is_first_sent:
        pho_seq = [" "] + pho_seq  # add space to separate
    return pho_seq


def pre_process_sent(sent, do_filter, lower_case, res_wrds):
    if do_filter:
        sent = re.sub("-", " ", sent)
        sent = re.sub("—", " ", sent)
    if len(res_wrds) > 0:
        wrds = sent.split()
        wrds = ["SPLIT_ME " + w + " SPLIT_ME" if w in res_wrds else w for w in wrds]
        sents = [x.strip() for x in " ".join(wrds).split("SPLIT_ME") if x.strip() != ""]
    else:
        sents = [sent]
    if lower_case:
        sents = [s.lower() if s not in res_wrds else s for s in sents]
    return sents


def dup_pho(sent, dup_v_num, dup_c_num):
    """
    duplicate phoneme defined as cmudict
    http://www.speech.cs.cmu.edu/cgi-bin/cmudict
    """
    if dup_v_num == 1 and dup_c_num == 1:
        return sent
    ns = []
    for p in sent:
        ns.append(p)
        if re.search(r"\d$", p):
            for i in range(1, dup_v_num):
                ns.append(f"{p}-{i}P")
        elif re.search(r"\w", p):
            for i in range(1, dup_c_num):
                ns.append(f"{p}-{i}P")
    return ns


def add_word_start(sent):
    ns = []
    do_add = True
    ws = "▁"
    for p in sent:
        if do_add:
            p = ws + p
            do_add = False
        if p == " ":
            do_add = True
        else:
            ns.append(p)
    return ns


def load_reserve_word(reserve_word):
    if reserve_word == "":
        return []
    with open(reserve_word, "r") as fp:
        res_wrds = [x.strip().split() for x in fp.readlines() if x.strip() != ""]
        assert sum([0 if len(x) == 2 else 1 for x in res_wrds]) == 0
        res_wrds = dict(res_wrds)
    return res_wrds


def process_sents(sents, args):
    g2p = G2p()
    out_sents = []
    res_wrds = load_reserve_word(args.reserve_word)
    for sent in sents:
        col1 = ""
        if args.reserve_first_column:
            col1, sent = sent.split(None, 1)
        sent = process_sent(sent, g2p, res_wrds, args)
        if args.reserve_first_column and col1 != "":
            sent = f"{col1} {sent}"
        out_sents.append(sent)
    return out_sents


def main():
    args = parse()
    out_sents = []
    with open(args.data_path, "r") as fp:
        sent_list = [x.strip() for x in fp.readlines()]
    if args.parallel_process_num > 1:
        try:
            import submitit
        except ImportError:
            logger.warn(
                "submitit is not found and only one job is used to process the data"
            )
            submitit = None

    if args.parallel_process_num == 1 or submitit is None:
        out_sents = process_sents(sent_list, args)
    else:
        # process sentences with parallel computation
        lsize = len(sent_list) // args.parallel_process_num + 1
        executor = submitit.AutoExecutor(folder=args.logdir)
        executor.update_parameters(timeout_min=1000, cpus_per_task=4)
        jobs = []
        for i in range(args.parallel_process_num):
            job = executor.submit(
                process_sents, sent_list[lsize * i : lsize * (i + 1)], args
            )
            jobs.append(job)
        is_running = True
        while is_running:
            time.sleep(5)
            is_running = sum([job.done() for job in jobs]) < len(jobs)
        out_sents = list(itertools.chain.from_iterable([job.result() for job in jobs]))
    with open(args.out_path, "w") as fp:
        fp.write("\n".join(out_sents) + "\n")


if __name__ == "__main__":
    main()