| """Publish durable codec speech artifacts.""" |
| from __future__ import annotations |
| import fcntl |
| import hashlib |
| import json |
| import os |
| import sys |
| import tempfile |
| import threading |
| import time |
| from collections.abc import Iterator |
| from contextlib import contextmanager |
| from pathlib import Path |
| from typing import BinaryIO, Literal, TextIO |
| from pydantic import Field |
| from .configuration import Specification |
| from .provenance import code_stamp |
| Split = Literal['train', 'validation', 'test'] |
|
|
| class Recording(Specification): |
| record_id: str |
| shard: str |
| row: int = Field(ge=0) |
| audio_sha256: str |
| speaker: str = Field(min_length=1) |
| text: str = Field(min_length=1) |
| text_key: str = Field(min_length=1) |
| stored_rate_hz: int = Field(gt=0) |
| samples: int = Field(gt=0) |
| prepared_samples: int = Field(gt=0) |
| duration_s: float = Field(gt=0) |
| peak: float = Field(ge=0) |
| rms_dbfs: float | None |
| clipping_fraction: float = Field(ge=0, le=1) |
| source_id: str |
| source_revision: str |
| license: str |
| retrieval_date: str |
| language_decision: Literal['review'] = 'review' |
| language_reason: str = 'upstream Kabyle label; project acoustic/transcript review pending' |
|
|
| class Rejection(Specification): |
| record_id: str |
| shard: str |
| row: int = Field(ge=0) |
| reason: str |
|
|
| class Pair(Specification): |
| target: str |
| reference: str |
| speaker: str |
| split: Split |
|
|
| def digest_file(path: Path) -> str: |
| with path.open('rb') as stream: |
| return hashlib.file_digest(stream, 'sha256').hexdigest() |
|
|
| def sync_directory(path: Path) -> None: |
| descriptor = os.open(path, os.O_RDONLY) |
| try: |
| os.fsync(descriptor) |
| finally: |
| os.close(descriptor) |
|
|
| @contextmanager |
| def atomic_binary(path: Path) -> Iterator[BinaryIO]: |
| path.parent.mkdir(parents=True, exist_ok=True) |
| descriptor, temporary = tempfile.mkstemp(prefix=f'.{path.name}.', dir=path.parent) |
| staged = Path(temporary) |
| try: |
| with os.fdopen(descriptor, 'wb') as stream: |
| yield stream |
| stream.flush() |
| os.fsync(stream.fileno()) |
| staged.replace(path) |
| sync_directory(path.parent) |
| finally: |
| staged.unlink(missing_ok=True) |
|
|
| def write_json(path: Path, value: object) -> None: |
| encoded = json.dumps(value, ensure_ascii=False, sort_keys=True, allow_nan=False) + '\n' |
| with atomic_binary(path) as stream: |
| stream.write(encoded.encode()) |
|
|
| @contextmanager |
| def run_lock(root: Path) -> Iterator[None]: |
| root.mkdir(parents=True, exist_ok=True) |
| with (root / '.lock').open('a+b') as stream: |
| try: |
| fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB) |
| except BlockingIOError as error: |
| message = f"another writer holds {root / '.lock'}" |
| raise RuntimeError(message) from error |
| try: |
| yield |
| finally: |
| fcntl.flock(stream, fcntl.LOCK_UN) |
|
|
| def read_recordings(path: Path) -> Iterator[Recording]: |
| with path.open(encoding='utf-8') as stream: |
| for number, line in enumerate(stream, start=1): |
| try: |
| yield Recording.model_validate_json(line) |
| except ValueError as error: |
| message = f'invalid recording at {path}:{number}: {error}' |
| raise ValueError(message) from error |
|
|
| def provenance(config_hash: str) -> dict[str, object]: |
| return {'version': 1, 'version_tag': None, 'config_hash': config_hash, **code_stamp()} |
|
|
| def verify_conservation(input_rows: int, output_rows: int, rejected_rows: int) -> None: |
| if min(input_rows, output_rows, rejected_rows) < 0 or input_rows != output_rows + rejected_rows: |
| message = f'unaccounted records: input={input_rows}, output={output_rows}' |
| message += f', rejected={rejected_rows}' |
| raise ValueError(message) |
|
|
| class EventLog: |
|
|
| def __init__(self, stream: TextIO, root: Path, identity: str) -> None: |
| self.stream = stream |
| self.root = root |
| self.identity = identity |
| self.lock = threading.Lock() |
| self.status: dict[str, object] = {} |
|
|
| def emit(self, event: str, **values: object) -> None: |
| payload = {'schema_version': 1, 'family': 'codec_tts', 'config_hash': self.identity, 'event': event, 'time': time.time(), **values} |
| line = json.dumps(payload, allow_nan=False) + '\n' |
| with self.lock: |
| self.stream.write(line) |
| self.stream.flush() |
| sys.stderr.write(line) |
| sys.stderr.flush() |
| self.status.update(payload) |
| write_json(self.root / 'status.json', self.status) |
|
|
| @contextmanager |
| def event_log(root: Path, identity: str) -> Iterator[EventLog]: |
| root.mkdir(parents=True, exist_ok=True) |
| with (root / 'log.jsonl').open('a', encoding='utf-8') as stream: |
| yield EventLog(stream, root, identity) |
|
|
