RayeRen's picture
init
d1b91e7
raw
history blame contribute delete
No virus
30 kB
import atexit
import sys
import os
import time
import argparse
from datetime import datetime
import multiprocessing as mp
from montreal_forced_aligner import __version__
from montreal_forced_aligner.utils import get_available_acoustic_languages, get_available_g2p_languages, \
get_available_dict_languages, get_available_lm_languages, get_available_ivector_languages
from montreal_forced_aligner.command_line.align import run_align_corpus
from mfa_usr.adapt import run_adapt_model
from montreal_forced_aligner.command_line.train_and_align import run_train_corpus
from montreal_forced_aligner.command_line.g2p import run_g2p
from montreal_forced_aligner.command_line.train_g2p import run_train_g2p
from montreal_forced_aligner.command_line.validate import run_validate_corpus
from montreal_forced_aligner.command_line.download import run_download
from montreal_forced_aligner.command_line.train_lm import run_train_lm
from montreal_forced_aligner.command_line.thirdparty import run_thirdparty
from montreal_forced_aligner.command_line.train_ivector_extractor import run_train_ivector_extractor
from montreal_forced_aligner.command_line.classify_speakers import run_classify_speakers
from montreal_forced_aligner.command_line.transcribe import run_transcribe_corpus
from montreal_forced_aligner.command_line.train_dictionary import run_train_dictionary
from montreal_forced_aligner.command_line.create_segments import run_create_segments
from montreal_forced_aligner.exceptions import MFAError
from montreal_forced_aligner.config import update_global_config, load_global_config, update_command_history, \
load_command_history
class ExitHooks(object):
def __init__(self):
self.exit_code = None
self.exception = None
def hook(self):
self._orig_exit = sys.exit
sys.exit = self.exit
sys.excepthook = self.exc_handler
def exit(self, code=0):
self.exit_code = code
self._orig_exit(code)
def exc_handler(self, exc_type, exc, *args):
self.exception = exc
hooks = ExitHooks()
hooks.hook()
BEGIN = time.time()
BEGIN_DATE = datetime.now()
def history_save_handler():
history_data = {
'command': ' '.join(sys.argv),
'execution_time': time.time() - BEGIN,
'date': BEGIN_DATE,
'version': __version__
}
if hooks.exit_code is not None:
history_data['exit_code'] = hooks.exit_code
history_data['exception'] = ''
elif hooks.exception is not None:
history_data['exit_code'] = 1
history_data['exception'] = hooks.exception
else:
history_data['exception'] = ''
history_data['exit_code'] = 0
update_command_history(history_data)
atexit.register(history_save_handler)
def fix_path():
from montreal_forced_aligner.config import TEMP_DIR
thirdparty_dir = os.path.join(TEMP_DIR, 'thirdparty', 'bin')
old_path = os.environ.get('PATH', '')
if sys.platform == 'win32':
os.environ['PATH'] = thirdparty_dir + ';' + old_path
else:
os.environ['PATH'] = thirdparty_dir + ':' + old_path
os.environ['LD_LIBRARY_PATH'] = thirdparty_dir + ':' + os.environ.get('LD_LIBRARY_PATH', '')
def unfix_path():
if sys.platform == 'win32':
sep = ';'
os.environ['PATH'] = sep.join(os.environ['PATH'].split(sep)[1:])
else:
sep = ':'
os.environ['PATH'] = sep.join(os.environ['PATH'].split(sep)[1:])
os.environ['LD_LIBRARY_PATH'] = sep.join(os.environ['PATH'].split(sep)[1:])
acoustic_languages = get_available_acoustic_languages()
ivector_languages = get_available_ivector_languages()
lm_languages = get_available_lm_languages()
g2p_languages = get_available_g2p_languages()
dict_languages = get_available_dict_languages()
def create_parser():
GLOBAL_CONFIG = load_global_config()
def add_global_options(subparser, textgrid_output=False):
subparser.add_argument('-t', '--temp_directory', type=str, default=GLOBAL_CONFIG['temp_directory'],
help=f"Temporary directory root to store MFA created files, default is {GLOBAL_CONFIG['temp_directory']}")
subparser.add_argument('--disable_mp',
help=f"Disable any multiprocessing during alignment (not recommended), default is {not GLOBAL_CONFIG['use_mp']}",
action='store_true',
default=not GLOBAL_CONFIG['use_mp'])
subparser.add_argument('-j', '--num_jobs', type=int, default=GLOBAL_CONFIG['num_jobs'],
help=f"Number of data splits (and cores to use if multiprocessing is enabled), defaults "
f"is {GLOBAL_CONFIG['num_jobs']}")
subparser.add_argument('-v', '--verbose', help=f"Output debug messages, default is {GLOBAL_CONFIG['verbose']}",
action='store_true',
default=GLOBAL_CONFIG['verbose'])
subparser.add_argument('--clean', help=f"Remove files from previous runs, default is {GLOBAL_CONFIG['clean']}",
action='store_true',
default=GLOBAL_CONFIG['clean'])
subparser.add_argument('--overwrite',
help=f"Overwrite output files when they exist, default is {GLOBAL_CONFIG['overwrite']}",
action='store_true',
default=GLOBAL_CONFIG['overwrite'])
subparser.add_argument('--debug',
help=f"Run extra steps for debugging issues, default is {GLOBAL_CONFIG['debug']}",
action='store_true',
default=GLOBAL_CONFIG['debug'])
if textgrid_output:
subparser.add_argument('--disable_textgrid_cleanup',
help=f"Disable extra clean up steps on TextGrid output, default is {not GLOBAL_CONFIG['cleanup_textgrids']}",
action='store_true',
default=not GLOBAL_CONFIG['cleanup_textgrids'])
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest="subcommand")
subparsers.required = True
version_parser = subparsers.add_parser('version')
align_parser = subparsers.add_parser('align')
align_parser.add_argument('corpus_directory', help="Full path to the directory to align")
align_parser.add_argument('dictionary_path', help="Full path to the pronunciation dictionary to use")
align_parser.add_argument('acoustic_model_path',
help=f"Full path to the archive containing pre-trained model or language ({', '.join(acoustic_languages)})")
align_parser.add_argument('output_directory',
help="Full path to output directory, will be created if it doesn't exist")
align_parser.add_argument('--config_path', type=str, default='',
help="Path to config file to use for alignment")
align_parser.add_argument('-s', '--speaker_characters', type=str, default='0',
help="Number of characters of file names to use for determining speaker, "
'default is to use directory names')
align_parser.add_argument('-a', '--audio_directory', type=str, default='',
help="Audio directory root to use for finding audio files")
add_global_options(align_parser, textgrid_output=True)
adapt_parser = subparsers.add_parser('adapt')
adapt_parser.add_argument('corpus_directory', help="Full path to the directory to align")
adapt_parser.add_argument('dictionary_path', help="Full path to the pronunciation dictionary to use")
adapt_parser.add_argument('acoustic_model_path',
help=f"Full path to the archive containing pre-trained model or language ({', '.join(acoustic_languages)})")
adapt_parser.add_argument('output_model_path',
help="Full path to save adapted_model")
adapt_parser.add_argument('output_directory',
help="Full path to output directory, will be created if it doesn't exist")
adapt_parser.add_argument('--config_path', type=str, default='',
help="Path to config file to use for alignment")
adapt_parser.add_argument('-s', '--speaker_characters', type=str, default='0',
help="Number of characters of file names to use for determining speaker, "
'default is to use directory names')
adapt_parser.add_argument('-a', '--audio_directory', type=str, default='',
help="Audio directory root to use for finding audio files")
add_global_options(adapt_parser, textgrid_output=True)
train_parser = subparsers.add_parser('train')
train_parser.add_argument('corpus_directory', help="Full path to the source directory to align")
train_parser.add_argument('dictionary_path', help="Full path to the pronunciation dictionary to use",
default='')
train_parser.add_argument('output_directory',
help="Full path to output directory, will be created if it doesn't exist")
train_parser.add_argument('--config_path', type=str, default='',
help="Path to config file to use for training and alignment")
train_parser.add_argument('-o', '--output_model_path', type=str, default='',
help="Full path to save resulting acoustic and dictionary model")
train_parser.add_argument('-s', '--speaker_characters', type=str, default='0',
help="Number of characters of filenames to use for determining speaker, "
'default is to use directory names')
train_parser.add_argument('-a', '--audio_directory', type=str, default='',
help="Audio directory root to use for finding audio files")
train_parser.add_argument('-m', '--acoustic_model_path', type=str, default='',
help="Full path to save adapted_model")
add_global_options(train_parser, textgrid_output=True)
validate_parser = subparsers.add_parser('validate')
validate_parser.add_argument('corpus_directory', help="Full path to the source directory to align")
validate_parser.add_argument('dictionary_path', help="Full path to the pronunciation dictionary to use",
default='')
validate_parser.add_argument('acoustic_model_path', nargs='?', default='',
help=f"Full path to the archive containing pre-trained model or language ({', '.join(acoustic_languages)})")
validate_parser.add_argument('-s', '--speaker_characters', type=str, default='0',
help="Number of characters of file names to use for determining speaker, "
'default is to use directory names')
validate_parser.add_argument('--test_transcriptions', help="Test accuracy of transcriptions", action='store_true')
validate_parser.add_argument('--ignore_acoustics',
help="Skip acoustic feature generation and associated validation",
action='store_true')
add_global_options(validate_parser)
g2p_model_help_message = f'''Full path to the archive containing pre-trained model or language ({', '.join(g2p_languages)})
If not specified, then orthographic transcription is split into pronunciations.'''
g2p_parser = subparsers.add_parser('g2p')
g2p_parser.add_argument("g2p_model_path", help=g2p_model_help_message, nargs='?')
g2p_parser.add_argument("input_path",
help="Corpus to base word list on or a text file of words to generate pronunciations")
g2p_parser.add_argument("output_path", help="Path to save output dictionary")
g2p_parser.add_argument('--include_bracketed', help="Included words enclosed by brackets, i.e. [...], (...), <...>",
action='store_true')
g2p_parser.add_argument('--config_path', type=str, default='',
help="Path to config file to use for G2P")
add_global_options(g2p_parser)
train_g2p_parser = subparsers.add_parser('train_g2p')
train_g2p_parser.add_argument("dictionary_path", help="Location of existing dictionary")
train_g2p_parser.add_argument("output_model_path", help="Desired location of generated model")
train_g2p_parser.add_argument('--config_path', type=str, default='',
help="Path to config file to use for G2P")
train_g2p_parser.add_argument("--validate", action='store_true',
help="Perform an analysis of accuracy training on "
"most of the data and validating on an unseen subset")
add_global_options(train_g2p_parser)
download_parser = subparsers.add_parser('download')
download_parser.add_argument("model_type",
help="Type of model to download, one of 'acoustic', 'g2p', or 'dictionary'")
download_parser.add_argument("language", help="Name of language code to download, if not specified, "
"will list all available languages", nargs='?')
train_lm_parser = subparsers.add_parser('train_lm')
train_lm_parser.add_argument('source_path', help="Full path to the source directory to train from, alternatively "
'an ARPA format language model to convert for MFA use')
train_lm_parser.add_argument('output_model_path', type=str,
help="Full path to save resulting language model")
train_lm_parser.add_argument('-m', '--model_path', type=str,
help="Full path to existing language model to merge probabilities")
train_lm_parser.add_argument('-w', '--model_weight', type=float, default=1.0,
help="Weight factor for supplemental language model, defaults to 1.0")
train_lm_parser.add_argument('--dictionary_path', help="Full path to the pronunciation dictionary to use",
default='')
train_lm_parser.add_argument('--config_path', type=str, default='',
help="Path to config file to use for training and alignment")
add_global_options(train_lm_parser)
train_dictionary_parser = subparsers.add_parser('train_dictionary')
train_dictionary_parser.add_argument('corpus_directory', help="Full path to the directory to align")
train_dictionary_parser.add_argument('dictionary_path', help="Full path to the pronunciation dictionary to use")
train_dictionary_parser.add_argument('acoustic_model_path',
help=f"Full path to the archive containing pre-trained model or language ({', '.join(acoustic_languages)})")
train_dictionary_parser.add_argument('output_directory',
help="Full path to output directory, will be created if it doesn't exist")
train_dictionary_parser.add_argument('--config_path', type=str, default='',
help="Path to config file to use for alignment")
train_dictionary_parser.add_argument('-s', '--speaker_characters', type=str, default='0',
help="Number of characters of file names to use for determining speaker, "
'default is to use directory names')
add_global_options(train_dictionary_parser)
train_ivector_parser = subparsers.add_parser('train_ivector')
train_ivector_parser.add_argument('corpus_directory', help="Full path to the source directory to "
'train the ivector extractor')
train_ivector_parser.add_argument('dictionary_path', help="Full path to the pronunciation dictionary to use")
train_ivector_parser.add_argument('acoustic_model_path', type=str, default='',
help="Full path to acoustic model for alignment")
train_ivector_parser.add_argument('output_model_path', type=str, default='',
help="Full path to save resulting ivector extractor")
train_ivector_parser.add_argument('-s', '--speaker_characters', type=str, default='0',
help="Number of characters of filenames to use for determining speaker, "
'default is to use directory names')
train_ivector_parser.add_argument('--config_path', type=str, default='',
help="Path to config file to use for training")
add_global_options(train_ivector_parser)
classify_speakers_parser = subparsers.add_parser('classify_speakers')
classify_speakers_parser.add_argument('corpus_directory', help="Full path to the source directory to "
'run speaker classification')
classify_speakers_parser.add_argument('ivector_extractor_path', type=str, default='',
help="Full path to ivector extractor model")
classify_speakers_parser.add_argument('output_directory',
help="Full path to output directory, will be created if it doesn't exist")
classify_speakers_parser.add_argument('-s', '--num_speakers', type=int, default=0,
help="Number of speakers if known")
classify_speakers_parser.add_argument('--cluster', help="Using clustering instead of classification",
action='store_true')
classify_speakers_parser.add_argument('--config_path', type=str, default='',
help="Path to config file to use for ivector extraction")
add_global_options(classify_speakers_parser)
create_segments_parser = subparsers.add_parser('create_segments')
create_segments_parser.add_argument('corpus_directory', help="Full path to the source directory to "
'run VAD segmentation')
create_segments_parser.add_argument('output_directory',
help="Full path to output directory, will be created if it doesn't exist")
create_segments_parser.add_argument('--config_path', type=str, default='',
help="Path to config file to use for segmentation")
add_global_options(create_segments_parser)
transcribe_parser = subparsers.add_parser('transcribe')
transcribe_parser.add_argument('corpus_directory', help="Full path to the directory to transcribe")
transcribe_parser.add_argument('dictionary_path', help="Full path to the pronunciation dictionary to use")
transcribe_parser.add_argument('acoustic_model_path',
help=f"Full path to the archive containing pre-trained model or language ({', '.join(acoustic_languages)})")
transcribe_parser.add_argument('language_model_path',
help=f"Full path to the archive containing pre-trained model or language ({', '.join(lm_languages)})")
transcribe_parser.add_argument('output_directory',
help="Full path to output directory, will be created if it doesn't exist")
transcribe_parser.add_argument('--config_path', type=str, default='',
help="Path to config file to use for transcription")
transcribe_parser.add_argument('-s', '--speaker_characters', type=str, default='0',
help="Number of characters of file names to use for determining speaker, "
'default is to use directory names')
transcribe_parser.add_argument('-a', '--audio_directory', type=str, default='',
help="Audio directory root to use for finding audio files")
transcribe_parser.add_argument('-e', '--evaluate', help="Evaluate the transcription "
"against golden texts", action='store_true')
add_global_options(transcribe_parser)
config_parser = subparsers.add_parser('configure',
help="The configure command is used to set global defaults for MFA so "
"you don't have to set them every time you call an MFA command.")
config_parser.add_argument('-t', '--temp_directory', type=str, default='',
help=f"Set the default temporary directory, default is {GLOBAL_CONFIG['temp_directory']}")
config_parser.add_argument('-j', '--num_jobs', type=int,
help=f"Set the number of processes to use by default, defaults to {GLOBAL_CONFIG['num_jobs']}")
config_parser.add_argument('--always_clean', help="Always remove files from previous runs by default",
action='store_true')
config_parser.add_argument('--never_clean', help="Don't remove files from previous runs by default",
action='store_true')
config_parser.add_argument('--always_verbose', help="Default to verbose output", action='store_true')
config_parser.add_argument('--never_verbose', help="Default to non-verbose output", action='store_true')
config_parser.add_argument('--always_debug', help="Default to running debugging steps", action='store_true')
config_parser.add_argument('--never_debug', help="Default to not running debugging steps", action='store_true')
config_parser.add_argument('--always_overwrite', help="Always overwrite output files", action='store_true')
config_parser.add_argument('--never_overwrite', help="Never overwrite output files (if file already exists, "
"the output will be saved in the temp directory)",
action='store_true')
config_parser.add_argument('--disable_mp', help="Disable all multiprocessing (not recommended as it will usually "
"increase processing times)", action='store_true')
config_parser.add_argument('--enable_mp', help="Enable multiprocessing (recommended and enabled by default)",
action='store_true')
config_parser.add_argument('--disable_textgrid_cleanup', help="Disable postprocessing of TextGrids that cleans up "
"silences and recombines compound words and clitics",
action='store_true')
config_parser.add_argument('--enable_textgrid_cleanup', help="Enable postprocessing of TextGrids that cleans up "
"silences and recombines compound words and clitics",
action='store_true')
history_parser = subparsers.add_parser('history')
history_parser.add_argument('depth', help='Number of commands to list', nargs='?', default=10)
history_parser.add_argument('--verbose', help="Flag for whether to output additional information",
action='store_true')
annotator_parser = subparsers.add_parser('annotator')
anchor_parser = subparsers.add_parser('anchor')
thirdparty_parser = subparsers.add_parser('thirdparty')
thirdparty_parser.add_argument("command",
help="One of 'download', 'validate', or 'kaldi'")
thirdparty_parser.add_argument('local_directory',
help="Full path to the built executables to collect", nargs="?",
default='')
return parser
parser = create_parser()
def main():
parser = create_parser()
mp.freeze_support()
args, unknown = parser.parse_known_args()
for short in ['-c', '-d']:
if short in unknown:
print(f'Due to the number of options that `{short}` could refer to, it is not accepted. '
'Please specify the full argument')
sys.exit(1)
try:
fix_path()
if args.subcommand in ['align', 'train', 'train_ivector']:
from montreal_forced_aligner.thirdparty.kaldi import validate_alignment_binaries
if not validate_alignment_binaries():
print("There was an issue validating Kaldi binaries, please ensure you've downloaded them via the "
"'mfa thirdparty download' command. See 'mfa thirdparty validate' for more detailed information "
"on why this check failed.")
sys.exit(1)
elif args.subcommand in ['transcribe']:
from montreal_forced_aligner.thirdparty.kaldi import validate_transcribe_binaries
if not validate_transcribe_binaries():
print("There was an issue validating Kaldi binaries, please ensure you've downloaded them via the "
"'mfa thirdparty download' command. See 'mfa thirdparty validate' for more detailed information "
"on why this check failed. If you are on MacOS, please note that the thirdparty binaries available "
"via the download command do not contain the transcription ones. To get this functionality working "
"for the time being, please build kaldi locally and follow the instructions for running the "
"'mfa thirdparty kaldi' command.")
sys.exit(1)
elif args.subcommand in ['train_dictionary']:
from montreal_forced_aligner.thirdparty.kaldi import validate_train_dictionary_binaries
if not validate_train_dictionary_binaries():
print("There was an issue validating Kaldi binaries, please ensure you've downloaded them via the "
"'mfa thirdparty download' command. See 'mfa thirdparty validate' for more detailed information "
"on why this check failed. If you are on MacOS, please note that the thirdparty binaries available "
"via the download command do not contain the train_dictionary ones. To get this functionality working "
"for the time being, please build kaldi locally and follow the instructions for running the "
"'mfa thirdparty kaldi' command.")
sys.exit(1)
elif args.subcommand in ['g2p', 'train_g2p']:
try:
import pynini
except ImportError:
print("There was an issue importing Pynini, please ensure that it is installed. If you are on Windows, "
"please use the Windows Subsystem for Linux to use g2p functionality.")
sys.exit(1)
if args.subcommand == 'align':
run_align_corpus(args, unknown, acoustic_languages)
elif args.subcommand == 'adapt':
run_adapt_model(args, unknown, acoustic_languages)
elif args.subcommand == 'train':
run_train_corpus(args, unknown)
elif args.subcommand == 'g2p':
run_g2p(args, unknown, g2p_languages)
elif args.subcommand == 'train_g2p':
run_train_g2p(args, unknown)
elif args.subcommand == 'validate':
run_validate_corpus(args, unknown)
elif args.subcommand == 'download':
run_download(args)
elif args.subcommand == 'train_lm':
run_train_lm(args, unknown)
elif args.subcommand == 'train_dictionary':
run_train_dictionary(args, unknown)
elif args.subcommand == 'train_ivector':
run_train_ivector_extractor(args, unknown)
elif args.subcommand == 'classify_speakers':
run_classify_speakers(args, unknown)
elif args.subcommand in ['annotator', 'anchor']:
from montreal_forced_aligner.command_line.anchor import run_anchor
run_anchor(args)
elif args.subcommand == 'thirdparty':
run_thirdparty(args)
elif args.subcommand == 'transcribe':
run_transcribe_corpus(args, unknown)
elif args.subcommand == 'create_segments':
run_create_segments(args, unknown)
elif args.subcommand == 'configure':
update_global_config(args)
global GLOBAL_CONFIG
GLOBAL_CONFIG = load_global_config()
elif args.subcommand == 'history':
depth = args.depth
history = load_command_history()[-depth:]
for h in history:
if args.verbose:
print('command\tDate\tExecution time\tVersion\tExit code\tException')
for h in history:
execution_time = time.strftime('%H:%M:%S', time.gmtime(h['execution_time']))
d = h['date'].isoformat()
print(
f"{h['command']}\t{d}\t{execution_time}\t{h['version']}\t{h['exit_code']}\t{h['exception']}")
pass
else:
for h in history:
print(h['command'])
elif args.subcommand == 'version':
print(__version__)
except MFAError as e:
if getattr(args, 'debug', False):
raise
print(e)
sys.exit(1)
finally:
unfix_path()
if __name__ == '__main__':
main()