Alignment API
Public API for forced-alignment operations.
- class pstrain.api.alignment.AlignmentCoverage(outcomes=<factory>, utterances=<factory>, seconds=<factory>, n_duration_unknown=0, speakers=<factory>, phones=<factory>, triphones=<factory>, phone_carriers=<factory>, unexpanded_words=<factory>, speakers_without_aligned_audio=(), retry_only_phones=(), retry_only_triphones=(), lost_phones=(), lost_triphones=(), thin_phones=(), run_failure_rate=0.0, thin_tokens=50, thin_min_carriers=3, thin_rate_ratio=2.0, phones_from_alignment=True)[source]
Bases:
objectMass and coverage of one corpus alignment, by outcome.
Reporting only: nothing here feeds back into which alignments are accepted. Alignment mass and coverage report defines each count and flag.
- Parameters:
outcomes (dict[str, Literal['first_pass', 'retry_accepted', 'retry_rejected', 'not_recovered']])
utterances (OutcomeSplit)
seconds (OutcomeSplit)
n_duration_unknown (int)
thin_phones (tuple[ThinPhone, ...])
run_failure_rate (float)
thin_tokens (int)
thin_min_carriers (int)
thin_rate_ratio (float)
phones_from_alignment (bool)
- __init__(outcomes=<factory>, utterances=<factory>, seconds=<factory>, n_duration_unknown=0, speakers=<factory>, phones=<factory>, triphones=<factory>, phone_carriers=<factory>, unexpanded_words=<factory>, speakers_without_aligned_audio=(), retry_only_phones=(), retry_only_triphones=(), lost_phones=(), lost_triphones=(), thin_phones=(), run_failure_rate=0.0, thin_tokens=50, thin_min_carriers=3, thin_rate_ratio=2.0, phones_from_alignment=True)
- Parameters:
outcomes (dict[str, Literal['first_pass', 'retry_accepted', 'retry_rejected', 'not_recovered']])
utterances (OutcomeSplit)
seconds (OutcomeSplit)
n_duration_unknown (int)
thin_phones (tuple[ThinPhone, ...])
run_failure_rate (float)
thin_tokens (int)
thin_min_carriers (int)
thin_rate_ratio (float)
phones_from_alignment (bool)
- Return type:
None
- format()[source]
A bounded, human-readable report.
Speakers are tabulated in full up to a limit and summarized beyond it; flagged speakers are always named. Flag lists name units up to a limit and count the rest;
to_dict()names them all.- Return type:
- lost_triphones: tuple[str, ...] = ()
Triphones that appear only in rejected or unrecovered utterances.
- outcomes: dict[str, Literal['first_pass', 'retry_accepted', 'retry_rejected', 'not_recovered']]
Each utterance’s outcome, in corpus order.
- phones_from_alignment: bool = True
Whether aligned utterances’ phones came from phone segments (
True) or from their words expanded through the dictionary (False).
- retry_only_phones: tuple[str, ...] = ()
Phones whose aligned tokens all come from accepted retries.
- retry_only_triphones: tuple[str, ...] = ()
Triphones whose aligned tokens all come from accepted retries.
- seconds: OutcomeSplit
Audio seconds by outcome.
- speakers: dict[str, tuple[OutcomeSplit, OutcomeSplit]]
Per speaker,
(utterances, seconds)by outcome.
- speakers_without_aligned_audio: tuple[str, ...] = ()
Speakers whose every utterance was rejected or not recovered.
- to_dict()[source]
A JSON-serializable form: every aggregate and every flagged unit.
Per-utterance outcomes are left out; they are in
outcomes.
- unexpanded_words: dict[str, int]
Transcript words with no pronunciation, over the utterances that did not align.
- utterances: OutcomeSplit
Utterance counts by outcome.
- pstrain.api.alignment.align_corpus(transcripts, audio_dir, model_dir, dict_path, filler_dict=None, audio_ext='.wav', include_phones=True, beam=1e-64, retry_beam_factor=1e+136, failed_alignment='recover', verbatim_tokens=False, phone_report=None, retry_acceptance_target=0.05, retry_acceptance_threshold=None, coverage_report=True)[source]
Align an entire corpus.
Loads the acoustic model once, keeps it resident for all utterances.
- Parameters:
transcripts (dict[str, str]) – Dict mapping utterance_id to transcript text.
audio_dir (Path) – Directory containing audio files.
model_dir (Path) – Path to acoustic model directory.
dict_path (Path) – Path to pronunciation dictionary.
filler_dict (Path | None) – Path to filler dictionary (optional).
audio_ext (str) – Audio file extension (default
".wav").include_phones (bool) – Capture phone-level segmentation.
beam (float) – Viterbi pruning beam.
retry_beam_factor (float | Sequence[float]) – Factor for one wider-beam final-state retry, or an ascending sequence of factors tried in order until one succeeds.
failed_alignment (Literal['recover', 'abort', 'omit']) – Whether final-state failures are retried before being recorded.
verbatim_tokens (bool) – Honor explicit pronunciation variants exactly.
phone_report (UnsupportedPhoneReport | None) – An already-collected report of pronunciations the model cannot support, so a caller that reported it before the run does not pay for the check or report it twice. When omitted, the check runs here and any finding is logged.
retry_acceptance_target (float | None) – The retry acceptance check’s quantile (default 0.05);
Noneaccepts retries unchecked. A retry-recovered alignment is accepted only if its speech score reaches the threshold for its rung’s beam; otherwise it is recorded as failed, with the reason. First-pass alignments are never checked. Unless a threshold is supplied, the run calibrates its own after the corpus pass, and only if a retry recovered anything: it realigns at mostRETRY_ACCEPTANCE_MAX_SAMPLES(200) of its first-pass alignments, evenly spaced, at each such rung’s beam and takes this quantile of their scores. With fewer thanRETRY_ACCEPTANCE_MIN_SAMPLES(20) scored, every recovery at that rung is rejected.retry_acceptance_threshold (float | Sequence[float] | None) – A threshold per rung, in nats per speech frame, to use instead of calibrating.
coverage_report (bool) – Attach the mass-and-coverage report (
AlignmentJob.coverage). It is built after the pass from the outcomes already decided and never changes one.
- Returns:
AlignmentJobwith all alignment results.- Return type:
AlignmentJob
Example
>>> transcripts = {"utt001": "hello world", "utt002": "goodbye"} >>> job = align_corpus(transcripts, audio_dir, model_dir, dict_path) >>> print(f"Aligned {job.n_aligned}/{job.n_utterances}")
- pstrain.api.alignment.alignment_coverage(job, transcripts, dict_path, filler_dict=None, audio_dir=None, audio_ext='.wav', *, speaker_of=<function speaker_from_id>, thin_tokens=50, thin_min_carriers=3, thin_rate_ratio=2.0)[source]
Report a finished corpus alignment’s mass and coverage by outcome.
Reads
joband never changes it.- Parameters:
job (AlignmentJob) – The finished alignment.
transcripts (Mapping[str, str]) – The transcripts it aligned, by utterance ID.
dict_path (Path) – The pronunciation dictionary it used.
filler_dict (Path | None) – The filler dictionary it used, if any.
audio_dir (Path | None) – Where the audio is, to read durations of utterances that did not align. Without it they have no duration.
audio_ext (str) – Audio file extension.
speaker_of (Callable[[str], str]) – Maps an utterance ID to its speaker. Defaults to the text before the first
/.thin_tokens (int) – Thin-phone rule: fewer first-pass tokens than this.
thin_min_carriers (int) – Thin-phone rule: at least this many carrying utterances.
thin_rate_ratio (float) – Thin-phone rule: carrier failure rate at least this multiple of the run’s.
- Returns:
The
AlignmentCoverage.- Return type:
- pstrain.api.alignment.collect_phone_report(model_dir, dict_path, filler_dict=None)[source]
Collect the pronunciations this model’s phone inventory cannot support.
- Parameters:
- Returns:
The collected report, or
Nonewhen the model definition or a dictionary could not be read. An unreadable input is the aligner’s problem to report, not this check’s: it must never be the reason a corpus pass does not start. Skipping is said out loud, because a silent skip turns the whole check off with no trace.- Return type:
UnsupportedPhoneReport | None
- pstrain.api.alignment.load_transcripts(transcript_file)[source]
Load transcripts from a Sphinx-format transcription file.
Format: <s> word word word </s> (utterance_id)