Library API 

map_pronunciation(phones, mapping, passthrough_unmapped=True)[source]

Map a pronunciation (list of phones).

Parameters:

phones (list[str]) – List of phones to map
mapping (dict[str, str]) – Phone mapping dictionary
passthrough_unmapped (bool) – If True, unmapped phones pass through

Returns:

List of mapped phones (flattened)

Return type:

phones()[source]

Get all phones in phoneset.

Return type:: set[str]

to_file(path, silence_phone='SIL')[source]

Save phoneset to file.

Parameters:

path (Path) – Output file path
silence_phone (str) – Silence phone to list first if present

Return type:

None

validate_dictionary(dictionary)[source]

Validate that dictionary phones are in phoneset.

Parameters:: dictionary (Dictionary) – Dictionary to validate
Returns:: Tuple of (is_valid, missing_phones)
Return type:: tuple[bool, set[str]]

class pstrain.api.Profile(*, description='', features=<factory>, training=<factory>, split=<factory>, runner=<factory>, sharding=<factory>, alignment=<factory>)[source]

Bases: StrictModel

One complete named model-training profile.

Parameters:

description (Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Human-readable profile purpose')])
features (FeatureConfig)
training (TrainingConfig)
split (SplitConfig)
runner (RunnerConfig)
sharding (ShardingConfig)
alignment (AlignmentConfig)

alignment: AlignmentConfig

description: Annotated[str, Field(description='Human-readable profile purpose')]

features: FeatureConfig

model_config = {'extra': 'forbid', 'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

runner: RunnerConfig

sharding: ShardingConfig

split: SplitConfig

training: TrainingConfig

class pstrain.api.PstrainPaths(bin_dir, lib_path, include_dir, project_root, data_dir)[source]

Bases: object

Collection of pstrain installation paths.

Parameters:

bin_dir (Path | None)
lib_path (Path | None)
include_dir (Path | None)
project_root (Path | None)
data_dir (Path)

__init__(bin_dir, lib_path, include_dir, project_root, data_dir)

Parameters:

bin_dir (Path | None)
lib_path (Path | None)
include_dir (Path | None)
project_root (Path | None)
data_dir (Path)

Return type:

None

bin_dir: Path | None: Directory containing pstrain C binaries (bw, norm, etc.)

data_dir: Path: Directory containing package data files

include_dir: Path | None: Directory containing pstrain C headers

lib_path: Path | None: Path to libpstrainc shared library

project_root: Path | None: Project root (development only)

to_dict()[source]

Convert to dictionary for JSON output.

Return type:: dict[str, str | None]

class pstrain.api.TrainingConfig(*, n_state=3, skip_state=False, n_senones=200, a_beam=1e-90, b_beam=1e-10, ci=<factory>, tied=<factory>, untied=<factory>, split_variance_floor_fraction=0.0, max_skip_fraction=0.05, retry_beam_factor=10000000000.0, failed_alignment='recover', bw_checkpoint_iterations=False, arctic_a0302_zero_codebook_band=None, accept_arctic_a0587_known_skip=False, tree_state_weights=(1.0, 0.05, 0.0), tree_rotate_state_weights=True, tree_directional_questions=True, tree_ssplitmax=7, tree_ssplitthr=0.0, tree_csplitmax=2000, tree_csplitthr=0.0, tree_mwfloor=1e-08, tree_intermediate_dumps=False, question_npermute=12, question_quests_per_state=20, question_niter=1, multipron_training=True, optional_final_silence=True, untied_inventory='transcript-reachable', exclusion_schedule=<factory>)[source]

Bases: StrictModel

Acoustic-model training parameters.

Parameters:

n_state (Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Emitting states per HMM', metadata=[Ge(ge=1)])])
skip_state (Annotated[bool, FieldInfo(annotation=NoneType, required=True, description="Enable SphinxTrain's $CFG_SKIPSTATE topology, adding an arc from each eligible emitting state to the state two positions ahead so a phone can be realized with fewer frames than states. SphinxTrain writes raw 3/1/1 weights and normalizes them on read; pstrain writes the behaviorally equivalent normalized values")])
n_senones (Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Target tied-state count', metadata=[Ge(ge=1)])])
a_beam (Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Forward alignment beam', metadata=[Gt(gt=0)])])
b_beam (Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Backward alignment beam', metadata=[Gt(gt=0)])])
ci (TrainingScheduleConfig)
tied (TrainingScheduleConfig)
untied (TrainingScheduleConfig)
split_variance_floor_fraction (Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Experimental variance lower bound for split training stages, as a fraction of each matching coordinate in the fixed CI-1g or CD-1g variance reference. Zero disables regularization. The reference never advances with later splits; zero reference coordinates contribute no positive floor. Select a nonzero fraction explicitly after evaluation; no universal nonzero value is assumed', metadata=[Strict(strict=True), Ge(ge=0), Le(le=1), _PydanticGeneralMetadata(allow_inf_nan=False)])])
max_skip_fraction (Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Maximum skipped-update fraction', metadata=[Ge(ge=0), Le(le=1)])])
retry_beam_factor (Annotated[float | list[float], PlainValidator(func=~pstrain.lib.config.models._validate_retry_beam_factor, json_schema_input_type=~typing.Any), WithJsonSchema(json_schema={'anyOf': [{'type': 'number', 'exclusiveMinimum': 0}, {'type': 'array', 'items': {'type': 'number', 'exclusiveMinimum': 1}, 'minItems': 1}]}, mode=None), FieldInfo(annotation=NoneType, required=True, description='Factor that widens the forward beam for one retry after an utterance fails to reach its final state, or an ascending list of factors, each greater than 1 and relative to the nominal beam, tried in order until one succeeds; a retry is counted only when that attempt runs')])
failed_alignment (Annotated[Literal['recover', 'abort', 'omit'], FieldInfo(annotation=NoneType, required=True, description='Action when an utterance fails to reach its final state: ``recover`` runs the wider-beam retries in ``retry_beam_factor`` and, if they all fail, reports the utterance and continues without it; ``abort`` fails the run on the first failure; and ``omit`` reports and excludes it without retrying. Skips are counted either way, and ``max_skip_fraction`` still fails the run when they stop being incidental')])
bw_checkpoint_iterations (Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Retain the compact model files from every completed Baum-Welch pass under ``iterations/NN``; costs roughly one additional model copy per pass and does not retain the much larger ``.bw-accum`` shard accumulators or change which checkpoint is loaded by training. The deprecated ``PSTRAIN_BW_CHECKPOINTS=1`` environment variable can also enable retention, but cannot disable a true profile setting')])
arctic_a0302_zero_codebook_band (Annotated[tuple[int, int] | None, FieldInfo(annotation=NoneType, required=True, description='Accepted inclusive exact-zero codebook occupancy band for the singular Arctic a0302 terminal-alignment exception')])
accept_arctic_a0587_known_skip (Annotated[bool, FieldInfo(annotation=NoneType, required=True, description="Deprecated for live profiles: retained solely for the Arctic pin's retired off-profile provenance; live benchmark cells run exception-free")])
tree_state_weights (Annotated[tuple[float, ...], FieldInfo(annotation=NoneType, required=True, description='Decision-tree state weights', metadata=[MinLen(min_length=1)])])
tree_rotate_state_weights (Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Apply target-relative tree state weights; disable only for isolation measurements')])
tree_directional_questions (Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Honor _L/_R tree-question suffixes; disable only for isolation measurements')])
tree_ssplitmax (Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Maximum state splits', metadata=[Ge(ge=0)])])
tree_ssplitthr (Annotated[float, FieldInfo(annotation=NoneType, required=True, description='State split threshold', metadata=[Ge(ge=0)])])
tree_csplitmax (Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Maximum phone-context splits', metadata=[Ge(ge=0)])])
tree_csplitthr (Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Phone-context split threshold', metadata=[Ge(ge=0)])])
tree_mwfloor (Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Tree mixture-weight floor', metadata=[Gt(gt=0)])])
tree_intermediate_dumps (Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Dump intermediate decision trees to worker diagnostics')])
question_npermute (Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Question permutations', metadata=[Ge(ge=1)])])
question_quests_per_state (Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Questions generated per state', metadata=[Ge(ge=1)])])
question_niter (Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Question generation iterations', metadata=[Ge(ge=1)])])
multipron_training (Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Sum posteriors over pronunciation variants; when disabled without an explicit inventory policy, untied inventory resolves to upstream-compatible ``linear``')])
optional_final_silence (Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Permit final transcript silence to consume zero frames; stock SphinxTrain requires that silence to consume at least one frame')])
untied_inventory (Annotated[Literal['all-triphone', 'transcript-reachable', 'linear'], FieldInfo(annotation=NoneType, required=True, description="Untied-model phone inventory policy: ``transcript-reachable`` includes contexts reachable through every pronunciation when multipron training is enabled; upstream-compatible ``linear`` includes contexts observed through each transcript word's first pronunciation; ``all-triphone`` includes the complete phoneset cross-product")])
exclusion_schedule (Annotated[dict[str, dict[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Strict(strict=True)])] | str, list[str]]], FieldInfo(annotation=NoneType, required=True, description='Experimental stage/pass utterance exclusions')])

a_beam: Annotated[float, Field(gt=0, description='Forward alignment beam')]

accept_arctic_a0587_known_skip: Annotated[bool, Field(description="Deprecated for live profiles: retained solely for the Arctic pin's retired off-profile provenance; live benchmark cells run exception-free")]

arctic_a0302_zero_codebook_band: Annotated[tuple[int, int] | None, Field(description='Accepted inclusive exact-zero codebook occupancy band for the singular Arctic a0302 terminal-alignment exception')]

b_beam: Annotated[float, Field(gt=0, description='Backward alignment beam')]

bw_checkpoint_iterations: Annotated[bool, Field(description='Retain the compact model files from every completed Baum-Welch pass under ``iterations/NN``; costs roughly one additional model copy per pass and does not retain the much larger ``.bw-accum`` shard accumulators or change which checkpoint is loaded by training. The deprecated ``PSTRAIN_BW_CHECKPOINTS=1`` environment variable can also enable retention, but cannot disable a true profile setting')]

ci: TrainingScheduleConfig

exclusion_schedule: Annotated[dict[str, dict[Annotated[int, Field(strict=True)] | str, list[str]]], Field(description='Experimental stage/pass utterance exclusions')]

failed_alignment: Annotated[Literal['recover', 'abort', 'omit'], Field(description='Action when an utterance fails to reach its final state: ``recover`` runs the wider-beam retries in ``retry_beam_factor`` and, if they all fail, reports the utterance and continues without it; ``abort`` fails the run on the first failure; and ``omit`` reports and excludes it without retrying. Skips are counted either way, and ``max_skip_fraction`` still fails the run when they stop being incidental')]

max_skip_fraction: Annotated[float, Field(ge=0, le=1, description='Maximum skipped-update fraction')]

model_config = {'extra': 'forbid', 'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

multipron_training: Annotated[bool, Field(description='Sum posteriors over pronunciation variants; when disabled without an explicit inventory policy, untied inventory resolves to upstream-compatible ``linear``')]

n_senones: Annotated[int, Field(ge=1, description='Target tied-state count')]

n_state: Annotated[int, Field(ge=1, description='Emitting states per HMM')]

optional_final_silence: Annotated[bool, Field(description='Permit final transcript silence to consume zero frames; stock SphinxTrain requires that silence to consume at least one frame')]

question_niter: Annotated[int, Field(ge=1, description='Question generation iterations')]

question_npermute: Annotated[int, Field(ge=1, description='Question permutations')]

question_quests_per_state: Annotated[int, Field(ge=1, description='Questions generated per state')]

retry_beam_factor: Annotated[RetryBeamFactorSetting, Field(description='Factor that widens the forward beam for one retry after an utterance fails to reach its final state, or an ascending list of factors, each greater than 1 and relative to the nominal beam, tried in order until one succeeds; a retry is counted only when that attempt runs')]

classmethod select_inventory_default(data)[source]

Parameters:: data (Any)
Return type:: Any

skip_state: Annotated[bool, Field(description="Enable SphinxTrain's $CFG_SKIPSTATE topology, adding an arc from each eligible emitting state to the state two positions ahead so a phone can be realized with fewer frames than states. SphinxTrain writes raw 3/1/1 weights and normalizes them on read; pstrain writes the behaviorally equivalent normalized values")]

split_variance_floor_fraction: Annotated[float, Field(ge=0, le=1, allow_inf_nan=False, strict=True, description='Experimental variance lower bound for split training stages, as a fraction of each matching coordinate in the fixed CI-1g or CD-1g variance reference. Zero disables regularization. The reference never advances with later splits; zero reference coordinates contribute no positive floor. Select a nonzero fraction explicitly after evaluation; no universal nonzero value is assumed')]

tied: TrainingScheduleConfig

tree_csplitmax: Annotated[int, Field(ge=0, description='Maximum phone-context splits')]

tree_csplitthr: Annotated[float, Field(ge=0, description='Phone-context split threshold')]

tree_directional_questions: Annotated[bool, Field(description='Honor _L/_R tree-question suffixes; disable only for isolation measurements')]

tree_intermediate_dumps: Annotated[bool, Field(description='Dump intermediate decision trees to worker diagnostics')]

tree_mwfloor: Annotated[float, Field(gt=0, description='Tree mixture-weight floor')]

tree_rotate_state_weights: Annotated[bool, Field(description='Apply target-relative tree state weights; disable only for isolation measurements')]

tree_ssplitmax: Annotated[int, Field(ge=0, description='Maximum state splits')]

tree_ssplitthr: Annotated[float, Field(ge=0, description='State split threshold')]

tree_state_weights: Annotated[tuple[float, ...], Field(min_length=1, description='Decision-tree state weights')]

untied: TrainingScheduleConfig

untied_inventory: Annotated[Literal['all-triphone', 'transcript-reachable', 'linear'], Field(description="Untied-model phone inventory policy: ``transcript-reachable`` includes contexts reachable through every pronunciation when multipron training is enabled; upstream-compatible ``linear`` includes contexts observed through each transcript word's first pronunciation; ``all-triphone`` includes the complete phoneset cross-product")]

classmethod validate_exclusions(value)[source]

Parameters:: value (dict[str, dict[int | str, list[str]]])
Return type:: dict[str, dict[int | str, list[str]]]

validate_training()[source]

Return type:: TrainingConfig

validate_tree_state_weights()[source]

Return type:: TrainingConfig

exception pstrain.api.TutorialExistsError(path)[source]

Bases: FileExistsError

Raised when a tutorial destination may not be replaced.

Parameters:: path (Path)
Return type:: None

__init__(path)[source]

Parameters:: path (Path)
Return type:: None

class pstrain.api.TutorialResult[source]

Bases: TypedDict

JSON-serializable result of a tutorial copy request.

path: str

status: str

class pstrain.api.ValidationReport(errors=<factory>, warnings=<factory>, train_utterances=0, test_utterances=0, dev_utterances=0, total_utterances=0, vocabulary_size=0, missing_words=<factory>, dictionary_entries=0, dictionary_base_words=0, phoneset_size=0, has_silence=False, missing_phones=<factory>, audio_files=0, missing_audio=<factory>)[source]

Bases: object

Validation report with errors and stats.

Parameters:

errors (list[str])
warnings (list[str])
train_utterances (int)
test_utterances (int)
dev_utterances (int)
total_utterances (int)
vocabulary_size (int)
missing_words (list[str])
dictionary_entries (int)
dictionary_base_words (int)
phoneset_size (int)
has_silence (bool)
missing_phones (list[str])
audio_files (int)
missing_audio (list[str])

__init__(errors=<factory>, warnings=<factory>, train_utterances=0, test_utterances=0, dev_utterances=0, total_utterances=0, vocabulary_size=0, missing_words=<factory>, dictionary_entries=0, dictionary_base_words=0, phoneset_size=0, has_silence=False, missing_phones=<factory>, audio_files=0, missing_audio=<factory>)

Parameters:

errors (list[str])
warnings (list[str])
train_utterances (int)
test_utterances (int)
dev_utterances (int)
total_utterances (int)
vocabulary_size (int)
missing_words (list[str])
dictionary_entries (int)
dictionary_base_words (int)
phoneset_size (int)
has_silence (bool)
missing_phones (list[str])
audio_files (int)
missing_audio (list[str])

Return type:

None

audio_files: int = 0

dev_utterances: int = 0

dictionary_base_words: int = 0

dictionary_entries: int = 0

errors: list[str]

has_silence: bool = False

property is_valid: bool: True if no errors.

missing_audio: list[str]

missing_phones: list[str]

missing_words: list[str]

phoneset_size: int = 0

save_json(path)[source]

Save report as JSON file.

Parameters:: path (Path)
Return type:: None

summary()[source]

Generate human-readable summary.

Return type:: str

test_utterances: int = 0

to_dict()[source]

Convert to dictionary (JSON-serializable).

Return type:: dict[str, object]

to_json(indent=2)[source]

Convert to JSON string.

Parameters:: indent (int)
Return type:: str

total_utterances: int = 0

train_utterances: int = 0

vocabulary_size: int = 0

warnings: list[str]

pstrain.api.build_lm(transcripts, output_path, max_order=3, smoothing='auto')[source]

Build an ARPA language model from transcripts using arpabo.

Parameters:

transcripts (list[str] | dict[str, str]) – List of transcript strings or dict mapping utt_id to transcript
output_path (Path) – Path to write the ARPA LM file
max_order (int) – N-gram order (default 3 for trigrams)
smoothing (str) – Smoothing method - “auto” (default), “good_turing”, “kneser_ney”

Returns:

Path to the created LM file

Return type:

Note

“auto” mode uses optimized Katz backoff, which works well for typical speech corpus sizes. For very small corpora, “good_turing” may be better.

pstrain.api.build_lm_from_file(transcript_file, output_path, max_order=3, smoothing='auto')[source]

Build an ARPA LM from a supported transcription file.

Parameters:

transcript_file (Path) – Path to a simple or Sphinx-format transcription file
output_path (Path) – Path to write ARPA LM
max_order (int) – N-gram order (default 3)
smoothing (str) – Smoothing method (default “auto”)

Returns:

Path to created LM file

Return type:

pstrain.api.compare_auto(file_a, file_b, rtol=1e-05, atol=1e-08)[source]

Auto-detect file types and compare.

Parameters:

file_a (Path) – First file or directory
file_b (Path) – Second file or directory
rtol (float) – Relative tolerance
atol (float) – Absolute tolerance

Returns:

Tuple of (detected_type, comparison_result)

Raises:

ValueError – If file types don’t match or are unknown

Return type:

tuple[FileType, CompareResult | ModelCompareResult]

pstrain.api.compare_features(file_a, file_b, rtol=1e-05, atol=1e-08)[source]

Compare two feature files (.mfc format).

Parameters:

file_a (Path) – First feature file
file_b (Path) – Second feature file
rtol (float) – Relative tolerance for comparison
atol (float) – Absolute tolerance for comparison

Returns:

CompareResult with comparison details

Return type:

CompareResult

pstrain.api.compare_gaussians(file_a, file_b, rtol=1e-05, atol=1e-08)[source]

Compare two Gaussian parameter files (means or variances).

Parameters:

file_a (Path) – First Gaussian file
file_b (Path) – Second Gaussian file
rtol (float) – Relative tolerance
atol (float) – Absolute tolerance

Returns:

CompareResult with comparison details

Return type:

CompareResult

pstrain.api.compare_mixw(file_a, file_b, rtol=1e-05, atol=1e-08)[source]

Compare two mixture weight files.

Parameters:

file_a (Path) – First mixw file
file_b (Path) – Second mixw file
rtol (float) – Relative tolerance
atol (float) – Absolute tolerance

Returns:

CompareResult with comparison details

Return type:

CompareResult

pstrain.api.compare_models(dir_a, dir_b, rtol=1e-05, atol=1e-08)[source]

Compare two complete model directories.

Compares all model components: - Critical: mdef (topology), feat.params (feature config) - Parameters: means, variances, mixture_weights, transition_matrices, sendump - Other: noisedict, README, etc.

Parameters:

dir_a (Path) – First model directory
dir_b (Path) – Second model directory
rtol (float) – Relative tolerance for numeric comparisons
atol (float) – Absolute tolerance for numeric comparisons

Returns:

ModelCompareResult with detailed comparison of all components

Return type:

ModelCompareResult

pstrain.api.compare_tmat(file_a, file_b, rtol=1e-05, atol=1e-08)[source]

Compare two transition matrix files.

Parameters:

file_a (Path) – First tmat file
file_b (Path) – Second tmat file
rtol (float) – Relative tolerance
atol (float) – Absolute tolerance

Returns:

CompareResult with comparison details

Return type:

CompareResult

pstrain.api.copy_tutorial(output='arctic_hmm_gmm_tutorial.ipynb', *, force=False, dry_run=False)[source]

Copy the bundled tutorial notebook to output.

If output is an existing directory, the standard tutorial filename is appended. Existing files are protected unless force is true.

Parameters:

output (str | Path)
force (bool)
dry_run (bool)

Return type:

TutorialResult

pstrain.api.create_model(model_type, config='baseline')[source]

Create a model instance.

Parameters:

model_type (str) – Model type identifier (e.g., “ci”, “cd”)
config (str) – Model configuration name (default: “baseline”)

Returns:

Model instance of the specified type

Raises:

ValueError – If model type is unknown

Return type:

Model

pstrain.api.default_cmudict_cache()[source]

Return the CMUdict cache within pstrain’s benchmark cache hierarchy.

Return type:: Path

pstrain.api.describe_file(path)[source]

Get human-readable description of file type.

Parameters:: path (Path)
Return type:: str

pstrain.api.detect_file_type(path)[source]

Detect file type from path and content.

Detection strategy: 1. Check if it’s a directory (model) 2. Check filename against known patterns 3. Check file extension 4. Check file content/magic bytes

Parameters:: path (Path) – Path to file or directory
Returns:: FileType enum value
Return type:: FileType

pstrain.api.extract_features(audio_path, output_path, *, fmt='sphinx', **config)[source]

Extract features from audio file and write to output file.

Parameters:

audio_path (str | Path) – Path to input audio file (WAV format)
output_path (str | Path) – Path to output feature file
fmt (str) – Output format (“sphinx” or “numpy”)
**config (Any) – Feature extraction parameters (samprate, ncep, etc.)

Returns:

Number of frames extracted

Return type:

pstrain.api.fetch_cmudict(ref='HEAD', cache=None)[source]

Fetch, convert, and cache CMUdict from its original upstream source.

ref may be a tag, branch, or commit SHA. Network access happens only in a fresh child process. On macOS, an in-process HTTPS download followed by a native worker spawn crashes the interpreter, so the download is kept out of any process that later spawns workers.

Parameters:

ref (str)
cache (Path | None)

Return type:

CMUDictSource

pstrain.api.get_fileids(transcription_path)[source]

Get list of fileids from transcription file.

Parameters:

transcription_path (Path) – Path to transcription file

Returns:

List of file IDs (utterance identifiers)

Raises:

FileNotFoundError – If transcription file does not exist
UnicodeDecodeError – If file is not UTF-8 encoded

Return type:

pstrain.api.get_model_class(model_type)[source]

Get model class from model type string.

Parameters:: model_type (str) – Model type identifier (e.g., “ci”, “cd”)
Returns:: Model class
Raises:: ValueError – If model type is unknown
Return type:: type[Model]

pstrain.api.get_paths()[source]

Get all pstrain paths.

Returns:: PstrainPaths with discovered locations (None if not found)
Return type:: PstrainPaths

pstrain.api.init_flat_model(phones, output_dir, n_state=3, n_density=1, ctl_path=None, cep_dir=None, cep_ext='.mfc', feat_type='1s_c_d_dd', ceplen=13, max_skip_fraction=0.05, skip_state=False)[source]

Initialize a complete flat model.

If ctl_path and cep_dir are provided, computes means/variances from the feature data using init_gau. Otherwise creates placeholder files.

Parameters:

phones (list[str]) – List of phone names
output_dir (Path) – Directory to write model files
n_state (int) – Number of emitting states per phone
n_density (int) – Number of Gaussians per state
ctl_path (Path | None) – Control file for feature computation
cep_dir (Path | None) – Feature directory for computation
cep_ext (str) – Feature file extension
feat_type (str) – Feature type string
ceplen (int) – Cepstral dimension
skip_state (bool) – Add state-to-state-plus-two transition arcs
max_skip_fraction (float)

Returns:

Dict mapping file type to path

Return type:

dict[str, Path]

pstrain.api.package_model(model_dir, output_dir, model_name=None, dictionary_path=None, filler_dict_path=None, include_dict=True, *, overwrite=False)[source]

Package a trained model for distribution.

Creates a complete, self-contained model directory that can be used directly with PocketSphinx and other Sphinx decoders.

Parameters:

model_dir (Path) – Source model directory
output_dir (Path) – Output directory for packaged model
model_name (str | None) – Name for the model (used in output path)
dictionary_path (Path | None) – Path to pronunciation dictionary
filler_dict_path (Path | None) – Path to filler dictionary
include_dict (bool) – Whether to include dictionary in package
overwrite (bool) – Allow replacement of a recognizable package without a marker

Returns:

Dict mapping file types to output paths

Return type:

dict[str, Path]

Notes

A supported package marker permits replacement by default; a recognizable legacy package without a marker requires overwrite=True. Unrecognized destinations and invalid or unsupported markers are never replaced.

On macOS and Linux, replacement opens the source, destination parent, staging directory, retained package, and recovery directory without following their final names. Identities come from those descriptors; validation and traversal remain descriptor-relative; and every rename is reconciled from the filesystem even when its call raises. Windows retains the path-based transaction and makes no guarantee against an active process substituting names during packaging. An asynchronous interruption can also leave a mixed unnamed package on Windows because its path transaction cannot reconcile a rename that completed before raising. No supported platform promises safety against every active same-filesystem race because final directory-entry deletion has no portable conditional-by-descriptor primitive. See docs/package-safety.md for the exact guarantee, recovery instructions, and remaining seams.

With no model name, acoustic, dict, README.txt, and pstrain-package.json transition separately. On a handled failure the implementation reconciles each open identity and attempts to restore the previous public set. Individual paths can be absent during that recovery; failures are attached to the initiating exception and recovery directories are retained when certainty is lost. Unrelated entries in output_dir are preserved.

Example output structure:

dist/models/my-model/
├── acoustic/
│   ├── feat.params
│   ├── mdef
│   ├── means
│   ├── variances
│   ├── mixture_weights
│   ├── transition_matrices
│   └── noisedict
├── dict/
│   ├── cmudict.dict
│   └── filler.dict
└── README.txt

pstrain.api.parse_transcription_file(transcription_path)[source]

Parse transcription file (word-level text, not time-aligned).

Parameters:

transcription_path (Path) – Path to transcription file

Returns:

Dict mapping fileid -> transcript text (words only)

Raises:

FileNotFoundError – If transcription file does not exist
UnicodeDecodeError – If file is not UTF-8 encoded
ValueError – If a nonempty line does not match a supported format

Return type:

dict[str, str]

Supports two formats: 1. Simple: <fileid> <word1> <word2> ... 2. Sphinx: [<s>] <word1> <word2> [</s>] (<fileid>)

The Sphinx begin/end silence markers are optional, following the convention used by the arpabo language-model tools. A trailing (<fileid>) token selects this form; otherwise the first whitespace token is the file ID.

pstrain.api.print_stats(file_path)[source]

Print statistics for a single file.

Detects file type and prints relevant statistics.

Parameters:: file_path (Path) – Path to file
Return type:: None

pstrain.api.resolve_config(project_dir, *, profile_name='default', experiment='default', cli_overrides=None, user_config_path=None)[source]

Resolve built-in < user < project < experiment < CLI.

Parameters:

project_dir (Path | str)
profile_name (str)
experiment (str)
cli_overrides (dict[str, Any] | None)
user_config_path (Path | None)

Return type:

ResolvedConfig

pstrain.api.run_build_lm(train_transcripts, output_path, max_order=3, smoothing='auto')[source]

Build an ARPA language model from training transcripts.

Uses arpabo with auto mode (optimized Katz backoff) by default.

Parameters:

train_transcripts (Path) – Path to a simple or Sphinx-format training transcription file
output_path (Path) – Path to write ARPA LM file
max_order (int) – N-gram order (default 3 for trigrams)
smoothing (str) – Smoothing method - “auto” (default), “good_turing”, “kneser_ney”

Returns:

Path to created LM file

Return type:

pstrain.api.run_step_cd_hmm_untied(project_dir, experiment='default', config='baseline', dry_run=False, **params)[source]

Run CD HMM untied training.

Parameters:

project_dir (Path | str) – Project directory path
experiment (str) – Experiment name
config (str) – Model configuration name
dry_run (bool) – If True, show what would be done without executing
**params (Any) – Training parameters

Returns:

Exit code (0 for success, non-zero for failure)

Return type:

pstrain.api.run_step_ci_hmm(project_dir, experiment='default', config='baseline', dry_run=False, **params)[source]

Run CI HMM training.

Parameters:

project_dir (Path | str) – Project directory path
experiment (str) – Experiment name
config (str) – Model configuration name
dry_run (bool) – If True, show what would be done without executing
**params (Any) – Training parameters

Returns:

Exit code (0 for success, non-zero for failure)

Return type:

pstrain.api.run_step_features(project_dir, experiment='default', config='baseline', dry_run=False, **params)[source]

Run feature extraction step.

Parameters:

project_dir (Path | str) – Project directory path
experiment (str) – Experiment name
config (str) – Configuration name
dry_run (bool) – If True, show what would be done without executing
**params (Any) – Additional parameters

Returns:

Exit code (0 for success, non-zero for failure)

Return type:

pstrain.api.setup_project(project_dir, transcription_path=None, audio_path=None, dictionary_path=None, phoneset_path=None, filler_dict_path=None, config_path=None, link_audio=False, clobber=False)[source]

Set up a new pstrain project.

Parameters:

project_dir (Path) – Project directory (create if needed)
transcription_path (Path | None) – Path to transcription file
audio_path (Path | None) – Path to audio directory or file (optional)
dictionary_path (Path | None) – Path to dictionary file
phoneset_path (Path | None) – Path to phoneset file (or extract from dictionary)
filler_dict_path (Path | None) – Path to filler dictionary (optional)
config_path (Path | None) – Path to config file (or create default)
link_audio (bool) – If True and audio_path provided, symlink instead of copying
clobber (bool) – If True, overwrite existing files; if False, skip existing files

Returns:

Dict with setup status and paths

Return type:

pstrain.api.step_cd_hmm_untied(project_dir, experiment='default', config='baseline', **params)[source]

Get the rule definition for CD HMM untied training.

Parameters:

project_dir (Path | str) – Project directory path
experiment (str) – Experiment name
config (str) – Model configuration name
**params (Any) – Training parameters

Returns:

StepDefinition with rule metadata

Return type:

pstrain.api.step_ci_hmm(project_dir, experiment='default', config='baseline', **params)[source]

Get the rule definition for CI HMM training.

Parameters:

project_dir (Path | str) – Project directory path
experiment (str) – Experiment name
config (str) – Model configuration name
**params (Any) – Training parameters

Returns:

StepDefinition with rule metadata

Return type:

pstrain.api.step_features(project_dir, experiment='default', config='baseline', **params)[source]

Get the rule definition for feature extraction.

Parameters:

project_dir (Path | str) – Project directory path
experiment (str) – Experiment name
config (str) – Configuration name
**params (Any) – Additional parameters

Returns:

StepDefinition with rule metadata

Return type:

pstrain.api.strip_dictionary_stress(input_dict, output_dict)[source]

Strip stress, merge duplicate pronunciations, and renumber variants.

This mirrors CMUdict’s PocketSphinx conversion: entries are grouped by their base word, the first occurrence of each distinct stripped pronunciation is retained, words are sorted, and surviving variants are numbered from two. Comments, blank lines, and malformed lines are omitted.

Parameters:

input_dict (Path) – Input dictionary (with stress)
output_dict (Path) – Output dictionary (without stress)

Returns:

Tuple of (entries_processed, unique_phones)

Return type:

tuple[int, int]

pstrain.api.strip_stress(phone)[source]

Remove ARPABET stress marker (trailing digit) from phone.

Parameters:: phone (str) – Phone possibly with stress marker (e.g., “AH0”, “OW1”)
Returns:: Phone without stress marker (e.g., “AH”, “OW”)
Return type:: str

Examples

strip_stress(“AA1”) -> “AA” strip_stress(“AE0”) -> “AE” strip_stress(“HH”) -> “HH” (no change, consonants don’t have stress)

pstrain.api.validate_file_type(path, expected, deep=False)[source]

Validate that a file matches the expected type.

Parameters:

path (Path) – Path to file or directory
expected (FileType) – Expected FileType
deep (bool) – If True, attempt to load the file to validate (slower but more thorough)

Returns:

Tuple of (is_valid, message)

Return type:

tuple[bool, str]

pstrain.api.validate_package_destination(model_dir, output_dir, model_name=None, *, include_dict=True, overwrite=False)[source]

Validate a package destination without changing the filesystem.

Named packages replace their complete destination. Unnamed packages replace only their generated entries and preserve unrelated entries in output_dir. Existing generated entries must belong to a recognizable package in either case. A supported package marker establishes ownership; replacing a legacy package without one requires explicit opt-in.

Parameters:

model_dir (Path) – Source model directory
output_dir (Path) – Output directory for packaged model
model_name (str | None) – Name for the model, as one ordinary path component
include_dict (bool) – Whether packaging will replace the generated dictionary directory
overwrite (bool) – Allow replacement of a recognizable package without a marker

Returns:

The package directory that packaging will write

Raises:

ValueError – If the name, source relationship, or existing destination is unsafe

Return type:

pstrain.api.validate_project(project_dir, experiment='default')[source]

Validate project structure and files, return detailed report.

Parameters:

project_dir (Path) – Project directory to validate
experiment (str) – Experiment name (default: “default”)

Returns:

ValidationReport with errors, warnings, and stats

Return type:

ValidationReport

pstrain.api.TUTORIAL_FILENAME: The filename used for the bundled tutorial notebook.

Project Setup

Project setup implementation for pstrain.

pstrain.lib.setup.setup_project(project_dir, transcription_path=None, audio_path=None, dictionary_path=None, phoneset_path=None, filler_dict_path=None, config_path=None, link_audio=False, clobber=False)[source]

Set up a new pstrain project.

Parameters:

project_dir (Path) – Project directory (create if needed)
transcription_path (Path | None) – Path to transcription file
audio_path (Path | None) – Path to audio directory or file (optional)
dictionary_path (Path | None) – Path to dictionary file
phoneset_path (Path | None) – Path to phoneset file (or extract from dictionary)
filler_dict_path (Path | None) – Path to filler dictionary (optional)
config_path (Path | None) – Path to config file (or create default)
link_audio (bool) – If True and audio_path provided, symlink instead of copying
clobber (bool) – If True, overwrite existing files; if False, skip existing files

Returns:

Dict with setup status and paths

Return type:

Project Validation

Project validation for pstrain.

exception pstrain.lib.validate.ValidationError[source]

Bases: Exception

Validation error.

class pstrain.lib.validate.ValidationReport(errors=<factory>, warnings=<factory>, train_utterances=0, test_utterances=0, dev_utterances=0, total_utterances=0, vocabulary_size=0, missing_words=<factory>, dictionary_entries=0, dictionary_base_words=0, phoneset_size=0, has_silence=False, missing_phones=<factory>, audio_files=0, missing_audio=<factory>)[source]

Bases: object

Validation report with errors and stats.

Parameters:

errors (list[str])
warnings (list[str])
train_utterances (int)
test_utterances (int)
dev_utterances (int)
total_utterances (int)
vocabulary_size (int)
missing_words (list[str])
dictionary_entries (int)
dictionary_base_words (int)
phoneset_size (int)
has_silence (bool)
missing_phones (list[str])
audio_files (int)
missing_audio (list[str])

__init__(errors=<factory>, warnings=<factory>, train_utterances=0, test_utterances=0, dev_utterances=0, total_utterances=0, vocabulary_size=0, missing_words=<factory>, dictionary_entries=0, dictionary_base_words=0, phoneset_size=0, has_silence=False, missing_phones=<factory>, audio_files=0, missing_audio=<factory>)

Parameters:

errors (list[str])
warnings (list[str])
train_utterances (int)
test_utterances (int)
dev_utterances (int)
total_utterances (int)
vocabulary_size (int)
missing_words (list[str])
dictionary_entries (int)
dictionary_base_words (int)
phoneset_size (int)
has_silence (bool)
missing_phones (list[str])
audio_files (int)
missing_audio (list[str])

Return type:

None

audio_files: int = 0

dev_utterances: int = 0

dictionary_base_words: int = 0

dictionary_entries: int = 0

errors: list[str]

has_silence: bool = False

property is_valid: bool: True if no errors.

missing_audio: list[str]

missing_phones: list[str]

missing_words: list[str]

phoneset_size: int = 0

save_json(path)[source]

Save report as JSON file.

Parameters:: path (Path)
Return type:: None

summary()[source]

Generate human-readable summary.

Return type:: str

test_utterances: int = 0

to_dict()[source]

Convert to dictionary (JSON-serializable).

Return type:: dict[str, object]

to_json(indent=2)[source]

Convert to JSON string.

Parameters:: indent (int)
Return type:: str

total_utterances: int = 0

train_utterances: int = 0

vocabulary_size: int = 0

warnings: list[str]

pstrain.lib.validate.validate_files_exist(files, context='')[source]

Validate that all files in a list exist.

Parameters:

files (list[Path]) – List of file paths to check
context (str) – Optional context string for error messages

Raises:

FileNotFoundError – If any file does not exist

Return type:

None

pstrain.lib.validate.validate_project(project_dir, experiment='default')[source]

Validate project structure and files, return detailed report.

Parameters:

project_dir (Path) – Project directory to validate
experiment (str) – Experiment name (default: “default”)

Returns:

ValidationReport with errors, warnings, and stats

Return type:

ValidationReport

Configuration

Canonical pstrain configuration API.

class pstrain.lib.config.FeatureConfig(*, samprate=16000, ncep=13, nfilt=25, nfft=512, lowerf=130.0, upperf=6800.0, alpha=0.97, dither=True, seed=-1, remove_dc=True, remove_noise=True, frate=100, wlen=0.025625, feat_type='1s_c_d_dd', lifter=22, transform='dct', agc='none', cmn='batch', cmninit='40,3,-1', varnorm='no')[source]

Bases: StrictModel

Acoustic front-end parameters.

Parameters:

samprate (Annotated[int, Gt(gt=0)])
ncep (Annotated[int, Gt(gt=0)])
nfilt (Annotated[int, Gt(gt=0)])
nfft (Annotated[int, Gt(gt=0)])
lowerf (Annotated[float, Ge(ge=0)])
upperf (Annotated[float, Gt(gt=0)])
alpha (float)
dither (bool)
seed (int)
remove_dc (bool)
remove_noise (bool)
frate (Annotated[int, Gt(gt=0)])
wlen (Annotated[float, Gt(gt=0)])
feat_type (str)
lifter (Annotated[int, Ge(ge=0)])
transform (str)
agc (str)
cmn (str)
cmninit (str)
varnorm (str)

agc: Annotated[str, Field(description='Automatic gain-control mode')]

alpha: Annotated[float, Field(description='Pre-emphasis coefficient')]

cmn: Annotated[str, Field(description='Cepstral mean-normalization mode')]

cmninit: Annotated[str, Field(description='Initial cepstral mean vector for live CMN')]

dither: Annotated[bool, Field(description='Add half-bit dither to input audio')]

feat_type: Annotated[str, Field(description='Sphinx feature stream type')]

frate: Annotated[int, Field(gt=0, description='Feature frame rate in Hz')]

lifter: Annotated[int, Field(ge=0, description='Cepstral lifter window')]

lowerf: Annotated[float, Field(ge=0, description='Lower filter-bank frequency in Hz')]

model_config = {'extra': 'forbid', 'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

ncep: Annotated[int, Field(gt=0, description='Number of cepstral coefficients')]

nfft: Annotated[int, Field(gt=0, description='FFT size')]

nfilt: Annotated[int, Field(gt=0, description='Number of mel filters')]

remove_dc: Annotated[bool, Field(description='Remove DC offset from each frame')]

remove_noise: Annotated[bool, Field(description='Remove noise with spectral subtraction')]

samprate: Annotated[int, Field(gt=0, description='Audio sample rate in Hz')]

seed: Annotated[int, Field(description='Seed for deterministic input dithering')]

transform: Annotated[str, Field(description='Filter-bank transform')]

upperf: Annotated[float, Field(gt=0, description='Upper filter-bank frequency in Hz')]

validate_band()[source]

Return type:: FeatureConfig

varnorm: Annotated[str, Field(description='Cepstral variance-normalization mode')]

wlen: Annotated[float, Field(gt=0, description='Analysis window length in seconds')]

class pstrain.lib.config.FieldExplanation(field_path: 'str', value: 'Any', canonical_type: 'str', winner: 'Candidate', overridden: 'tuple[Candidate, ...]', default: 'Any', constraints: 'dict[str, Any]', consumer: 'str', provenance_scope: 'str')[source]

Bases: object

Parameters:

field_path (str)
value (Any)
canonical_type (str)
winner (Candidate)
overridden (tuple[Candidate, ...])
default (Any)
constraints (dict[str, Any])
consumer (str)
provenance_scope (str)

__init__(field_path, value, canonical_type, winner, overridden, default, constraints, consumer, provenance_scope)

Parameters:

field_path (str)
value (Any)
canonical_type (str)
winner (Candidate)
overridden (tuple[Candidate, ...])
default (Any)
constraints (dict[str, Any])
consumer (str)
provenance_scope (str)

Return type:

None

canonical_type: str

constraints: dict[str, Any]

consumer: str

default: Any

field_path: str

overridden: tuple[Candidate, ...]

provenance_scope: str

value: Any

winner: Candidate

class pstrain.lib.config.OverlayDocument(*, config_version=1, profile=None, features=None, training=None, split=None, runner=None, sharding=None, alignment=None)[source]

Bases: StrictModel

Canonical user, project, or experiment field overlay.

Parameters:

config_version (Literal[1])
profile (str | None)
features (dict[str, Any] | None)
training (dict[str, Any] | None)
split (dict[str, Any] | None)
runner (dict[str, Any] | None)
sharding (dict[str, Any] | None)
alignment (dict[str, Any] | None)

alignment: dict[str, Any] | None

config_version: Literal[1]

features: dict[str, Any] | None

model_config = {'extra': 'forbid', 'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

profile: str | None

runner: dict[str, Any] | None

sharding: dict[str, Any] | None

split: dict[str, Any] | None

training: dict[str, Any] | None

class pstrain.lib.config.ParameterInfo(key, type, default, description, required)[source]

Bases: object

Information about a configuration parameter.

Parameters:

key (str)
type (str)
default (Any)
description (str)
required (bool)

__init__(key, type, default, description, required)

Parameters:

key (str)
type (str)
default (Any)
description (str)
required (bool)

Return type:

None

default: Any

description: str

key: str

required: bool

type: str

class pstrain.lib.config.Profile(*, description='', features=<factory>, training=<factory>, split=<factory>, runner=<factory>, sharding=<factory>, alignment=<factory>)[source]

Bases: StrictModel

One complete named model-training profile.

Parameters:

description (str)
features (FeatureConfig)
training (TrainingConfig)
split (SplitConfig)
runner (RunnerConfig)
sharding (ShardingConfig)
alignment (AlignmentConfig)

alignment: AlignmentConfig

description: Annotated[str, Field(description='Human-readable profile purpose')]

features: FeatureConfig

model_config = {'extra': 'forbid', 'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

runner: RunnerConfig

sharding: ShardingConfig

split: SplitConfig

training: TrainingConfig

class pstrain.lib.config.ProfilesDocument(*, config_version=1, profiles)[source]

Bases: StrictModel

Canonical etc/configs.yaml document.

Parameters:

config_version (Literal[1])
profiles (dict[str, ProfileDefinition])

config_version: Literal[1]

model_config = {'extra': 'forbid', 'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

profiles: dict[str, ProfileDefinition]

class pstrain.lib.config.ResolvedConfig(profile: 'Profile', profile_name: 'str', config_version: 'int', fields: 'dict[str, FieldExplanation]', warnings: 'tuple[str, ...]' = ())[source]

Bases: object

Parameters:

profile (Profile)
profile_name (str)
config_version (int)
fields (dict[str, FieldExplanation])
warnings (tuple[str, ...])

__init__(profile, profile_name, config_version, fields, warnings=())

Parameters:

profile (Profile)
profile_name (str)
config_version (int)
fields (dict[str, FieldExplanation])
warnings (tuple[str, ...])

Return type:

None

as_dict()[source]

Return type:: dict[str, Any]

benchmark_document()[source]

Return a deterministic snapshot of values and winning source kinds.

Return type:: dict[str, Any]

config_version: int

fields: dict[str, FieldExplanation]

profile: Profile

profile_name: str

warnings: tuple[str, ...] = ()

class pstrain.lib.config.RunnerConfig(*, jobs=None, nice=5)[source]

Bases: StrictModel

Local pipeline execution policy.

Parameters:

jobs (Annotated[int | None, Ge(ge=1)])
nice (Annotated[int, Ge(ge=0)])

jobs: Annotated[int | None, Field(ge=1, description='Parallel workers; null means auto')]

model_config = {'extra': 'forbid', 'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

nice: Annotated[int, Field(ge=0, description='Worker niceness increment')]

class pstrain.lib.config.SplitConfig(*, train_ratio=None, test_count=None, seed=42)[source]

Bases: StrictModel

Train/test split parameters.

Parameters:

train_ratio (Annotated[float | None, Gt(gt=0), Lt(lt=1)])
test_count (Annotated[int | None, Ge(ge=0)])
seed (int)

model_config = {'extra': 'forbid', 'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

seed: Annotated[int, Field(description='Deterministic split seed')]

test_count: Annotated[int | None, Field(ge=0, description='Fixed test utterance count; zero disables an additional holdout')]

train_ratio: Annotated[float | None, Field(gt=0, lt=1, description='Training fraction')]

validate_choice()[source]

Return type:: SplitConfig

class pstrain.lib.config.TrainingConfig(*, n_state=3, skip_state=False, n_senones=200, a_beam=1e-90, b_beam=1e-10, ci=<factory>, tied=<factory>, untied=<factory>, split_variance_floor_fraction=0.0, max_skip_fraction=0.05, retry_beam_factor=10000000000.0, failed_alignment='recover', bw_checkpoint_iterations=False, arctic_a0302_zero_codebook_band=None, accept_arctic_a0587_known_skip=False, tree_state_weights=(1.0, 0.05, 0.0), tree_rotate_state_weights=True, tree_directional_questions=True, tree_ssplitmax=7, tree_ssplitthr=0.0, tree_csplitmax=2000, tree_csplitthr=0.0, tree_mwfloor=1e-08, tree_intermediate_dumps=False, question_npermute=12, question_quests_per_state=20, question_niter=1, multipron_training=True, optional_final_silence=True, untied_inventory='transcript-reachable', exclusion_schedule=<factory>)[source]

Bases: StrictModel

Acoustic-model training parameters.

Parameters:

n_state (Annotated[int, Ge(ge=1)])
skip_state (bool)
n_senones (Annotated[int, Ge(ge=1)])
a_beam (Annotated[float, Gt(gt=0)])
b_beam (Annotated[float, Gt(gt=0)])
ci (TrainingScheduleConfig)
tied (TrainingScheduleConfig)
untied (TrainingScheduleConfig)
split_variance_floor_fraction (Annotated[float, Strict(strict=True), Ge(ge=0), Le(le=1), _PydanticGeneralMetadata(allow_inf_nan=False)])
max_skip_fraction (Annotated[float, Ge(ge=0), Le(le=1)])
retry_beam_factor (Annotated[float | list[float], PlainValidator(func=~pstrain.lib.config.models._validate_retry_beam_factor, json_schema_input_type=~typing.Any), WithJsonSchema(json_schema={'anyOf': [{'type': 'number', 'exclusiveMinimum': 0}, {'type': 'array', 'items': {'type': 'number', 'exclusiveMinimum': 1}, 'minItems': 1}]}, mode=None)])
failed_alignment (Literal['recover', 'abort', 'omit'])
bw_checkpoint_iterations (bool)
arctic_a0302_zero_codebook_band (tuple[int, int] | None)
accept_arctic_a0587_known_skip (bool)
tree_state_weights (Annotated[tuple[float, ...], MinLen(min_length=1)])
tree_rotate_state_weights (bool)
tree_directional_questions (bool)
tree_ssplitmax (Annotated[int, Ge(ge=0)])
tree_ssplitthr (Annotated[float, Ge(ge=0)])
tree_csplitmax (Annotated[int, Ge(ge=0)])
tree_csplitthr (Annotated[float, Ge(ge=0)])
tree_mwfloor (Annotated[float, Gt(gt=0)])
tree_intermediate_dumps (bool)
question_npermute (Annotated[int, Ge(ge=1)])
question_quests_per_state (Annotated[int, Ge(ge=1)])
question_niter (Annotated[int, Ge(ge=1)])
multipron_training (bool)
optional_final_silence (bool)
untied_inventory (Literal['all-triphone', 'transcript-reachable', 'linear'])
exclusion_schedule (dict[str, dict[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Strict(strict=True)])] | str, list[str]]])

a_beam: Annotated[float, Field(gt=0, description='Forward alignment beam')]

accept_arctic_a0587_known_skip: Annotated[bool, Field(description="Deprecated for live profiles: retained solely for the Arctic pin's retired off-profile provenance; live benchmark cells run exception-free")]

arctic_a0302_zero_codebook_band: Annotated[tuple[int, int] | None, Field(description='Accepted inclusive exact-zero codebook occupancy band for the singular Arctic a0302 terminal-alignment exception')]

b_beam: Annotated[float, Field(gt=0, description='Backward alignment beam')]

bw_checkpoint_iterations: Annotated[bool, Field(description='Retain the compact model files from every completed Baum-Welch pass under ``iterations/NN``; costs roughly one additional model copy per pass and does not retain the much larger ``.bw-accum`` shard accumulators or change which checkpoint is loaded by training. The deprecated ``PSTRAIN_BW_CHECKPOINTS=1`` environment variable can also enable retention, but cannot disable a true profile setting')]

ci: TrainingScheduleConfig

exclusion_schedule: Annotated[dict[str, dict[Annotated[int, Field(strict=True)] | str, list[str]]], Field(description='Experimental stage/pass utterance exclusions')]

failed_alignment: Annotated[Literal['recover', 'abort', 'omit'], Field(description='Action when an utterance fails to reach its final state: ``recover`` runs the wider-beam retries in ``retry_beam_factor`` and, if they all fail, reports the utterance and continues without it; ``abort`` fails the run on the first failure; and ``omit`` reports and excludes it without retrying. Skips are counted either way, and ``max_skip_fraction`` still fails the run when they stop being incidental')]

max_skip_fraction: Annotated[float, Field(ge=0, le=1, description='Maximum skipped-update fraction')]

model_config = {'extra': 'forbid', 'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

multipron_training: Annotated[bool, Field(description='Sum posteriors over pronunciation variants; when disabled without an explicit inventory policy, untied inventory resolves to upstream-compatible ``linear``')]

n_senones: Annotated[int, Field(ge=1, description='Target tied-state count')]

n_state: Annotated[int, Field(ge=1, description='Emitting states per HMM')]

optional_final_silence: Annotated[bool, Field(description='Permit final transcript silence to consume zero frames; stock SphinxTrain requires that silence to consume at least one frame')]

question_niter: Annotated[int, Field(ge=1, description='Question generation iterations')]

question_npermute: Annotated[int, Field(ge=1, description='Question permutations')]

question_quests_per_state: Annotated[int, Field(ge=1, description='Questions generated per state')]

retry_beam_factor: Annotated[RetryBeamFactorSetting, Field(description='Factor that widens the forward beam for one retry after an utterance fails to reach its final state, or an ascending list of factors, each greater than 1 and relative to the nominal beam, tried in order until one succeeds; a retry is counted only when that attempt runs')]

classmethod select_inventory_default(data)[source]

Parameters:: data (Any)
Return type:: Any

skip_state: Annotated[bool, Field(description="Enable SphinxTrain's $CFG_SKIPSTATE topology, adding an arc from each eligible emitting state to the state two positions ahead so a phone can be realized with fewer frames than states. SphinxTrain writes raw 3/1/1 weights and normalizes them on read; pstrain writes the behaviorally equivalent normalized values")]

split_variance_floor_fraction: Annotated[float, Field(ge=0, le=1, allow_inf_nan=False, strict=True, description='Experimental variance lower bound for split training stages, as a fraction of each matching coordinate in the fixed CI-1g or CD-1g variance reference. Zero disables regularization. The reference never advances with later splits; zero reference coordinates contribute no positive floor. Select a nonzero fraction explicitly after evaluation; no universal nonzero value is assumed')]

tied: TrainingScheduleConfig

tree_csplitmax: Annotated[int, Field(ge=0, description='Maximum phone-context splits')]

tree_csplitthr: Annotated[float, Field(ge=0, description='Phone-context split threshold')]

tree_directional_questions: Annotated[bool, Field(description='Honor _L/_R tree-question suffixes; disable only for isolation measurements')]

tree_intermediate_dumps: Annotated[bool, Field(description='Dump intermediate decision trees to worker diagnostics')]

tree_mwfloor: Annotated[float, Field(gt=0, description='Tree mixture-weight floor')]

tree_rotate_state_weights: Annotated[bool, Field(description='Apply target-relative tree state weights; disable only for isolation measurements')]

tree_ssplitmax: Annotated[int, Field(ge=0, description='Maximum state splits')]

tree_ssplitthr: Annotated[float, Field(ge=0, description='State split threshold')]

tree_state_weights: Annotated[tuple[float, ...], Field(min_length=1, description='Decision-tree state weights')]

untied: TrainingScheduleConfig

untied_inventory: Annotated[Literal['all-triphone', 'transcript-reachable', 'linear'], Field(description="Untied-model phone inventory policy: ``transcript-reachable`` includes contexts reachable through every pronunciation when multipron training is enabled; upstream-compatible ``linear`` includes contexts observed through each transcript word's first pronunciation; ``all-triphone`` includes the complete phoneset cross-product")]

classmethod validate_exclusions(value)[source]

Parameters:: value (dict[str, dict[int | str, list[str]]])
Return type:: dict[str, dict[int | str, list[str]]]

validate_training()[source]

Return type:: TrainingConfig

validate_tree_state_weights()[source]

Return type:: TrainingConfig

class pstrain.lib.config.TrainingScheduleConfig(*, max_iterations=10, min_iterations=1, convergence_ratio=0.001)[source]

Bases: StrictModel

Convergence controller for one Baum-Welch stage family.

Parameters:

max_iterations (Annotated[int, Ge(ge=1)])
min_iterations (Annotated[int, Ge(ge=1)])
convergence_ratio (Annotated[float, Gt(gt=0)])

convergence_ratio: Annotated[float, Field(gt=0, description="Converge after min_iterations when the finite per-frame log-likelihood increase is between zero and this many nats, inclusive. Negative or nonfinite changes do not indicate convergence. Despite the name -- kept because SphinxTrain's $CFG_CONVERGENCE_RATIO is the same signed per-frame delta -- this is an absolute difference, not a ratio. At the default, corpora of Arctic's size run all ten passes in every schedule, which is the more accurate outcome as measured; treat max_iterations as the operative control. The sphinxtrain profile carries SphinxTrain's own 0.1")]

max_iterations: Annotated[int, Field(ge=1, description='Maximum training passes')]

min_iterations: Annotated[int, Field(ge=1, description='Minimum training passes')]

model_config = {'extra': 'forbid', 'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

validate_iterations()[source]

Return type:: TrainingScheduleConfig

pstrain.lib.config.generate_markdown_docs()[source]

Generate Markdown documentation for all config parameters.

Returns:: Markdown string suitable for docs
Return type:: str

pstrain.lib.config.generate_rst_docs()[source]

Generate reStructuredText documentation for all config parameters.

Returns:: RST string suitable for Sphinx docs
Return type:: str

pstrain.lib.config.get_parameter(key)[source]

Get info about a specific parameter.

Parameters:: key (str) – Dot-separated parameter path (e.g., “audio.sample_rate”)
Returns:: ParameterInfo or None if not found
Return type:: ParameterInfo | None

pstrain.lib.config.get_schema()[source]

Get the canonical profile JSON schema.

Returns:: JSON Schema dict (for docs generation, validation, etc.)
Return type:: dict[str, Any]

pstrain.lib.config.list_parameters(prefix='')[source]

List all configuration parameters with descriptions.

Parameters:: prefix (str) – Optional prefix to filter parameters (e.g., “audio”, “features”)
Returns:: List of ParameterInfo objects
Return type:: list[ParameterInfo]

pstrain.lib.config.list_profiles(project_dir)[source]

Parameters:: project_dir (Path)
Return type:: list[dict[str, Any]]

pstrain.lib.config.migrate_project(project_dir, *, check)[source]

Render or atomically write the canonical profiles document.

Parameters:

project_dir (Path)
check (bool)

Return type:

tuple[Path, str, Path | None]

pstrain.lib.config.resolve_config(project_dir, *, profile_name='default', experiment='default', cli_overrides=None, user_config_path=None)[source]

Resolve built-in < user < project < experiment < CLI.

Parameters:

project_dir (Path | str)
profile_name (str)
experiment (str)
cli_overrides (dict[str, Any] | None)
user_config_path (Path | None)

Return type:

ResolvedConfig

Data Structures

Dictionary

Dictionary utilities.

Core classes: - Dictionary: Base class for pronunciation dictionaries (phoneset-agnostic) - CMUDict: Subclass with ARPABET stress handling

For Phoneset and phone mapping, see pstrain.lib.phoneset.

class pstrain.lib.dictionary.CMUDict[source]

Bases: Dictionary

CMUDict-style dictionary with ARPABET stress handling.

Extends Dictionary with ARPABET-specific features: - Stress marker parsing and manipulation - Stress-stripped variants for training - Vowel/consonant classification

classmethod from_file(path)[source]

Load CMUDict from file.

Overrides parent to return CMUDict instance.

Parameters:: path (Path)
Return type:: Self

get_primary_stress_position(word)[source]

Get the syllable position (0-indexed) of primary stress.

Parameters:: word (str) – Word to look up
Returns:: Position of primary stress (0 = first vowel), or None if not found
Return type:: int | None

get_stressed_vowels(word)[source]

Get vowels with their stress levels for a word.

Parameters:: word (str) – Word to look up
Returns:: List of (vowel, stress_level) tuples stress_level is 0, 1, 2, or -1 if no stress marker
Return type:: list[tuple[str, int]]

Example

get_stressed_vowels(“HELLO”) -> [(“AH”, 0), (“OW”, 1)] # for HH AH0 L OW1

strip_stress_from_entries()[source]

Create a copy with stress markers removed from all phones.

Returns:: New CMUDict with stress-free pronunciations
Return type:: CMUDict

Note

This may create duplicate entries (e.g., “read” R IY D and R EH D both become R IY D and R EH D without stress). Duplicates are automatically deduplicated.

class pstrain.lib.dictionary.Dictionary[source]

Bases: object

Pronunciation dictionary with Unicode and case-sensitive support.

Supports: - UTF-8 encoding - Case-sensitive words (hello != Hello != HELLO) - Sphinx-style variants: word, word(2), word(3), … - Multi-word entries: New_York, ice_cream - Any phoneset (ARPABET, IPA, X-SAMPA, custom)

__init__()[source]

Initialize empty dictionary.

Return type:: None

__len__()[source]

Number of dictionary entries (including variants).

Return type:: int

__repr__()[source]

String representation.

Return type:: str

add_entry(word, phonemes)[source]

Add pronunciation entry to dictionary.

Automatically handles duplicate pronunciations: - Same word + same pronunciation → Skip silently (true duplicate) - Same word + different pronunciation → Create variant (word(2), word(3), etc.)

Parameters:

word (str) – Word (may include variant suffix like (2), (3))
phonemes (list[str]) – List of phoneme strings

Return type:

None

Examples

READ R EH D # Stored as READ READ R IY D # Automatically stored as READ(2) READ R EH D # Skipped (duplicate of first)

base_words()[source]

Get all base words (no variant suffixes).

Return type:: list[str]

contains(word)[source]

Check if word exists in dictionary (exact match, case-sensitive).

Parameters:: word (str)
Return type:: bool

contains_base(base_word)[source]

Check if base word exists (any variant).

Parameters:: base_word (str)
Return type:: bool

filter_to_vocabulary(vocabulary, include_variants=True)[source]

Create filtered dictionary containing only words in vocabulary.

Parameters:

vocabulary (set[str]) – Set of words to keep (case-sensitive)
include_variants (bool) – If True (default), “word” in vocab gets all variants (word, word(2), word(3)). If False, only exact matches.

Returns:

New Dictionary containing matched words

Return type:

Dictionary

Examples

Given dict: read, read(2), hello, world vocab = {“read”, “hello”}

include_variants=True: read, read(2), hello include_variants=False: read, hello (no read(2))

vocab = {“read(2)”} include_variants=True: read(2) only (exact variant specified) include_variants=False: read(2) only

classmethod from_file(path)[source]

Load dictionary from file.

Parameters:: path (Path) – Path to dictionary file (UTF-8 encoded)
Returns:: Dictionary instance
Raises:: ValueError – If file has encoding or format errors
Return type:: Self

get(word)[source]

Get pronunciation for exact word match.

Parameters:: word (str) – Word to look up (case-sensitive, exact match)
Returns:: List of phonemes, or None if not found
Return type:: list[str] | None

get_variants(base_word)[source]

Get all pronunciation variants for a base word.

Parameters:: base_word (str) – Base word (without variant suffix)
Returns:: List of pronunciations (each is a list of phonemes) Returns empty list if word not found
Return type:: list[list[str]]

Examples

get_variants(“read”) -> [[“R”, “IY”, “D”], [“R”, “EH”, “D”]]

merge(other)[source]

Merge another dictionary into this one.

Parameters:: other (Dictionary) – Dictionary to merge in
Return type:: None

Note

Uses add_entry which automatically handles: - Deduplicating identical pronunciations - Creating variants for different pronunciations - Renumbering variants sequentially

phonemes()[source]

Get set of all phonemes used in dictionary.

Return type:: set[str]

property pronunciations: dict[str, list[list[str]]]

Get all pronunciations as a dictionary mapping base words to variant lists.

Returns:: Dict mapping base word to list of pronunciations Each pronunciation is a list of phonemes

Examples

{“read”: [[“R”, “IY”, “D”], [“R”, “EH”, “D”]],: “hello”: [[“HH”, “AH”, “L”, “OW”]]}

save(path)[source]

Save dictionary to file.

Parameters:: path (Path) – Output file path
Return type:: None

Note

Writes in CMU Sphinx format: - One entry per line: word phone1 phone2 … - Variants: word(2), word(3), etc. - UTF-8 encoding

words()[source]

Get all word keys (including variants).

Return type:: list[str]

pstrain.lib.dictionary.create_standard_filler_dict(output_path, silence_phone='SIL')[source]

Create standard filler dictionary with configurable silence phone.

Parameters:

output_path (Path) – Path to write filler dictionary
silence_phone (str) – Silence phone symbol (default: SIL)

Return type:

None

pstrain.lib.dictionary.get_filler_phones(silence_phone='SIL')[source]

Get set of filler phones.

Parameters:: silence_phone (str) – Silence phone symbol (default: SIL)
Returns:: Set of filler phone symbols
Return type:: set[str]

pstrain.lib.dictionary.get_standard_fillers(silence_phone='SIL')[source]

Get standard filler words and their phones.

Parameters:: silence_phone (str) – Silence phone symbol (default: SIL)
Returns:: Dict mapping filler words to phone lists
Return type:: dict[str, list[str]]

pstrain.lib.dictionary.get_standard_fillers_text(silence_phone='SIL')[source]

Get standard filler dictionary text with configurable silence phone.

Note: NO COMMENTS - PocketSphinx fails to parse comment lines correctly.

Parameters:: silence_phone (str) – Silence phone symbol (default: SIL)
Returns:: Filler dictionary text (no comments)
Return type:: str

pstrain.lib.dictionary.get_stress(phone)[source]

Get ARPABET stress level from phone.

Parameters:: phone (str) – Phone possibly with stress marker
Returns:: 0 (no stress), 1 (primary), 2 (secondary), or None if no stress marker
Return type:: int | None

Examples

get_stress(“AH0”) -> 0 get_stress(“OW1”) -> 1 get_stress(“AE2”) -> 2 get_stress(“HH”) -> None

pstrain.lib.dictionary.is_vowel(phone)[source]

Check if phone is an ARPABET vowel (stress stripped).

Parameters:: phone (str) – Phone with or without stress marker
Returns:: True if vowel
Return type:: bool

pstrain.lib.dictionary.strip_dictionary_stress(input_dict, output_dict)[source]

Strip stress, merge duplicate pronunciations, and renumber variants.

This mirrors CMUdict’s PocketSphinx conversion: entries are grouped by their base word, the first occurrence of each distinct stripped pronunciation is retained, words are sorted, and surviving variants are numbered from two. Comments, blank lines, and malformed lines are omitted.

Parameters:

input_dict (Path) – Input dictionary (with stress)
output_dict (Path) – Output dictionary (without stress)

Returns:

Tuple of (entries_processed, unique_phones)

Return type:

tuple[int, int]

pstrain.lib.dictionary.strip_phoneset_stress(input_phoneset, output_phoneset)[source]

Strip stress from phoneset file.

Parameters:

input_phoneset (Path) – Input phoneset (with stress)
output_phoneset (Path) – Output phoneset (without stress)

Returns:

Number of unique phones

Return type:

pstrain.lib.dictionary.strip_stress(phone)[source]

Remove ARPABET stress marker (trailing digit) from phone.

Parameters:: phone (str) – Phone possibly with stress marker (e.g., “AH0”, “OW1”)
Returns:: Phone without stress marker (e.g., “AH”, “OW”)
Return type:: str

Examples

strip_stress(“AA1”) -> “AA” strip_stress(“AE0”) -> “AE” strip_stress(“HH”) -> “HH” (no change, consonants don’t have stress)

Phoneset

Phoneset handling and phone mapping utilities.

A Phoneset defines the inventory of valid phones in an acoustic model. Phone mapping functions convert between different phonetic representations.

Phone Mapping

Maps can be loaded from JSON or text files:

JSON format:: {“AA”: “ɑ”, “AE”: “æ”, “_description”: “ignored”}
Text format (-> delimiter):: AA -> ɑ AE -> æ # Comments start with #

Mapping types: - One-to-one: “AA” -> “ɑ” (simple substitution) - Expansion: “CH” -> “t ʃ” (one phone expands to sequence, space-separated) - Deletion: “X” -> “” (phone removed from output)

Mappings are directional (A -> B does not imply B -> A). For bidirectional conversion, provide separate forward and reverse mapping files.

Note: True one-to-many (ambiguous) mappings where one source has multiple possible targets are not supported. Each source phone maps to exactly one target (which may be a sequence or empty).

Stress Removal via Maps

ARPABET stress can be stripped using a map:: AA0 -> AA AA1 -> AA AA2 -> AA …

This is more general than regex-based stripping and works with any phoneset that uses numeric suffixes for stress/tone.

class pstrain.lib.phoneset.Phoneset(phones)[source]

Bases: object

Phone inventory with validation and mapping capabilities.

Supports: - Loading from file (one phone per line) - Extracting from dictionary - Validating dictionaries against phoneset - Phone mapping between phonesets

Parameters:: phones (set[str])

__init__(phones)[source]

Initialize phoneset.

Parameters:: phones (set[str]) – Set of phone strings (UTF-8, case-sensitive)

__len__()[source]

Number of phones in phoneset.

Return type:: int

__repr__()[source]

String representation.

Return type:: str

contains(phone)[source]

Check if phone is in phoneset.

Parameters:: phone (str)
Return type:: bool

create_mapped_phoneset(mapping)[source]

Create new phoneset by mapping all phones.

Parameters:: mapping (dict[str, str]) – Phone mapping dictionary
Returns:: New Phoneset with mapped phones
Return type:: Phoneset

classmethod from_dictionary(dictionary, include_sil=True, silence_phone='SIL')[source]

Extract phoneset from dictionary.

Parameters:

dictionary (Dictionary) – Dictionary to extract phones from
include_sil (bool) – If True, add silence phone
silence_phone (str) – Symbol for silence (default: “SIL”)

Returns:

Phoneset instance

Return type:

Self

classmethod from_file(path)[source]

Load phoneset from file.

File format: - One phone per line - Comments start with # - UTF-8 encoding, case-sensitive

Parameters:: path (Path) – Path to phoneset file
Returns:: Phoneset instance
Raises:: ValueError – If file has encoding or format errors
Return type:: Self

has_sil(silence_phone='SIL')[source]

Check if phoneset includes silence phone.

Parameters:: silence_phone (str)
Return type:: bool

map_phone(phone, mapping, passthrough_unmapped=True)[source]

Map a single phone.

Supports: - One-to-one: “AA” -> “ɑ” produces [“ɑ”] - Expansion: “CH” -> “t ʃ” produces [“t”, “ʃ”] (splits on space) - Deletion: “X” -> “” produces []

Parameters:

phone (str) – Phone to map
mapping (dict[str, str]) – Phone mapping dictionary
passthrough_unmapped (bool) – If True, unmapped phones pass through

Returns:

List of mapped phones (may be empty for deletion, multiple for expansion)

Return type:

map_pronunciation(phones, mapping, passthrough_unmapped=True)[source]

Map a pronunciation (list of phones).

Parameters:

phones (list[str]) – List of phones to map
mapping (dict[str, str]) – Phone mapping dictionary
passthrough_unmapped (bool) – If True, unmapped phones pass through

Returns:

List of mapped phones (flattened)

Return type:

phones()[source]

Get all phones in phoneset.

Return type:: set[str]

to_file(path, silence_phone='SIL')[source]

Save phoneset to file.

Parameters:

path (Path) – Output file path
silence_phone (str) – Silence phone to list first if present

Return type:

None

validate_dictionary(dictionary)[source]

Validate that dictionary phones are in phoneset.

Parameters:: dictionary (Dictionary) – Dictionary to validate
Returns:: Tuple of (is_valid, missing_phones)
Return type:: tuple[bool, set[str]]

pstrain.lib.phoneset.create_stress_strip_map(vowels, stress_markers='012')[source]

Create a phone map that strips stress markers from vowels.

This is a general approach that works with any phoneset using numeric suffixes for stress (like ARPABET) or tone markers.

Parameters:

vowels (set[str]) – Set of base vowel phones (without stress)
stress_markers (str) – Characters used as stress suffixes (default: “012”)

Returns:

{“AA0”: “AA”, “AA1”: “AA”, “AA2”: “AA”, …}

Return type:

Phone map

Example

>>> vowels = {"AA", "AE", "AH", "AO", "AW", "AY", "EH", "ER", "EY",
...           "IH", "IY", "OW", "OY", "UH", "UW"}
>>> stress_map = create_stress_strip_map(vowels)
>>> stress_map["AA0"]
'AA'

pstrain.lib.phoneset.load_phone_map_json(map_file)[source]

Load phone mapping from JSON file.

JSON format:

{
  "_description": "Optional (ignored)",
  "PHONE1": "target1",
  "PHONE2": "target2"
}

Parameters:: map_file (Path) – Path to JSON mapping file
Returns:: Dictionary mapping source phones to target phones
Return type:: dict[str, str]

pstrain.lib.phoneset.load_phone_map_text(map_file)[source]

Load phone mapping from text file.

Format: SOURCE -> TARGET .. rubric:: Example

AA -> ɑ # Comment

Parameters:: map_file (Path) – Path to mapping file
Returns:: Dictionary mapping source phones to target phones
Return type:: dict[str, str]

pstrain.lib.phoneset.reverse_phone_map(phone_map)[source]

Attempt to create reverse mapping from a phone map.

Mappings are directional by design. This utility tries to invert a mapping but may lose information: - Expansion mappings (A -> “x y”) become multiple entries (x -> A, y -> A) - Multiple sources mapping to same target: only last one preserved - Deletions (A -> “”) cannot be reversed

For reliable bidirectional conversion, provide separate forward and reverse mapping files rather than relying on this function.

Parameters:: phone_map (dict[str, str]) – Original phone mapping
Returns:: Best-effort reversed mapping (may be lossy)
Return type:: dict[str, str]

Transcription

Transcription file handling for pstrain projects.

Transcription = word-level text (what was said), not time-aligned. Alignment = time-aligned phone/word boundaries (when each occurs).

pstrain.lib.transcription.get_fileids(transcription_path)[source]

Get list of fileids from transcription file.

Parameters:

transcription_path (Path) – Path to transcription file

Returns:

List of file IDs (utterance identifiers)

Raises:

FileNotFoundError – If transcription file does not exist
UnicodeDecodeError – If file is not UTF-8 encoded

Return type:

pstrain.lib.transcription.parse_transcription_file(transcription_path)[source]

Parse transcription file (word-level text, not time-aligned).

Parameters:

transcription_path (Path) – Path to transcription file

Returns:

Dict mapping fileid -> transcript text (words only)

Raises:

FileNotFoundError – If transcription file does not exist
UnicodeDecodeError – If file is not UTF-8 encoded
ValueError – If a nonempty line does not match a supported format

Return type:

dict[str, str]

Supports two formats: 1. Simple: <fileid> <word1> <word2> ... 2. Sphinx: [<s>] <word1> <word2> [</s>] (<fileid>)

The Sphinx begin/end silence markers are optional, following the convention used by the arpabo language-model tools. A trailing (<fileid>) token selects this form; otherwise the first whitespace token is the file ID.

Models

Base model class and model implementations.

Each model type (CI, CD) is a class that knows about its own training process, parameters, directory structure, and requirements.

Models provide metadata for the pipeline runner: - File paths (inputs, outputs) - Parameters (training settings) - Dependencies (what stages must run first)

Complete-model validation follows native value ranges and enumerations conservatively, but deliberately requires whole-token numeric spellings. Native command-line parsing accepts numeric prefixes and silently truncates fractional spellings for integer options; those lossy spellings are rejected here so a recorded front end cannot describe a value different from the one the native parser actually used.

class pstrain.lib.model.CDModel(config='baseline')[source]

Bases: Model

Context-Dependent (triphone) acoustic model.

Parameters:: config (str)

property default_topn: int: Default top-n Gaussians for this model type.

property display_name: str: Human-readable name for the model type.

classmethod from_string(value)[source]

Get model class from string identifier.

Parameters:: value (str) – Model type string (e.g., “cd”, “CD”, “context-dependent”)
Returns:: Model class (CDModel)
Raises:: ValueError – If model type is unknown
Return type:: type[Model]

get_default_training_params()[source]

Get default training parameters for CD models.

Returns:: Dictionary of parameter names to default values for CD model training
Return type:: dict[str, Any]

get_training_dependencies()[source]

Get list of dependencies required for CD model training.

Returns:: [“ci”, “features”, “dictionary”, “split”] Note: CD models depend on CI models, so “ci” is included
Return type:: List of dependency names

property model_type: str: Model type identifier (e.g., “ci”, “cd”).

class pstrain.lib.model.CIModel(config='baseline')[source]

Bases: Model

Context-Independent (monophone) acoustic model.

Parameters:: config (str)

property default_topn: int: Default top-n Gaussians for this model type.

property display_name: str: Human-readable name for the model type.

classmethod from_string(value)[source]

Get model class from string identifier.

Parameters:: value (str) – Model type string (e.g., “ci”, “CD”, “context-independent”)
Returns:: Model class (CIModel)
Raises:: ValueError – If model type is unknown
Return type:: type[Model]

get_default_training_params()[source]

Get default training parameters for CI models.

Returns:: Dictionary of parameter names to default values for CI model training
Return type:: dict[str, Any]

get_training_dependencies()[source]

Get list of dependencies required for CI model training.

Returns:: [“flat”, “features”, “dictionary”, “split”]
Return type:: List of dependency names

property model_type: str: Model type identifier (e.g., “ci”, “cd”).

class pstrain.lib.model.Model(config='baseline')[source]

Bases: ABC

Base class for acoustic models.

Each model type (CI, CD) should inherit from this class and implement the abstract methods to define its specific behavior.

Parameters:: config (str)

__init__(config='baseline')[source]

Initialize model.

Parameters:: config (str) – Model configuration name (e.g., “baseline”, “1g”, “lda”)
Return type:: None

abstract property default_topn: int: Default top-n Gaussians for this model type.

abstract property display_name: str: Human-readable name for the model type.

abstractmethod classmethod from_string(value)[source]

Get model class from string identifier.

Parameters:: value (str) – Model type string (e.g., “ci”, “CD”, “context-independent”)
Returns:: Model class
Raises:: ValueError – If model type is unknown
Return type:: type[Model]

abstractmethod get_default_training_params()[source]

Get default training parameters for this model type.

Returns:: Dictionary of parameter names to default values
Return type:: dict[str, Any]

get_flat_dir(experiment_dir)[source]

Get the flat model directory for this model.

Parameters:: experiment_dir (str | Path)
Return type:: Path

get_hmm_dir(experiment_dir)[source]

Get the trained HMM model directory for this model.

Parameters:: experiment_dir (str | Path)
Return type:: Path

get_model_dir(experiment_dir)[source]

Get the model directory for this model.

Parameters:: experiment_dir (str | Path) – Experiment directory path
Returns:: {experiment_dir}/models/{model_type}/{config}/model/
Return type:: Path to model directory

abstractmethod get_training_dependencies()[source]

Get list of dependencies required for training.

Returns:: List of dependency names (e.g., [“flat”, “features”, “dictionary”, “split”])
Return type:: list[str]

abstract property model_type: str: Model type identifier (e.g., “ci”, “cd”).

validate_training_params(params)[source]

Validate and normalize training parameters.

Parameters:: params (dict[str, Any]) – Training parameters to validate
Returns:: Validated parameters with defaults filled in
Raises:: ValueError – If parameters are invalid
Return type:: dict[str, Any]

pstrain.lib.model.create_model(model_type, config='baseline')[source]

Create a model instance.

Parameters:

model_type (str) – Model type identifier (e.g., “ci”, “cd”)
config (str) – Model configuration name (default: “baseline”)

Returns:

Model instance of the specified type

Raises:

ValueError – If model type is unknown

Return type:

Model

pstrain.lib.model.get_model_class(model_type)[source]

Get model class from model type string.

Parameters:: model_type (str) – Model type identifier (e.g., “ci”, “cd”)
Returns:: Model class
Raises:: ValueError – If model type is unknown
Return type:: type[Model]

pstrain.lib.model.read_ci_phones(mdef_path)[source]

Read a model’s context-independent phone inventory from its mdef.

The inventory is the set of phones the trained model actually defines. A lexicon pronunciation that uses a phone outside it cannot be loaded: the native lexicon reader drops that pronunciation, and the word then fails much later as an unresolvable transcript token.

Parameters:

mdef_path (str | Path) – Path to a model definition file.

Returns:

The base (context-independent) phone names, in mdef order.

Raises:

FileNotFoundError – If the model definition does not exist.
ValueError – If the file is not a model definition this reader understands.

Return type:

pstrain.lib.model.read_complete_model_feat_params(model_dir)[source]

Require, validate, and return the complete pstrain front-end record.

Range and enumeration rejection is a conservative subset of native rejection. Numeric spelling is deliberately stricter: the complete token must parse without native-style prefix acceptance or integer truncation. Native parsing remains authoritative otherwise. This does not prove compatibility between the record and the model’s binary tensors.

Parameters:: model_dir (str | Path)
Return type:: dict[str, str]

pstrain.lib.model.require_complete_model(model_dir)[source]

Require and validate the complete pstrain front-end record.

Parameters:: model_dir (str | Path)
Return type:: Path

Native Worker

Crash containment for the guarded native operations. See Native boundary: what is contained, and what is not for which operations are guarded, which are not, and what each exception means.

Crash containment for the contained-all-operations native surface.

One-shot wrappers execute as complete Python/CFFI operations in the helper; BW, alignment, feature extraction, and logmath use opaque remote objects. See docs/design/native-boundary.md for the phase contract.

Each Python process owns at most one lazily spawned helper process, created through an explicit spawn context and reused across calls. The helper services one request at a time; the owning process classifies every outcome (clean result, native error return, signal death, nonzero exit, clean exit mid-request, transport EOF) into the exception hierarchy below.

Diagnostics are captured worker-side: the helper redirects its own stderr – where every E_INFO/E_ERROR/E_FATAL line lands – into a private temporary file, and the owner attaches the tail of that file to the raised exception.

pstrain.lib.native_worker.GUARDED_OPERATIONS = frozenset({'make_quests', 'mdef_gen_ci', 'object_call', 'object_close', 'object_create', 'prune_tree', 'python_call', 'stdout_redirect', 'stdout_restore'}): Worker protocol operations accepted from containment wrappers and proxies.

class pstrain.lib.native_worker.NativeObjectProxy(module_name, target_name, args, kwargs, inputs=())[source]

Bases: object

Proxy a stateful CFFI-backed Python object living in the helper.

Parameters:

module_name (str)
target_name (str)
args (tuple[Any, ...])
kwargs (dict[str, Any])
inputs (tuple[Path | str, ...])

__init__(module_name, target_name, args, kwargs, inputs=())[source]

Parameters:

module_name (str)
target_name (str)
args (tuple[Any, ...])
kwargs (dict[str, Any])
inputs (tuple[Path | str, ...])

Return type:

None

call(method_name, *args, **kwargs)[source]

Parameters:

method_name (str)
args (Any)
kwargs (Any)

Return type:

Any

close()[source]

Return type:: None

exception pstrain.lib.native_worker.PstrainError[source]

Bases: RuntimeError

Base class for public pstrain failures.

exception pstrain.lib.native_worker.PstrainInvalidInputError(operation, input_paths, diagnostic='', returncode=None)[source]

Bases: PstrainNativeError

Python-side validation rejected the request before it reached a worker.

Parameters:

operation (str)
input_paths (tuple[str, ...])
diagnostic (str)
returncode (int | None)

Return type:

None

exception pstrain.lib.native_worker.PstrainNativeCrashError(*args, signal, **kwargs)[source]

Bases: PstrainNativeError

The native worker died from a signal.

Parameters:

args (Any)
signal (int)
kwargs (Any)

Return type:

None

__init__(*args, signal, **kwargs)[source]

Parameters:

args (Any)
signal (int)
kwargs (Any)

Return type:

None

exception pstrain.lib.native_worker.PstrainNativeError(operation, input_paths, diagnostic='', returncode=None)[source]

Bases: PstrainError

A contained native operation failed.

Parameters:

operation (str)
input_paths (tuple[str, ...])
diagnostic (str)
returncode (int | None)

Return type:

None

__init__(operation, input_paths, diagnostic='', returncode=None)[source]

Parameters:

operation (str)
input_paths (tuple[str, ...])
diagnostic (str)
returncode (int | None)

Return type:

None

__reduce__()[source]

Preserve constructor-independent exception state when pickled.

Return type:: tuple[object, tuple[object, …]]

exception pstrain.lib.native_worker.PstrainNativeFatalError(operation, input_paths, diagnostic='', returncode=None)[source]

Bases: PstrainNativeError

The native worker exited nonzero or returned a diagnosed failure.

Parameters:

operation (str)
input_paths (tuple[str, ...])
diagnostic (str)
returncode (int | None)

Return type:

None

exception pstrain.lib.native_worker.PstrainWorkerError[source]

Bases: PstrainError

The contained worker or process-pool infrastructure is unavailable.

exception pstrain.lib.native_worker.PstrainWorkerProtocolError(operation, input_paths, diagnostic='', returncode=None)[source]

Bases: PstrainNativeError

The worker violated the request/response protocol.

Raised in particular when the helper exits cleanly (status 0) with a request still outstanding: a successful-looking exit is not a result.

Parameters:

operation (str)
input_paths (tuple[str, ...])
diagnostic (str)
returncode (int | None)

Return type:

None

pstrain.lib.native_worker.call(operation, arguments, inputs)[source]

Execute one guarded native operation in this process’s helper.

Parameters:

operation (str) – One of GUARDED_OPERATIONS.
arguments (tuple[Any, ...]) – Picklable positional arguments for the native entry point.
inputs (tuple[Path | str, ...]) – Input paths, recorded on any raised exception.

Raises:

PstrainInvalidInputError – The operation is not routed through the helper. Requests whose serialized representation exceeds 64 KiB are also rejected before anything is sent.
PstrainWorkerError – The helper could not be started, or timed out.
PstrainNativeCrashError – The helper died on a signal.
PstrainWorkerProtocolError – The helper exited cleanly mid-request.
PstrainNativeFatalError – The operation failed with a diagnostic.
PstrainNativeError – The operation failed without one.

Return type:

Any

pstrain.lib.native_worker.call_python(module_name, target_name, args, kwargs, inputs=())[source]

Run a complete Python/CFFI operation in the contained helper.

Parameters:

module_name (str)
target_name (str)
args (tuple[Any, ...])
kwargs (dict[str, Any])
inputs (tuple[Path | str, ...])

Return type:

Any

pstrain.lib.native_worker.close_helper_before_children_are_joined()[source]

Arrange for this process’s helper to be closed early enough at exit.

Call in any process that multiprocessing itself started – a pool worker, or any other BaseProcess – before it can spawn a helper. Calling it more than once in the same process does nothing further.

BaseProcess._bootstrap runs multiprocessing.util._exit_function when the process body returns, and that joins every non-daemon child with no timeout. The helper is such a child, and it waits on its request pipe until it is told to exit, so a process that leaves one running parks in waitpid forever, and whatever is waiting on that process parks with it. Registering _shutdown with atexit() is too late: ordinary atexit handlers do not run until sys.exit afterwards. A multiprocessing.util.Finalize runs inside _exit_function itself, ahead of the joins, which is early enough.

Return type:: None

pstrain.lib.native_worker.contained(function)[source]

Route a complete stateless public CFFI operation through the helper.

Parameters:: function (Callable[[~P], R])
Return type:: Callable[[~P], R]

pstrain.lib.native_worker.in_worker()[source]

Return whether code is executing inside the contained native helper.

Return type:: bool

Guarded Logmath Wrapper

LogMath is the supported wrapper for native log-domain arithmetic. Other contents of the private _pstrainc module, including raw library handles and symbols, are implementation details and are not supported application APIs.

Low-level CFFI bindings to libpstrainc.

The supported model-I/O and logmath operations exported here are coarse operations routed through the persistent native worker. get_lib() and dynamic raw-symbol access are private implementation/testing escape hatches; applications should use the higher-level wrappers in pstrain.lib.

Implementation is organized into submodules: - _cffi.cdef: C type definitions - _cffi.core: FFI initialization and helpers - _cffi.io: Model file I/O - _cffi.logmath: Log-domain math wrapper

class pstrain.lib._pstrainc.LogMath(base=1.0001, shift=0, use_table=True)[source]

Bases: object

Log-domain math using C library for numerical stability.

This wraps the C logmath functions which use integer log representations for fast, stable computation of probabilities in log domain.

Parameters:

base (float)
shift (int)
use_table (bool)

__del__()[source]

Free C resources.

Return type:: None

__init__(base=1.0001, shift=0, use_table=True)[source]

Initialize log math.

Parameters:

base (float) – Log base (default 1.0001 for high precision)
shift (int) – Bit shift for table lookup
use_table (bool) – Whether to use lookup table for speed

Return type:

None

add(logp, logq)[source]

Add two probabilities in log domain.

Computes log(exp(logp) + exp(logq)) efficiently.

Parameters:

logp (int) – First log probability
logq (int) – Second log probability

Returns:

log(p + q)

Return type:

property base: float: Get the log base.

exp(logp)[source]

Convert log-domain value back to probability.

Parameters:: logp (int) – Log-domain integer
Returns:: Probability value
Return type:: float

get_base()[source]

Return the base for worker-side property access.

Return type:: float

log(p)[source]

Convert probability to log domain.

Parameters:: p (float) – Probability value (0 < p <= 1)
Returns:: Log-domain integer representation
Return type:: int

pstrain.lib._pstrainc.get_ffi()[source]

Get the FFI instance.

Return type:: FFI

pstrain.lib._pstrainc.get_lib()[source]

Get the loaded library with all C functions.

Return type:: Any

pstrain.lib._pstrainc.path_or_null(path)[source]

Convert a path to bytes, or return NULL if None.

Helper to reduce boilerplate when passing optional path arguments to C.

Parameters:: path (Path | str | None) – Path to encode, or None
Returns:: Encoded path bytes, or ffi.NULL if path is None
Return type:: bytes | Any

pstrain.lib._pstrainc.read_dnom(filename)[source]

Read Gaussian density counts from S3 format file.

Parameters:: filename (str) – Input file path
Returns:: Tuple of (dnom_array, n_cb, n_feat, n_density) dnom_array has shape (n_cb, n_feat, n_density)
Raises:: RuntimeError – If file cannot be read
Return type:: tuple[npt.NDArray[np.float32], int, int, int]

pstrain.lib._pstrainc.read_gau(filename)[source]

Read Gaussian parameters from S3 format file.

Parameters:: filename (str) – Input file path
Returns:: Tuple of (gau_array, n_mgau, n_feat, n_density, veclen_list) gau_array has shape (n_mgau, n_feat, n_density, max_veclen)
Raises:: RuntimeError – If file cannot be read
Return type:: tuple[npt.NDArray[np.float32], int, int, int, list[int]]

pstrain.lib._pstrainc.read_mixw(filename)[source]

Read mixture weights as probabilities, normalizing occupied rows.

Parameters:: filename (str)
Return type:: tuple[npt.NDArray[np.float32], int, int, int]

pstrain.lib._pstrainc.read_mixw_counts(filename)[source]

Read stored mixture-weight values without normalization.

Parameters:: filename (str) – Input file path
Returns:: Tuple of (mixw_array, n_mixw, n_feat, n_density) mixw_array has shape (n_mixw, n_feat, n_density)
Raises:: RuntimeError – If file cannot be read
Return type:: tuple[npt.NDArray[np.float32], int, int, int]

pstrain.lib._pstrainc.read_tmat(filename)[source]

Read transition matrices as probabilities, normalizing occupied rows.

Parameters:: filename (str)
Return type:: tuple[npt.NDArray[np.float32], int, int]

pstrain.lib._pstrainc.read_tmat_counts(filename)[source]

Read stored transition-matrix values without normalization.

Parameters:: filename (str) – Input file path
Returns:: Tuple of (tmat_array, n_tmat, n_state) tmat_array has shape (n_tmat, n_state-1, n_state)
Raises:: RuntimeError – If file cannot be read
Return type:: tuple[npt.NDArray[np.float32], int, int]

pstrain.lib._pstrainc.write_dnom(filename, dnom)[source]

Write Gaussian density counts to S3 format file.

These counts track how often each Gaussian is used during BW training. Used by inc_comp to decide which Gaussians to split.

Parameters:

filename (str) – Output file path
dnom (npt.NDArray[np.float32]) – Density counts array of shape (n_cb, n_feat, n_density) or (n_cb, n_density) which will be reshaped

Returns:

0 on success, non-zero on error

Return type:

pstrain.lib._pstrainc.write_gau(filename, gau)[source]

Write Gaussian parameters (means or variances) to S3 format file.

Parameters:

filename (str) – Output file path
gau (npt.NDArray[np.float32]) – Gaussian array of shape (n_mgau, n_feat, n_density, veclen) or (n_mgau, n_density, veclen) which will be reshaped

Returns:

0 on success, non-zero on error

Return type:

pstrain.lib._pstrainc.write_mixw(filename, mixw)[source]

Write mixture weights to S3 format file.

Parameters:

filename (str) – Output file path
mixw (npt.NDArray[np.float32]) – Mixture weights array of shape (n_mixw, n_feat, n_density) or (n_mixw, n_density) which will be reshaped

Returns:

0 on success, non-zero on error

Return type:

pstrain.lib._pstrainc.write_tmat(filename, tmat)[source]

Write transition matrices to S3 format file.

Parameters:

filename (str) – Output file path
tmat (npt.NDArray[np.float32]) – Transition matrices, either square (n_tmat, n_state, n_state) including an exit-state row, or stored rectangular shape (n_tmat, n_state-1, n_state). The exit-state row, when present, is excluded; values are stored without row normalization.

Returns:

0 on success, non-zero on error

Return type: