| |
| |
| |
| |
| |
| |
| |
|
|
| """Resolve safe WAV paths and probe lightweight PCM metadata with the standard library.""" |
|
|
| from __future__ import annotations |
|
|
| import math |
| import wave |
| from dataclasses import dataclass |
| from pathlib import Path, PurePath |
|
|
| from speech_text_data_aligners.core.errors import AudioError |
| from speech_text_data_aligners.core.models import AlignmentRequest, Attributes |
|
|
|
|
| @dataclass(frozen=True, slots=True) |
| class WavInfo: |
| """Describe lightweight WAV header facts needed by neutral services. |
| |
| Parameters |
| ---------- |
| path: Probed WAV path. |
| frames: Audio frame count. |
| sample_rate_hz: Frames per second. |
| channels: Channel count. |
| sample_width_bytes: Bytes per sample per channel. |
| duration_s: Exact frame-count/sample-rate duration in seconds. |
| |
| """ |
|
|
| path: Path |
| frames: int |
| sample_rate_hz: int |
| channels: int |
| sample_width_bytes: int |
| duration_s: float |
|
|
|
|
| def resolve_audio_path(audio_dir: Path, audio_id: str) -> Path: |
| """Resolve an audio identifier strictly as ``<audio_dir>/<audio_id>.wav``. |
| |
| Parameters |
| ---------- |
| audio_dir: Directory containing source WAVs. |
| audio_id: Metadata identifier without path components or extension. |
| |
| Returns |
| ------- |
| Lexically resolved WAV path; existence is checked by :func:`probe_wav`. |
| |
| Raises |
| ------ |
| AudioError: If the identifier is empty, absolute, traversal-like, or contains |
| path separators. |
| |
| """ |
| candidate = PurePath(audio_id) |
| if ( |
| not audio_id |
| or audio_id in {".", ".."} |
| or candidate.is_absolute() |
| or candidate.name != audio_id |
| or "/" in audio_id |
| or "\\" in audio_id |
| ): |
| raise AudioError("audio_id must be a non-empty path-component-free identifier") |
| return audio_dir / f"{audio_id}.wav" |
|
|
|
|
| def probe_wav(path: Path) -> WavInfo: |
| """Read a local WAV header and calculate a finite positive duration. |
| |
| Parameters |
| ---------- |
| path: WAV file to inspect. |
| |
| Returns |
| ------- |
| Immutable standard-library WAV facts. |
| |
| Raises |
| ------ |
| AudioError: If the path is missing, unreadable, corrupt, empty, or has invalid |
| header dimensions. |
| |
| """ |
| try: |
| with wave.open(str(path), "rb") as wav: |
| frames = wav.getnframes() |
| sample_rate = wav.getframerate() |
| channels = wav.getnchannels() |
| sample_width = wav.getsampwidth() |
| except (OSError, EOFError, wave.Error) as error: |
| raise AudioError(f"cannot read WAV audio: {path}") from error |
| if frames <= 0 or sample_rate <= 0 or channels <= 0 or sample_width <= 0: |
| raise AudioError(f"WAV has invalid or empty header values: {path}") |
| duration = frames / sample_rate |
| if not math.isfinite(duration) or duration <= 0: |
| raise AudioError(f"WAV duration is invalid: {path}") |
| return WavInfo(path, frames, sample_rate, channels, sample_width, duration) |
|
|
|
|
| def make_alignment_request( |
| *, |
| request_id: str, |
| audio_path: Path, |
| transcript: str, |
| context: Attributes = (), |
| ) -> AlignmentRequest: |
| """Probe a WAV and form a complete backend-neutral alignment request. |
| |
| Parameters |
| ---------- |
| request_id: Caller-defined stable unique identity. |
| audio_path: Existing readable WAV path. |
| transcript: Exact authoritative transcript. |
| context: Immutable JSON-safe caller identifiers. |
| |
| Returns |
| ------- |
| Alignment request populated with the probed WAV duration. |
| |
| Raises |
| ------ |
| AudioError: If WAV probing fails. |
| |
| """ |
| info = probe_wav(audio_path) |
| return AlignmentRequest( |
| request_id=request_id, |
| audio_path=audio_path, |
| transcript=transcript, |
| audio_duration_s=info.duration_s, |
| context=context, |
| ) |
|
|
