Pipelines

pipeline

Expose the three independent project execution pipelines.

The full workflow is deliberately implemented as composition of the typo- generation-only and AutoCorrect2-only workflows. Neither subsystem depends on the other, which keeps generation policy separate from AutoHotkey source inspection and conflict detection.

GENERATED_CANDIDATE_OPTIONS module-attribute

GENERATED_CANDIDATE_OPTIONS: Final[HotstringOptions] = (
    HotstringOptions("B0X")
)

Options assigned when the full pipeline converts typo mappings to hotstrings.

FullPipelineResult dataclass

Represent the two results produced by the composed full pipeline.

Attributes:

Name	Type	Description
typo_generation	TypoGenerationResult	Result of typo generation and internal ambiguity filtering.
autocorrect2	AutoCorrect2CheckResult	Result of checking the surviving candidates against AutoCorrect2.

Source code in hotstring\pipeline.py

@dataclass(frozen=True, slots=True)
class FullPipelineResult:
    """Represent the two results produced by the composed full pipeline.

    Attributes:
        typo_generation:
            Result of typo generation and internal ambiguity filtering.
        autocorrect2:
            Result of checking the surviving candidates against AutoCorrect2.
    """

    typo_generation: TypoGenerationResult
    autocorrect2: AutoCorrect2CheckResult

run_typo_generation

run_typo_generation(
    word_list: Sequence[str],
    tasks: Sequence[TypoGenerationTask],
    config: TypoGenerationConfig,
    *,
    n_workers: int | None = None,
    report_path: Path | None = None,
    logger: Logger | None = None,
) -> TypoGenerationResult

Run typo generation and internal ambiguity filtering only.

Parameters:

Name	Type	Description	Default
word_list	Sequence[str]	Source words to corrupt.	required
tasks	Sequence[TypoGenerationTask]	Ordered typo-generation tasks to execute.	required
config	TypoGenerationConfig	Shared MULTYPO generator configuration.	required
n_workers	int | None	Optional process-pool size. None uses the executor default when parallel execution is selected.	None
report_path	Path | None	Optional destination for a typo-generation-only report.	None
logger	Logger | None	Optional orchestration logger forwarded to the execution layer.	None

Returns:

Type	Description
TypoGenerationResult	Aggregated typo-generation result containing valid mappings and
TypoGenerationResult	internally ambiguous noisy forms.

Raises:

Type	Description
TypeError	If task/execution inputs have invalid types.
ValueError	If no generation tasks are supplied or a generation setting is invalid.
RuntimeError	If a parallel generation task fails.
OSError	If a requested report cannot be written.

Source code in hotstring\pipeline.py

def run_typo_generation(
    word_list: Sequence[str],
    tasks: Sequence[TypoGenerationTask],
    config: TypoGenerationConfig,
    *,
    n_workers: int | None = None,
    report_path: Path | None = None,
    logger: logging.Logger | None = None,
) -> TypoGenerationResult:
    """Run typo generation and internal ambiguity filtering only.

    Args:
        word_list:
            Source words to corrupt.
        tasks:
            Ordered typo-generation tasks to execute.
        config:
            Shared MULTYPO generator configuration.
        n_workers:
            Optional process-pool size. `None` uses the executor default when
            parallel execution is selected.
        report_path:
            Optional destination for a typo-generation-only report.
        logger:
            Optional orchestration logger forwarded to the execution layer.

    Returns:
        Aggregated typo-generation result containing valid mappings and
        internally ambiguous noisy forms.

    Raises:
        TypeError:
            If task/execution inputs have invalid types.
        ValueError:
            If no generation tasks are supplied or a generation setting is
            invalid.
        RuntimeError:
            If a parallel generation task fails.
        OSError:
            If a requested report cannot be written.
    """
    words = list(word_list)
    task_tuple = tuple(tasks)
    samples = execute_typo_generation_tasks(
        words,
        task_tuple,
        config,
        n_workers=n_workers,
        logger=logger,
    )
    result = aggregate_typo_samples(
        samples,
        config=config,
        tasks=task_tuple,
        source_word_count=len(words),
    )

    if report_path is not None:
        _write_report(
            report_path,
            "TYPO GENERATION REPORT",
            create_typo_generation_report(result),
        )

    return result

run_autocorrect2_check

run_autocorrect2_check(
    candidates: Sequence[AutoCorrect2CandidateHotstring],
    *,
    project_dir: Path,
    report_path: Path | None = None,
    write_accepted: bool = False,
) -> AutoCorrect2CheckResult

Check manually supplied candidates against active AutoCorrect2 hotstrings.

Trigger conflict detection supports every combination of the recognition options currently modeled by the project: ending-character-free matching (*), inside-word matching (?), and case-sensitive matching (C). No candidate is rejected merely for using one of those semantics.

Parameters:

Name	Type	Description	Default
candidates	Sequence[AutoCorrect2CandidateHotstring]	AutoCorrect2 candidates supplied directly by the caller.	required
project_dir	Path	AutoCorrect2 project directory containing the configured source files.	required
report_path	Path | None	Optional destination for an AutoCorrect2-only report.	None
write_accepted	bool	Append accepted candidates to the generated include file when True.	False

Returns:

Type	Description
AutoCorrect2CheckResult	Conflict-check result partitioning candidates into accepted and
AutoCorrect2CheckResult	rejected groups.

Raises:

Type	Description
FileNotFoundError	If a required AutoCorrect2 source file is missing.
UnicodeDecodeError	If a source requiring parsing is not valid UTF-8 text.
ValueError	If source data is invalid or writing is requested for a candidate that violates the writer's explicit B0X contract.
OSError	If authoritative sources, reports, or generated files cannot be read or written.

Source code in hotstring\pipeline.py

def run_autocorrect2_check(
    candidates: Sequence[AutoCorrect2CandidateHotstring],
    *,
    project_dir: Path,
    report_path: Path | None = None,
    write_accepted: bool = False,
) -> AutoCorrect2CheckResult:
    """Check manually supplied candidates against active AutoCorrect2 hotstrings.

    Trigger conflict detection supports every combination of the recognition
    options currently modeled by the project: ending-character-free matching
    (`*`), inside-word matching (`?`), and case-sensitive matching (`C`). No
    candidate is rejected merely for using one of those semantics.

    Args:
        candidates:
            AutoCorrect2 candidates supplied directly by the caller.
        project_dir:
            AutoCorrect2 project directory containing the configured source
            files.
        report_path:
            Optional destination for an AutoCorrect2-only report.
        write_accepted:
            Append accepted candidates to the generated include file when
            `True`.

    Returns:
        Conflict-check result partitioning candidates into accepted and
        rejected groups.

    Raises:
        FileNotFoundError:
            If a required AutoCorrect2 source file is missing.
        UnicodeDecodeError:
            If a source requiring parsing is not valid UTF-8 text.
        ValueError:
            If source data is invalid or writing is requested for a candidate
            that violates the writer's explicit `B0X` contract.
        OSError:
            If authoritative sources, reports, or generated files cannot be
            read or written.
    """
    candidate_tuple = tuple(candidates)
    existing_hotstrings = load_existing_hotstrings(project_dir)
    assessments = assess_candidates(candidate_tuple, existing_hotstrings)

    accepted: list[AutoCorrect2CandidateHotstring] = []
    rejected = []
    for candidate, assessment in zip(candidate_tuple, assessments, strict=True):
        if assessment.is_accepted:
            accepted.append(candidate)
        else:
            rejected.append(assessment)

    result = AutoCorrect2CheckResult(
        accepted=tuple(accepted),
        rejected=tuple(rejected),
    )

    if write_accepted:
        append_candidates(result.accepted, project_dir=project_dir)

    if report_path is not None:
        _write_report(
            report_path,
            "AUTOCORRECT2 CONFLICT CHECK REPORT",
            create_autocorrect2_report(result),
        )

    return result

run_full_pipeline

run_full_pipeline(
    word_list: Sequence[str],
    tasks: Sequence[TypoGenerationTask],
    config: TypoGenerationConfig,
    *,
    project_dir: Path,
    n_workers: int | None = None,
    report_path: Path | None = None,
    write_accepted: bool = False,
    logger: Logger | None = None,
) -> FullPipelineResult

Run typo generation followed by AutoCorrect2 conflict checking.

The function composes the two independent stage pipelines without asking either stage to emit its own report. Surviving typo mappings are converted to AutoCorrect2CandidateHotstring objects with the project's fixed generated B0X option set.

Parameters:

Name	Type	Description	Default
word_list	Sequence[str]	Source words to corrupt.	required
tasks	Sequence[TypoGenerationTask]	Ordered typo-generation tasks to execute.	required
config	TypoGenerationConfig	Shared MULTYPO generator configuration.	required
project_dir	Path	AutoCorrect2 project directory.	required
n_workers	int | None	Optional process-pool size.	None
report_path	Path | None	Optional destination for the combined report.	None
write_accepted	bool	Append final accepted candidates to the generated include file.	False
logger	Logger | None	Optional typo-generation orchestration logger.	None

Returns:

Type	Description
FullPipelineResult	Combined result containing both stage results.

Raises:

Type	Description
TypeError	If generation or source-loading inputs have invalid types.
ValueError	If generation/source data is invalid or a writable candidate violates the generated-file contract.
FileNotFoundError	If a required AutoCorrect2 source file is missing.
UnicodeDecodeError	If an authoritative source requiring parsing is not valid UTF-8.
RuntimeError	If a parallel generation task fails.
OSError	If source, report, or generated files cannot be read or written.

Source code in hotstring\pipeline.py

def run_full_pipeline(
    word_list: Sequence[str],
    tasks: Sequence[TypoGenerationTask],
    config: TypoGenerationConfig,
    *,
    project_dir: Path,
    n_workers: int | None = None,
    report_path: Path | None = None,
    write_accepted: bool = False,
    logger: logging.Logger | None = None,
) -> FullPipelineResult:
    """Run typo generation followed by AutoCorrect2 conflict checking.

    The function composes the two independent stage pipelines without asking
    either stage to emit its own report. Surviving typo mappings are converted
    to [`AutoCorrect2CandidateHotstring`]
    [hotstring.autocorrect2.models.AutoCorrect2CandidateHotstring] objects with
    the project's fixed generated `B0X` option set.

    Args:
        word_list:
            Source words to corrupt.
        tasks:
            Ordered typo-generation tasks to execute.
        config:
            Shared MULTYPO generator configuration.
        project_dir:
            AutoCorrect2 project directory.
        n_workers:
            Optional process-pool size.
        report_path:
            Optional destination for the combined report.
        write_accepted:
            Append final accepted candidates to the generated include file.
        logger:
            Optional typo-generation orchestration logger.

    Returns:
        Combined result containing both stage results.

    Raises:
        TypeError:
            If generation or source-loading inputs have invalid types.
        ValueError:
            If generation/source data is invalid or a writable candidate
            violates the generated-file contract.
        FileNotFoundError:
            If a required AutoCorrect2 source file is missing.
        UnicodeDecodeError:
            If an authoritative source requiring parsing is not valid UTF-8.
        RuntimeError:
            If a parallel generation task fails.
        OSError:
            If source, report, or generated files cannot be read or written.
    """
    typo_result = run_typo_generation(
        word_list,
        tasks,
        config,
        n_workers=n_workers,
        report_path=None,
        logger=logger,
    )

    candidates = tuple(
        AutoCorrect2CandidateHotstring(
            trigger=noisy_word,
            options_input=GENERATED_CANDIDATE_OPTIONS,
            replacement=target_word,
        )
        for noisy_word, target_word in typo_result.candidates.items()
    )

    autocorrect2_result = run_autocorrect2_check(
        candidates,
        project_dir=project_dir,
        report_path=None,
        write_accepted=write_accepted,
    )
    result = FullPipelineResult(
        typo_generation=typo_result,
        autocorrect2=autocorrect2_result,
    )

    if report_path is not None:
        _write_report(
            report_path,
            "AUTOHOTKEY HOTSTRING GENERATION REPORT",
            create_full_pipeline_report(typo_result, autocorrect2_result),
        )

    return result

_write_report

_write_report(
    path: Path, title: str, body_lines: Sequence[str]
) -> None

Build and write one complete report document.

Parameters:

Name	Type	Description	Default
path	Path	Destination report file.	required
title	str	Top-level report title.	required
body_lines	Sequence[str]	Pipeline-specific report body lines.	required

Raises:

Type	Description
ValueError	If the report title is empty.
OSError	If the report cannot be written.

Source code in hotstring\pipeline.py

def _write_report(path: Path, title: str, body_lines: Sequence[str]) -> None:
    """Build and write one complete report document.

    Args:
        path:
            Destination report file.
        title:
            Top-level report title.
        body_lines:
            Pipeline-specific report body lines.

    Raises:
        ValueError:
            If the report title is empty.
        OSError:
            If the report cannot be written.
    """
    document_lines = build_report_document(title, body_lines)
    write_text(path, "\n".join(document_lines) + "\n")