"""Acquire frozen small human speech inputs and prepare controls without inference.""" import argparse import hashlib import json from pathlib import Path import subprocess import urllib.request import wave def sha(data): return hashlib.sha256(data).hexdigest() def execute(args): return subprocess.run(args, capture_output=True, check=True, timeout=60).stdout def write_wav(path, samples, prefix_ms=0, maximum_ms=30000): if not 0 < maximum_ms <= 45000 or len(samples) % 2 or not samples or len(samples) // 2 + prefix_ms * 16 > maximum_ms * 16: raise ValueError("Invalid or oversized 30-second source") if path.exists(): raise ValueError("Preserve existing prepared clip") with wave.open(str(path), "wb") as output: output.setnchannels(1) output.setsampwidth(2) output.setframerate(16000) output.writeframes(bytes(prefix_ms * 32) + samples) def main(): parser = argparse.ArgumentParser() parser.add_argument("fixture", type=Path) parser.add_argument("output", type=Path) parser.add_argument("ffmpeg", type=Path) parser.add_argument("ffprobe", type=Path) args = parser.parse_args() raw = args.fixture.read_bytes() fixture = json.loads(raw) lock = json.loads(args.fixture.with_name("labels.lock.json").read_text()) if sha(raw) != lock["fixture_sha256"]: raise ValueError("Frozen plan changed") args.output.mkdir(parents=True, exist_ok=True) plan_path = args.output / "prepared.json" if plan_path.exists(): raise ValueError("Preserve existing preparation") prepared = [] for asset in fixture["assets"]: path = args.output / (asset["id"] + "-source.ogg") if not path.exists(): request = urllib.request.Request(asset["source_url"] + "?download=1", headers={"User-Agent": "SearchiMediaEvaluation/1.0"}) with urllib.request.urlopen(request, timeout=60) as response: data = response.read(asset["bytes"] + 1) if len(data) != asset["bytes"] or hashlib.sha1(data).hexdigest() != asset["sha1"]: raise ValueError("Publisher source changed") path.write_bytes(data) original = path.read_bytes() if len(original) != asset["bytes"] or hashlib.sha1(original).hexdigest() != asset["sha1"]: raise ValueError("Cached source changed") source_probe = json.loads(execute([str(args.ffprobe), "-v", "error", "-show_format", "-show_streams", "-of", "json", str(path)])) audio = next(s for s in source_probe["streams"] if s["codec_type"] == "audio") if "duration" not in audio or float(audio["duration"]) + asset["prefix_silence_ms"] / 1000 > 30: raise ValueError("Full primary speech stream must fit 30 seconds; never silently crop reference speech") samples = execute([str(args.ffmpeg), "-v", "error", "-nostdin", "-threads", "1", "-i", str(path), "-ac", "1", "-ar", "16000", "-f", "s16le", "pipe:1"]) output = args.output / (asset["id"] + ".wav") write_wav(output, samples, asset["prefix_silence_ms"]) prepared.append({"id": asset["id"], "path": str(output.resolve()), "sha256": sha(output.read_bytes()), "samples": len(samples) // 2 + asset["prefix_silence_ms"] * 16, "source_sha256": sha(original), "source_probe": source_probe, "source_audio_start_ms": float(audio.get("start_time", "0")) * 1000, "prefix_silence_ms": asset["prefix_silence_ms"], "fine_utterance_timestamps_available": False}) music_source = Path(".searchi/qa/music-validation-v1-canonical/unseen03-source.ogg") expected_music = fixture["controls"][0] if sha(music_source.read_bytes()) != expected_music["source_sha256"]: raise ValueError("Frozen instrumental control changed") samples = execute([str(args.ffmpeg), "-v", "error", "-nostdin", "-threads", "1", "-i", str(music_source), "-t", "30", "-ac", "1", "-ar", "16000", "-af", "aresample=16000,atrim=end_sample=480000,asetpts=PTS-STARTPTS", "-f", "s16le", "pipe:1"]) if len(samples) != 960000: raise ValueError("Instrumental control must cover full 30 seconds") for id, data in [("music-control", samples), ("silence-control", bytes(960000))]: output = args.output / (id + ".wav") write_wav(output, data) prepared.append({"id": id, "path": str(output.resolve()), "sha256": sha(output.read_bytes()), "samples": 480000, "source_audio_start_ms": 0, "expected_recognized_segments": 0}) for item in prepared: item["actual_probe"] = json.loads(execute([str(args.ffprobe), "-v", "error", "-show_format", "-show_streams", "-of", "json", item["path"]])) seam_raw = args.fixture.with_name("seam-plan.json").read_bytes() seam_plan = json.loads(seam_raw) seam_lock = json.loads(args.fixture.with_name("seams.lock.json").read_text()) if sha(seam_raw) != seam_lock["seam_plan_sha256"]: raise ValueError("Frozen seam plan changed") seams = [] for recipe in seam_plan["recipes"]: source = next(item for item in prepared if item["id"] == recipe["source"]) with wave.open(source["path"], "rb") as clip: clip.readframes(source["prefix_silence_ms"] * 16) samples = clip.readframes(clip.getnframes()) path = args.output / (recipe["id"] + ".wav") write_wav(path, samples * recipe["copies"], recipe["prefix_silence_ms"], recipe["maximum_duration_ms"]) count = len(samples) // 2 prefix = recipe["prefix_silence_ms"] * 16 mapping = [{"derived_samples": [prefix + index * count, prefix + (index + 1) * count], "source": recipe["source"], "source_samples": [0, count], "copy_index": index} for index in range(recipe["copies"])] seams.append({"id": recipe["id"], "path": str(path.resolve()), "sha256": sha(path.read_bytes()), "source_sha256": source["source_sha256"], "samples": prefix + count * recipe["copies"], "piecewise_map": mapping, "prefix_silence_samples": prefix, "artificial_repeat_control": True}) result = {"fixture_sha256": sha(raw), "seam_plan_sha256": sha(seam_raw), "frozen_before_inference": True, "model_inference_performed": False, "ffmpeg_sha256": sha(args.ffmpeg.read_bytes()), "ffprobe_sha256": sha(args.ffprobe.read_bytes()), "clips": prepared, "seam_controls": seams} with plan_path.open("x", encoding="utf-8") as stream: json.dump(result, stream, ensure_ascii=False, indent=2) print("Prepared four clips without inference; plan SHA256", sha(plan_path.read_bytes())) if __name__ == "__main__": main()