Training Steps API

Public API for training steps.

This module exposes step functions from lib.steps through the public API. CLI and web clients should use these functions, not call lib.steps directly.

Steps can be run via CLI: pstrain step ci_hmm [args]

class pstrain.api.steps.CDHMMUntiedStep[source]

Bases: Step

CD HMM untied training step.

add_arguments(parser)[source]

Add CD HMM untied specific arguments.

Parameters:: parser (ArgumentParser)
Return type:: None

default_params: dict[str, Any] = {}

description: str = 'Train context-dependent untied HMM models using Baum-Welch'

execute(ctx, **params)[source]

Execute CD HMM untied training by delegating to the pipeline runner.

Upstream tasks (flat, ci-1g, cd-untied-init) will run automatically if their outputs are stale. Use pstrain build cd-untied from the CLI.

Parameters:

ctx (StepContext)
params (Any)

Return type:

int

get_inputs(ctx)[source]

Get input files for CD HMM untied training.

Parameters:: ctx (StepContext)
Return type:: list[Path]

get_outputs(ctx)[source]

Get output files from CD HMM untied training.

Parameters:: ctx (StepContext)
Return type:: list[Path]

name: str = 'cd_hmm_untied'

script: str = 'bw'

class pstrain.api.steps.CIHMMStep[source]

Bases: Step

CI HMM training step.

add_arguments(parser)[source]

Add CI HMM specific arguments.

Parameters:: parser (ArgumentParser)
Return type:: None

default_params: dict[str, Any] = {}

description: str = 'Train context-independent HMM models using Baum-Welch'

execute(ctx, **params)[source]

Execute CI HMM training by delegating to the pipeline runner.

The pipeline does the right thing for stale dependencies: if flat or features are missing/older than their inputs, those tasks will run too. Use pstrain build ci-1g for the same effect from the CLI.

Parameters:

ctx (StepContext)
params (Any)

Return type:

int

get_inputs(ctx)[source]

Get input files for CI HMM training.

Parameters:: ctx (StepContext)
Return type:: list[Path]

get_outputs(ctx)[source]

Get output files from CI HMM training.

Parameters:: ctx (StepContext)
Return type:: list[Path]

get_params_from_args(args)[source]

Extract training parameters from args.

Parameters:: args (Namespace)
Return type:: dict[str, Any]

name: str = 'ci_hmm'

script: str = 'bw'

class pstrain.api.steps.FeaturesStep[source]

Bases: Step

Feature extraction step.

add_arguments(parser)[source]

Add feature extraction specific arguments.

Parameters:: parser (ArgumentParser)
Return type:: None

default_params: dict[str, Any] = {'agc': 'none', 'alpha': 0.97, 'cmn': 'batch', 'cmninit': '40,3,-1', 'dither': True, 'frate': 100, 'lifter': 22, 'lowerf': 130.0, 'ncep': 13, 'nfft': 512, 'nfilt': 25, 'remove_dc': True, 'remove_noise': True, 'samprate': 16000, 'seed': -1, 'transform': 'dct', 'upperf': 6800.0, 'varnorm': 'no', 'wlen': 0.025625}

description: str = 'Extract acoustic features from audio using sphinx_fe'

execute(ctx, **params)[source]

Execute feature extraction via the pipeline runner.

This is the single-step variant of pstrain build features. It builds a pipeline scoped to the current project/experiment/config and runs the “features” target.

Parameters:

ctx (StepContext)
params (Any)

Return type:

int

get_inputs(ctx)[source]

Get input files for feature extraction.

Parameters:: ctx (StepContext)
Return type:: list[Path]

get_outputs(ctx)[source]

Get output files from feature extraction.

Parameters:: ctx (StepContext)
Return type:: list[Path]

name: str = 'features'

script: str = 'sphinx_fe'

class pstrain.api.steps.Step[source]

Bases: ABC

Base class for training steps.

Subclasses must implement: - name, description, script (class attributes) - get_inputs(), get_outputs() - return file paths - execute() - perform the step

add_arguments(parser)[source]

Add step-specific arguments to parser.

Override in subclasses to add custom arguments.

Parameters:: parser (ArgumentParser)
Return type:: None

default_params: dict[str, Any] = {}

description: str = ''

abstractmethod execute(ctx, **params)[source]

Execute the step.

This is the single code path - ctx methods emit shell in dry-run, execute otherwise.

Parameters:

ctx (StepContext) – Step context (with dry_run support)
**params (Any) – Step parameters

Returns:

Exit code (0 for success)

Return type:

int

get_definition(ctx, **params)[source]

Get this step’s declarative definition.

Parameters:

ctx (StepContext) – Step context
**params (Any) – Parameter overrides

Returns:

StepDefinition with inputs, outputs, params, and script.

Return type:

StepDefinition

abstractmethod get_inputs(ctx)[source]

Get input file paths.

Parameters:: ctx (StepContext) – Step context with project/experiment info
Returns:: List of input file paths (must exist before step runs)
Return type:: list[Path]

abstractmethod get_outputs(ctx)[source]

Get output file paths.

Parameters:: ctx (StepContext) – Step context with project/experiment info
Returns:: List of output file paths (created by this step)
Return type:: list[Path]

get_params(ctx, **overrides)[source]

Get step parameters.

Parameters:

ctx (StepContext) – Step context
**overrides (Any) – Parameter overrides

Returns:

Merged parameters

Return type:

dict[str, Any]

get_params_from_args(args)[source]

Extract step parameters from parsed args.

Override in subclasses to handle custom arguments.

Parameters:: args (Namespace)
Return type:: dict[str, Any]

main(args=None)[source]

CLI entry point.

Parameters:: args (list[str] | None) – Command line arguments (defaults to sys.argv[1:])
Returns:: Exit code
Return type:: int

name: str = ''

run(project_dir, experiment='default', config='baseline', dry_run=False, **params)[source]

Run the step (convenience wrapper).

Parameters:

project_dir (Path | str) – Project directory
experiment (str) – Experiment name
config (str) – Model configuration name
dry_run (bool) – If True, emit shell commands instead of executing
**params (Any) – Step parameters

Returns:

Exit code

Return type:

int

script: str = ''

to_dict(ctx, **params)[source]

Get this step’s definition as a dict.

Parameters:

ctx (StepContext) – Step context
**params (Any) – Parameter overrides

Returns:

Dict with rule definition

Return type:

dict[str, Any]

class pstrain.api.steps.StepContext(project_dir, experiment='default', config='baseline', dry_run=False, _header_emitted=False)[source]

Bases: object

Context for step execution.

Similar to CLI’s CommandContext but for library-level steps. Supports dry-run mode where actions emit shell instead of executing.

Parameters:

project_dir (Path)
experiment (str)
config (str)
dry_run (bool)
_header_emitted (bool)

__init__(project_dir, experiment='default', config='baseline', dry_run=False, _header_emitted=False)

Parameters:

project_dir (Path)
experiment (str)
config (str)
dry_run (bool)
_header_emitted (bool)

Return type:

None

blank()[source]

Add blank line.

Return type:: None

comment(text)[source]

Add a comment.

Parameters:: text (str)
Return type:: None

config: str = 'baseline'

copy(src, dst)[source]

Copy file.

Parameters:

src (Path)
dst (Path)

Return type:

None

dry_run: bool = False

property etc_dir: Path: Get etc directory.

experiment: str = 'default'

property experiment_dir: Path: Get experiment directory.

flat_dir(model_type='ci')[source]

Get flat model directory.

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

hmm_dir(model_type='ci')[source]

Get HMM model directory.

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

log(message)[source]

Log a message.

Parameters:: message (str)
Return type:: None

log_comment(text)[source]

Alias for comment.

Parameters:: text (str)
Return type:: None

mkdir(path, parents=True)[source]

Create directory.

Parameters:

path (Path)
parents (bool)

Return type:

None

model_dir(model_type)[source]

Get model directory for a model type and config.

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

project_dir: Path

run_cmd(program, *args, **kwargs)[source]

Run a command.

Parameters:

program (str)
args (str)
kwargs (Any)

Return type:

int

property shared_dir: Path: Get shared directory.

symlink(src, dst)[source]

Create symlink.

Parameters:

src (Path)
dst (Path)

Return type:

None

class pstrain.api.steps.StepDefinition(name, description, inputs, outputs, params, script)[source]

Bases: object

Declarative rule definition for a step (inputs/outputs/params/script).

Parameters:

name (str)
description (str)
inputs (list[str])
outputs (list[str])
params (dict[str, Any])
script (str)

__init__(name, description, inputs, outputs, params, script)

Parameters:

name (str)
description (str)
inputs (list[str])
outputs (list[str])
params (dict[str, Any])
script (str)

Return type:

None

description: str

inputs: list[str]

name: str

outputs: list[str]

params: dict[str, Any]

script: str

pstrain.api.steps.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:

Path

pstrain.api.steps.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:

int

pstrain.api.steps.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:

int

pstrain.api.steps.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:

int

pstrain.api.steps.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:

dict[str, Any]

pstrain.api.steps.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:

dict[str, Any]

pstrain.api.steps.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:

dict[str, Any]

pstrain.api.steps.features_step: Public feature-extraction step instance.

pstrain.api.steps.ci_hmm_step: Public context-independent HMM training step instance.

pstrain.api.steps.cd_hmm_untied_step: Public untied context-dependent HMM training step instance.