diff --git a/kalpy/aligner.py b/kalpy/aligner.py index ece9642..51a554c 100644 --- a/kalpy/aligner.py +++ b/kalpy/aligner.py @@ -86,6 +86,10 @@ def __init__( self.self_loop_scale = self_loop_scale self.boost_silence = boost_silence self.careful = careful + if self.has_multiple_lexicons: + silence_symbols = next(iter(self.lexicon_compiler.values())).silence_symbols + else: + silence_symbols = self.lexicon_compiler.silence_symbols self.aligner = GmmAligner( self.acoustic_model.model_path, beam=beam, @@ -94,6 +98,7 @@ def __init__( acoustic_scale=acoustic_scale, self_loop_scale=self_loop_scale, careful=careful, + silence_phones=silence_symbols, ) self.ali_aligner = None if self.acoustic_model.alignment_model_path != self.acoustic_model.model_path: @@ -105,15 +110,12 @@ def __init__( acoustic_scale=acoustic_scale, self_loop_scale=self_loop_scale, careful=careful, + silence_phones=silence_symbols, ) if self.boost_silence != 1.0: - if self.has_multiple_lexicons: - silence_symbols = next(iter(self.lexicon_compiler.values())).silence_symbols - else: - silence_symbols = self.lexicon_compiler.silence_symbols if self.ali_aligner is not None: - self.ali_aligner.boost_silence(boost_silence, silence_symbols) - self.aligner.boost_silence(boost_silence, silence_symbols) + self.ali_aligner.boost_silence(boost_silence) + self.aligner.boost_silence(boost_silence) def _align_utterance( self, diff --git a/kalpy/data.py b/kalpy/data.py index 2c91dd2..4af61d8 100644 --- a/kalpy/data.py +++ b/kalpy/data.py @@ -1,7 +1,7 @@ """Data classes for kalpy""" from __future__ import annotations -import os.path +import os import pathlib import typing from dataclasses import dataclass @@ -18,8 +18,6 @@ ) from kalpy.utils import generate_read_specifier -PathLike = typing.Union[str, pathlib.Path] - @dataclass class Segment: @@ -27,7 +25,7 @@ class Segment: Data class for information about acoustic segments """ - file_path: PathLike + file_path: os.PathLike begin: typing.Optional[float] = 0.0 end: typing.Optional[float] = None channel: typing.Optional[int] = 0 @@ -102,7 +100,7 @@ def __init__(self, *args, list_mapping: bool = False, **kwargs): self.list_mapping = list_mapping super().__init__(*args, **kwargs) - def export(self, file_path: typing.Union[str, pathlib.Path], skip_safe: bool = False) -> None: + def export(self, file_path: os.PathLike, skip_safe: bool = False) -> None: with open(file_path, "w", encoding="utf8") as f: for k in sorted(self.keys()): v = self[k] @@ -140,7 +138,7 @@ class MatrixArchive: Path to archive or SCP file to read from """ - def __init__(self, file_name: typing.Union[pathlib.Path, str], double: bool = False): + def __init__(self, file_name: os.PathLike, double: bool = False): if not os.path.exists(file_name): raise OSError(f"Specified file does not exist: {file_name}") self.file_name = str(file_name) diff --git a/kalpy/decoder/data.py b/kalpy/decoder/data.py index 34850e7..e696004 100644 --- a/kalpy/decoder/data.py +++ b/kalpy/decoder/data.py @@ -19,7 +19,7 @@ class FstArchive: Path to archive or SCP file to read from """ - def __init__(self, file_name: typing.Union[pathlib.Path, str]): + def __init__(self, file_name: os.PathLike): if not os.path.exists(file_name): raise OSError(f"Specified file does not exist: {file_name}") self.file_name = str(file_name) diff --git a/kalpy/decoder/decode_graph.py b/kalpy/decoder/decode_graph.py index 4faad40..3f815ea 100644 --- a/kalpy/decoder/decode_graph.py +++ b/kalpy/decoder/decode_graph.py @@ -37,9 +37,9 @@ class DecodeGraphCompiler: """ Parameters ---------- - acoustic_model_path: str + acoustic_model_path: :class:`~os.PathLike` Path to model file - tree_path: str + tree_path: :class:`~os.PathLike` Path to tree file lexicon_compiler: :class:`~kalpy.fstext.lexicon.LexiconCompiler` Lexicon compiler to use in generating training graphs @@ -58,10 +58,10 @@ class DecodeGraphCompiler: def __init__( self, - acoustic_model_path: typing.Union[str, pathlib.Path], - tree_path: typing.Union[str, pathlib.Path], + acoustic_model_path: os.PathLike, + tree_path: os.PathLike, lexicon_compiler: LexiconCompiler, - arpa_path: typing.Union[str, pathlib.Path] = None, + arpa_path: typing.Optional[os.PathLike] = None, transition_scale: float = 1.0, self_loop_scale: float = 0.1, ): @@ -83,8 +83,8 @@ def __init__( def export_hclg( self, - arpa_path: typing.Union[pathlib.Path, str], - file_name: typing.Union[pathlib.Path, str], + arpa_path: os.PathLike, + file_name: os.PathLike, ) -> None: """ Export HCLG.fst @@ -103,7 +103,7 @@ def export_hclg( def export_g( self, - file_name: typing.Union[pathlib.Path, str], + file_name: os.PathLike, ) -> None: """ Export g.fst @@ -120,7 +120,7 @@ def export_g( def load_from_file( self, - hclg_fst_path: typing.Union[pathlib.Path, str], + hclg_fst_path: os.PathLike, ) -> None: """ Read HCLG.fst from file @@ -135,7 +135,7 @@ def load_from_file( def load_g_from_file( self, - g_fst_path: typing.Union[pathlib.Path, str], + g_fst_path: os.PathLike, ) -> None: """ Read g.fst from file @@ -148,13 +148,13 @@ def load_g_from_file( g_fst_path = str(g_fst_path) self.g_fst = VectorFst.Read(g_fst_path) - def compile_g_fst(self, arpa_path: typing.Union[str, pathlib.Path]) -> VectorFst: + def compile_g_fst(self, arpa_path: os.PathLike) -> VectorFst: """ Compile G.fst for a language model Parameters ---------- - arpa_path: str + arpa_path: :class:`~os.PathLike` Path to ARPA format language model Returns @@ -175,17 +175,17 @@ def g_carpa(self) -> ConstArpaLm: def compile_g_carpa( self, - arpa_path: typing.Union[str, pathlib.Path], - compiled_path: typing.Union[str, pathlib.Path], + arpa_path: os.PathLike, + compiled_path: os.PathLike, ) -> None: """ Compile G.carpa for a language model Parameters ---------- - arpa_path: str + arpa_path: :class:`~os.PathLike` Path to ARPA format language model - compiled_path: str + compiled_path: :class:`~os.PathLike` Path to compiled G.carpa """ compiled_path = str(compiled_path) @@ -239,13 +239,13 @@ def compile_g_carpa( else: os.remove(temp_carpa_path) - def compile_lg_fst(self, arpa_path: typing.Union[str, pathlib.Path]) -> VectorFst: + def compile_lg_fst(self, arpa_path: os.PathLike) -> VectorFst: """ Compile LG.fst for a language model Parameters ---------- - arpa_path: str + arpa_path: :class:`~os.PathLike` Path to ARPA format language model Returns @@ -263,14 +263,14 @@ def compile_lg_fst(self, arpa_path: typing.Union[str, pathlib.Path]) -> VectorFs return lg_fst def compile_clg_fst( - self, arpa_path: typing.Union[str, pathlib.Path] + self, arpa_path: os.PathLike ) -> typing.Tuple[VectorFst, typing.List[typing.List[int]]]: """ Compile CLG.fst for a language model Parameters ---------- - arpa_path: str + arpa_path: :class:`~os.PathLike` Path to ARPA format language model Returns @@ -292,13 +292,13 @@ def compile_clg_fst( fst_arc_sort(clg_fst, sort_type="ilabel") return clg_fst, ilabels - def compile_hclg_fst(self, arpa_path: typing.Union[str, pathlib.Path]): + def compile_hclg_fst(self, arpa_path: os.PathLike): """ Compile HCLG.fst for a language model Parameters ---------- - arpa_path: str + arpa_path: :class:`~os.PathLike` Path to ARPA format language model Returns diff --git a/kalpy/decoder/training_graphs.py b/kalpy/decoder/training_graphs.py index 884907f..2096ca7 100644 --- a/kalpy/decoder/training_graphs.py +++ b/kalpy/decoder/training_graphs.py @@ -1,5 +1,6 @@ """Classes for generating training graphs""" import logging +import os import pathlib import typing @@ -61,8 +62,8 @@ class TrainingGraphCompiler: def __init__( self, - model_path: typing.Union[pathlib.Path, str], - tree_path: typing.Union[pathlib.Path, str], + model_path: os.PathLike, + tree_path: os.PathLike, lexicon_compiler: LexiconCompiler, transition_scale: float = 0.0, self_loop_scale: float = 0.0, @@ -131,7 +132,7 @@ def compiler(self): def export_graphs( self, - file_name: typing.Union[pathlib.Path, str], + file_name: os.PathLike, transcripts: typing.Iterable[typing.Tuple[str, str]], write_scp: bool = False, callback: typing.Callable = None, diff --git a/kalpy/feat/cmvn.py b/kalpy/feat/cmvn.py index 8ee8551..57521a0 100644 --- a/kalpy/feat/cmvn.py +++ b/kalpy/feat/cmvn.py @@ -2,6 +2,7 @@ from __future__ import annotations import logging +import os import pathlib import typing @@ -100,7 +101,7 @@ def compute_cmvn_for_export( def export_cmvn( self, - file_name: typing.Union[pathlib.Path, str], + file_name: os.PathLike, feature_archive: FeatureArchive, spk2utt: KaldiMapping, write_scp: bool = False, diff --git a/kalpy/feat/data.py b/kalpy/feat/data.py index 899706f..4c61747 100644 --- a/kalpy/feat/data.py +++ b/kalpy/feat/data.py @@ -19,12 +19,12 @@ class FeatureArchive: def __init__( self, - file_name: typing.Union[pathlib.Path, str], - utt2spk: KaldiMapping = None, - cmvn_file_name: typing.Union[pathlib.Path, str] = None, - lda_mat_file_name: typing.Union[pathlib.Path, str] = None, - transform_file_name: typing.Union[pathlib.Path, str] = None, - vad_file_name: typing.Union[pathlib.Path, str] = None, + file_name: os.PathLike, + utt2spk: typing.Optional[KaldiMapping] = None, + cmvn_file_name: typing.Optional[os.PathLike] = None, + lda_mat_file_name: typing.Optional[os.PathLike] = None, + transform_file_name: typing.Optional[os.PathLike] = None, + vad_file_name: typing.Optional[os.PathLike] = None, use_sliding_cmvn: bool = False, cmvn_norm_vars: bool = False, cmvn_reverse: bool = False, @@ -35,7 +35,7 @@ def __init__( sliding_cmvn_window: int = 300, sliding_cmvn_center_window: bool = True, double: bool = False, - callback: typing.Callable = None, + callback: typing.Optional[typing.Callable] = None, ): self.cmvn_reader = None self.transform_reader = None diff --git a/kalpy/feat/fmllr.py b/kalpy/feat/fmllr.py index 7f1a621..514c1e1 100644 --- a/kalpy/feat/fmllr.py +++ b/kalpy/feat/fmllr.py @@ -2,7 +2,7 @@ from __future__ import annotations import logging -import pathlib +import os import threading import typing @@ -23,8 +23,8 @@ class FmllrComputer: def __init__( self, - alignment_acoustic_model_path: typing.Union[pathlib.Path, str], - acoustic_model_path: typing.Union[pathlib.Path, str], + alignment_acoustic_model_path: os.PathLike, + acoustic_model_path: os.PathLike, silence_phones: typing.List[int], spk2utt: KaldiMapping = None, weight_distribute: bool = False, @@ -207,7 +207,7 @@ def compute_fmllr( def export_transforms( self, - file_name: typing.Union[pathlib.Path, str], + file_name: os.PathLike, feature_archive: FeatureArchive, alignment_archive: typing.Union[AlignmentArchive, LatticeArchive], previous_transform_archive: MatrixArchive = None, diff --git a/kalpy/feat/lda.py b/kalpy/feat/lda.py index 7d0cd69..8f72c05 100644 --- a/kalpy/feat/lda.py +++ b/kalpy/feat/lda.py @@ -2,7 +2,7 @@ from __future__ import annotations import logging -import pathlib +import os import typing from _kalpy import transform @@ -20,7 +20,7 @@ class LdaStatsAccumulator: def __init__( self, - acoustic_model_path: typing.Union[pathlib.Path, str], + acoustic_model_path: os.PathLike, silence_phones: typing.List[int], weight_distribute: bool = False, rand_prune: float = 0.0, @@ -80,7 +80,7 @@ def export_transform( class MlltStatsAccumulator: def __init__( self, - acoustic_model_path: typing.Union[pathlib.Path, str], + acoustic_model_path: os.PathLike, silence_phones: typing.List[int], weight_distribute: bool = False, rand_prune: float = 0.0, diff --git a/kalpy/feat/mfcc.py b/kalpy/feat/mfcc.py index 482f345..2893a0b 100644 --- a/kalpy/feat/mfcc.py +++ b/kalpy/feat/mfcc.py @@ -2,10 +2,10 @@ from __future__ import annotations import logging +import os import pathlib import typing -import librosa import numpy as np from _kalpy import feat @@ -249,7 +249,7 @@ def compute_mfccs_for_export( def export_feats( self, - file_name: typing.Union[pathlib.Path, str], + file_name: os.PathLike, segments: typing.Iterable[typing.Tuple[str, Segment]], write_scp: bool = False, compress: bool = True, diff --git a/kalpy/feat/pitch.py b/kalpy/feat/pitch.py index 740624b..befbc29 100644 --- a/kalpy/feat/pitch.py +++ b/kalpy/feat/pitch.py @@ -1,6 +1,7 @@ """Classes for computing pitch features""" from __future__ import annotations +import os import pathlib import typing @@ -298,7 +299,7 @@ def compute_pitch_for_wave( def export_feats( self, - file_name: typing.Union[pathlib.Path, str], + file_name: os.PathLike, segments: typing.Iterable[typing.Tuple[str, Segment]], write_scp: bool = False, compress: bool = True, diff --git a/kalpy/feat/vad.py b/kalpy/feat/vad.py index 496fb1c..077488f 100644 --- a/kalpy/feat/vad.py +++ b/kalpy/feat/vad.py @@ -2,6 +2,7 @@ from __future__ import annotations import logging +import os import pathlib import typing @@ -101,7 +102,7 @@ def compute_vads( def export_vad( self, - file_name: typing.Union[pathlib.Path, str], + file_name: os.PathLike, feature_archive: FeatureArchive, write_scp: bool = False, callback: typing.Callable = None, diff --git a/kalpy/fstext/lexicon.py b/kalpy/fstext/lexicon.py index 95e80d9..135de83 100644 --- a/kalpy/fstext/lexicon.py +++ b/kalpy/fstext/lexicon.py @@ -3,6 +3,7 @@ import collections import math +import os import pathlib import re import typing @@ -35,7 +36,7 @@ class Pronunciation: def parse_dictionary_file( - path: typing.Union[pathlib.Path, str], + path: os.PathLike, ) -> typing.Generator[Pronunciation]: """ Parses a lexicon file and yields parsed pronunciation lines @@ -230,7 +231,7 @@ def specials_set(self) -> typing.Set[str]: "", } - def load_pronunciations(self, file_name: typing.Union[pathlib.Path, str]) -> None: + def load_pronunciations(self, file_name: os.PathLike) -> None: """ Load pronunciations from a dictionary file and calculate necessary disambiguation symbols @@ -793,7 +794,7 @@ def kaldi_fst(self) -> VectorFst: def load_l_from_file( self, - l_fst_path: typing.Union[pathlib.Path, str], + l_fst_path: os.PathLike, ) -> None: """ Read g.fst from file @@ -807,7 +808,7 @@ def load_l_from_file( def load_l_align_from_file( self, - l_fst_path: typing.Union[pathlib.Path, str], + l_fst_path: os.PathLike, ) -> None: """ Read g.fst from file diff --git a/kalpy/gmm/align.py b/kalpy/gmm/align.py index f5911cf..b9bdb1a 100644 --- a/kalpy/gmm/align.py +++ b/kalpy/gmm/align.py @@ -2,6 +2,7 @@ from __future__ import annotations import logging +import os import pathlib import sys import traceback @@ -53,7 +54,7 @@ class GmmAligner: def __init__( self, - acoustic_model_path: typing.Union[pathlib.Path, str], + acoustic_model_path: os.PathLike, acoustic_scale: float = 1.0, transition_scale: float = 1.0, self_loop_scale: float = 1.0, @@ -316,12 +317,12 @@ def align_utterances( def export_alignments( self, - file_name: typing.Union[pathlib.Path, str], + file_name: os.PathLike, training_graph_archive: FstArchive, feature_archive: FeatureArchive, reference_phone_archive: RandomAccessInt32VectorVectorReader = None, - word_file_name: typing.Union[pathlib.Path, str] = None, - likelihood_file_name: typing.Union[pathlib.Path, str] = None, + word_file_name: typing.Optional[os.PathLike] = None, + likelihood_file_name: typing.Optional[os.PathLike] = None, write_scp: bool = False, callback: typing.Callable = None, utterance_parameters: typing.Dict[str, typing.Dict[str, typing.Any]] = None, diff --git a/kalpy/gmm/data.py b/kalpy/gmm/data.py index 8960c7d..b6d4ea2 100644 --- a/kalpy/gmm/data.py +++ b/kalpy/gmm/data.py @@ -76,7 +76,7 @@ class HierarchicalCtm: def export_textgrid( self, - file_name: typing.Union[str, pathlib.Path], + file_name: os.PathLike, file_duration: float = None, output_format: typing.Literal[ "short_textgrid", "long_textgrid", "json", "textgrid_json" @@ -290,9 +290,9 @@ class AlignmentArchive: def __init__( self, - file_name: typing.Union[pathlib.Path, str], - words_file_name: typing.Union[pathlib.Path, str] = None, - likelihood_file_name: typing.Union[pathlib.Path, str] = None, + file_name: os.PathLike, + words_file_name: typing.Optional[os.PathLike] = None, + likelihood_file_name: typing.Optional[os.PathLike] = None, ): if not os.path.exists(file_name): raise OSError(f"Specified file does not exist: {file_name}") @@ -400,7 +400,7 @@ class TranscriptionArchive: def __init__( self, - file_name: typing.Union[pathlib.Path, str], + file_name: os.PathLike, lm_scale: float = 1.0, acoustic_scale: float = 1.0, ): @@ -453,7 +453,7 @@ class LatticeArchive: Path to archive or SCP file to read from """ - def __init__(self, file_name: typing.Union[pathlib.Path, str], determinized: bool = True): + def __init__(self, file_name: os.PathLike, determinized: bool = True): if not os.path.exists(file_name): raise OSError(f"Specified file does not exist: {file_name}") self.file_name = str(file_name) diff --git a/kalpy/gmm/decode.py b/kalpy/gmm/decode.py index dcae21d..d893aec 100644 --- a/kalpy/gmm/decode.py +++ b/kalpy/gmm/decode.py @@ -2,6 +2,7 @@ from __future__ import annotations import logging +import os import pathlib import typing @@ -32,7 +33,7 @@ class GmmDecoder: def __init__( self, - acoustic_model_path: typing.Union[pathlib.Path, str], + acoustic_model_path: os.PathLike, hclg_fst: ConstFst, acoustic_scale: float = 0.1, beam: float = 16.0, @@ -195,11 +196,11 @@ def decode_utterances(self, feature_archive: FeatureArchive) -> typing.Generator def export_lattices( self, - file_name: typing.Union[str, pathlib.Path], + file_name: os.PathLike, feature_archive: FeatureArchive, write_scp: bool = False, - alignment_file_name: typing.Union[str, pathlib.Path] = None, - word_file_name: typing.Union[str, pathlib.Path] = None, + alignment_file_name: typing.Optional[os.PathLike] = None, + word_file_name: typing.Optional[os.PathLike] = None, callback: typing.Callable = None, ): write_specifier = generate_write_specifier(file_name, write_scp) @@ -235,7 +236,7 @@ def export_lattices( class GmmRescorer: def __init__( self, - acoustic_model_path: typing.Union[pathlib.Path, str], + acoustic_model_path: os.PathLike, acoustic_scale: float = 0.1, lattice_beam: float = 6.0, ): @@ -275,7 +276,7 @@ def rescore_utterances( def export_lattices( self, - file_name: typing.Union[str, pathlib.Path], + file_name: os.PathLike, lattice_archive: LatticeArchive, feature_archive: FeatureArchive, write_scp: bool = False, diff --git a/kalpy/gmm/train.py b/kalpy/gmm/train.py index 61892b5..7c12c13 100644 --- a/kalpy/gmm/train.py +++ b/kalpy/gmm/train.py @@ -2,7 +2,7 @@ from __future__ import annotations import logging -import pathlib +import os import typing from _kalpy import gmm, hmm, tree @@ -19,7 +19,7 @@ class GmmStatsAccumulator: - def __init__(self, acoustic_model_path: typing.Union[pathlib.Path, str]): + def __init__(self, acoustic_model_path: os.PathLike): self.acoustic_model_path = str(acoustic_model_path) self.transition_accs = DoubleVector() self.transition_model, self.acoustic_model = read_gmm_model(self.acoustic_model_path) @@ -76,7 +76,7 @@ def accumulate_stats( def export_stats( self, - file_name: typing.Union[pathlib.Path, str], + file_name: os.PathLike, feature_archive: FeatureArchive, alignment_archive: AlignmentArchive, ): @@ -92,7 +92,7 @@ def export_stats( class TreeStatsAccumulator: def __init__( self, - acoustic_model_path: typing.Union[pathlib.Path, str], + acoustic_model_path: os.PathLike, var_floor: float = 0.01, context_width: int = 3, central_position: int = 1, @@ -146,7 +146,7 @@ def accumulate_stats( def export_stats( self, - file_name: typing.Union[pathlib.Path, str], + file_name: os.PathLike, feature_archive: FeatureArchive, alignment_archive: AlignmentArchive, ): @@ -159,7 +159,7 @@ def export_stats( class TwoFeatsStatsAccumulator: - def __init__(self, acoustic_model_path: typing.Union[pathlib.Path, str]): + def __init__(self, acoustic_model_path: os.PathLike): self.acoustic_model_path = str(acoustic_model_path) self.transition_accs = DoubleVector() self.transition_model, self.acoustic_model = read_gmm_model(self.acoustic_model_path) diff --git a/kalpy/gmm/utils.py b/kalpy/gmm/utils.py index 681983c..264b070 100644 --- a/kalpy/gmm/utils.py +++ b/kalpy/gmm/utils.py @@ -1,7 +1,6 @@ from __future__ import annotations -import os.path -import pathlib +import os import typing from _kalpy.gmm import AmDiagGmm @@ -11,7 +10,7 @@ from kalpy.exceptions import ReadError -def read_transition_model(model_path: typing.Union[str, pathlib.Path]) -> TransitionModel: +def read_transition_model(model_path: os.PathLike) -> TransitionModel: if not os.path.exists(model_path): raise ReadError(f"The specified model path {model_path} does not exist.") ki = Input() @@ -22,9 +21,7 @@ def read_transition_model(model_path: typing.Union[str, pathlib.Path]) -> Transi return transition_model -def read_gmm_model( - model_path: typing.Union[str, pathlib.Path] -) -> typing.Tuple[TransitionModel, AmDiagGmm]: +def read_gmm_model(model_path: os.PathLike) -> typing.Tuple[TransitionModel, AmDiagGmm]: if not os.path.exists(model_path): raise ReadError(f"The specified model path {model_path} does not exist.") ki = Input() @@ -37,7 +34,7 @@ def read_gmm_model( return transition_model, acoustic_model -def read_topology(topo_path: typing.Union[str, pathlib.Path]) -> HmmTopology: +def read_topology(topo_path: os.PathLike) -> HmmTopology: if not os.path.exists(topo_path): raise ReadError(f"The specified topo path {topo_path} does not exist.") ki = Input() @@ -49,7 +46,7 @@ def read_topology(topo_path: typing.Union[str, pathlib.Path]) -> HmmTopology: def write_gmm_model( - model_path: typing.Union[str, pathlib.Path], + model_path: os.PathLike, transition_model: TransitionModel, acoustic_model: AmDiagGmm, binary: bool = True, @@ -60,15 +57,13 @@ def write_gmm_model( ko.Close() -def write_tree( - tree_path: typing.Union[str, pathlib.Path], tree: ContextDependency, binary: bool = True -) -> None: +def write_tree(tree_path: os.PathLike, tree: ContextDependency, binary: bool = True) -> None: ko = Output(str(tree_path), binary) tree.Write(ko.Stream(), binary) ko.Close() -def read_tree(tree_path: typing.Union[str, pathlib.Path]) -> ContextDependency: +def read_tree(tree_path: os.PathLike) -> ContextDependency: if not os.path.exists(tree_path): raise ReadError(f"The specified tree path {tree_path} does not exist.") tree = ContextDependency() diff --git a/kalpy/ivector/data.py b/kalpy/ivector/data.py index 6cd586d..c2d3c47 100644 --- a/kalpy/ivector/data.py +++ b/kalpy/ivector/data.py @@ -11,7 +11,6 @@ SequentialBaseDoubleVectorReader, SequentialInt32VectorVectorReader, ) -from kalpy.data import PathLike from kalpy.utils import generate_read_specifier, read_kaldi_object @@ -21,11 +20,13 @@ class IvectorArchive: Parameters ---------- - file_name: :class:`~pathlib.Path` or str + file_name: :class:`~os.PathLike` Path to archive or SCP file to read from """ - def __init__(self, file_name: PathLike, num_utterances_file_name: PathLike = None): + def __init__( + self, file_name: os.PathLike, num_utterances_file_name: typing.Optional[os.PathLike] = None + ): if not os.path.exists(file_name): raise OSError(f"Specified file does not exist: {file_name}") self.file_name = str(file_name) @@ -92,7 +93,7 @@ class GselectArchive: Path to archive or SCP file to read from """ - def __init__(self, file_name: PathLike): + def __init__(self, file_name: os.PathLike): if not os.path.exists(file_name): raise OSError(f"Specified file does not exist: {file_name}") self.file_name = str(file_name) diff --git a/kalpy/ivector/extractor.py b/kalpy/ivector/extractor.py index d676f6a..7d9ca3f 100644 --- a/kalpy/ivector/extractor.py +++ b/kalpy/ivector/extractor.py @@ -2,6 +2,7 @@ from __future__ import annotations import logging +import os import pathlib import typing @@ -21,8 +22,8 @@ class IvectorExtractor: def __init__( self, - dubm_path: typing.Union[str, pathlib.Path], - ivector_extractor_path: typing.Union[str, pathlib.Path], + dubm_path: os.PathLike, + ivector_extractor_path: os.PathLike, acoustic_weight: float = 1.0, max_count: float = 0.0, num_gselect: int = 50, @@ -67,7 +68,7 @@ def extract_ivectors(self, feature_archive: FeatureArchive): def export_ivectors( self, - file_name: typing.Union[pathlib.Path, str], + file_name: os.PathLike, feature_archive: FeatureArchive, write_scp: bool = False, callback: typing.Callable = None, diff --git a/kalpy/ivector/plda.py b/kalpy/ivector/plda.py index e2c4530..9bb2959 100644 --- a/kalpy/ivector/plda.py +++ b/kalpy/ivector/plda.py @@ -1,18 +1,18 @@ -import pathlib +import os import typing import numpy as np from _kalpy.ivector import Plda, ivector_normalize_length, ivector_subtract_mean from _kalpy.matrix import DoubleVector, FloatVector -from kalpy.utils import read_kaldi_object from kalpy.ivector.data import IvectorArchive +from kalpy.utils import read_kaldi_object class PldaScorer: def __init__( self, - plda_path: typing.Union[str, pathlib.Path], + plda_path: os.PathLike, normalize_length: bool = True, simple_length_norm: bool = True, ): @@ -37,8 +37,10 @@ def load_speaker_ivectors(self, speaker_archive_path, num_utts_path=None): if self.normalize_length: ivector_normalize_length(ivector) speaker_ivectors.append(DoubleVector(ivector)) - ivector_subtract_mean(speaker_ivectors,normalize=self.normalize_length) - self.speaker_ivectors = self.plda.transform_ivectors(speaker_ivectors, self.num_speaker_examples) + ivector_subtract_mean(speaker_ivectors, normalize=self.normalize_length) + self.speaker_ivectors = self.plda.transform_ivectors( + speaker_ivectors, self.num_speaker_examples + ) def transform_ivector(self, ivector: np.ndarray, num_examples: int = 1): return self.plda.transform_ivector(ivector, num_examples) @@ -68,6 +70,8 @@ def classify_speaker( if isinstance(utterance_ivector, np.ndarray): utterance_ivector = DoubleVector() utterance_ivector.from_numpy(utterance_ivector) - ind, score = self.plda.classify_utterance(utterance_ivector, self.speaker_ivectors, self.num_speaker_examples) + ind, score = self.plda.classify_utterance( + utterance_ivector, self.speaker_ivectors, self.num_speaker_examples + ) speaker = self.speaker_ids[ind] return speaker, score diff --git a/kalpy/ivector/train.py b/kalpy/ivector/train.py index 1a2db60..c0dc939 100644 --- a/kalpy/ivector/train.py +++ b/kalpy/ivector/train.py @@ -2,7 +2,7 @@ from __future__ import annotations import logging -import pathlib +import os import typing from _kalpy import gmm, ivector @@ -18,7 +18,7 @@ class GlobalGmmStatsAccumulator: - def __init__(self, model_path: typing.Union[pathlib.Path, str]): + def __init__(self, model_path: os.PathLike): self.model_path = str(model_path) self.model: gmm.DiagGmm = read_kaldi_object(gmm.DiagGmm, self.model_path) self.gmm_accs = gmm.AccumDiagGmm() @@ -59,7 +59,7 @@ def accumulate_stats( class IvectorExtractorStatsAccumulator: - def __init__(self, ivector_extractor_path: typing.Union[pathlib.Path, str]): + def __init__(self, ivector_extractor_path: os.PathLike): self.ivector_extractor_path = str(ivector_extractor_path) self.model: ivector.IvectorExtractor = read_kaldi_object( diff --git a/kalpy/lm/rescore.py b/kalpy/lm/rescore.py index 20511d5..bf72030 100644 --- a/kalpy/lm/rescore.py +++ b/kalpy/lm/rescore.py @@ -2,7 +2,7 @@ from __future__ import annotations import logging -import pathlib +import os import typing from _kalpy.fstext import VectorFst @@ -85,7 +85,7 @@ def rescore_utterances( def export_lattices( self, - file_name: typing.Union[str, pathlib.Path], + file_name: os.PathLike, lattice_archive: LatticeArchive, add_lm: typing.Union[VectorFst, ConstArpaLm], write_scp: bool = False, diff --git a/kalpy/utils.py b/kalpy/utils.py index cecc9de..91bdca0 100644 --- a/kalpy/utils.py +++ b/kalpy/utils.py @@ -1,14 +1,14 @@ from __future__ import annotations import logging -import pathlib +import os import typing from contextlib import contextmanager from _kalpy.util import Input, Output -def generate_read_specifier(file_name: typing.Union[str, pathlib.Path], sorted=True) -> str: +def generate_read_specifier(file_name: os.PathLike, sorted=True) -> str: file_name = str(file_name) if sorted: read_identifier = "ark,s,cs" @@ -23,7 +23,7 @@ def generate_read_specifier(file_name: typing.Union[str, pathlib.Path], sorted=T def generate_write_specifier( - file_name: typing.Union[str, pathlib.Path], write_scp: bool = False, text: bool = False + file_name: os.PathLike, write_scp: bool = False, text: bool = False ) -> str: file_name = str(file_name) if not file_name.endswith(".ark"): @@ -52,7 +52,7 @@ def read_kaldi_object(obj_type, path, binary=True): @contextmanager -def kalpy_logger(log_name: str, log_path: typing.Union[pathlib.Path, str]) -> logging.Logger: +def kalpy_logger(log_name: str, log_path: os.PathLike) -> typing.Generator[logging.Logger]: kalpy_logging = logging.getLogger(log_name) file_handler = logging.FileHandler(log_path, encoding="utf8") file_handler.setLevel(logging.DEBUG)