processors/phone_extractor.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
from tqdm import tqdm
from text.g2p_module import G2PModule, LexiconModule
from text.symbol_table import SymbolTable

'''
    phoneExtractor: extract phone from text
'''
class phoneExtractor:
    def __init__(self, cfg, dataset_name=None, phone_symbol_file=None):
        '''
            Args:
                cfg: config
                dataset_name: name of dataset
        '''
        self.cfg = cfg

        #  phone symbols dict
        self.phone_symbols = set()
        
        # phone symbols dict file
        if phone_symbol_file is not None:
            self.phone_symbols_file = phone_symbol_file
        elif dataset_name is not None:
            self.dataset_name = dataset_name
            self.phone_symbols_file = os.path.join(cfg.preprocess.processed_dir, 
                                            dataset_name, 
                                            cfg.preprocess.symbols_dict)

        
        # initialize g2p module
        if cfg.preprocess.phone_extractor in ["espeak", "pypinyin", "pypinyin_initials_finals"]:
            self.g2p_module = G2PModule(backend=cfg.preprocess.phone_extractor)
        elif cfg.preprocess.phone_extractor == 'lexicon':
            assert cfg.preprocess.lexicon_path != ""
            self.g2p_module = LexiconModule(cfg.preprocess.lexicon_path)
        else:
            print('No suppert to', cfg.preprocess.phone_extractor)
            raise

    
    def extract_phone(self, text):
        '''
            Extract phone from text
            Args:

                text:  text of utterance
                
            Returns:    
                phone_symbols: set of phone symbols
                phone_seq: list of phone sequence of each utterance
        '''
        
        if self.cfg.preprocess.phone_extractor in ["espeak", "pypinyin", "pypinyin_initials_finals"]:
            text = text.replace("”", '"').replace("“", '"')
            phone = self.g2p_module.g2p_conversion(text=text)  
            self.phone_symbols.update(phone)    
            phone_seq = [phn for phn in phone]
            
        elif self.cfg.preprocess.phone_extractor == 'lexicon':
            phone_seq = self.g2p_module.g2p_conversion(text)
            phone = phone_seq
            if not isinstance(phone_seq, list):
                phone_seq = phone_seq.split()
           
        return phone_seq

    def save_dataset_phone_symbols_to_table(self):
        # load and merge saved phone symbols                
        if os.path.exists(self.phone_symbols_file):
            phone_symbol_dict_saved = SymbolTable.from_file(self.phone_symbols_file)._sym2id.keys()
            self.phone_symbols.update(set(phone_symbol_dict_saved))

        # save phone symbols
        phone_symbol_dict = SymbolTable()
        for s in sorted(list(self.phone_symbols)):
            phone_symbol_dict.add(s)
        phone_symbol_dict.to_file(self.phone_symbols_file)    

                
def extract_utt_phone_sequence(cfg, metadata):
    '''
        Extract phone sequence from text
        Args:
            cfg: config
            metadata: list of dict, each dict contains "Uid", "Text"
            
    '''
    
    dataset_name = cfg.dataset[0]
    
    # output path
    out_path = os.path.join(cfg.preprocess.processed_dir, dataset_name, cfg.preprocess.phone_dir)
    os.makedirs(out_path, exist_ok=True)
        
    phone_extractor = phoneExtractor(cfg, dataset_name)

    for utt in tqdm(metadata):  
        uid = utt["Uid"]
        text = utt["Text"]    
                 
        phone_seq = phone_extractor.extract_phone(text)
                 
        phone_path = os.path.join(out_path, uid+'.phone')
        with open(phone_path, 'w') as fin:
            fin.write(' '.join(phone_seq))
    
    if cfg.preprocess.phone_extractor != 'lexicon':
        phone_extractor.save_dataset_phone_symbols_to_table()
    
    
        
def save_all_dataset_phone_symbols_to_table(self, cfg, dataset):
    #  phone symbols dict
    phone_symbols = set()
    
    for dataset_name in dataset:
        phone_symbols_file = os.path.join(cfg.preprocess.processed_dir, 
                                          dataset_name, 
                                          cfg.preprocess.symbols_dict)
        
        # load and merge saved phone symbols                
        assert os.path.exists(phone_symbols_file)
        phone_symbol_dict_saved = SymbolTable.from_file(phone_symbols_file)._sym2id.keys()
        phone_symbols.update(set(phone_symbol_dict_saved))
        
    # save all phone symbols to each dataset
    phone_symbol_dict = SymbolTable()
    for s in sorted(list(phone_symbols)):
        phone_symbol_dict.add(s)
    for dataset_name in dataset:
        phone_symbols_file = os.path.join(cfg.preprocess.processed_dir, 
                                          dataset_name, 
                                          cfg.preprocess.symbols_dict)
        phone_symbol_dict.to_file(phone_symbols_file)