| """Based on https://pytorch.org/audio/stable/tutorials/ctc_forced_alignment_api_tutorial.html""" |
| import csv |
| import os |
| import re |
| import sys |
| import typing |
|
|
| import sqlalchemy |
| import torch |
| import torchaudio |
| import torchaudio.functional as F |
| import tqdm |
| from kalpy.fstext.lexicon import CtmInterval as KalpyCtmInterval |
| from kalpy.fstext.lexicon import HierarchicalCtm, WordCtmInterval |
| from montreal_forced_aligner import config |
| from montreal_forced_aligner.command_line.mfa import mfa_cli |
| from montreal_forced_aligner.corpus.acoustic_corpus import AcousticCorpus |
| from montreal_forced_aligner.data import CtmInterval |
| from montreal_forced_aligner.db import File, Utterance, Word, WordInterval |
| from montreal_forced_aligner.exceptions import TextGridParseError |
| from montreal_forced_aligner.helper import mfa_open |
| from praatio import textgrid as tgio |
|
|
| root_dir = r"D:\Data\experiments\alignment_benchmarking" |
| mfa10_dir = r"D:\Data\models\1.0_archived" |
| mfa20_dir = r"D:\Data\models\2.0_archived" |
| mfa20a_dir = r"D:\Data\models\2.0.0a_archived" |
| mfa21_dir = r"D:\Data\models\2.1_trained" |
| mfa22_dir = r"D:\Data\models\2.2_trained" |
| mfa30_dir = r"D:\Data\models\3.0_trained" |
| trained22_dir = r"D:\Data\models\2.2_trained\buckeye" |
| trained30_dir = r"D:\Data\models\3.0_trained\buckeye" |
| mapping_directory = os.path.join( |
| os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "mapping_files" |
| ) |
| nemo_tools_path = r"C:\Users\micha\Documents\Dev\NeMo\tools" |
| config.CLEAN = True |
| config.USE_POSTGRES = True |
|
|
| root_dir = r"D:\Data\experiments\alignment_benchmarking" |
|
|
| corpus_directories = { |
| "timit": r"D:\Data\speech\benchmark_datasets\timit", |
| "buckeye": r"D:\Data\speech\Buckeye", |
| } |
|
|
| conditions = { |
| "torchaudio_mms_fa": (None, None), |
| "nemo_forced_aligner": (None, None), |
| "whisperx": (None, None), |
| "arpa_1.0": ( |
| os.path.join(mfa10_dir, "english.dict"), |
| os.path.join(mfa10_dir, "english.zip"), |
| ), |
| "arpa_2.0": ( |
| os.path.join(mfa20_dir, "english_us_arpa.dict"), |
| os.path.join(mfa20_dir, "english_us_arpa.zip"), |
| ), |
| "arpa_2.0a": ( |
| os.path.join(mfa20a_dir, "english_us_arpa.dict"), |
| os.path.join(mfa20a_dir, "english_us_arpa.zip"), |
| ), |
| "mfa_2.0": ( |
| os.path.join(mfa20_dir, "english_us_mfa.dict"), |
| os.path.join(mfa20_dir, "english_mfa.zip"), |
| ), |
| "mfa_2.0a": ( |
| os.path.join(mfa20a_dir, "english_us_mfa.dict"), |
| os.path.join(mfa20a_dir, "english_mfa.zip"), |
| ), |
| "mfa_2.1": ( |
| os.path.join(mfa21_dir, "english_us_mfa.dict"), |
| os.path.join(mfa21_dir, "english_mfa.zip"), |
| ), |
| "mfa_2.2": ( |
| os.path.join(mfa22_dir, "english_us_mfa.dict"), |
| os.path.join(mfa22_dir, "english_mfa.zip"), |
| ), |
| "mfa_3.0": ( |
| os.path.join(mfa30_dir, "english_us_mfa.dict"), |
| os.path.join(mfa30_dir, "english_mfa.zip"), |
| ), |
| "arpa_3.0": ( |
| os.path.join(mfa30_dir, "english_us_arpa.dict"), |
| os.path.join(mfa30_dir, "english_us_arpa.zip"), |
| ), |
| "trained_2.2": ( |
| os.path.join(trained22_dir, "english_us_mfa.dict"), |
| os.path.join(trained22_dir, "english_mfa.zip"), |
| ), |
| "trained_3.0": ( |
| os.path.join(trained30_dir, "english_us_mfa.dict"), |
| os.path.join(trained30_dir, "english_mfa.zip"), |
| ), |
| "arpa_2.2": ( |
| os.path.join(mfa20a_dir, "english_us_arpa.dict"), |
| os.path.join(mfa20a_dir, "english_us_arpa.zip"), |
| ), |
| } |
|
|
|
|
| def align(emission, tokens): |
| targets = torch.tensor([tokens], dtype=torch.int32, device=device) |
| alignments, scores = F.forced_align(emission, targets, blank=0) |
|
|
| alignments, scores = ( |
| alignments[0], |
| scores[0], |
| ) |
| scores = scores.exp() |
| return alignments, scores |
|
|
|
|
| def unflatten(list_, lengths): |
| assert len(list_) == sum(lengths) |
| i = 0 |
| ret = [] |
| for l in lengths: |
| ret.append(list_[i : i + l]) |
| i += l |
| return ret |
|
|
|
|
| def generate_ctm( |
| token_spans, transcript, ratio, sample_rate, utterance_begin, reference |
| ) -> HierarchicalCtm: |
| i = 0 |
| reference_index = 0 |
| word_intervals = [] |
| for word in transcript: |
| character_intervals = [] |
| for j in range(len(word)): |
| span = token_spans[i] |
| begin = int(ratio * span.start) / sample_rate |
| begin += utterance_begin |
| end = int(ratio * span.end) / sample_rate |
| end += utterance_begin |
| character_intervals.append( |
| KalpyCtmInterval(begin, end, LABELS[span.token], span.token) |
| ) |
| i += 1 |
| if word == reference[reference_index].label: |
| reference_index += 1 |
| elif ( |
| word_intervals |
| and f"{word_intervals[-1].label}-{word}" in reference[reference_index].label |
| ): |
| word_intervals[-1].label = f"{word_intervals[-1].label}-{word}" |
| word_intervals[-1].symbol = f"{word_intervals[-1].label}-{word}" |
| word_intervals[-1].phones.extend(character_intervals) |
| if reference[reference_index].label == word_intervals[-1].label: |
| reference_index += 1 |
| continue |
| word_intervals.append(WordCtmInterval(word, word, character_intervals)) |
| return HierarchicalCtm(word_intervals) |
|
|
|
|
| def align_words( |
| ref: typing.List[CtmInterval], |
| test: typing.List[CtmInterval], |
| ): |
| try: |
| assert len(ref) == len(test) |
| except AssertionError: |
| print(ref) |
| print(test) |
| raise |
| error_sum = 0.0 |
| boundary_count = 0 |
| for i, r in enumerate(ref): |
| t = test[i] |
| |
| error_sum += abs(r.begin - t.begin) |
| error_sum += abs(r.end - t.end) |
| boundary_count += 2 |
| return error_sum / boundary_count |
|
|
|
|
| def parse_aligned_textgrid(path: str, exclude_unknowns=True) -> typing.List[CtmInterval]: |
| """ |
| Load a TextGrid as a dictionary of speaker's phone tiers |
| |
| Parameters |
| ---------- |
| path: :class:`~pathlib.Path` |
| TextGrid file to parse |
| |
| Returns |
| ------- |
| dict[str, list[:class:`~montreal_forced_aligner.data.CtmInterval`]] |
| Parsed phone tier |
| """ |
| tg = tgio.openTextgrid(path, includeEmptyIntervals=False, reportingMode="silence") |
| num_tiers = len(tg.tiers) |
| if num_tiers == 0: |
| raise TextGridParseError(path, "Number of tiers parsed was zero") |
| data = [] |
| for tier_name in tg.tierNames: |
| ti = tg._tierDict[tier_name] |
| if not isinstance(ti, tgio.IntervalTier): |
| continue |
| if "words" not in tier_name: |
| continue |
| for begin, end, text in ti.entries: |
| text = text.lower().strip().replace("?", "") |
| if exclude_unknowns and re.match(r"^[\[(<].*", text): |
| continue |
| if not text: |
| continue |
| begin, end = round(begin, 4), round(end, 4) |
| |
| |
| interval = CtmInterval(begin, end, text) |
| data.append(interval) |
| return data |
|
|
|
|
| if __name__ == "__main__": |
| csv_header = [ |
| "file", |
| "begin", |
| "end", |
| "speaker", |
| "duration", |
| "normalized_text", |
| "filtered_text", |
| "alignment_score", |
| "alignment_likelihood", |
| "word_count", |
| "frames_per_second", |
| "words_per_second", |
| ] |
| device = torch.device("cuda" if torch.cuda.is_available() else "cpu") |
| print(device) |
|
|
| for condition, (dictionary_path, model_path) in conditions.items(): |
| if condition == "torchaudio_mms_fa": |
| with torch.inference_mode(): |
| for corpus, root in corpus_directories.items(): |
| output_directory = os.path.join(root_dir, "alignments", condition, corpus) |
| if os.path.exists(output_directory): |
| continue |
| try: |
| bundle = torchaudio.pipelines.MMS_FA |
| except AttributeError: |
| print("Incorrect version of torchaudio, skipping torchaudio_mms_fa") |
| continue |
|
|
| model = bundle.get_model(with_star=False).to(device) |
|
|
| LABELS = bundle.get_labels(star=None) |
| DICTIONARY = bundle.get_dict(star=None) |
| corpus = AcousticCorpus(corpus_directory=os.path.join(root, "benchmark")) |
| corpus.delete_database() |
| corpus._load_corpus() |
| os.makedirs(output_directory, exist_ok=True) |
| csv_path = os.path.join(output_directory, "word_alignment.csv") |
| with ( |
| corpus.session() as session, |
| mfa_open(csv_path, "w") as f, |
| tqdm.tqdm(total=corpus.num_utterances) as progress_bar, |
| ): |
| writer = csv.DictWriter(f, fieldnames=csv_header) |
| writer.writeheader() |
| file_query = ( |
| session.query(File) |
| .options( |
| sqlalchemy.orm.joinedload(File.sound_file, innerjoin=True), |
| ) |
| .order_by(File.name) |
| ) |
| for file in file_query: |
| file_ctm = HierarchicalCtm([]) |
| if file.relative_path: |
| output_path = os.path.join(output_directory, file.relative_path) |
| os.makedirs(output_path, exist_ok=True) |
| else: |
| output_path = output_directory |
| output_path = os.path.join(output_path, file.name + ".TextGrid") |
| query = ( |
| session.query(Utterance) |
| .filter(Utterance.file_id == file.id) |
| .options( |
| sqlalchemy.orm.joinedload( |
| Utterance.file, innerjoin=True |
| ).joinedload(File.sound_file, innerjoin=True), |
| ) |
| .order_by(Utterance.begin) |
| ) |
| reference_path = os.path.join(root, "reference") |
| if file.relative_path: |
| reference_path = os.path.join(reference_path, file.relative_path) |
| reference_path = os.path.join(reference_path, file.name + ".TextGrid") |
| reference_word_intervals = parse_aligned_textgrid(reference_path) |
| reference_interval_index = 0 |
| for utterance in query: |
| kalpy_utterance = utterance.to_kalpy() |
| waveform = torch.unsqueeze( |
| torch.Tensor(kalpy_utterance.segment.load_audio()), |
| 0, |
| ) |
| text = utterance.text.lower().replace("-", " ") |
| transcript = [ |
| re.sub(rf'[^{"".join(LABELS[1:])}]', "", x) |
| for x in text.split() |
| if not re.match(r"^[\[(<].*", x) |
| ] |
| transcript = [x for x in transcript if x] |
| tokenized_transcript = [ |
| DICTIONARY[c] for word in transcript for c in word |
| ] |
| emission, _ = model(waveform.to(device)) |
| aligned_tokens, alignment_scores = align( |
| emission, tokenized_transcript |
| ) |
| token_spans = F.merge_tokens(aligned_tokens, alignment_scores) |
| alignment_likelihood = sum(alignment_scores) / len( |
| alignment_scores |
| ) |
| word_spans = unflatten( |
| token_spans, [len(word) for word in transcript] |
| ) |
| ratio = waveform.size(1) / emission.size(1) |
| frames_per_second = emission.size(1) / utterance.duration |
| reference = [] |
| while True: |
| try: |
| interval = reference_word_intervals[ |
| reference_interval_index |
| ] |
| except IndexError: |
| break |
| if interval.begin >= utterance.end: |
| break |
| if interval.begin >= utterance.begin: |
| reference.append(interval) |
| reference_interval_index += 1 |
| ctm = generate_ctm( |
| token_spans, |
| transcript, |
| ratio, |
| bundle.sample_rate, |
| utterance.begin, |
| reference, |
| ) |
| file_ctm.word_intervals.extend(ctm.word_intervals) |
| try: |
| alignment_score = align_words(reference, ctm.word_intervals) |
| except AssertionError: |
| print(file.name) |
| raise |
| continue |
| words_per_second = len(reference) / utterance.duration |
|
|
| data = { |
| "file": file.name, |
| "begin": utterance.begin, |
| "end": utterance.end, |
| "duration": utterance.duration, |
| "speaker": utterance.speaker_name, |
| "normalized_text": text.replace(",", ""), |
| "filtered_text": " ".join(transcript), |
| "alignment_score": alignment_score, |
| "alignment_likelihood": float(alignment_likelihood), |
| "word_count": len(ctm.word_intervals), |
| "frames_per_second": frames_per_second, |
| "words_per_second": words_per_second, |
| } |
| writer.writerow(data) |
| progress_bar.update(1) |
| file_ctm.export_textgrid( |
| output_path, |
| file_duration=file.duration, |
| output_format="short_textgrid", |
| ) |
| elif condition == "nemo_forced_aligner": |
| try: |
| from dataclasses import dataclass, field, is_dataclass |
|
|
| from nemo.collections.asr.models.ctc_models import EncDecCTCModel |
| from nemo.collections.asr.models.hybrid_rnnt_ctc_models import ( |
| EncDecHybridRNNTCTCModel, |
| ) |
| from nemo.collections.asr.parts.utils.streaming_utils import FrameBatchASR |
| from nemo.collections.asr.parts.utils.transcribe_utils import setup_model |
| from nemo.core.config import hydra_runner |
| except ImportError: |
| raise |
| continue |
| sys.path.append(nemo_tools_path) |
| sys.path.insert(0, os.path.join(nemo_tools_path, "nemo_forced_aligner")) |
| import tempfile |
|
|
| from nemo_forced_aligner.align import AlignmentConfig, ASSFileConfig, CTMFileConfig |
| from nemo_forced_aligner.utils.data_prep import ( |
| V_NEGATIVE_NUM, |
| Segment, |
| Token, |
| Word, |
| add_t_start_end_to_utt_obj, |
| get_utt_obj, |
| ) |
| from nemo_forced_aligner.utils.viterbi_decoding import viterbi_decoding |
|
|
| cfg = AlignmentConfig() |
| cfg.pretrained_name = "stt_en_conformer_ctc_medium" |
| if cfg.transcribe_device is None: |
| transcribe_device = torch.device("cuda" if torch.cuda.is_available() else "cpu") |
| else: |
| transcribe_device = torch.device(cfg.transcribe_device) |
| if cfg.viterbi_device is None: |
| viterbi_device = torch.device("cuda" if torch.cuda.is_available() else "cpu") |
| else: |
| viterbi_device = torch.device(cfg.viterbi_device) |
| model, _ = setup_model(cfg, transcribe_device) |
| model.eval() |
| if isinstance(model, EncDecHybridRNNTCTCModel): |
| model.change_decoding_strategy(decoder_type="ctc") |
| if cfg.use_local_attention: |
| model.change_attention_model( |
| self_attention_model="rel_pos_local_attn", att_context_size=[64, 64] |
| ) |
|
|
| with torch.inference_mode(): |
| for corpus, root in corpus_directories.items(): |
| output_directory = os.path.join(root_dir, "alignments", condition, corpus) |
| if os.path.exists(output_directory): |
| continue |
| corpus = AcousticCorpus(corpus_directory=os.path.join(root, "benchmark")) |
| corpus.delete_database() |
| corpus._load_corpus() |
| os.makedirs(output_directory, exist_ok=True) |
| csv_path = os.path.join(output_directory, "word_alignment.csv") |
| with ( |
| corpus.session() as session, |
| mfa_open(csv_path, "w") as f, |
| tempfile.TemporaryDirectory() as tmpdir, |
| tqdm.tqdm(total=corpus.num_utterances) as progress_bar, |
| ): |
| writer = csv.DictWriter(f, fieldnames=csv_header) |
| writer.writeheader() |
| file_query = ( |
| session.query(File) |
| .options( |
| sqlalchemy.orm.joinedload(File.sound_file, innerjoin=True), |
| ) |
| .order_by(File.name) |
| ) |
| for file in file_query: |
| file_ctm = HierarchicalCtm([]) |
| if file.relative_path: |
| output_path = os.path.join(output_directory, file.relative_path) |
| os.makedirs(output_path, exist_ok=True) |
| else: |
| output_path = output_directory |
| output_path = os.path.join(output_path, file.name + ".TextGrid") |
| query = ( |
| session.query(Utterance) |
| .filter(Utterance.file_id == file.id) |
| .options( |
| sqlalchemy.orm.joinedload( |
| Utterance.file, innerjoin=True |
| ).joinedload(File.sound_file, innerjoin=True), |
| ) |
| .order_by(Utterance.begin) |
| ) |
| reference_path = os.path.join(root, "reference") |
| if file.relative_path: |
| reference_path = os.path.join(reference_path, file.relative_path) |
| reference_path = os.path.join(reference_path, file.name + ".TextGrid") |
| reference_word_intervals = parse_aligned_textgrid(reference_path) |
| reference_interval_index = 0 |
| for utterance in query: |
| kalpy_utterance = utterance.to_kalpy() |
| reference = [] |
| while True: |
| try: |
| interval = reference_word_intervals[ |
| reference_interval_index |
| ] |
| except IndexError: |
| break |
| if interval.begin >= utterance.end: |
| break |
| if interval.begin >= utterance.begin: |
| reference.append(interval) |
| reference_interval_index += 1 |
| text = " ".join(x.label for x in reference) |
| with torch.no_grad(): |
| output_timestep_duration = None |
| path = os.path.join(tmpdir, f"sound_file.wav") |
| torchaudio.save( |
| path, |
| torch.unsqueeze( |
| torch.Tensor(kalpy_utterance.segment.load_audio()), |
| 0, |
| ), |
| 16000, |
| ) |
| hypotheses = model.transcribe( |
| [path], |
| return_hypotheses=True, |
| verbose=False, |
| batch_size=1, |
| ) |
| |
| if type(hypotheses) == tuple and len(hypotheses) == 2: |
| hypotheses = hypotheses[0] |
|
|
| for hypothesis in hypotheses: |
| log_probs_list_batch = [hypothesis.y_sequence] |
| T_list_batch = [hypothesis.y_sequence.shape[0]] |
| pred_text_batch = [hypothesis.text] |
| utt_obj = get_utt_obj( |
| text, |
| model, |
| cfg.additional_segment_grouping_separator, |
| T_list_batch[0], |
| file.sound_file.sound_file_path, |
| utterance.kaldi_id, |
| ) |
| alignment_likelihood = float(hypothesis.score) |
| y_list_batch = [utt_obj.token_ids_with_blanks] |
| U_list_batch = [len(utt_obj.token_ids_with_blanks)] |
| utt_obj_batch = [utt_obj] |
| T_max = max(T_list_batch) |
| U_max = max(U_list_batch) |
| if hasattr(model, "tokenizer"): |
| V = len(model.tokenizer.vocab) + 1 |
| else: |
| V = len(model.decoder.vocabulary) + 1 |
| T_batch = torch.tensor(T_list_batch) |
| U_batch = torch.tensor(U_list_batch) |
| log_probs_batch = V_NEGATIVE_NUM * torch.ones((1, T_max, V)) |
| for b, log_probs_utt in enumerate(log_probs_list_batch): |
| t = log_probs_utt.shape[0] |
| log_probs_batch[b, :t, :] = log_probs_utt |
| y_batch = V * torch.ones((1, U_max), dtype=torch.int64) |
| for b, y_utt in enumerate(y_list_batch): |
| U_utt = U_batch[b] |
| y_batch[b, :U_utt] = torch.tensor(y_utt) |
| |
| if output_timestep_duration is None: |
| if not "window_stride" in model.cfg.preprocessor: |
| raise ValueError( |
| "Don't have attribute 'window_stride' in " |
| "'model.cfg.preprocessor' => cannot calculate " |
| " model_downsample_factor => stopping process" |
| ) |
|
|
| if not "sample_rate" in model.cfg.preprocessor: |
| raise ValueError( |
| "Don't have attribute 'sample_rate' in " |
| "'model.cfg.preprocessor' => cannot calculate start " |
| " and end time of segments => stopping process" |
| ) |
|
|
| audio_dur = utterance.duration |
| n_input_frames = ( |
| audio_dur / model.cfg.preprocessor.window_stride |
| ) |
| model_downsample_factor = round( |
| n_input_frames / int(T_batch[0]) |
| ) |
|
|
| output_timestep_duration = ( |
| model.preprocessor.featurizer.hop_length |
| * model_downsample_factor |
| / model.cfg.preprocessor.sample_rate |
| ) |
| alignments_batch = viterbi_decoding( |
| log_probs_batch, |
| y_batch, |
| T_batch, |
| U_batch, |
| viterbi_device, |
| ) |
| utt_obj, alignment_utt = ( |
| utt_obj_batch[0], |
| alignments_batch[0], |
| ) |
|
|
| utt_obj = add_t_start_end_to_utt_obj( |
| utt_obj, alignment_utt, output_timestep_duration |
| ) |
| segment = [ |
| x |
| for x in utt_obj.segments_and_tokens |
| if isinstance(x, Segment) |
| ][0] |
| words = [ |
| x for x in segment.words_and_tokens if isinstance(x, Word) |
| ] |
| word_intervals = [] |
| for word in words: |
| character_intervals = [] |
| word_label = word.text |
| for token in word.tokens: |
| if token.text == "<b>": |
| continue |
| begin = token.t_start + utterance.begin |
| end = token.t_end + utterance.begin |
| if end > utterance.end: |
| end = utterance.end |
| label = token.text |
| character_intervals.append( |
| KalpyCtmInterval(begin, end, label, label) |
| ) |
| word_intervals.append( |
| WordCtmInterval( |
| word_label, word_label, character_intervals |
| ) |
| ) |
| ctm = HierarchicalCtm(word_intervals) |
|
|
| frames_per_second = ( |
| log_probs_list_batch[0].size(0) / utterance.duration |
| ) |
| file_ctm.word_intervals.extend(ctm.word_intervals) |
| try: |
| alignment_score = align_words(reference, ctm.word_intervals) |
| except AssertionError: |
| print(file.name) |
| raise |
| continue |
| words_per_second = len(reference) / utterance.duration |
|
|
| data = { |
| "file": file.name, |
| "begin": utterance.begin, |
| "end": utterance.end, |
| "duration": utterance.duration, |
| "speaker": utterance.speaker_name, |
| "normalized_text": utterance.normalized_text, |
| "filtered_text": text, |
| "alignment_score": alignment_score, |
| "alignment_likelihood": float(alignment_likelihood), |
| "word_count": len(ctm.word_intervals), |
| "frames_per_second": frames_per_second, |
| "words_per_second": words_per_second, |
| } |
| writer.writerow(data) |
| progress_bar.update(1) |
| file_ctm.export_textgrid( |
| output_path, |
| file_duration=file.duration, |
| output_format="short_textgrid", |
| ) |
| elif condition == "whisperx": |
| try: |
| import whisperx |
| except ImportError: |
| print("whisperx not installed, skipping") |
| continue |
| compute_type = "float16" |
| device = "cuda" if torch.cuda.is_available() else "cpu" |
| model = whisperx.load_model( |
| "large-v2", device, compute_type=compute_type, language="en" |
| ) |
| model_a, metadata = whisperx.load_align_model(language_code="en", device=device) |
|
|
| with torch.inference_mode(): |
| for corpus, root in corpus_directories.items(): |
| output_directory = os.path.join(root_dir, "alignments", condition, corpus) |
| if os.path.exists(output_directory): |
| continue |
| corpus = AcousticCorpus(corpus_directory=os.path.join(root, "benchmark")) |
| corpus.delete_database() |
| corpus._load_corpus() |
| os.makedirs(output_directory, exist_ok=True) |
| csv_path = os.path.join(output_directory, "word_alignment.csv") |
| with ( |
| corpus.session() as session, |
| mfa_open(csv_path, "w") as f, |
| tqdm.tqdm(total=corpus.num_utterances) as progress_bar, |
| ): |
| writer = csv.DictWriter(f, fieldnames=csv_header) |
| writer.writeheader() |
| file_query = ( |
| session.query(File) |
| .options( |
| sqlalchemy.orm.joinedload(File.sound_file, innerjoin=True), |
| ) |
| .order_by(File.name) |
| ) |
| for file in file_query: |
| file_ctm = HierarchicalCtm([]) |
| if file.relative_path: |
| output_path = os.path.join(output_directory, file.relative_path) |
| os.makedirs(output_path, exist_ok=True) |
| else: |
| output_path = output_directory |
| output_path = os.path.join(output_path, file.name + ".TextGrid") |
| query = ( |
| session.query(Utterance) |
| .filter(Utterance.file_id == file.id) |
| .options( |
| sqlalchemy.orm.joinedload( |
| Utterance.file, innerjoin=True |
| ).joinedload(File.sound_file, innerjoin=True), |
| ) |
| .order_by(Utterance.begin) |
| ) |
| reference_path = os.path.join(root, "reference") |
| if file.relative_path: |
| reference_path = os.path.join(reference_path, file.relative_path) |
| reference_path = os.path.join(reference_path, file.name + ".TextGrid") |
| reference_word_intervals = parse_aligned_textgrid(reference_path) |
| reference_interval_index = 0 |
| for utterance in query: |
| kalpy_utterance = utterance.to_kalpy() |
| reference = [] |
| while True: |
| try: |
| interval = reference_word_intervals[ |
| reference_interval_index |
| ] |
| except IndexError: |
| break |
| if interval.begin >= utterance.end: |
| break |
| if interval.begin >= utterance.begin: |
| reference.append(interval) |
| reference_interval_index += 1 |
| text = " ".join(x.label for x in reference) |
|
|
| audio = kalpy_utterance.segment.load_audio() |
|
|
| |
| |
| segments = [ |
| { |
| "text": text, |
| "start": 0.0, |
| "end": utterance.duration, |
| } |
| ] |
| result = whisperx.align( |
| segments, |
| model_a, |
| metadata, |
| audio, |
| device, |
| return_char_alignments=True, |
| ) |
|
|
| word_intervals = [] |
| words = result["segments"][0]["words"] |
| chars = result["segments"][0]["chars"] |
| character_index = 0 |
| score_sum = 0 |
| for word in words: |
| character_intervals = [] |
| word_label = word["word"] |
| while True: |
| try: |
| c = chars[character_index] |
| except IndexError: |
| break |
| if c["char"] == " ": |
| character_index += 1 |
| continue |
| if c["start"] >= word["end"]: |
| break |
| begin = c["start"] + utterance.begin |
| end = c["end"] + utterance.begin |
| character_intervals.append( |
| KalpyCtmInterval(begin, end, c["char"], c["char"]) |
| ) |
| character_index += 1 |
| word_intervals.append( |
| WordCtmInterval( |
| word_label, word_label, character_intervals |
| ) |
| ) |
| score_sum += word["score"] |
| ctm = HierarchicalCtm(word_intervals) |
| alignment_likelihood = score_sum / len(words) |
| file_ctm.word_intervals.extend(ctm.word_intervals) |
| try: |
| alignment_score = align_words(reference, ctm.word_intervals) |
| except AssertionError: |
| print(file.name) |
| raise |
| continue |
| words_per_second = len(reference) / utterance.duration |
|
|
| data = { |
| "file": file.name, |
| "begin": utterance.begin, |
| "end": utterance.end, |
| "duration": utterance.duration, |
| "speaker": utterance.speaker_name, |
| "normalized_text": utterance.normalized_text, |
| "filtered_text": text, |
| "alignment_score": alignment_score, |
| "alignment_likelihood": float(alignment_likelihood), |
| "word_count": len(ctm.word_intervals), |
| "frames_per_second": "NA", |
| "words_per_second": words_per_second, |
| } |
| writer.writerow(data) |
| progress_bar.update(1) |
| file_ctm.export_textgrid( |
| output_path, |
| file_duration=file.duration, |
| output_format="short_textgrid", |
| ) |
| else: |
| if not os.path.exists(model_path): |
| continue |
| for corpus, root in corpus_directories.items(): |
| output_directory = os.path.join(root_dir, "alignments", condition, corpus) |
| csv_path = os.path.join(output_directory, "word_alignment.csv") |
| if os.path.exists(csv_path): |
| continue |
| command = [ |
| "align", |
| os.path.join(root, "benchmark"), |
| dictionary_path, |
| model_path, |
| output_directory, |
| "-j", |
| "10", |
| "--clean", |
| "--debug", |
| "--use_mp", |
| "--use_cutoff_model", |
| "--use_postgres", |
| "--cleanup_textgrids", |
| "--beam", |
| "10", |
| "--retry_beam", |
| "40", |
| ] |
| print(command) |
| mfa_cli(command, standalone_mode=False) |
| corpus = AcousticCorpus(corpus_directory=os.path.join(root, "benchmark")) |
| with ( |
| corpus.session() as session, |
| mfa_open(csv_path, "w") as f, |
| tqdm.tqdm(total=corpus.num_utterances) as progress_bar, |
| ): |
| writer = csv.DictWriter(f, fieldnames=csv_header) |
| writer.writeheader() |
| file_query = ( |
| session.query(File) |
| .options( |
| sqlalchemy.orm.joinedload(File.sound_file, innerjoin=True), |
| ) |
| .order_by(File.name) |
| ) |
| for file in file_query: |
| query = ( |
| session.query(Utterance) |
| .filter(Utterance.file_id == file.id) |
| .options( |
| sqlalchemy.orm.joinedload( |
| Utterance.file, innerjoin=True |
| ).joinedload(File.sound_file, innerjoin=True), |
| ) |
| .order_by(Utterance.begin) |
| ) |
| reference_path = os.path.join(root, "reference") |
| if file.relative_path: |
| reference_path = os.path.join(reference_path, file.relative_path) |
| reference_path = os.path.join(reference_path, file.name + ".TextGrid") |
| reference_word_intervals = parse_aligned_textgrid( |
| reference_path, exclude_unknowns=True |
| ) |
| reference_interval_index = 0 |
| for utterance in query: |
| word_interval_query = ( |
| session.query(WordInterval) |
| .join(WordInterval.word) |
| .filter(WordInterval.utterance_id == utterance.id) |
| .filter(Word.word.op("~")(r"^[^\[<{(]")) |
| .options( |
| sqlalchemy.orm.joinedload(WordInterval.word, innerjoin=True), |
| ) |
| .order_by(WordInterval.begin) |
| ) |
| reference = [] |
| while True: |
| try: |
| interval = reference_word_intervals[reference_interval_index] |
| except IndexError: |
| break |
| if interval.begin >= utterance.end: |
| break |
| if interval.begin >= utterance.begin: |
| reference.append(interval) |
| reference_interval_index += 1 |
| word_intervals = [] |
| reference_index = 0 |
| for i, wi in enumerate(word_interval_query): |
| if reference[reference_index].label == wi.word.word: |
| word_intervals.append(wi.as_ctm()) |
| reference_index += 1 |
| elif ( |
| word_intervals |
| and reference[reference_index].label |
| == word_intervals[-1].label + wi.word.word |
| ): |
| word_intervals[-1].label += wi.word.word |
| word_intervals[-1].end = wi.end |
| reference_index += 1 |
| elif ( |
| word_intervals |
| and word_intervals[-1].label + "-" + wi.word.word |
| in reference[reference_index].label |
| ): |
| word_intervals[-1].label += "-" + wi.word.word |
| word_intervals[-1].end = wi.end |
| if ( |
| word_intervals[-1].label |
| == reference[reference_index].label |
| ): |
| reference_index += 1 |
| else: |
| word_intervals.append(wi.as_ctm()) |
| if not word_intervals: |
| continue |
| word_index = 0 |
| alignment_score = align_words(reference, word_intervals) |
| words_per_second = len(reference) / utterance.duration |
| data = { |
| "file": file.name, |
| "begin": utterance.begin, |
| "end": utterance.end, |
| "duration": utterance.duration, |
| "speaker": utterance.speaker_name, |
| "normalized_text": utterance.normalized_text, |
| "filtered_text": " ".join(x.label for x in word_intervals), |
| "alignment_score": alignment_score, |
| "alignment_likelihood": utterance.alignment_log_likelihood, |
| "word_count": len(word_intervals), |
| "frames_per_second": 100, |
| "words_per_second": words_per_second, |
| } |
| writer.writerow(data) |
| progress_bar.update(1) |
|
|
