File size: 4,119 Bytes
51e84d2 fc1dcbc | 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 | # 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
"""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,
)
|