File size: 4,069 Bytes
51e84d2 fc1dcbc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 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 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | # License: Apache-2.0 License
# Created by: Patrick Lumbantobing, VertoX-AI
# Copyright (c) 2026 VertoX-AI. All rights reserved.
#
# This work is licensed under the Apache-2.0 License.
# To view a copy of this license, visit
# https://www.apache.org/licenses/LICENSE-2.0
"""Adapt benchmark CLI arguments to reusable application/report services."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
from speech_text_data_aligners.application.benchmark import (
BenchmarkOptions,
benchmark_backends,
)
from speech_text_data_aligners.backends.registry import (
BackendRegistry,
default_backend_registry,
)
from speech_text_data_aligners.benchmarking.reports import (
preflight_report_bundle,
render_report,
write_report_bundle,
)
from speech_text_data_aligners.core.errors import ConfigurationError
def configure_parser(parser: argparse.ArgumentParser) -> None:
"""Add benchmark composition/presentation arguments to a subcommand parser.
Parameters
----------
parser: Parser owned by the CLI composition root.
Side Effects:
Registers arguments and the command handler on ``parser``.
"""
parser.add_argument("metadata_csv", type=Path)
parser.add_argument("reference_csv", type=Path)
parser.add_argument("audio_dir", type=Path)
parser.add_argument("--backend", action="append", required=True)
parser.add_argument("--backend-config-json", default="{}")
parser.add_argument("--strict", action="store_true")
parser.add_argument("--output-base", type=Path)
parser.add_argument("--overwrite", action="store_true")
parser.add_argument(
"--format", choices=("json", "csv", "markdown"), default="markdown"
)
parser.set_defaults(handler=run)
def run(
args: argparse.Namespace,
*,
registry: BackendRegistry | None = None,
) -> int:
"""Construct selected backends and delegate benchmark work to library services.
Parameters
----------
args: Parsed benchmark arguments.
registry: Optional injected lazy registry for tests/custom composition.
Returns
-------
Zero after writing a bundle or printing the requested report format.
Raises
------
ConfigurationError: If backend names repeat or configuration JSON is invalid.
SpeechTextAlignerError: Propagated typed construction/report/service failures.
Side Effects:
Calls backends and either writes a report bundle or prints one report to stdout.
"""
if len(args.backend) != len(set(args.backend)):
raise ConfigurationError("benchmark backend names must be unique")
try:
config = json.loads(args.backend_config_json)
except json.JSONDecodeError as error:
raise ConfigurationError("backend configuration is not valid JSON") from error
if not isinstance(config, dict):
raise ConfigurationError("backend configuration JSON must be an object")
if args.output_base is not None:
# Reserve all report paths before factories can initialize costly backends.
preflight_report_bundle(
args.output_base,
metadata_csv=args.metadata_csv,
reference_csv=args.reference_csv,
overwrite=args.overwrite,
)
selected_registry = registry or default_backend_registry()
backends = {
name: selected_registry.create(name, config=config.get(name, {}))
for name in args.backend
}
report = benchmark_backends(
backends=backends,
metadata_csv=args.metadata_csv,
reference_csv=args.reference_csv,
audio_dir=args.audio_dir,
options=BenchmarkOptions(strict=args.strict),
)
if args.output_base is not None:
write_report_bundle(
report,
args.output_base,
metadata_csv=args.metadata_csv,
reference_csv=args.reference_csv,
overwrite=args.overwrite,
)
else:
print(render_report(report, args.format), end="")
return 0
|