Skip to content

Command execution

commands

Dispatch validated CLI commands to the public pipeline APIs.

The runtime layer resolves all user input before this module is called. Each handler therefore performs only workflow dispatch and concise terminal reporting; report generation and optional writes remain owned by hotstring.pipeline.

execute_command

execute_command(
    command: CommandConfig,
    *,
    logger: Logger | None = None,
    output: TextIO | None = None,
) -> int

Execute one validated command and print its summary.

Parameters:

Name Type Description Default
command CommandConfig

Runtime command created by create_command_config().

required
logger Logger | None

Optional orchestration logger forwarded to typo-generation pipelines.

None
output TextIO | None

Optional output stream. None uses the current sys.stdout.

None

Returns:

Type Description
int

Zero after successful pipeline execution.

Source code in hotstring\cli\commands.py
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def execute_command(
    command: CommandConfig,
    *,
    logger: logging.Logger | None = None,
    output: TextIO | None = None,
) -> int:
    """Execute one validated command and print its summary.

    Args:
        command:
            Runtime command created by
            [`create_command_config()`][hotstring.cli.runtime.create_command_config].
        logger:
            Optional orchestration logger forwarded to typo-generation
            pipelines.
        output:
            Optional output stream. `None` uses the current `sys.stdout`.

    Returns:
        Zero after successful pipeline execution.
    """
    stream = sys.stdout if output is None else output

    if isinstance(command, TypoGenerationCommand):
        generation = command.generation
        result = run_typo_generation(
            generation.words,
            generation.tasks,
            generation.config,
            n_workers=generation.n_workers,
            report_path=command.report_path,
            logger=logger,
        )
        _print_typo_summary(result, output=stream)
        _print_report_path(command.report_path, output=stream)
        return 0

    if isinstance(command, AutoCorrect2CheckCommand):
        autocorrect2 = command.autocorrect2
        result = run_autocorrect2_check(
            command.candidates,
            project_dir=autocorrect2.project_dir,
            report_path=command.report_path,
            write_accepted=autocorrect2.write_accepted,
        )
        _print_autocorrect2_summary(result, output=stream)
        _print_write_destination(
            write_accepted=autocorrect2.write_accepted,
            project_dir=autocorrect2.project_dir,
            output=stream,
        )
        _print_report_path(command.report_path, output=stream)
        return 0

    if isinstance(command, FullPipelineCommand):
        generation = command.generation
        autocorrect2 = command.autocorrect2
        result = run_full_pipeline(
            generation.words,
            generation.tasks,
            generation.config,
            n_workers=generation.n_workers,
            project_dir=autocorrect2.project_dir,
            report_path=command.report_path,
            write_accepted=autocorrect2.write_accepted,
            logger=logger,
        )
        _print_typo_summary(result.typo_generation, output=stream)
        _print_autocorrect2_summary(result.autocorrect2, output=stream)
        _print_write_destination(
            write_accepted=autocorrect2.write_accepted,
            project_dir=autocorrect2.project_dir,
            output=stream,
        )
        _print_report_path(command.report_path, output=stream)
        return 0

    assert_never(command)

_print_typo_summary

_print_typo_summary(
    result: TypoGenerationResult, *, output: TextIO
) -> None

Print the stable summary fields of a typo-generation result.

Parameters:

Name Type Description Default
result TypoGenerationResult

Typo-generation result returned by the pipeline.

required
output TextIO

Destination stream.

required
Source code in hotstring\cli\commands.py
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def _print_typo_summary(result: TypoGenerationResult, *, output: TextIO) -> None:
    """Print the stable summary fields of a typo-generation result.

    Args:
        result:
            Typo-generation result returned by the pipeline.
        output:
            Destination stream.
    """
    print("Typo generation complete.", file=output)
    print(f"  Source words: {result.source_word_count}", file=output)
    print(f"  Raw samples: {result.generated_sample_count}", file=output)
    print(f"  Candidates: {len(result.candidates)}", file=output)
    print(f"  Internal clashes: {len(result.clashes)}", file=output)

_print_autocorrect2_summary

_print_autocorrect2_summary(
    result: AutoCorrect2CheckResult, *, output: TextIO
) -> None

Print the stable summary fields of an AutoCorrect2 check result.

Parameters:

Name Type Description Default
result AutoCorrect2CheckResult

AutoCorrect2 result returned by the pipeline.

required
output TextIO

Destination stream.

required
Source code in hotstring\cli\commands.py
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def _print_autocorrect2_summary(
    result: AutoCorrect2CheckResult,
    *,
    output: TextIO,
) -> None:
    """Print the stable summary fields of an AutoCorrect2 check result.

    Args:
        result:
            AutoCorrect2 result returned by the pipeline.
        output:
            Destination stream.
    """
    print("AutoCorrect2 check complete.", file=output)
    print(f"  Candidates checked: {result.candidate_count}", file=output)
    print(f"  Accepted: {len(result.accepted)}", file=output)
    print(f"  Rejected: {len(result.rejected)}", file=output)

_print_write_destination

_print_write_destination(
    *,
    write_accepted: bool,
    project_dir: Path,
    output: TextIO,
) -> None

Print the generated-file destination when writing was enabled.

Parameters:

Name Type Description Default
write_accepted bool

Whether the selected command enabled writing.

required
project_dir Path

AutoCorrect2 project directory.

required
output TextIO

Destination stream.

required
Source code in hotstring\cli\commands.py
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def _print_write_destination(
    *,
    write_accepted: bool,
    project_dir: Path,
    output: TextIO,
) -> None:
    """Print the generated-file destination when writing was enabled.

    Args:
        write_accepted:
            Whether the selected command enabled writing.
        project_dir:
            AutoCorrect2 project directory.
        output:
            Destination stream.
    """
    if write_accepted:
        print(
            f"  Generated include: {project_dir / GENERATED_HOTSTRINGS_RELATIVE_PATH}",
            file=output,
        )

_print_report_path

_print_report_path(
    report_path: Path | None, *, output: TextIO
) -> None

Print the report destination when one was requested.

Parameters:

Name Type Description Default
report_path Path | None

Optional report destination.

required
output TextIO

Destination stream.

required
Source code in hotstring\cli\commands.py
166
167
168
169
170
171
172
173
174
175
176
def _print_report_path(report_path: Path | None, *, output: TextIO) -> None:
    """Print the report destination when one was requested.

    Args:
        report_path:
            Optional report destination.
        output:
            Destination stream.
    """
    if report_path is not None:
        print(f"  Report: {report_path}", file=output)