File size: 7,822 Bytes
35d483e | 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 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 | """Audio payload inspection and exact hashing without mandatory DSP packages."""
from __future__ import annotations
from dataclasses import dataclass
import hashlib
import io
import json
from pathlib import Path
import struct
from typing import Any, Mapping, Sequence
import wave
class AudioPayloadError(ValueError):
"""Raised when an audio value cannot be resolved to bytes or samples."""
@dataclass(frozen=True)
class AudioInfo:
"""Content fingerprint and inexpensive container metadata."""
sha256: str
num_bytes: int
format: str
sample_rate: int | None = None
num_channels: int | None = None
num_frames: int | None = None
bits_per_sample: int | None = None
duration_seconds: float | None = None
path: str | None = None
def to_dict(self) -> dict[str, Any]:
return {
"audio_sha256": self.sha256,
"audio_num_bytes": self.num_bytes,
"audio_format": self.format,
"sample_rate": self.sample_rate,
"num_channels": self.num_channels,
"num_frames": self.num_frames,
"bits_per_sample": self.bits_per_sample,
"duration_seconds": self.duration_seconds,
"audio_path": self.path,
}
def _resolve_path(path_value: str | Path, base_dir: str | Path | None) -> Path:
path = Path(path_value).expanduser()
if not path.is_absolute() and base_dir is not None:
path = Path(base_dir) / path
return path
def _read_path(path: Path) -> bytes:
try:
return path.read_bytes()
except OSError as exc:
raise AudioPayloadError(f"cannot read audio path {path}: {exc}") from exc
def extract_audio_bytes(
audio: Any,
*,
base_dir: str | Path | None = None,
) -> tuple[bytes, str | None]:
"""Resolve common Hugging Face/local audio representations to exact bytes.
Hugging Face Parquet rows normally contain ``{"bytes": ..., "path": ...}``.
Byte content is preferred over the often-virtual path. Decoded arrays are
represented canonically when no encoded payload is available.
"""
if isinstance(audio, bytes):
return audio, None
if isinstance(audio, (bytearray, memoryview)):
return bytes(audio), None
if isinstance(audio, (str, Path)):
path = _resolve_path(audio, base_dir)
return _read_path(path), str(audio)
if isinstance(audio, Mapping):
path_value = audio.get("path")
encoded = audio.get("bytes")
if encoded is not None:
if not isinstance(encoded, (bytes, bytearray, memoryview)):
raise AudioPayloadError("audio['bytes'] is not bytes-like")
return bytes(encoded), str(path_value) if path_value is not None else None
if path_value:
path = _resolve_path(str(path_value), base_dir)
return _read_path(path), str(path_value)
if audio.get("array") is not None:
return _canonical_array_bytes(audio["array"], audio.get("sampling_rate")), None
if hasattr(audio, "read"):
stream = audio
try:
position = stream.tell()
except (AttributeError, OSError):
position = None
payload = stream.read()
if position is not None:
try:
stream.seek(position)
except (AttributeError, OSError):
pass
if isinstance(payload, str):
payload = payload.encode("utf-8")
if isinstance(payload, (bytes, bytearray, memoryview)):
return bytes(payload), None
raise AudioPayloadError(
"unsupported audio value; expected bytes, a path, an Audio mapping, or a decoded array"
)
def _canonical_array_bytes(array: Any, sampling_rate: Any) -> bytes:
header: dict[str, Any] = {"sampling_rate": sampling_rate}
if hasattr(array, "dtype"):
header["dtype"] = str(array.dtype)
if hasattr(array, "shape"):
header["shape"] = tuple(int(value) for value in array.shape)
if hasattr(array, "tobytes"):
body = array.tobytes(order="C")
elif isinstance(array, Sequence) and not isinstance(array, (str, bytes, bytearray)):
# JSON is slower than ndarray.tobytes but deterministic and dependency-free.
try:
body = json.dumps(array, separators=(",", ":"), allow_nan=False).encode("utf-8")
except (TypeError, ValueError) as exc:
raise AudioPayloadError(f"cannot serialize decoded audio array: {exc}") from exc
else:
raise AudioPayloadError("audio['array'] cannot be serialized")
return b"TURN_ARRAY\0" + json.dumps(header, sort_keys=True).encode("utf-8") + b"\0" + body
def _infer_format(payload: bytes, path: str | None) -> str:
if payload.startswith(b"RIFF") and payload[8:12] == b"WAVE":
return "wav"
if payload.startswith(b"fLaC"):
return "flac"
if payload.startswith(b"OggS"):
return "ogg"
if payload.startswith(b"ID3") or payload[:2] in (b"\xff\xfb", b"\xff\xf3", b"\xff\xf2"):
return "mp3"
if payload.startswith(b"TURN_ARRAY\0"):
return "decoded_array"
if path:
suffix = Path(path).suffix.lower().lstrip(".")
if suffix:
return suffix
return "unknown"
def _wav_metadata(payload: bytes) -> tuple[int, int, int, int, float] | None:
try:
with wave.open(io.BytesIO(payload), "rb") as wav_file:
sample_rate = wav_file.getframerate()
channels = wav_file.getnchannels()
frames = wav_file.getnframes()
bits = wav_file.getsampwidth() * 8
except (wave.Error, EOFError, OSError, struct.error):
return None
duration = frames / sample_rate if sample_rate else None
return sample_rate, channels, frames, bits, duration
def _flac_metadata(payload: bytes) -> tuple[int, int, int, int, float] | None:
# FLAC STREAMINFO is always the first metadata block and has a 34-byte body.
if len(payload) < 42 or payload[:4] != b"fLaC":
return None
block_type = payload[4] & 0x7F
block_length = int.from_bytes(payload[5:8], "big")
if block_type != 0 or block_length < 34 or len(payload) < 8 + block_length:
return None
streaminfo = payload[8 : 8 + block_length]
packed = int.from_bytes(streaminfo[10:18], "big")
sample_rate = packed >> 44
channels = ((packed >> 41) & 0x7) + 1
bits = ((packed >> 36) & 0x1F) + 1
frames = packed & ((1 << 36) - 1)
duration = frames / sample_rate if sample_rate else None
return sample_rate, channels, frames, bits, duration
def inspect_audio(audio: Any, *, base_dir: str | Path | None = None) -> AudioInfo:
"""Hash exact audio bytes and extract WAV/FLAC STREAMINFO metadata."""
payload, path = extract_audio_bytes(audio, base_dir=base_dir)
if not payload:
raise AudioPayloadError("audio payload is empty")
digest = hashlib.sha256(payload).hexdigest()
audio_format = _infer_format(payload, path)
metadata = _wav_metadata(payload) if audio_format == "wav" else None
if audio_format == "flac":
metadata = _flac_metadata(payload)
if metadata is None:
return AudioInfo(digest, len(payload), audio_format, path=path)
sample_rate, channels, frames, bits, duration = metadata
return AudioInfo(
sha256=digest,
num_bytes=len(payload),
format=audio_format,
sample_rate=sample_rate,
num_channels=channels,
num_frames=frames,
bits_per_sample=bits,
duration_seconds=duration,
path=path,
)
def exact_audio_sha256(audio: Any, *, base_dir: str | Path | None = None) -> str:
"""Return the SHA-256 digest of the exact encoded payload/canonical array."""
return inspect_audio(audio, base_dir=base_dir).sha256
|