"""Reading and checking wav deliverables.""" from __future__ import annotations import hashlib import os import urllib.request from pathlib import Path import numpy as np from scipy.io import wavfile def resolve_output(output: str, json_dir: Path, cache_dir: Path | None = None) -> Path: """Return a local path for a solution output, downloading URLs into cache_dir.""" if output.startswith("http://") or output.startswith("https://"): cache_dir = cache_dir or Path(os.environ.get("CHALLENGE_CACHE", ".challenge-cache")) cache_dir.mkdir(parents=True, exist_ok=True) name = hashlib.sha256(output.encode()).hexdigest()[:16] + ".wav" dest = cache_dir / name if not dest.exists(): urllib.request.urlretrieve(output, dest) return dest p = (json_dir / output).resolve() return p def read_wav(path: Path, spec: dict) -> np.ndarray: """Read a wav and check it against the challenge's output spec. Returns float64 samples (mono).""" rate, data = wavfile.read(path) problems = [] if rate != spec["sample_rate_hz"]: problems.append(f"sample rate {rate} != {spec['sample_rate_hz']}") if spec.get("sample_format") == "float32" and data.dtype != np.float32: problems.append(f"sample format {data.dtype} != float32") channels = 1 if data.ndim == 1 else data.shape[1] if channels != spec["channels"]: problems.append(f"channels {channels} != {spec['channels']}") n = data.shape[0] if n != spec["num_samples"]: problems.append(f"num samples {n} != {spec['num_samples']}") if not np.all(np.isfinite(data)): problems.append("non-finite samples") if problems: raise ValueError(f"{path}: " + "; ".join(problems)) if data.ndim > 1: data = data[:, 0] return data.astype(np.float64) def write_wav(path: Path, samples: np.ndarray, sample_rate: int) -> None: path.parent.mkdir(parents=True, exist_ok=True) wavfile.write(path, sample_rate, np.asarray(samples, dtype=np.float32))