File size: 3,541 Bytes
c9761c3 | 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 | # 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
"""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
|