| |
| |
| |
| |
| |
| |
| |
|
|
| """Compose W1-T providers and delegate offline timing to the reusable service.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| from pathlib import Path |
|
|
| from speech_text_data_aligners.application.readiness import run_readiness |
| from speech_text_data_aligners.backends.registry import ( |
| BackendRegistry, |
| default_backend_registry, |
| ) |
| from speech_text_data_aligners.core.errors import ConfigurationError |
| from speech_text_data_aligners.runtime.alignment_evidence import EvidenceCandidate |
|
|
|
|
| def configure_parser(parser: argparse.ArgumentParser) -> None: |
| """Register source paths and an explicit evidence/provider catalog on ``parser``. |
| |
| Side Effects: |
| Adds W1-T arguments and the command handler without loading a backend. |
| """ |
| parser.add_argument("metadata_csv", type=Path) |
| parser.add_argument("boundaries_jsonl", type=Path) |
| parser.add_argument("audio_dir", type=Path) |
| parser.add_argument("output_dir", type=Path) |
| parser.add_argument("--catalog", type=Path, required=True) |
| parser.set_defaults(handler=run) |
|
|
|
|
| def run(args: argparse.Namespace, *, registry: BackendRegistry | None = None) -> int: |
| """Build catalog-selected providers and delegate W1-T to the application service. |
| |
| Parameters |
| ---------- |
| args: Parsed W1-T arguments. |
| registry: Optional injectable lazy provider registry. |
| |
| Returns |
| ------- |
| Zero for the timing coverage gate, three for a packaged BLOCKED timing result. |
| |
| Raises |
| ------ |
| ConfigurationError: If the catalog is malformed or unavailable. |
| SpeechTextAlignerError: If provider construction or source validation fails. |
| |
| Side Effects: |
| The service writes sidecar/evidence and runs only missing backend pairs. |
| |
| """ |
| selected = registry or default_backend_registry() |
| try: |
| catalog = json.loads(args.catalog.read_text(encoding="utf-8")) |
| specs = catalog["backends"] |
| candidates = { |
| (row["audio_id"], row["backend"]): tuple( |
| EvidenceCandidate(Path(c["path"]), c["sha256"]) |
| for c in row["candidates"] |
| ) |
| for row in catalog.get("evidence", []) |
| } |
| identities = {name: spec["identity"] for name, spec in specs.items()} |
| unavailable = { |
| name: spec.get("unavailable_reason", "runtime was not supplied") |
| for name, spec in specs.items() |
| } |
| configs = {name: spec.get("config") for name, spec in specs.items()} |
| except ( |
| OSError, |
| UnicodeError, |
| ValueError, |
| KeyError, |
| TypeError, |
| AttributeError, |
| ) as error: |
| raise ConfigurationError(f"invalid W1-T catalog: {error}") from error |
| backends = { |
| name: selected.create(name, config=config) if config is not None else None |
| for name, config in configs.items() |
| } |
| receipt = run_readiness( |
| metadata_csv=args.metadata_csv, |
| boundaries_jsonl=args.boundaries_jsonl, |
| audio_dir=args.audio_dir, |
| output_dir=args.output_dir, |
| backends=backends, |
| identities=identities, |
| candidates=candidates, |
| unavailable_reasons=unavailable, |
| ) |
| return 0 if receipt["status"] == "COMPLETE" else 3 |
|
|
