"""Thin adapter over immutable public packages; no teacher or research weights.""" import hashlib import importlib import importlib.util import io import json import os from pathlib import Path import sys import threading import time import uuid from catalog import MODELS LOCK = threading.RLock() EVENT_REVISION = "7b6c0cb1ce98d8330dc96cc2685bdb71b1016837" EVENT_MANIFEST = "55ecf645c7aa2b28d2112a0d25e000f48a723e19638c8f5ba160fdc5e76f4045" EVENT_BANK_MANIFEST = "46e01c1bc394ce60785652a021e53a4ebd9fa0e7d0b7a8508ea87dc1ef3d3395" EVENT_BANK_RUNTIME = "f163b56415fb84f5dcd135fdb85157fa16445c352fc2fac897f8b854d1a79604" class NativeCapture: """Keep the exact native chunks passed to the existing enhancer.""" def __init__(self, enhancer): self.enhancer = enhancer self.chunks = [] def enhance(self, wave, sample_rate): self.chunks.append(wave.detach().cpu().numpy().copy()) return self.enhancer.enhance(wave, sample_rate) def module(path, name): spec = importlib.util.spec_from_file_location(name, path) value = importlib.util.module_from_spec(spec) sys.modules[name] = value spec.loader.exec_module(value) return value def verify_files(root, manifest, nested=False): root = Path(root).resolve() data = json.loads((root / manifest).read_text()) for name, value in data["files"].items(): path = root / name expected = value["sha256"] if isinstance(value, dict) else value if not path.resolve().is_relative_to(root) or path.is_symlink() or not path.is_file(): raise ValueError("Release file is missing or linked: " + name) with path.open("rb") as stream: if hashlib.file_digest(stream, "sha256").hexdigest() != expected: raise ValueError("Release integrity check failed: " + name) return data class Studio: def __init__(self, cache, output, device="cpu"): from huggingface_hub import snapshot_download import torch self.torch = torch self.device = device self.output = Path(output) self.output.mkdir(parents=True, exist_ok=True) self.paths = {} for key, spec in MODELS.items(): patterns = ["core/**"] if key == "mini" else None path = snapshot_download(spec.repo, revision=spec.revision, local_dir=Path(cache) / key, allow_patterns=patterns, token=False, max_workers=4) self.paths[key] = Path(path) mini = self.paths["mini"] / "core" micro = self.paths["micro"] self.mini_cli = module(mini / "synthesize.py", "_demo_mini_cli") self.mini_cli.verify(mini) self.micro_cli = module(micro / "synthesize.py", "_demo_micro_cli") self.micro_cli.verify_candidate(micro) verify_files(self.paths["atom"], "manifest.json") # The two neural packages intentionally share these exact source bytes. a = json.loads((mini / "assembly-manifest.json").read_text())["files"] b = json.loads((micro / "release-manifest.json").read_text())["files"] for name, digest in a.items(): other = name.replace("baseline/models/mini/code/", "models/micro/code/") if name.startswith("baseline/models/mini/code/") and other in b and digest != b[other]: raise ValueError("Model packages have incompatible shared modules.") # Micro enforces module origins; load it first, then reuse the verified # byte-identical common modules when constructing Mini. self.micro_runtime = self.micro_cli.load_runtime(micro, "micro", device) self.base_cli, self.mini_runtime, _ = self.mini_cli.load_selected("mini", device, profile="pitch", runtime="canonical", root=mini) self.mini_runtime[2]._load_pitch() selected = module(mini / "selected_mini.py", "_demo_mini_selected") self.enhancer = selected.Native48Enhancer(mini, device) self.core, self.frontend = self.mini_runtime[:2] self.numeric = self.base_cli.load_numeric(mini / "baseline") atom_path = self.paths["atom"] self.atom_code = module(atom_path / "atom_runtime_20261024.py", "_demo_atom") self.atom = self.atom_code.AtomRuntime(atom_path / "export.pt", atom_path / "vocab.json", device=device, g2p=self.frontend, dsp_backend="torch") event_snapshot = snapshot_download(MODELS["mini"].repo, revision=EVENT_REVISION, local_dir=Path(cache) / "events", allow_patterns=["extras/events/component/**"], token=False, max_workers=4) events = Path(event_snapshot) / "extras/events/component" self.event_code = module(events / "sdk.py", "_demo_events") base_events = self.event_code.ApprovedEvents(self._event_speech, events, EVENT_MANIFEST) base_events._load() bank = Path(__file__).resolve().parent / "event_bank" if hashlib.sha256((bank / "runtime.py").read_bytes()).hexdigest() != EVENT_BANK_RUNTIME: raise ValueError("Event bank runtime integrity check failed.") self.event_bank_code = module(bank / "runtime.py", "_demo_multivoice_events") # Owner-requested beta rollout; preserve listening-approval metadata rather than inventing approvals. self.event_adapter = self.event_bank_code.MultiVoiceEvents(base_events, bank, EVENT_BANK_MANIFEST, allow_unreviewed=True) if not all(p["source_approved"] for p in self.event_adapter.manifest["profiles"]): raise ValueError("The event bank contains a source that has not been approved.") for voice in self.event_bank_code.VOICES: for tag in self.event_bank_code.TAGS: if not (voice == "cassidy" and tag in ("sigh", "gasp", "cough", "laugh")) and not self.event_adapter.profiles.get((voice, tag)): raise ValueError("Incomplete event bank: " + voice + "/" + tag) torch.set_num_threads(2) def prepare(self, request): with LOCK: if request["events"]: self.event_bank_code.parse(request["text"]) return request prepared = self.base_cli.prepare_text(request["text"], self.numeric, "conservative") # Preflight the same offline frontend before requesting shared GPU time. for chunk in prepared["chunks"]: self.frontend(chunk) return dict(request, prepared=prepared) def _speech(self, text, voice, seed, enhanced=False, prepared=None): import soundfile as sf prepared = prepared or self.base_cli.prepare_text(text, self.numeric, "conservative") raw, metadata = self.mini_cli.render(prepared, voice, self.mini_runtime, self.enhancer if enhanced else None, seed, "FLOAT") audio, rate = sf.read(io.BytesIO(raw), dtype="float32") return audio, rate, metadata def _event_speech(self, text, voice="cassidy", seed=42, enhanced=False): audio, rate, metadata = self._speech(text, voice, seed, enhanced) return dict(audio=self.torch.from_numpy(audio), sample_rate=rate, metadata=metadata) def _save_take(self, audio, rate, key, request, begin, **extra): import numpy as np import soundfile as sf if audio.ndim != 1 or not audio.size or not np.isfinite(audio).all() or np.abs(audio).max() >= 1: raise ValueError("The model returned unsafe audio. Please try a shorter passage.") for old in self.output.glob("inflect-*.wav"): if old.is_file() and not old.is_symlink() and time.time() - old.stat().st_mtime > 3600: old.unlink(missing_ok=True) path = self.output / f"inflect-{key}-{request['voice']}-{uuid.uuid4().hex}.wav" sf.write(path, audio, rate, subtype="PCM_16") return dict(path=str(path), model=key, voice=request["voice"], rate=rate, seconds=len(audio)/rate, compute_seconds=time.perf_counter()-begin, seed=request["seed"], characters=len(request["text"]), **extra) def generate_quality_pair(self, request): import numpy as np import soundfile as sf if request["model"] != "mini" or request["events"]: raise ValueError("24/48 kHz comparison uses Mini with plain text.") begin = time.perf_counter() with LOCK, self.torch.inference_mode(): capture = NativeCapture(self.enhancer) raw, _ = self.mini_cli.render(request["prepared"], request["voice"], self.mini_runtime, capture, request["seed"], "FLOAT") enhanced, rate = sf.read(io.BytesIO(raw), dtype="float32") chunks = [] for chunk in capture.chunks: if chunks: chunks.append(np.zeros(2880, dtype=np.float32)) chunks.append(chunk) if not chunks: raise ValueError("The native performance could not be captured.") native = np.concatenate(chunks) if rate != 48000 or len(enhanced) != 2 * len(native): raise ValueError("The comparison did not preserve the native performance length.") return [self._save_take(native, 24000, "mini", request, begin, variant="native", shared_performance=True), self._save_take(enhanced, 48000, "mini", request, begin, variant="enhanced", shared_performance=True)] def generate(self, request, key=None): import numpy as np import soundfile as sf key = key or request["model"] begin = time.perf_counter() with LOCK, self.torch.inference_mode(): if key == "mini": if request["events"]: result = self.event_adapter.synthesize(request["text"], voice=request["voice"], seed=request["seed"], events=True, sample_rate=24000) if request["enhanced"]: count = max(1, (result["audio"].numel() + 479999) // 480000) chunks = result["audio"].tensor_split(count) result["audio"] = self.torch.cat([self.enhancer.enhance(chunk, 24000)[0] for chunk in chunks]) result["sample_rate"] = 48000 audio, rate = result["audio"].detach().cpu().numpy(), result["sample_rate"] else: audio, rate, _ = self._speech(request["text"], request["voice"], request["seed"], request["enhanced"], request["prepared"]) elif key == "micro": raw, _ = self.micro_cli.render_prepared(request["prepared"], request["voice"], self.micro_runtime, request["seed"]) audio, rate = sf.read(io.BytesIO(raw), dtype="float32") else: pieces = [] for i, chunk in enumerate(request["prepared"]["chunks"]): result = self.atom.synthesize_text(chunk, request["voice"], seed=(request["seed"] + i) % 2**32) if pieces: pieces.append(np.zeros(2880, dtype=np.float32)) pieces.append(result["audio"].detach().cpu().numpy()) audio, rate = np.concatenate(pieces), 24000 return self._save_take(audio, rate, key, request, begin)