File size: 4,842 Bytes
3699b0a | 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 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 | """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)
|