"""
Alignment multiprocessing functions
-----------------------------------
"""
from __future__ import annotations
import collections
import json
import logging
import math
import multiprocessing as mp
import os
import shutil
import statistics
import sys
import time
import traceback
import typing
from dataclasses import dataclass
from pathlib import Path
from queue import Empty
from typing import TYPE_CHECKING
import librosa.feature
import numpy as np
import sqlalchemy
from _kalpy.gmm import gmm_compute_likes
from _kalpy.hmm import TransitionModel
from _kalpy.util import (
RandomAccessBaseDoubleMatrixReader,
RandomAccessBaseFloatMatrixReader,
RandomAccessInt32VectorVectorReader,
)
from kalpy.aligner import KalpyAligner
from kalpy.decoder.data import FstArchive
from kalpy.decoder.training_graphs import TrainingGraphCompiler
try:
from kalpy.evaluation import align_phones, align_words, fix_unk_words, naive_boundary_f1
except ImportError:
from kalpy.evaluation import align_phones, align_words, fix_unk_words
naive_boundary_f1 = None
from kalpy.fstext.lexicon import LexiconCompiler
from kalpy.gmm.align import GmmAligner
from kalpy.gmm.data import AlignmentArchive, CtmInterval, TranscriptionArchive
from kalpy.gmm.train import GmmStatsAccumulator
from kalpy.gmm.utils import read_gmm_model
from kalpy.utils import generate_read_specifier
from sqlalchemy.orm import joinedload, selectinload, subqueryload
from montreal_forced_aligner.abc import KaldiFunction
from montreal_forced_aligner.data import (
WORD_BEGIN_SYMBOL,
WORD_END_SYMBOL,
MfaArguments,
PhoneType,
PronunciationProbabilityCounter,
WordType,
WorkflowType,
)
from montreal_forced_aligner.db import (
CorpusWorkflow,
File,
Job,
Phone,
PhoneInterval,
PhoneMapping,
ReferencePhoneInterval,
SoundFile,
Speaker,
TextFile,
Utterance,
Word,
WordInterval,
)
from montreal_forced_aligner.exceptions import AlignmentCollectionError, AlignmentExportError
from montreal_forced_aligner.helper import mfa_open, split_phone_position
from montreal_forced_aligner.models import AcousticModel
from montreal_forced_aligner.textgrid import construct_textgrid_output
from montreal_forced_aligner.utils import thread_logger
if TYPE_CHECKING:
from montreal_forced_aligner.abc import MetaDict
__all__ = [
"AlignmentExtractionFunction",
"ExportTextGridProcessWorker",
"AlignmentExtractionArguments",
"ExportTextGridArguments",
"AlignFunction",
"AlignArguments",
"AnalyzeAlignmentsFunction",
"AnalyzeAlignmentsArguments",
"AnalyzeTranscriptsFunction",
"AccStatsFunction",
"AccStatsArguments",
"CompileTrainGraphsFunction",
"CompileTrainGraphsArguments",
"GeneratePronunciationsArguments",
"GeneratePronunciationsFunction",
"FineTuneArguments",
"FineTuneFunction",
"PhoneConfidenceArguments",
"PhoneConfidenceFunction",
]
logger = logging.getLogger("mfa")
[docs]
@dataclass
class GeneratePronunciationsArguments(MfaArguments):
"""
Arguments for :func:`~montreal_forced_aligner.alignment.multiprocessing.GeneratePronunciationsFunction`
Parameters
----------
job_name: int
Integer ID of the job
session: :class:`sqlalchemy.orm.scoped_session` or str
SqlAlchemy scoped session or string for database connections
log_path: :class:`~pathlib.Path`
Path to save logging information during the run
aligner: :class:`kalpy.gmm.align.GmmAligner`
GmmAligner to use
lexicon_compilers: dict[int, :class:`kalpy.fstext.lexicon.LexiconCompiler`]
Lexicon compilers for each pronunciation dictionary
for_g2p: bool
Flag for training a G2P model with acoustic information
"""
aligner: GmmAligner
lexicon_compilers: typing.Dict[int, LexiconCompiler]
for_g2p: bool
[docs]
@dataclass
class ExportTextGridArguments(MfaArguments):
"""
Arguments for :class:`~montreal_forced_aligner.alignment.multiprocessing.ExportTextGridProcessWorker`
Parameters
----------
job_name: int
Integer ID of the job
session: :class:`sqlalchemy.orm.scoped_session` or str
SqlAlchemy scoped session or string for database connections
log_path: :class:`~pathlib.Path`
Path to save logging information during the run
export_frame_shift: float
Frame shift in seconds
cleanup_textgrids: bool
Flag to cleanup silences and recombine words
clitic_marker: str
Marker indicating clitics
output_directory: :class:`~pathlib.Path`
Directory for exporting
output_format: str
Format to export
include_original_text: bool
Flag for including original unnormalized text as a tier
"""
export_frame_shift: float
cleanup_textgrids: bool
clitic_marker: str
output_directory: Path
output_format: str
include_original_text: bool
[docs]
@dataclass
class CompileTrainGraphsArguments(MfaArguments):
"""
Arguments for :class:`~montreal_forced_aligner.alignment.multiprocessing.CompileTrainGraphsFunction`
Parameters
----------
job_name: int
Integer ID of the job
session: :class:`sqlalchemy.orm.scoped_session` or str
SqlAlchemy scoped session or string for database connections
log_path: :class:`~pathlib.Path`
Path to save logging information during the run
working_directory: :class:`~pathlib.Path`
Working directory
tree_path: :class:`~pathlib.Path`
Path to tree file
model_path: :class:`~pathlib.Path`
Path to model file
"""
working_directory: Path
lexicon_compilers: typing.Dict[int, LexiconCompiler]
tree_path: Path
model_path: Path
[docs]
@dataclass
class AlignArguments(MfaArguments):
"""
Arguments for :class:`~montreal_forced_aligner.alignment.multiprocessing.AlignFunction`
Parameters
----------
job_name: int
Integer ID of the job
session: :class:`sqlalchemy.orm.scoped_session` or str
SqlAlchemy scoped session or string for database connections
log_path: :class:`~pathlib.Path`
Path to save logging information during the run
working_directory: :class:`~pathlib.Path`
Working directory
model_path: :class:`~pathlib.Path`
Path to model file
align_options: dict[str, Any]
Alignment options
final: bool
Flag for final alignment pass
"""
working_directory: Path
model_path: Path
align_options: MetaDict
confidence: bool
final: bool
@dataclass
class AnalyzeAlignmentsArguments(MfaArguments):
"""
Arguments for :class:`~montreal_forced_aligner.alignment.multiprocessing.AnalyzeAlignmentsFunction`
Parameters
----------
job_name: int
Integer ID of the job
session: :class:`sqlalchemy.orm.scoped_session` or str
SqlAlchemy scoped session or string for database connections
log_path: :class:`~pathlib.Path`
Path to save logging information during the run
"""
analyze_intensity: bool
@dataclass
class EvaluateAlignmentsArguments(MfaArguments):
"""
Arguments for :class:`~montreal_forced_aligner.alignment.multiprocessing.AnalyzeAlignmentsFunction`
Parameters
----------
job_name: int
Integer ID of the job
session: :class:`sqlalchemy.orm.scoped_session` or str
SqlAlchemy scoped session or string for database connections
log_path: :class:`~pathlib.Path`
Path to save logging information during the run
"""
use_cutoff_model: bool
naive: bool
[docs]
@dataclass
class FineTuneArguments(MfaArguments):
"""
Arguments for :class:`~montreal_forced_aligner.alignment.multiprocessing.AlignFunction`
Parameters
----------
job_name: int
Integer ID of the job
session: :class:`sqlalchemy.orm.scoped_session` or str
SqlAlchemy scoped session or string for database connections
log_path: :class:`~pathlib.Path`
Path to save logging information during the run
acoustic_model: :class:`~montreal_forced_aligner.models.AcousticModel`
Acoustic model
lexicon_compilers: dict[int, :class:`~kalpy.lexicon.LexiconCompiler`]
Lexicon compilers
boundary_tolerance: float, optional
Boundary tolerance, defaults to half the feature generation time step
"""
acoustic_model: AcousticModel
lexicon_compilers: typing.Dict[int, LexiconCompiler]
boundary_tolerance: typing.Optional[float]
[docs]
@dataclass
class PhoneConfidenceArguments(MfaArguments):
"""
Arguments for :class:`~montreal_forced_aligner.alignment.multiprocessing.AlignFunction`
Parameters
----------
job_name: int
Integer ID of the job
session: :class:`sqlalchemy.orm.scoped_session` or str
SqlAlchemy scoped session or string for database connections
log_path: :class:`~pathlib.Path`
Path to save logging information during the run
working_directory: :class:`~pathlib.Path`
Working directory
model_path: :class:`~pathlib.Path`
Path to model file
phone_pdf_counts_path: :class:`~pathlib.Path`
Path to output PDF counts
"""
working_directory: Path
model_path: Path
phone_pdf_counts_path: Path
[docs]
@dataclass
class AccStatsArguments(MfaArguments):
"""
Arguments for :class:`~montreal_forced_aligner.alignment.multiprocessing.AccStatsFunction`
Parameters
----------
job_name: int
Integer ID of the job
session: :class:`sqlalchemy.orm.scoped_session` or str
SqlAlchemy scoped session or string for database connections
log_path: :class:`~pathlib.Path`
Path to save logging information during the run
working_directory: :class:`~pathlib.Path`
Path to working directory
model_path: :class:`~pathlib.Path`
Path to model file
"""
working_directory: Path
model_path: Path
filter_likely_errors: bool = False
[docs]
class CompileTrainGraphsFunction(KaldiFunction):
"""
Multiprocessing function to compile training graphs
See Also
--------
:meth:`.AlignMixin.compile_train_graphs`
Main function that calls this function in parallel
:meth:`.AlignMixin.compile_train_graphs_arguments`
Job method for generating arguments for this function
:kaldi_src:`compile-train-graphs`
Relevant Kaldi binary
Parameters
----------
args: :class:`~montreal_forced_aligner.alignment.multiprocessing.CompileTrainGraphsArguments`
Arguments for the function
"""
def __init__(self, args: CompileTrainGraphsArguments):
super().__init__(args)
self.tree_path = args.tree_path
self.lexicon_compilers = args.lexicon_compilers
self.model_path = args.model_path
def _run(self):
"""Run the function"""
with self.session() as session, thread_logger(
"kalpy.graphs", self.log_path, job_name=self.job_name
) as graph_logger:
graph_logger.debug(f"Tree path: {self.tree_path}")
graph_logger.debug(f"Model path: {self.model_path}")
job = (
session.query(Job)
.options(joinedload(Job.corpus, innerjoin=True), subqueryload(Job.dictionaries))
.filter(Job.id == self.job_name)
.first()
)
workflow: CorpusWorkflow = (
session.query(CorpusWorkflow)
.filter(CorpusWorkflow.current == True) # noqa
.first()
)
interjection_costs = {}
if workflow.workflow_type is WorkflowType.transcript_verification:
interjection_words = (
session.query(Word).filter(Word.word_type == WordType.interjection).all()
)
if interjection_words:
max_count = max(math.log(x.count) for x in interjection_words)
for w in interjection_words:
count = math.log(w.count)
if count == 0:
count = 0.01
cost = max_count / count
interjection_costs[w.word] = cost
for d in job.training_dictionaries:
begin = time.time()
if self.lexicon_compilers and d.id in self.lexicon_compilers:
lexicon = self.lexicon_compilers[d.id]
else:
lexicon = d.lexicon_compiler
if workflow.workflow_type is WorkflowType.transcript_verification:
if interjection_words and d.oov_word not in interjection_costs:
interjection_costs[d.oov_word] = min(interjection_costs.values())
# interjection_costs[d.cutoff_word] = min(interjection_costs.values())
compiler = TrainingGraphCompiler(
self.model_path,
self.tree_path,
lexicon,
batch_size=500
if workflow.workflow_type is not WorkflowType.transcript_verification
else 250,
)
graph_logger.debug(f"Set up took {time.time() - begin} seconds")
query = (
session.query(Utterance.kaldi_id, Utterance.normalized_text)
.join(Utterance.speaker)
.filter(Utterance.job_id == self.job_name, Speaker.dictionary_id == d.id)
.filter(Utterance.ignored == False) # noqa
.order_by(Utterance.kaldi_id)
)
if job.corpus.current_subset > 0:
query = query.filter(Utterance.in_subset == True) # noqa
graph_logger.info(f"Compiling {query.count()} graphs for {d.name}")
fst_ark_path = job.construct_path(
workflow.working_directory, "fsts", "ark", d.name
)
compiler.export_graphs(
fst_ark_path,
query,
# callback=self.callback,
interjection_words=interjection_costs,
# cutoff_pattern = d.cutoff_word
)
graph_logger.debug(f"Total compilation time: {time.time() - begin} seconds")
del compiler
del lexicon
[docs]
class AccStatsFunction(KaldiFunction):
"""
Multiprocessing function for accumulating stats in GMM training.
See Also
--------
:meth:`.AcousticModelTrainingMixin.acc_stats`
Main function that calls this function in parallel
:meth:`.AcousticModelTrainingMixin.acc_stats_arguments`
Job method for generating arguments for this function
:kaldi_src:`gmm-acc-stats-ali`
Relevant Kaldi binary
Parameters
----------
args: :class:`~montreal_forced_aligner.alignment.multiprocessing.AccStatsArguments`
Arguments for the function
"""
def __init__(self, args: AccStatsArguments):
super().__init__(args)
self.working_directory = args.working_directory
self.model_path = args.model_path
self.filter_likely_errors = args.filter_likely_errors
def _run(self) -> None:
"""Run the function"""
with self.session() as session, thread_logger(
"kalpy.train", self.log_path, job_name=self.job_name
) as train_logger:
job: Job = (
session.query(Job)
.options(joinedload(Job.corpus, innerjoin=True), subqueryload(Job.dictionaries))
.filter(Job.id == self.job_name)
.first()
)
for d in job.training_dictionaries:
train_logger.debug(f"Accumulating stats for dictionary {d.name} ({d.id})")
train_logger.debug(f"Accumulating stats for model: {self.model_path}")
accumulator = GmmStatsAccumulator(self.model_path)
ignored_keys = None
if self.filter_likely_errors:
ignored_keys = set(
x[0]
for x in session.query(Utterance.kaldi_id)
.filter(Utterance.job_id == self.job_name) # noqa
.filter(
sqlalchemy.or_(
sqlalchemy.and_(
Utterance.manual_alignments == True, # noqa
Utterance.alignment_log_likelihood < -1000,
),
sqlalchemy.and_(
Utterance.manual_alignments == False, # noqa
Utterance.duration_deviation > 10,
),
sqlalchemy.and_(
Utterance.manual_alignments == False, # noqa
Utterance.snr <= 1.0,
),
)
)
.all()
)
train_logger.debug(
f"Ignoring {len(ignored_keys)} utterances due to likely alignment errors"
)
try:
feature_archive = job.construct_feature_archive(self.working_directory, d.name)
except OSError:
continue
ali_path = job.construct_path(self.working_directory, "ali", "ark", d.name)
if not ali_path.exists():
continue
alignment_archive = AlignmentArchive(ali_path)
train_logger.debug("Feature Archive information:")
train_logger.debug(f"CMVN: {feature_archive.cmvn_read_specifier}")
train_logger.debug(f"Deltas: {feature_archive.use_deltas}")
train_logger.debug(f"Splices: {feature_archive.use_splices}")
train_logger.debug(f"LDA: {feature_archive.lda_mat_file_name}")
train_logger.debug(f"fMLLR: {feature_archive.transform_read_specifier}")
train_logger.debug(f"Alignment path: {ali_path}")
accumulator.accumulate_stats(
feature_archive,
alignment_archive,
callback=self.callback,
ignored_keys=ignored_keys,
)
self.callback((accumulator.transition_accs, accumulator.gmm_accs))
[docs]
class AlignFunction(KaldiFunction):
"""
Multiprocessing function for alignment.
See Also
--------
:meth:`.AlignMixin.align_utterances`
Main function that calls this function in parallel
:meth:`.AlignMixin.align_arguments`
Job method for generating arguments for this function
:kaldi_src:`align-gmm-compiled`
Relevant Kaldi binary
:kaldi_src:`gmm-boost-silence`
Relevant Kaldi binary
Parameters
----------
args: :class:`~montreal_forced_aligner.alignment.multiprocessing.AlignArguments`
Arguments for the function
"""
def __init__(self, args: AlignArguments):
super().__init__(args)
self.working_directory = args.working_directory
self.model_path = args.model_path
self.align_options = args.align_options
self.confidence = args.confidence
self.final = args.final
def _run(self) -> None:
"""Run the function"""
with self.session() as session, thread_logger(
"kalpy.align", self.log_path, job_name=self.job_name
) as align_logger:
align_logger.debug(f"Align options: {self.align_options}")
job: Job = (
session.query(Job)
.options(joinedload(Job.corpus, innerjoin=True), subqueryload(Job.dictionaries))
.filter(Job.id == self.job_name)
.first()
)
align_options = self.align_options
boost_silence = align_options.pop("boost_silence", 1.0)
silence_phones = [
x
for x, in session.query(Phone.mapping_id).filter(
Phone.phone_type == PhoneType.silence, Phone.phone != "<eps>"
)
]
aligner = GmmAligner(
self.model_path,
silence_phones=silence_phones,
**align_options,
)
aligner.boost_silence(boost_silence)
for d in job.training_dictionaries:
align_logger.debug(f"Aligning for dictionary {d.name} ({d.id})")
align_logger.debug(f"Aligning with model: {aligner.acoustic_model_path}")
fst_path = job.construct_path(self.working_directory, "fsts", "ark", d.name)
align_logger.debug(f"Training graph archive: {fst_path}")
try:
feature_archive = job.construct_feature_archive(self.working_directory, d.name)
except OSError:
continue
align_logger.debug("Feature Archive information:")
align_logger.debug(f"Archive: {feature_archive.file_name}")
align_logger.debug(f"CMVN: {feature_archive.cmvn_read_specifier}")
align_logger.debug(f"Deltas: {feature_archive.use_deltas}")
align_logger.debug(f"Splices: {feature_archive.use_splices}")
align_logger.debug(f"LDA: {feature_archive.lda_mat_file_name}")
align_logger.debug(f"fMLLR: {feature_archive.transform_read_specifier}")
training_graph_archive = FstArchive(fst_path)
ali_path = job.construct_path(self.working_directory, "ali", "ark", d.name)
words_path = job.construct_path(self.working_directory, "words", "ark", d.name)
likes_path = job.construct_path(
self.working_directory, "likelihoods", "ark", d.name
)
ali_path.unlink(missing_ok=True)
words_path.unlink(missing_ok=True)
likes_path.unlink(missing_ok=True)
reference_phones_path = job.construct_path(
job.corpus.current_subset_directory, "ref_phones", "ark", d.name
)
if reference_phones_path.exists():
align_logger.debug(f"Reference phones: {reference_phones_path}")
reference_phone_archive = RandomAccessInt32VectorVectorReader(
generate_read_specifier(reference_phones_path)
)
else:
reference_phone_archive = None
if aligner.acoustic_model_path.endswith(".alimdl"):
ali_path = job.construct_path(
self.working_directory, "ali_first_pass", "ark", d.name
)
words_path = job.construct_path(
self.working_directory, "words_first_pass", "ark", d.name
)
likes_path = job.construct_path(
self.working_directory, "likelihoods_first_pass", "ark", d.name
)
query = (
session.query(Utterance.kaldi_id, Utterance.boost_silence)
.join(Utterance.speaker)
.filter(Utterance.job_id == self.job_name, Speaker.dictionary_id == d.id)
.filter(Utterance.ignored == False) # noqa
.filter(Utterance.boost_silence != None) # noqa
.order_by(Utterance.kaldi_id)
)
utterance_parameters = {}
for u_id, u_boost in query:
utterance_parameters[u_id] = {"boost_silence": u_boost}
try:
aligner.export_alignments(
ali_path,
training_graph_archive,
feature_archive,
reference_phone_archive=reference_phone_archive,
word_file_name=words_path,
likelihood_file_name=likes_path,
callback=self.callback,
utterance_parameters=utterance_parameters,
)
except Exception:
exc_type, exc_value, exc_traceback = sys.exc_info()
error_text = "\n".join(
traceback.format_exception(exc_type, exc_value, exc_traceback)
)
align_logger.debug(error_text)
raise
finally:
if reference_phone_archive is not None:
reference_phone_archive.Close()
if aligner.acoustic_model_path.endswith(".alimdl"):
try:
job.construct_path(
self.working_directory, "ali", "ark", d.name
).symlink_to(ali_path)
job.construct_path(
self.working_directory, "words", "ark", d.name
).symlink_to(words_path)
job.construct_path(
self.working_directory, "likelihoods", "ark", d.name
).symlink_to(likes_path)
except OSError as e:
logger.debug(str(e))
shutil.copyfile(
ali_path,
job.construct_path(self.working_directory, "ali", "ark", d.name),
)
shutil.copyfile(
words_path,
job.construct_path(self.working_directory, "words", "ark", d.name),
)
shutil.copyfile(
likes_path,
job.construct_path(
self.working_directory, "likelihoods", "ark", d.name
),
)
class EvaluateAlignmentsFunction(KaldiFunction):
"""
Multiprocessing function for analyzing alignments.
See Also
--------
:meth:`.CorpusAligner.analyze_alignments`
Main function that calls this function in parallel
:meth:`.CorpusAligner.calculate_speech_post_arguments`
Job method for generating arguments for this function
:kaldi_src:`lattice-to-post`
Relevant Kaldi binary
:kaldi_src:`weight-silence-post`
Relevant Kaldi binary
Parameters
----------
args: :class:`~montreal_forced_aligner.alignment.multiprocessing.CalculateSpeechPostArguments`
Arguments for the function
"""
def __init__(self, args: EvaluateAlignmentsArguments):
super().__init__(args)
self.use_cutoff_model = args.use_cutoff_model
self.naive = args.naive
def _run(self):
"""Run the function"""
from montreal_forced_aligner import config
with self.session() as session, thread_logger(
"kalpy.evaluate", self.log_path, job_name=self.job_name
) as evaluation_logger:
# Set up
mapping = {}
for x in session.query(PhoneMapping).all():
if x.model_phone_string not in mapping:
mapping[x.model_phone_string] = set()
mapping[x.model_phone_string].add(x.reference_phone_string)
evaluation_logger.debug(f"Mapping: {mapping}")
silence_phones = {
x[0]
for x in session.query(Phone.phone).filter(Phone.phone_type == PhoneType.silence)
}
oov_phone = (
session.query(Phone.phone)
.filter(sqlalchemy.or_(Phone.phone_type == PhoneType.oov, Phone.phone == "spn"))
.first()
)
if oov_phone is not None:
oov_phone = oov_phone[0]
utterances: typing.List[Utterance] = (
session.query(Utterance)
.join(Utterance.file)
.filter(Utterance.job_id == self.job_name)
.options(
joinedload(Utterance.file, innerjoin=True),
joinedload(Utterance.speaker, innerjoin=True),
subqueryload(Utterance.phone_intervals).options(
joinedload(PhoneInterval.phone, innerjoin=True),
),
subqueryload(Utterance.reference_phone_intervals).options(
joinedload(ReferencePhoneInterval.phone, innerjoin=True),
),
subqueryload(Utterance.word_intervals).options(
joinedload(WordInterval.word, innerjoin=True),
),
)
)
for u in utterances:
reference_phones = [x.as_ctm() for x in u.reference_phone_intervals]
reference_phone_count = len(reference_phones)
if reference_phone_count == 0:
continue
comparison_intervals = u.phone_intervals
comparison_phones = [x.as_ctm() for x in comparison_intervals]
if self.use_cutoff_model:
for wi in u.word_intervals:
if wi.word.word_type is WordType.cutoff:
comparison_phones = [
x
for x in comparison_phones
if x.end <= wi.begin or x.begin >= wi.end
]
comparison_phones.append(
CtmInterval(
begin=wi.begin,
end=wi.end,
label=oov_phone,
)
)
comparison_phones = sorted(comparison_phones, key=lambda x: x.begin)
if not self.naive:
results = align_phones(
reference_phones,
comparison_phones,
silence_phones=silence_phones,
custom_mapping=mapping,
debug=config.DEBUG,
)
score, phone_error_rate, confusions = results[:3]
if score is None:
self.callback(
(
{
"id": u.id,
"alignment_score": None,
"phone_error_rate": 1.0,
"edit_distance": reference_phone_count,
},
u.file.name,
reference_phone_count,
{},
[],
[],
)
)
continue
edit_distance = None
boundary_errors = None
pi_update_mappings = []
if len(results) > 3:
alignment = results[3]
edit_distance = alignment.score
boundary_errors = results[4]
interval_index = 0
for i, (sa, sb) in enumerate(alignment.alignment):
if (
interval_index < len(comparison_phones)
and sb.label == comparison_phones[interval_index].label
):
if sa.label == "-":
pi_update_mappings.append(
{
"id": comparison_intervals[interval_index].id,
"begin_error": None,
"end_error": None,
}
)
else:
pi_update_mappings.append(
{
"id": comparison_intervals[interval_index].id,
"begin_error": sb.begin - sa.begin,
"end_error": sb.end - sa.end,
}
)
interval_index += 1
self.callback(
(
{
"id": u.id,
"alignment_score": score,
"phone_error_rate": phone_error_rate,
"edit_distance": edit_distance,
},
u.file.name,
reference_phone_count,
confusions,
boundary_errors,
pi_update_mappings,
)
)
else:
p, r, f1, edit_distance, precision, recall = naive_boundary_f1(
reference_phones, comparison_phones
)
self.callback(
(
{
"id": u.id,
"edit_distance": edit_distance,
},
p,
r,
f1,
precision,
recall,
)
)
class AnalyzeAlignmentsFunction(KaldiFunction):
"""
Multiprocessing function for analyzing alignments.
See Also
--------
:meth:`.CorpusAligner.analyze_alignments`
Main function that calls this function in parallel
:meth:`.CorpusAligner.calculate_speech_post_arguments`
Job method for generating arguments for this function
:kaldi_src:`lattice-to-post`
Relevant Kaldi binary
:kaldi_src:`weight-silence-post`
Relevant Kaldi binary
Parameters
----------
args: :class:`~montreal_forced_aligner.alignment.multiprocessing.CalculateSpeechPostArguments`
Arguments for the function
"""
def __init__(self, args: AnalyzeAlignmentsArguments):
super().__init__(args)
self.analyze_intensity = args.analyze_intensity
def _run(self):
"""Run the function"""
with self.session() as session, thread_logger(
"kalpy.align", self.log_path, job_name=self.job_name
) as extraction_logger:
job: Job = (
session.query(Job)
.options(joinedload(Job.corpus, innerjoin=True), subqueryload(Job.dictionaries))
.filter(Job.id == self.job_name)
.first()
)
phones = {
k: (m, sd)
for k, m, sd in session.query(
Phone.id, Phone.mean_duration, Phone.sd_duration
).filter(
Phone.phone_type.in_([PhoneType.non_silence, PhoneType.extra]),
Phone.sd_duration != None, # noqa
Phone.sd_duration != 0,
)
}
extraction_logger.debug(f"Phone stats: {phones}")
query = (
session.query(Utterance)
.join(Utterance.speaker)
.filter(
Utterance.job_id == job.id,
Utterance.alignment_log_likelihood != None, # noqa
)
.options(
joinedload(Utterance.speaker, innerjoin=True),
joinedload(Utterance.file, innerjoin=True).joinedload(
File.sound_file, innerjoin=True
),
)
)
silence_phone_ids = [
x[0] for x in session.query(Phone.id).filter(Phone.phone_type == PhoneType.silence)
]
max_running_short_interval = 0
running_short_interval_count = 0
for utterance in query:
phone_intervals = (
session.query(PhoneInterval)
.join(PhoneInterval.phone)
.filter(
PhoneInterval.utterance_id == utterance.id,
)
.all()
)
if not phone_intervals:
extraction_logger.debug(
f"Skipping {utterance.kaldi_id} due to missing phone intervals"
)
continue
audio = utterance.segment.load_audio()
sr = 16_000
rms = librosa.feature.rms(
S=librosa.magphase(librosa.stft(audio, hop_length=int(0.01 * sr)))[0]
)[0, ...]
rms[rms == 0] = sys.float_info.epsilon
intensity = 10 * np.log10(rms)
interval_count = 0
log_like_sum = 0
duration_zscore_max = 0
silence_energy_sum = 0
silence_frame_count = 0
nonsilence_energy_sum = 0
nonsilence_frame_count = 0
pi_update_mappings = []
for pi in phone_intervals:
begin_index = int((pi.begin - utterance.begin) / 0.01)
end_index = int((pi.end - utterance.begin) / 0.01)
if self.analyze_intensity:
pi_update_mappings.append(
{
"id": pi.id,
"intensity": float(
intensity[begin_index:end_index].sum()
/ (end_index - begin_index)
),
}
)
if pi.phone_id in silence_phone_ids:
silence_frame_count += end_index - begin_index
silence_energy_sum += rms[begin_index:end_index].sum()
else:
nonsilence_frame_count += end_index - begin_index
nonsilence_energy_sum += rms[begin_index:end_index].sum()
if pi.phone_id not in phones:
continue
interval_count += 1
if pi.phone_goodness is not None:
log_like_sum += pi.phone_goodness
m, sd = phones[pi.phone_id]
duration_zscore = abs((pi.duration - m) / sd)
if duration_zscore > duration_zscore_max:
duration_zscore_max = duration_zscore
if pi.duration < 0.011:
running_short_interval_count += 1
else:
if running_short_interval_count > max_running_short_interval:
max_running_short_interval = running_short_interval_count
running_short_interval_count = 0
utterance_duration_deviation = duration_zscore_max
utterance_speech_log_likelihood = None
try:
if log_like_sum != 0:
utterance_speech_log_likelihood = log_like_sum / interval_count
except ZeroDivisionError:
pass
if silence_frame_count == 0:
silence_energy = rms.min()
else:
silence_energy = silence_energy_sum / silence_frame_count
if nonsilence_frame_count == 0:
nonsilence_energy = rms.max()
else:
nonsilence_energy = nonsilence_energy_sum / nonsilence_frame_count
nonsilence_energy = 10 * math.log10(max(nonsilence_energy, sys.float_info.epsilon))
silence_energy = 10 * math.log10(max(silence_energy, sys.float_info.epsilon))
snr = nonsilence_energy - silence_energy
extraction_logger.debug(
f"{utterance.id}: Energy shape: {rms.shape}, Nonsilence energy: {nonsilence_energy} ({nonsilence_frame_count}), Silence energy: {silence_energy} ({silence_frame_count}), SNR: {snr}"
)
self.callback(
(
utterance.id,
utterance_speech_log_likelihood,
utterance_duration_deviation,
snr,
max_running_short_interval,
pi_update_mappings,
)
)
class AnalyzeIntensityDeviationFunction(KaldiFunction):
"""
Multiprocessing function for analyzing alignments.
See Also
--------
:meth:`.CorpusAligner.analyze_alignments`
Main function that calls this function in parallel
:meth:`.CorpusAligner.calculate_speech_post_arguments`
Job method for generating arguments for this function
:kaldi_src:`lattice-to-post`
Relevant Kaldi binary
:kaldi_src:`weight-silence-post`
Relevant Kaldi binary
Parameters
----------
args: :class:`~montreal_forced_aligner.alignment.multiprocessing.CalculateSpeechPostArguments`
Arguments for the function
"""
def __init__(self, args: AnalyzeAlignmentsArguments):
super().__init__(args)
def _run(self):
"""Run the function"""
with self.session() as session, thread_logger(
"kalpy.align", self.log_path, job_name=self.job_name
) as extraction_logger:
job: Job = (
session.query(Job)
.options(joinedload(Job.corpus, innerjoin=True), subqueryload(Job.dictionaries))
.filter(Job.id == self.job_name)
.first()
)
phones = {
k: (m, sd)
for k, m, sd in session.query(
Phone.id, Phone.mean_intensity, Phone.sd_intensity
).filter(
Phone.phone_type.in_([PhoneType.non_silence, PhoneType.extra]),
Phone.sd_intensity != None, # noqa
Phone.sd_intensity != 0,
)
}
extraction_logger.debug(f"Phone stats: {phones}")
query = (
session.query(Utterance)
.join(Utterance.speaker)
.filter(
Utterance.job_id == job.id,
Utterance.alignment_log_likelihood != None, # noqa
)
)
for utterance in query:
phone_intervals = (
session.query(PhoneInterval)
.join(PhoneInterval.phone)
.filter(
PhoneInterval.utterance_id == utterance.id,
)
.all()
)
if not phone_intervals:
extraction_logger.debug(
f"Skipping {utterance.kaldi_id} due to missing phone intervals"
)
continue
max_deviation = 0
for pi in phone_intervals:
if pi.phone_id not in phones:
continue
m, sd = phones[pi.phone_id]
intensity_zscore = abs((pi.intensity - m) / sd)
max_deviation = max(max_deviation, intensity_zscore)
self.callback(
(
utterance.id,
max_deviation,
)
)
class AnalyzeTranscriptsFunction(KaldiFunction):
"""
Multiprocessing function for analyzing alignments.
See Also
--------
:meth:`.CorpusAligner.analyze_alignments`
Main function that calls this function in parallel
:meth:`.CorpusAligner.calculate_speech_post_arguments`
Job method for generating arguments for this function
:kaldi_src:`lattice-to-post`
Relevant Kaldi binary
:kaldi_src:`weight-silence-post`
Relevant Kaldi binary
Parameters
----------
args: :class:`~montreal_forced_aligner.alignment.multiprocessing.CalculateSpeechPostArguments`
Arguments for the function
"""
def __init__(self, args: AnalyzeAlignmentsArguments):
super().__init__(args)
def _run(self):
"""Run the function"""
with self.session() as session:
job: Job = (
session.query(Job)
.options(joinedload(Job.corpus, innerjoin=True), subqueryload(Job.dictionaries))
.filter(Job.id == self.job_name)
.first()
)
query = session.query(Utterance).filter(
Utterance.job_id == job.id, Utterance.alignment_log_likelihood != None # noqa
)
for utterance in query:
word_intervals = [
x.as_ctm()
for x in (
session.query(WordInterval)
.join(WordInterval.word)
.filter(
WordInterval.utterance_id == utterance.id,
Word.word_type != WordType.silence,
WordInterval.end - WordInterval.begin > 0.03,
)
.options(
joinedload(WordInterval.word, innerjoin=True),
)
.order_by(WordInterval.begin)
)
]
if not word_intervals:
continue
extra_duration, wer, aligned_duration = align_words(
utterance.normalized_text.split(), word_intervals, "<eps>", debug=True
)
transcript = " ".join(x.label for x in word_intervals)
self.callback((utterance.id, wer, extra_duration, transcript))
[docs]
class FineTuneFunction(KaldiFunction):
"""
Multiprocessing function for fine-tuning alignment.
Parameters
----------
args: :class:`~montreal_forced_aligner.alignment.multiprocessing.FineTuneArguments`
Arguments for the function
"""
def __init__(self, args: FineTuneArguments):
super().__init__(args)
self.acoustic_model = args.acoustic_model
self.lexicon_compilers = args.lexicon_compilers
self.boundary_tolerance = args.boundary_tolerance
def _run(self):
"""Run the function"""
with self.session() as session, thread_logger(
"kalpy.align", self.log_path, job_name=self.job_name
):
job = (
session.query(Job)
.options(joinedload(Job.corpus, innerjoin=True), subqueryload(Job.dictionaries))
.filter(Job.id == self.job_name)
.first()
)
workflow: CorpusWorkflow = (
session.query(CorpusWorkflow)
.filter(CorpusWorkflow.current == True) # noqa
.first()
)
cmvn_paths = job.per_dictionary_cmvn_scp_paths
trans_paths = job.per_dictionary_trans_scp_paths
for d in job.dictionaries:
aligner = KalpyAligner(
acoustic_model=self.acoustic_model,
lexicon_compiler=self.lexicon_compilers[d.id],
)
ali_path = job.construct_path(workflow.working_directory, "ali", "ark", d.name)
alignment_archive = AlignmentArchive(ali_path)
utterance_query = (
session.query(Utterance, SoundFile.sound_file_path)
.join(Utterance.file)
.join(Utterance.speaker)
.join(File.sound_file)
.filter(Utterance.job_id == self.job_name, Speaker.dictionary_id == d.id)
.order_by(Utterance.kaldi_id)
)
cmvn_reader = None
cmvn_path = cmvn_paths[d.id]
if cmvn_path.exists():
cmvn_read_specifier = generate_read_specifier(cmvn_path)
cmvn_reader = RandomAccessBaseDoubleMatrixReader(cmvn_read_specifier)
fmllr_path = trans_paths[d.id]
transform_reader = None
if fmllr_path.exists():
transform_read_specifier = generate_read_specifier(fmllr_path)
transform_reader = RandomAccessBaseFloatMatrixReader(transform_read_specifier)
current_speaker = None
current_transform = None
current_cmvn = None
for utterance, sf_path in utterance_query:
try:
alignment = alignment_archive[utterance.id]
except KeyError:
continue
interval_query = (
session.query(PhoneInterval)
.filter(
PhoneInterval.utterance_id == utterance.id,
)
.order_by(PhoneInterval.begin)
)
if utterance.speaker_id != current_speaker:
current_speaker = utterance.speaker_id
if cmvn_reader is not None and cmvn_reader.HasKey(str(current_speaker)):
current_cmvn = cmvn_reader.Value(str(current_speaker))
if transform_reader is not None and transform_reader.HasKey(
str(current_speaker)
):
current_transform = transform_reader.Value(str(current_speaker))
ctm = aligner.fine_tune_alignments(
utterance.to_kalpy(),
alignment,
boundary_tolerance=self.boundary_tolerance,
cmvn=current_cmvn,
fmllr_trans=current_transform,
)
new_boundaries = ctm.phone_boundaries
interval_mapping = []
for i, interval in enumerate(interval_query):
begin = interval.begin
if i != 0:
begin = new_boundaries[i - 1]
if i < len(new_boundaries) - 1:
interval_mapping.append(
{"id": interval.id, "begin": begin, "end": new_boundaries[i]}
)
self.callback(interval_mapping)
[docs]
class PhoneConfidenceFunction(KaldiFunction):
"""
Multiprocessing function to calculate phone confidence metrics
See Also
--------
:kaldi_src:`gmm-compute-likes`
Relevant Kaldi binary
:kaldi_src:`transform-feats`
Relevant Kaldi binary
Parameters
----------
args: :class:`~montreal_forced_aligner.alignment.multiprocessing.PhoneConfidenceArguments`
Arguments for the function
"""
def __init__(self, args: PhoneConfidenceArguments):
super().__init__(args)
self.working_directory = args.working_directory
self.model_path = args.model_path
self.phone_pdf_counts_path = args.phone_pdf_counts_path
def _run(self):
"""Run the function"""
with mfa_open(self.phone_pdf_counts_path, "r") as f:
data = json.load(f)
phone_pdf_mapping = collections.defaultdict(collections.Counter)
for phone, pdf_counts in data.items():
phone = split_phone_position(phone)[0]
for pdf, count in pdf_counts.items():
phone_pdf_mapping[phone][int(pdf)] += count
phones = {p: i for i, p in enumerate(sorted(phone_pdf_mapping.keys()))}
reversed_phones = {k: v for v, k in phones.items()}
for phone, pdf_counts in phone_pdf_mapping.items():
phone_total = sum(pdf_counts.values())
for pdf, count in pdf_counts.items():
phone_pdf_mapping[phone][int(pdf)] = count / phone_total
_, acoustic_model = read_gmm_model(self.model_path)
with self.session() as session:
job: typing.Optional[Job] = session.get(
Job, self.job_name, options=[joinedload(Job.dictionaries), joinedload(Job.corpus)]
)
utterances = (
session.query(Utterance)
.filter(Utterance.job_id == self.job_name)
.options(
selectinload(Utterance.phone_intervals).joinedload(
PhoneInterval.phone, innerjoin=True
),
selectinload(Utterance.reference_phone_intervals).options(
joinedload(ReferencePhoneInterval.phone, innerjoin=True),
),
)
)
utterances = {u.id: (u.begin, u.phone_intervals) for u in utterances}
for d in job.dictionaries:
try:
feature_archive = job.construct_feature_archive(self.working_directory, d.name)
except OSError:
continue
interval_mappings = []
for utterance_id, feats in feature_archive:
utterance_id = int(utterance_id.split("-")[-1])
likelihoods = gmm_compute_likes(acoustic_model, feats).numpy()
phone_likes = np.zeros((likelihoods.shape[0], len(phones)))
for i, p in reversed_phones.items():
like = likelihoods[:, [x for x in phone_pdf_mapping[p].keys()]]
weight = np.array([x for x in phone_pdf_mapping[p].values()])
phone_likes[:, i] = np.dot(like, weight)
top_phone_inds = np.argmax(phone_likes, axis=1)
utt_begin, intervals = utterances[utterance_id]
for pi in intervals:
if pi.phone.phone == "sil":
continue
frame_begin = int(((pi.begin - utt_begin) * 1000) / 10)
frame_end = int(((pi.end - utt_begin) * 1000) / 10)
if frame_begin == frame_end:
frame_end += 1
frame_end = min(frame_end, top_phone_inds.shape[0])
alternate_labels = collections.Counter()
scores = []
for i in range(frame_begin, frame_end):
top_phone_ind = top_phone_inds[i]
alternate_label = reversed_phones[top_phone_ind]
alternate_label = split_phone_position(alternate_label)[0]
alternate_labels[alternate_label] += 1
if alternate_label == pi.phone.phone:
scores.append(0)
else:
actual_score = phone_likes[i, phones[pi.phone.phone]]
scores.append(phone_likes[i, top_phone_ind] - actual_score)
average_score = statistics.mean(scores)
interval_mappings.append(
{"id": pi.id, "phone_goodness": float(average_score)}
)
self.callback(interval_mappings)
interval_mappings = []
[docs]
class GeneratePronunciationsFunction(KaldiFunction):
"""
Multiprocessing function for generating pronunciations
See Also
--------
:meth:`.DictionaryTrainer.export_lexicons`
Main function that calls this function in parallel
:meth:`.CorpusAligner.generate_pronunciations_arguments`
Job method for generating arguments for this function
:kaldi_src:`linear-to-nbest`
Kaldi binary this uses
Parameters
----------
args: :class:`~montreal_forced_aligner.alignment.multiprocessing.AlignArguments`
Arguments for the function
"""
def __init__(self, args: GeneratePronunciationsArguments):
super().__init__(args)
self.aligner = args.aligner
self.lexicon_compilers = args.lexicon_compilers
self.for_g2p = args.for_g2p
self.silence_words = set()
def _process_pronunciations(
self, word_pronunciations: typing.List[typing.Tuple[str, str]]
) -> PronunciationProbabilityCounter:
"""
Process an utterance's pronunciations and extract relevant count information
Parameters
----------
word_pronunciations: list[tuple[str, tuple[str, ...]]]
List of tuples containing the word integer ID and a list of the integer IDs of the phones
"""
counter = PronunciationProbabilityCounter()
word_pronunciations = [("<s>", "")] + word_pronunciations + [("</s>", "")]
for i, w_p in enumerate(word_pronunciations):
if i != 0:
word = word_pronunciations[i - 1][0]
if word in self.silence_words:
counter.silence_before_counts[w_p] += 1
else:
counter.non_silence_before_counts[w_p] += 1
silence_check = w_p[0] in self.silence_words
if not silence_check:
counter.word_pronunciation_counts[w_p[0]][w_p[1]] += 1
if i != len(word_pronunciations) - 1:
word = word_pronunciations[i + 1][0]
if word in self.silence_words:
counter.silence_following_counts[w_p] += 1
if i != len(word_pronunciations) - 2:
next_w_p = word_pronunciations[i + 2]
counter.ngram_counts[w_p, next_w_p]["silence"] += 1
else:
next_w_p = word_pronunciations[i + 1]
counter.non_silence_following_counts[w_p] += 1
counter.ngram_counts[w_p, next_w_p]["non_silence"] += 1
return counter
def _run(self) -> None:
"""Run the function"""
with self.session() as session:
job = (
session.query(Job)
.options(joinedload(Job.corpus, innerjoin=True), subqueryload(Job.dictionaries))
.filter(Job.id == self.job_name)
.first()
)
workflow: CorpusWorkflow = (
session.query(CorpusWorkflow)
.filter(CorpusWorkflow.current == True) # noqa
.first()
)
silence_words = session.query(Word.word).filter(Word.word_type == WordType.silence)
self.silence_words.update(x for x, in silence_words)
for d in job.training_dictionaries:
ali_path = job.construct_path(workflow.working_directory, "ali", "ark", d.name)
if not os.path.exists(ali_path):
continue
if self.lexicon_compilers and d.id in self.lexicon_compilers:
lexicon_compiler = self.lexicon_compilers[d.id]
else:
lexicon_compiler = d.lexicon_compiler
words_path = job.construct_path(workflow.working_directory, "words", "ark", d.name)
alignment_archive = AlignmentArchive(ali_path, words_file_name=words_path)
for alignment in alignment_archive:
intervals = alignment.generate_ctm(
self.aligner.transition_model, lexicon_compiler.phone_table
)
utterance = int(alignment.utterance_id.split("-")[-1])
ctm = lexicon_compiler.phones_to_pronunciations(alignment.words, intervals)
word_pronunciations = []
for wi in ctm.word_intervals:
label = wi.label
pronunciation = wi.pronunciation
if label.startswith(d.cutoff_word[:-1]):
label = d.cutoff_word
if pronunciation != d.oov_phone:
pronunciation = "cutoff_model"
word_pronunciations.append((label, pronunciation))
if self.for_g2p:
phones = []
for i, x in enumerate(word_pronunciations):
if i > 0 and (
x[0].startswith(d.clitic_marker)
or word_pronunciations[i - 1][0].endswith(d.clitic_marker)
):
phones.pop(-1)
else:
phones.append(WORD_BEGIN_SYMBOL)
phones.extend(x[1].split())
phones.append(WORD_END_SYMBOL)
self.callback((d.id, utterance, " ".join(phones)))
else:
self.callback((d.id, self._process_pronunciations(word_pronunciations)))
[docs]
class ExportTextGridProcessWorker(mp.Process):
"""
Multiprocessing worker for exporting TextGrids
See Also
--------
:meth:`.CorpusAligner.collect_alignments`
Main function that runs this worker in parallel
Parameters
----------
for_write_queue: :class:`~multiprocessing.Queue`
Input queue of files to export
stopped: :class:`~multiprocessing.Event`
Stop check for processing
finished_adding: :class:`~multiprocessing.Event`
Input signal that all jobs have been added and no more new ones will come in
textgrid_errors: dict[str, str]
Dictionary for storing errors encountered
arguments: :class:`~montreal_forced_aligner.alignment.multiprocessing.ExportTextGridArguments`
Arguments to pass to the TextGrid export function
exported_file_count: :class:`~montreal_forced_aligner.utils.Counter`
Counter for exported files
"""
def __init__(
self,
db_string: str,
for_write_queue: mp.Queue,
return_queue: mp.Queue,
stopped: mp.Event,
finished_adding: mp.Event,
export_frame_shift: float,
cleanup_textgrids: bool,
clitic_marker: str,
output_directory: Path,
output_format: str,
include_original_text: bool,
):
super().__init__()
self.db_string = db_string
self.for_write_queue = for_write_queue
self.return_queue = return_queue
self.stopped = stopped
self.finished_adding = finished_adding
self.finished_processing = mp.Event()
self.output_directory = output_directory
self.output_format = output_format
self.export_frame_shift = export_frame_shift
self.include_original_text = include_original_text
self.cleanup_textgrids = cleanup_textgrids
self.clitic_marker = clitic_marker
[docs]
def run(self) -> None:
"""Run the exporter function"""
db_engine = sqlalchemy.create_engine(self.db_string)
with sqlalchemy.orm.Session(db_engine) as session:
while True:
try:
(file_batch) = self.for_write_queue.get(timeout=1)
except Empty:
if self.finished_adding.is_set():
self.finished_processing.set()
break
continue
if self.stopped.is_set():
continue
try:
for output_path in construct_textgrid_output(
session,
file_batch,
self.cleanup_textgrids,
self.clitic_marker,
self.output_directory,
self.export_frame_shift,
self.output_format,
self.include_original_text,
):
self.return_queue.put(1)
except Exception:
exc_type, exc_value, exc_traceback = sys.exc_info()
self.return_queue.put(
AlignmentExportError(
output_path,
traceback.format_exception(exc_type, exc_value, exc_traceback),
)
)
self.stopped.set()
