Spaces:
Paused
Paused
| #!/usr/bin/env python3 | |
| """ | |
| studio.py | |
| A local web app for managing cloned voices: write a script, pick a voice, generate takes, | |
| compare profiles, and approve audio. Runs on your own computer; nothing is uploaded anywhere. | |
| .venv-tts\\Scripts\\activate | |
| python studio.py | |
| Then open http://127.0.0.1:7860 (it opens automatically). The voice model loads once and | |
| stays loaded, so every generation after the first starts right away. | |
| Options: | |
| --model standard|turbo which Chatterbox model (voice profiles must match it) | |
| --port 7860 | |
| --host 127.0.0.1 use 0.0.0.0 to let others on your network open it (see README) | |
| """ | |
| import os | |
| os.environ.setdefault("HF_HUB_DISABLE_SYMLINKS_WARNING", "1") | |
| os.environ.setdefault("HF_HUB_DISABLE_SYMLINKS", "1") | |
| import argparse | |
| import csv | |
| import json | |
| import math | |
| import mimetypes | |
| import queue | |
| import random | |
| import re | |
| import shutil | |
| import tempfile | |
| import threading | |
| import time | |
| import traceback | |
| import uuid | |
| import webbrowser | |
| from datetime import datetime | |
| from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer | |
| from pathlib import Path | |
| from types import SimpleNamespace | |
| from urllib.parse import parse_qs, unquote, urlparse | |
| import soundfile as sf | |
| import speak | |
| ROOT = Path.cwd() | |
| OUTPUT = ROOT / "output" | |
| VOICES = ROOT / "voices" | |
| LEGACY_REFERENCE_DIR = ROOT / "voice_refs" | |
| NAMED_REFERENCE_SUFFIX = "_voice_refs" | |
| REFERENCE_DIR = ( | |
| ROOT / "brian_voice_refs" | |
| if (ROOT / "brian_voice_refs").exists() or not LEGACY_REFERENCE_DIR.exists() | |
| else LEGACY_REFERENCE_DIR | |
| ) | |
| PEOPLE_MANIFEST = VOICES / "people.json" | |
| REFERENCE_COUNT = 6 | |
| MAX_VIDEO_UPLOAD_BYTES = 8 * 1024 ** 3 | |
| MAX_REFERENCE_UPLOAD_BYTES = 100 * 1024 ** 2 | |
| VIDEO_UPLOAD_EXTS = {".mp4", ".mov", ".mkv", ".webm", ".avi", ".m4v", ".mts", ".wav", ".mp3", ".m4a"} | |
| REFERENCE_LOCK = threading.Lock() | |
| AUDIO_EXTS = {".wav", ".flac", ".mp3", ".ogg", ".m4a"} | |
| HTML = Path(__file__).with_name("studio.html") | |
| APP_ASSETS = { | |
| "/manifest.webmanifest": (Path(__file__).with_name("manifest.webmanifest"), "application/manifest+json"), | |
| "/mobile-manifest.webmanifest": (Path(__file__).with_name("mobile-manifest.webmanifest"), "application/manifest+json"), | |
| "/icon.svg": (Path(__file__).with_name("icon.svg"), "image/svg+xml"), | |
| "/sw.js": (Path(__file__).with_name("sw.js"), "text/javascript; charset=utf-8"), | |
| } | |
| def people_registry(): | |
| if not PEOPLE_MANIFEST.is_file(): | |
| return {"people": [], "sources": {}} | |
| data = json.loads(PEOPLE_MANIFEST.read_text(encoding="utf-8")) | |
| if not isinstance(data, dict) or not isinstance(data.get("people", []), list): | |
| raise ValueError("People registry is invalid.") | |
| data.setdefault("sources", {}) | |
| return data | |
| def save_people_registry(data): | |
| VOICES.mkdir(parents=True, exist_ok=True) | |
| with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=VOICES, | |
| suffix=".tmp", delete=False) as handle: | |
| temp_path = Path(handle.name) | |
| json.dump(data, handle, indent=2) | |
| try: | |
| temp_path.replace(PEOPLE_MANIFEST) | |
| except Exception: | |
| temp_path.unlink(missing_ok=True) | |
| raise | |
| def available_people(): | |
| registry = people_registry() | |
| records = [{"id": "brian", "name": "Brian"}] | |
| records.extend(item for item in registry["people"] if item.get("id") != "brian") | |
| known_ids = {person["id"] for person in records} | |
| for directory in named_reference_dirs(): | |
| folder_name = directory.name[:-len(NAMED_REFERENCE_SUFFIX)] | |
| person_id = re.sub(r"[^a-z0-9]+", "-", folder_name.lower()).strip("-")[:40].strip("-") | |
| if person_id and person_id not in known_ids: | |
| name = re.sub(r"[_-]+", " ", folder_name).strip().title() | |
| records.append({"id": person_id, "name": name}) | |
| known_ids.add(person_id) | |
| return records | |
| def get_person(person_id): | |
| if not isinstance(person_id, str) or not re.fullmatch(r"[a-z0-9][a-z0-9_-]{0,39}", person_id): | |
| raise ValueError("Choose a valid person.") | |
| for person in available_people(): | |
| if person["id"] == person_id: | |
| return person | |
| raise ValueError("That person does not exist. Refresh the page and try again.") | |
| def create_person(name): | |
| if not isinstance(name, str) or not name.strip(): | |
| raise ValueError("Enter a name for the person.") | |
| name = name.strip() | |
| slug = re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")[:40].strip("-") | |
| if not slug: | |
| raise ValueError("Use a name that includes at least one letter or number.") | |
| with REFERENCE_LOCK: | |
| registry = people_registry() | |
| if any(person["id"] == slug for person in available_people()): | |
| raise ValueError("A person with that name already exists.") | |
| person = {"id": slug, "name": name} | |
| registry["people"].append(person) | |
| registry["sources"].setdefault(slug, []) | |
| save_people_registry(registry) | |
| return person | |
| def save_person_sources(person_id, sources): | |
| person = get_person(person_id) | |
| if not isinstance(sources, list): | |
| raise ValueError("Source folders must be a list.") | |
| available = {item["path"] for item, _ in profile_sources()} | |
| if any(not isinstance(source, str) or source not in available for source in sources): | |
| raise ValueError("One or more selected source folders are no longer available.") | |
| registry = people_registry() | |
| registry.setdefault("sources", {})[person["id"]] = list(dict.fromkeys(sources)) | |
| save_people_registry(registry) | |
| return registry["sources"][person["id"]] | |
| def person_profile_paths(person, kind): | |
| if kind not in {"standard", "turbo"}: | |
| raise ValueError("Choose a supported voice model.") | |
| if person["id"] == "brian": | |
| stem = "brian" if kind == "standard" else "brian_turbo" | |
| directory = VOICES | |
| else: | |
| stem = kind | |
| directory = VOICES / "people" / person["id"] | |
| active = directory / f"{stem}.pt" | |
| candidate_stem = f"{stem}_candidate" | |
| return { | |
| "active": active, | |
| "candidate": directory / f"{candidate_stem}.pt", | |
| } | |
| def person_reference_dir(person): | |
| if person["id"] == "brian": | |
| return REFERENCE_DIR | |
| folder_name = re.sub(r"[^a-z0-9]+", "_", person["id"].lower()).strip("_") | |
| named_dir = ROOT / f"{folder_name}{NAMED_REFERENCE_SUFFIX}" | |
| if named_dir.is_dir(): | |
| return named_dir | |
| return VOICES / "people" / person["id"] / "references" | |
| def person_reference_manifest(person): | |
| return person_reference_dir(person) / "reference_pool.json" | |
| def person_video_dir(person): | |
| return VOICES / "people" / person["id"] / "uploads" / "videos" | |
| def person_dataset_dir(person): | |
| return VOICES / "people" / person["id"] / "datasets" / "prepared" | |
| def person_reference_audio(person): | |
| directory = person_reference_dir(person) | |
| return [ | |
| directory / name | |
| for name in [*(f"voice_ref{index}.wav" for index in range(1, 6)), | |
| *(f"ref_{index:02d}.wav" for index in range(1, 7))] | |
| if (directory / name).is_file() | |
| ][:5] | |
| def person_media_status(person_id): | |
| person = get_person(person_id) | |
| video_dir = person_video_dir(person) | |
| videos = sorted( | |
| path.name for path in video_dir.iterdir() | |
| if path.is_file() and path.suffix.lower() in | |
| {".mp4", ".mov", ".mkv", ".webm", ".avi", ".m4v", ".mts", ".wav", ".mp3", ".m4a"} | |
| ) if video_dir.is_dir() else [] | |
| references = person_reference_audio(person) | |
| return { | |
| "person": person, | |
| "videos": videos, | |
| "references": [path.name for path in references], | |
| "dataset": rel(person_dataset_dir(person)), | |
| "dataset_ready": (person_dataset_dir(person) / "report.csv").is_file(), | |
| } | |
| def delete_uploaded_voice_sample(person_id, name): | |
| person = get_person(person_id) | |
| if not isinstance(name, str) or not re.fullmatch(r"voice_ref[1-5]\.wav", name): | |
| raise ValueError("Choose one of this person's uploaded voice samples.") | |
| directory = person_reference_dir(person) | |
| target = directory / name | |
| with REFERENCE_LOCK: | |
| if not target.is_file(): | |
| raise ValueError("That voice sample is no longer available.") | |
| target.unlink() | |
| manifest_path = person_reference_manifest(person) | |
| if manifest_path.is_file(): | |
| manifest = json.loads(manifest_path.read_text(encoding="utf-8")) | |
| slots = manifest.get("slots", {}) | |
| if name in slots: | |
| del slots[name] | |
| save_reference_manifest(manifest, person_id) | |
| return {"deleted": name, "media": person_media_status(person_id)} | |
| def managed_reference_dirs(): | |
| return [person_reference_dir(person) for person in available_people()] | |
| def named_reference_dirs(): | |
| return sorted( | |
| path for path in ROOT.iterdir() | |
| if path.is_dir() and path.name.lower().endswith(NAMED_REFERENCE_SUFFIX) | |
| ) | |
| def clip_dirs(): | |
| return list(dict.fromkeys([ | |
| ROOT / "refs", | |
| LEGACY_REFERENCE_DIR, | |
| *named_reference_dirs(), | |
| *[person_reference_dir(person) for person in available_people() if person["id"] != "brian"], | |
| *[person_video_dir(person) for person in available_people()], | |
| ])) | |
| # ---------------------------------------------------------------- engine (one model, one job at a time) | |
| class Engine: | |
| def __init__(self, kind, device): | |
| self.kind = kind | |
| self.device = device | |
| self.model = None | |
| self.error = None | |
| self.voice_key = None | |
| self.jobs = {} | |
| self.rebuild_jobs = {} | |
| self.dataset_jobs = {} | |
| self.compare_jobs = {} | |
| self.rebuild_submit_lock = threading.Lock() | |
| self.dataset_submit_lock = threading.Lock() | |
| self.queue = queue.Queue() | |
| self.lock = threading.Lock() | |
| threading.Thread(target=self._load, daemon=True).start() | |
| threading.Thread(target=self._worker, daemon=True).start() | |
| # -- model | |
| def _load(self): | |
| try: | |
| self.model = speak.load_model(self.kind, self.device) | |
| except Exception as e: | |
| self.error = f"The voice model didn't load: {e}" | |
| traceback.print_exc() | |
| def status(self): | |
| busy = sum(1 for j in self.jobs.values() if j["state"] in ("queued", "running")) | |
| return {"ready": self.model is not None, "error": self.error, "model": self.kind, | |
| "device": self.device, "jobs_waiting": busy} | |
| # -- voices | |
| def use_voice(self, voice): | |
| """voice = {"type": "profile"|"clip", "path": ...}; only re-processes when it changes.""" | |
| path = safe_path(voice["path"], [VOICES] if voice["type"] == "profile" else clip_dirs()) | |
| file_version = None | |
| if voice["type"] == "clip": | |
| stat = path.stat() | |
| file_version = (stat.st_mtime_ns, stat.st_size) | |
| key = (voice["type"], voice["path"], voice.get("exaggeration") if voice["type"] == "clip" else None, | |
| file_version) | |
| if key == self.voice_key: | |
| return | |
| if voice["type"] == "profile": | |
| meta = path.with_suffix(".json") | |
| if meta.exists(): | |
| made_for = json.loads(meta.read_text(encoding="utf-8")).get("model", self.kind) | |
| if made_for != self.kind: | |
| raise ValueError(f"{path.name} was built for the {made_for} model, but the studio is " | |
| f"running {self.kind}. Restart with --model {made_for}.") | |
| speak.load_voice_profile(self.model, self.kind, path, self.device) | |
| else: | |
| if sf.info(str(path)).duration < 5.5: | |
| raise ValueError(f"{path.name} is too short to copy a voice from. Use a clip of 7-15 seconds.") | |
| speak.prepare_voice(self.model, self.kind, path, voice.get("exaggeration", 0.5)) | |
| self.voice_key = key | |
| # -- jobs | |
| def submit(self, req): | |
| job_id = uuid.uuid4().hex[:10] | |
| voice = req.get("voice") or {} | |
| safe_path(voice.get("path", ""), [VOICES] if voice.get("type") == "profile" else clip_dirs()) | |
| text = (req.get("text") or "").strip() | |
| if not text: | |
| raise ValueError("Write a script first.") | |
| if len(text) > 6000: | |
| raise ValueError("That script is very long. Split it into parts of about 6,000 characters.") | |
| takes = max(1, min(int(req.get("takes", 1)), 5)) | |
| seed = req.get("seed") | |
| seed = int(seed) if str(seed).strip().lstrip("-").isdigit() else random.randint(0, 2 ** 31 - 1) | |
| chunks = speak.split_text(text, int(req.get("max_chars", 280))) | |
| if not chunks: | |
| raise ValueError("Write a script first.") | |
| job = { | |
| "id": job_id, "state": "queued", "created": time.time(), "text": text, | |
| "voice": req["voice"], "takes_wanted": takes, "seed": seed, | |
| "settings": { | |
| "exaggeration": float(req.get("exaggeration", 0.5)), | |
| "cfg": float(req.get("cfg", 0.5)), | |
| "temperature": float(req.get("temperature", 0.8)), | |
| "pause": float(req.get("pause", 0.25)), | |
| "para_pause": float(req.get("para_pause", 0.7)), | |
| "max_chars": int(req.get("max_chars", 280)), | |
| }, | |
| "chunks": chunks, "take": 0, "piece": 0, "pieces": len(chunks) * takes, | |
| "done_pieces": 0, "results": [], "error": None, | |
| } | |
| self.jobs[job_id] = job | |
| self.queue.put(job_id) | |
| return job | |
| def _worker(self): | |
| while True: | |
| job_id = self.queue.get() | |
| job = self.jobs.get(job_id) | |
| if not job: | |
| continue | |
| while self.model is None and self.error is None: | |
| time.sleep(0.5) | |
| if self.error: | |
| job.update(state="error", error=self.error) | |
| continue | |
| job["state"] = "running" | |
| job["started"] = time.time() | |
| try: | |
| with self.lock: | |
| self._run(job) | |
| job["state"] = "done" | |
| except ValueError as e: # expected problems (bad voice file etc.): message only | |
| job.update(state="error", error=str(e)) | |
| except Exception as e: | |
| traceback.print_exc() | |
| job.update(state="error", error=f"Generation failed: {e}") | |
| job["finished"] = time.time() | |
| def _run(self, job): | |
| s = job["settings"] | |
| voice = dict(job["voice"], exaggeration=s["exaggeration"]) | |
| self.use_voice(voice) | |
| args = SimpleNamespace(exaggeration=s["exaggeration"], cfg=s["cfg"], temperature=s["temperature"]) | |
| stamp = datetime.now().strftime("%Y%m%d_%H%M%S") | |
| OUTPUT.mkdir(parents=True, exist_ok=True) | |
| for k in range(job["takes_wanted"]): | |
| seed = job["seed"] + k | |
| job["take"] = k + 1 | |
| speak.seed_everything(seed) | |
| pieces = [] | |
| for i, (chunk, pause_kind) in enumerate(job["chunks"], 1): | |
| job["piece"] = i | |
| wav = speak.generate_piece(self.model, self.kind, chunk, args) | |
| pieces.append((speak.trim(wav, self.model.sr), pause_kind)) | |
| job["done_pieces"] += 1 | |
| audio = speak.assemble(pieces, self.model.sr, s["pause"], s["para_pause"]) | |
| name = f"studio_{stamp}_{job['id']}_take{k + 1}.wav" | |
| path = OUTPUT / name | |
| sf.write(str(path), audio, self.model.sr, subtype="PCM_16") | |
| settings = { | |
| "text": job["text"], | |
| "reference": job["voice"]["path"] if job["voice"]["type"] == "clip" else None, | |
| "voice_profile": job["voice"]["path"] if job["voice"]["type"] == "profile" else None, | |
| "model": self.kind, "seed": seed, **{k2: s[k2] for k2 in | |
| ("exaggeration", "cfg", "temperature", "pause", "para_pause", "max_chars")}, | |
| "created": datetime.now().isoformat(timespec="seconds"), | |
| "take": k + 1, "of_takes": job["takes_wanted"], "source": "studio", | |
| "approved": False, | |
| "note": "AI-generated voice (Chatterbox, Perth watermark).", | |
| } | |
| path.with_suffix(".json").write_text(json.dumps(settings, indent=2), encoding="utf-8") | |
| job["results"].append(entry_for(path, settings)) | |
| def public(self, job): | |
| keys = ("id", "state", "take", "takes_wanted", "piece", "pieces", "done_pieces", | |
| "results", "error", "seed") | |
| out = {k: job.get(k) for k in keys} | |
| out["chunks"] = len(job["chunks"]) | |
| if job.get("started"): | |
| out["elapsed"] = round((job.get("finished") or time.time()) - job["started"], 1) | |
| ahead = [j for j in self.jobs.values() | |
| if j["state"] in ("queued", "running") and j["created"] < job["created"]] | |
| out["ahead"] = len(ahead) | |
| return out | |
| # -- voice profile rebuilds | |
| def submit_dataset(self, person_id): | |
| person = get_person(person_id) | |
| media = person_media_status(person["id"]) | |
| references = person_reference_audio(person) | |
| if not media["videos"]: | |
| raise ValueError(f"Upload one or more videos for {person['name']} first.") | |
| if not 1 <= len(references) <= 5: | |
| raise ValueError( | |
| f"Upload 1-5 clean 10-30 second WAV voice samples for {person['name']} first." | |
| ) | |
| with self.dataset_submit_lock: | |
| if any(job["state"] in ("queued", "running") for job in self.dataset_jobs.values()): | |
| raise ValueError("A video-to-dataset job is already in progress.") | |
| job_id = uuid.uuid4().hex[:10] | |
| job = { | |
| "id": job_id, "person": person["id"], "state": "queued", "progress": 0, | |
| "message": "Waiting to prepare clips…", "error": None, "result": None, | |
| "created": time.time(), | |
| } | |
| self.dataset_jobs[job_id] = job | |
| settings = { | |
| "person": person["id"], | |
| "input": person_video_dir(person), | |
| "references": references, | |
| "output": person_dataset_dir(person), | |
| } | |
| threading.Thread(target=self._dataset_worker, args=(job_id, settings), daemon=True).start() | |
| return self.public_dataset(job) | |
| def _dataset_worker(self, job_id, settings): | |
| job = self.dataset_jobs[job_id] | |
| job.update(state="running", started=time.time(), message="Starting clip preparation…") | |
| try: | |
| import prepare_dataset | |
| def report(message, progress): | |
| job.update(message=message, progress=progress) | |
| while self.model is None and self.error is None: | |
| job.update(message="Waiting for Studio to finish loading before preparing clips…") | |
| time.sleep(0.5) | |
| device = self.device if self.device in {"cpu", "cuda"} else \ | |
| ("cuda" if prepare_dataset.torch.cuda.is_available() else "cpu") | |
| args = SimpleNamespace( | |
| input=settings["input"], reference=settings["references"], out=settings["output"], | |
| threshold=0.60, sentence_threshold=0.55, min_unit=0.6, sample_rate=44100, | |
| min_len=3.0, max_len=15.0, max_gap=0.6, pad=0.10, min_words=2, | |
| min_dbfs=-40.0, whisper_model="large-v3", language="en", device=device, | |
| keep_temp=False, | |
| ) | |
| with self.lock: | |
| reload_voice_model = self.model is not None | |
| if reload_voice_model: | |
| job.update(message="Freeing GPU memory for video processing…") | |
| self.model = None | |
| self.voice_key = None | |
| import gc | |
| gc.collect() | |
| if prepare_dataset.torch.cuda.is_available(): | |
| prepare_dataset.torch.cuda.empty_cache() | |
| try: | |
| result = prepare_dataset.run_dataset(args, progress=report) | |
| finally: | |
| if reload_voice_model: | |
| job.update(message="Reloading Studio’s voice model…", progress=99) | |
| gc.collect() | |
| if prepare_dataset.torch.cuda.is_available(): | |
| prepare_dataset.torch.cuda.empty_cache() | |
| try: | |
| self.model = speak.load_model(self.kind, self.device) | |
| except Exception as e: | |
| self.error = f"The voice model didn't reload after video preparation: {e}" | |
| raise RuntimeError(self.error) from e | |
| self.voice_key = None | |
| if result["videos_processed"] == 0 and result["failures"]: | |
| failed = "; ".join( | |
| f"{item['video']}: {item['error']}" for item in result["failures"] | |
| ) | |
| raise RuntimeError(f"No videos were processed successfully. Failed: {failed}") | |
| registry = people_registry() | |
| sources = registry.setdefault("sources", {}).setdefault(settings["person"], []) | |
| dataset_path = rel(settings["output"]) | |
| if dataset_path not in sources: | |
| sources.append(dataset_path) | |
| save_people_registry(registry) | |
| job.update( | |
| state="done", progress=100, | |
| message=f"Prepared {result['clips_kept']} clips from {result['videos_processed']} videos.", | |
| result={**result, "dataset": dataset_path}, | |
| ) | |
| except (ValueError, RuntimeError) as e: | |
| job.update(state="error", error=str(e), message="Video clip preparation failed.") | |
| except Exception as e: | |
| traceback.print_exc() | |
| job.update(state="error", error=f"Video clip preparation failed: {e}", | |
| message="Video clip preparation failed.") | |
| finally: | |
| job["finished"] = time.time() | |
| def public_dataset(job): | |
| result = job.get("result") | |
| if result: | |
| result = {key: result.get(key) for key in ( | |
| "videos_found", "videos_processed", "videos_skipped", "clips_kept", | |
| "failures", "dataset", | |
| )} | |
| return { | |
| **{key: job.get(key) for key in ( | |
| "id", "person", "state", "progress", "message", "error", | |
| )}, | |
| "result": result, | |
| } | |
| def submit_rebuild(self, req): | |
| if not isinstance(req, dict): | |
| raise ValueError("Rebuild request must be an object.") | |
| person = get_person(req.get("person", "brian")) | |
| available_sources = {item["path"]: path for item, path in profile_sources()} | |
| requested_sources = req.get("sources") | |
| if not isinstance(requested_sources, list) or not requested_sources: | |
| raise ValueError("Select at least one audio source.") | |
| if any(not isinstance(source, str) or source not in available_sources for source in requested_sources): | |
| raise ValueError("One or more selected audio sources are no longer available. Refresh the page.") | |
| available_anchors = {item["path"]: path for item, path in profile_anchors(requested_sources)} | |
| anchor_key = req.get("anchor") | |
| if not isinstance(anchor_key, str) or anchor_key not in available_anchors: | |
| raise ValueError("Choose an available anchor clip.") | |
| try: | |
| max_clips = int(req.get("max_clips", 400)) | |
| min_sim = float(req.get("min_sim", 0.65)) | |
| blend = float(req.get("blend", 1.0)) | |
| except (TypeError, ValueError) as e: | |
| raise ValueError("Rebuild settings must be valid numbers.") from e | |
| if not 3 <= max_clips <= 4000 or not math.isfinite(min_sim) or not math.isfinite(blend) or \ | |
| not 0 <= min_sim <= 1 or not 0 <= blend <= 1: | |
| raise ValueError("Use 3-4000 clips, similarity 0-1, and blend 0-1.") | |
| with self.rebuild_submit_lock: | |
| if any(job["state"] in ("queued", "running") for job in self.rebuild_jobs.values()): | |
| raise ValueError("A voice rebuild is already in progress.") | |
| job_id = uuid.uuid4().hex[:10] | |
| job = { | |
| "id": job_id, "state": "queued", "progress": 0, "message": "Waiting for the voice model…", | |
| "error": None, "result": None, "created": time.time(), | |
| } | |
| self.rebuild_jobs[job_id] = job | |
| settings = { | |
| "person": person["id"], | |
| "sources": [available_sources[source] for source in dict.fromkeys(requested_sources)], | |
| "anchor": available_anchors[anchor_key], | |
| "max_clips": max_clips, "min_sim": min_sim, "blend": blend, | |
| } | |
| threading.Thread(target=self._rebuild_worker, args=(job_id, settings), daemon=True).start() | |
| return self.public_rebuild(job) | |
| def _rebuild_worker(self, job_id, settings): | |
| job = self.rebuild_jobs[job_id] | |
| while self.model is None and self.error is None: | |
| time.sleep(0.5) | |
| if self.error: | |
| job.update(state="error", error=self.error, message="The voice model failed to load.") | |
| return | |
| job.update(state="running", started=time.time(), message="Starting voice profile rebuild…") | |
| try: | |
| from build_voice import build_voice_profile | |
| def report(message, progress): | |
| job.update(message=message, progress=progress) | |
| profile_path = person_profile_paths(get_person(settings["person"]), self.kind)["candidate"] | |
| with self.lock: | |
| result = build_voice_profile( | |
| self.model, self.kind, self.device, settings["sources"], settings["anchor"], | |
| profile_path, max_clips=settings["max_clips"], min_sim=settings["min_sim"], | |
| blend=settings["blend"], progress=report, | |
| ) | |
| self.voice_key = None | |
| job.update(state="done", progress=100, message="Voice profile rebuilt.", | |
| result={**result, "path": rel(profile_path)}) | |
| except ValueError as e: | |
| job.update(state="error", error=str(e), message="Voice profile rebuild failed.") | |
| except Exception as e: | |
| traceback.print_exc() | |
| job.update(state="error", error=f"Voice profile rebuild failed: {e}", | |
| message="Voice profile rebuild failed.") | |
| finally: | |
| job["finished"] = time.time() | |
| def public_rebuild(job): | |
| result = job.get("result") | |
| if result: | |
| result = {key: result.get(key) for key in | |
| ("path", "clips_used", "clips_rejected_quality", "source_groups", | |
| "anchor_agreement", "blend", "created")} | |
| return { | |
| **{key: job.get(key) for key in ("id", "state", "progress", "message", "error")}, | |
| "result": result, | |
| } | |
| def submit_compare(self, req): | |
| if not isinstance(req, dict): | |
| raise ValueError("Comparison request must be an object.") | |
| text = req.get("text") | |
| if not isinstance(text, str) or not text.strip() or len(text) > 500: | |
| raise ValueError("Enter a comparison script of 1-500 characters.") | |
| person = get_person(req.get("person", "brian")) | |
| paths = person_profile_paths(person, self.kind) | |
| current, candidate = paths["active"], paths["candidate"] | |
| if not current.is_file() or not candidate.is_file(): | |
| raise ValueError("Build both the current voice profile and a candidate before comparing.") | |
| settings = { | |
| "text": text.strip(), | |
| "exaggeration": self._bounded_setting(req.get("exaggeration", 0.5), 0.25, 1.2, "Energy"), | |
| "cfg": self._bounded_setting(req.get("cfg", 0.5), 0.1, 0.9, "Pace"), | |
| "temperature": self._bounded_setting(req.get("temperature", 0.8), 0.4, 1.1, "Variation"), | |
| "seed": self._bounded_seed(req.get("seed")), | |
| "current": current, | |
| "candidate": candidate, | |
| } | |
| job_id = uuid.uuid4().hex[:10] | |
| job = {"id": job_id, "state": "queued", "progress": 0, "message": "Waiting for voice model…", | |
| "error": None, "result": None} | |
| self.compare_jobs[job_id] = job | |
| threading.Thread(target=self._compare_worker, args=(job_id, settings), daemon=True).start() | |
| return self.public_compare(job) | |
| def _bounded_setting(value, minimum, maximum, label): | |
| try: | |
| number = float(value) | |
| except (TypeError, ValueError) as e: | |
| raise ValueError(f"{label} must be a number.") from e | |
| if not math.isfinite(number) or not minimum <= number <= maximum: | |
| raise ValueError(f"{label} must be between {minimum} and {maximum}.") | |
| return number | |
| def _bounded_seed(value): | |
| try: | |
| seed = int(value) | |
| except (TypeError, ValueError) as e: | |
| raise ValueError("Comparison seed must be an integer.") from e | |
| if not 0 <= seed <= 2**31 - 1: | |
| raise ValueError("Comparison seed must be between 0 and 2147483647.") | |
| return seed | |
| def _compare_worker(self, job_id, settings): | |
| job = self.compare_jobs[job_id] | |
| while self.model is None and self.error is None: | |
| time.sleep(0.5) | |
| if self.error: | |
| job.update(state="error", error=self.error, message="The voice model failed to load.") | |
| return | |
| job.update(state="running", message="Generating matched A/B samples…") | |
| compare_dir = OUTPUT / ".compare" | |
| outputs = [] | |
| args = SimpleNamespace( | |
| exaggeration=settings["exaggeration"], cfg=settings["cfg"], temperature=settings["temperature"] | |
| ) | |
| chunks = speak.split_text(settings["text"], 280) | |
| if not chunks: | |
| job.update(state="error", error="Enter a comparison script.", message="A/B comparison failed.") | |
| return | |
| try: | |
| with self.lock: | |
| previous_conds = self.model.conds | |
| previous_voice_key = self.voice_key | |
| try: | |
| compare_dir.mkdir(parents=True, exist_ok=True) | |
| for index, profile in enumerate((settings["current"], settings["candidate"])): | |
| job.update(progress=index * 50, message=f"Generating {'A' if index == 0 else 'B'} sample…") | |
| self.use_voice({"type": "profile", "path": rel(profile)}) | |
| speak.seed_everything(settings["seed"]) | |
| pieces = [] | |
| for chunk, pause_kind in chunks: | |
| wav = speak.generate_piece(self.model, self.kind, chunk, args) | |
| pieces.append((speak.trim(wav, self.model.sr), pause_kind)) | |
| audio = speak.assemble(pieces, self.model.sr, 0.25, 0.7) | |
| label = "current" if index == 0 else "candidate" | |
| path = compare_dir / f"{job_id}_{label}.wav" | |
| sf.write(str(path), audio, self.model.sr, subtype="PCM_16") | |
| outputs.append({"label": label, "file": rel(path)}) | |
| finally: | |
| self.model.conds = previous_conds | |
| self.voice_key = previous_voice_key | |
| job.update(state="done", progress=100, message="A/B samples are ready.", result=outputs) | |
| except ValueError as e: | |
| job.update(state="error", error=str(e), message="A/B comparison failed.") | |
| except Exception as e: | |
| traceback.print_exc() | |
| job.update(state="error", error=f"A/B comparison failed: {e}", message="A/B comparison failed.") | |
| def public_compare(job): | |
| return {key: job.get(key) for key in | |
| ("id", "state", "progress", "message", "error", "result")} | |
| def adopt_candidate(self, person_id="brian"): | |
| person = get_person(person_id) | |
| paths = person_profile_paths(person, self.kind) | |
| current, candidate = paths["active"], paths["candidate"] | |
| candidate_meta = candidate.with_suffix(".json") | |
| if not candidate.is_file() or not candidate_meta.is_file(): | |
| raise ValueError("Build a candidate profile before using it.") | |
| try: | |
| metadata = json.loads(candidate_meta.read_text(encoding="utf-8")) | |
| except (OSError, json.JSONDecodeError) as e: | |
| raise ValueError("Candidate profile metadata is unreadable.") from e | |
| if metadata.get("model") != self.kind: | |
| raise ValueError(f"Candidate profile is not built for the {self.kind} model.") | |
| with self.lock: | |
| current.parent.mkdir(parents=True, exist_ok=True) | |
| temp_profile = current.with_name(current.name + ".tmp") | |
| temp_meta = current.with_suffix(".json.tmp") | |
| try: | |
| import shutil | |
| shutil.copy2(candidate, temp_profile) | |
| shutil.copy2(candidate_meta, temp_meta) | |
| os.replace(temp_profile, current) | |
| os.replace(temp_meta, current.with_suffix(".json")) | |
| finally: | |
| for path in (temp_profile, temp_meta): | |
| try: | |
| path.unlink() | |
| except FileNotFoundError: | |
| pass | |
| self.voice_key = None | |
| return {"path": rel(current), "created": metadata.get("created"), "clips_used": metadata.get("clips_used")} | |
| # ---------------------------------------------------------------- files | |
| def safe_path(rel, allowed_dirs): | |
| """Resolve a path the browser sent and make sure it stays inside the allowed folders.""" | |
| p = (ROOT / unquote(rel)).resolve() | |
| for d in allowed_dirs: | |
| try: | |
| p.relative_to(d.resolve()) | |
| if p.is_file(): | |
| return p | |
| except ValueError: | |
| continue | |
| raise ValueError("That file isn't available.") | |
| def rel(p): | |
| return p.resolve().relative_to(ROOT.resolve()).as_posix() | |
| def entry_for(wav, meta): | |
| try: | |
| seconds = round(sf.info(str(wav)).duration, 1) | |
| except Exception: | |
| seconds = None | |
| return { | |
| "file": rel(wav), "name": wav.name, "seconds": seconds, | |
| "text": meta.get("text", ""), "seed": meta.get("seed"), | |
| "voice": Path(meta.get("voice_profile") or meta.get("reference") or "").name or None, | |
| "created": meta.get("created"), "take": meta.get("take"), "of_takes": meta.get("of_takes"), | |
| "approved": bool(meta.get("approved")), "approved_at": meta.get("approved_at"), | |
| "settings": {k: meta.get(k) for k in ("exaggeration", "cfg", "temperature", "model")}, | |
| } | |
| def list_voices(): | |
| profiles = [] | |
| people = {person["id"]: person for person in available_people()} | |
| for p in sorted(VOICES.rglob("*.pt")) if VOICES.exists() else []: | |
| meta = {} | |
| if p.with_suffix(".json").exists(): | |
| try: | |
| meta = json.loads(p.with_suffix(".json").read_text(encoding="utf-8")) | |
| except Exception: | |
| pass | |
| person_id = p.parent.name if p.parent.parent == VOICES / "people" else "brian" | |
| person = people.get(person_id, {"name": p.stem}) | |
| model_name = meta.get("model") | |
| display_name = person["name"] if person_id == "brian" else \ | |
| f"{person['name']} ({model_name or p.stem})" | |
| profiles.append({"type": "profile", "path": rel(p), "name": display_name, | |
| "clips": meta.get("clips_used"), "model": meta.get("model"), | |
| "anchor": Path(meta.get("anchor", "")).name or None}) | |
| clips = [] | |
| managed_dirs = {path.resolve() for path in managed_reference_dirs()} | |
| for d in clip_dirs(): | |
| if not d.exists(): | |
| continue | |
| for p in sorted(d.glob("*.wav")): | |
| if p.parent == REFERENCE_DIR and not re.fullmatch(r"ref_0[1-6]\.wav", p.name): | |
| continue | |
| if p.parent != REFERENCE_DIR and p.name.startswith("ref_") and not re.fullmatch(r"ref_0[1-6]\.wav", p.name): | |
| continue | |
| try: | |
| dur = sf.info(str(p)).duration | |
| except Exception: | |
| continue | |
| active_reference = ( | |
| p.parent.resolve() in managed_dirs and | |
| re.fullmatch(r"(?:ref_0[1-6]|voice_ref[1-6])\.wav", p.name) | |
| ) | |
| if active_reference or 5.5 <= dur <= 40: | |
| clips.append({"type": "clip", "path": rel(p), "name": f"{d.name}/{p.stem}", | |
| "seconds": round(dur, 1)}) | |
| return {"profiles": profiles, "clips": clips} | |
| def profile_sources(): | |
| sources = [] | |
| candidates = [ | |
| ROOT / "refs", LEGACY_REFERENCE_DIR, *named_reference_dirs(), | |
| *managed_reference_dirs(), | |
| *[person_dataset_dir(person) for person in available_people()], | |
| ] | |
| candidates.extend( | |
| path for path in ROOT.iterdir() | |
| if path.is_dir() and (path / "report.csv").is_file() and (path / "wavs").is_dir() | |
| ) | |
| seen = set() | |
| for path in candidates: | |
| if not path.is_dir() or path.resolve() in seen: | |
| continue | |
| seen.add(path.resolve()) | |
| if (path / "report.csv").is_file() and (path / "wavs").is_dir(): | |
| with open(path / "report.csv", newline="", encoding="utf-8") as f: | |
| count = sum(1 for row in csv.DictReader(f) if row.get("status") == "kept") | |
| else: | |
| count = sum(1 for item in path.rglob("*") if item.is_file() and item.suffix.lower() in | |
| {".wav", ".flac", ".mp3", ".ogg", ".m4a"}) | |
| if count: | |
| sources.append(({"path": rel(path), "name": path.name, "clips": count}, path)) | |
| return sources | |
| def profile_anchors(requested_sources=None): | |
| sources = profile_sources() | |
| if requested_sources is None: | |
| folders = [path for _, path in sources] | |
| else: | |
| available = {item["path"]: path for item, path in sources} | |
| if any(source not in available for source in requested_sources): | |
| raise ValueError("One or more selected source folders are no longer available.") | |
| folders = [available[source] for source in dict.fromkeys(requested_sources)] | |
| anchors, seen = [], set() | |
| for folder in folders: | |
| search_root = folder / "wavs" if (folder / "report.csv").is_file() and (folder / "wavs").is_dir() else folder | |
| for path in search_root.rglob("*"): | |
| if not path.is_file() or path.suffix.lower() not in AUDIO_EXTS or path.resolve() in seen: | |
| continue | |
| seen.add(path.resolve()) | |
| try: | |
| duration = sf.info(str(path)).duration | |
| except (OSError, RuntimeError): | |
| continue | |
| if 5.5 <= duration <= 40: | |
| anchors.append(({"path": rel(path), "name": f"{path.parent.name}/{path.name}", | |
| "seconds": round(duration, 1)}, path)) | |
| return anchors | |
| def reference_candidates(requested_sources): | |
| available = {item["path"]: path for item, path in profile_sources()} | |
| if not isinstance(requested_sources, list) or not requested_sources: | |
| raise ValueError("Select at least one audio source.") | |
| if any(not isinstance(source, str) or source not in available for source in requested_sources): | |
| raise ValueError("One or more selected audio sources are no longer available. Refresh the page.") | |
| from pick_voice_ref import analyze | |
| candidates, seen = [], set() | |
| for source_key in dict.fromkeys(requested_sources): | |
| folder = available[source_key] | |
| if (folder / "report.csv").is_file() and (folder / "wavs").is_dir(): | |
| with open(folder / "report.csv", newline="", encoding="utf-8") as report: | |
| rows = ((row, folder / "wavs" / f"{row.get('clip_id', '')}.wav") | |
| for row in csv.DictReader(report) if row.get("status") == "kept") | |
| for row, path in rows: | |
| try: | |
| similarity = float(row["similarity"]) | |
| except (KeyError, TypeError, ValueError): | |
| continue | |
| add_reference_candidate(candidates, seen, path, similarity, row.get("text", ""), analyze) | |
| else: | |
| for path in folder.rglob("*"): | |
| if path.is_file() and path.suffix.lower() in AUDIO_EXTS: | |
| add_reference_candidate(candidates, seen, path, None, "", analyze) | |
| candidates.sort(key=lambda item: (-item["score"], item["path"])) | |
| return candidates | |
| def add_reference_candidate(candidates, seen, path, similarity, text, analyze): | |
| resolved = path.resolve() | |
| if resolved.parent in {directory.resolve() for directory in managed_reference_dirs()} and \ | |
| re.fullmatch(r"(?:ref_0[1-6]|voice_ref[1-5])\.wav", resolved.name): | |
| return | |
| try: | |
| resolved.relative_to(REFERENCE_DIR.resolve()) | |
| except ValueError: | |
| pass | |
| else: | |
| return | |
| if resolved in seen or not resolved.is_file(): | |
| return | |
| seen.add(resolved) | |
| try: | |
| quality = analyze(resolved) | |
| except (OSError, RuntimeError, ValueError): | |
| return | |
| if quality is None or not 7 <= quality["seconds"] <= 16 or quality["clipped"] > 0.0005: | |
| return | |
| match_score = similarity if similarity is not None else 0.5 | |
| length_bonus = 1.0 - min(abs(quality["seconds"] - 12.0) / 6.0, 1.0) | |
| candidates.append({ | |
| "path": rel(resolved), "name": resolved.name, "seconds": round(quality["seconds"], 1), | |
| "source": resolved.parent.parent.name if resolved.parent.name == "wavs" else resolved.parent.name, | |
| "similarity": round(similarity, 3) if similarity is not None else None, | |
| "score": round(3.0 * match_score + quality["speech_ratio"] + 0.5 * length_bonus, 4), | |
| "text": text, | |
| }) | |
| def reference_items(person_id="brian"): | |
| person = get_person(person_id) | |
| reference_dir = person_reference_dir(person) | |
| manifest_path = person_reference_manifest(person) | |
| manifest = json.loads(manifest_path.read_text(encoding="utf-8")) if manifest_path.is_file() else {} | |
| slots = manifest.get("slots", {}) | |
| items = [] | |
| for index in range(1, REFERENCE_COUNT + 1): | |
| standard_name = f"ref_{index:02d}.wav" | |
| legacy_name = f"voice_ref{index}.wav" | |
| filename = standard_name if (reference_dir / standard_name).is_file() else legacy_name | |
| path = reference_dir / filename | |
| if not path.is_file(): | |
| continue | |
| entry = slots.get(filename, {}) | |
| try: | |
| seconds = round(sf.info(str(path)).duration, 1) | |
| except (OSError, RuntimeError): | |
| seconds = None | |
| items.append({ | |
| "slot": filename, "file": rel(path), "name": filename, "seconds": seconds, | |
| "source": entry.get("source"), "similarity": entry.get("similarity"), | |
| "source_clip": Path(entry.get("path", "")).name or None, | |
| "text": entry.get("text", ""), | |
| }) | |
| return items | |
| def stage_reference_copy(source, person_id="brian"): | |
| reference_dir = person_reference_dir(get_person(person_id)) | |
| reference_dir.mkdir(parents=True, exist_ok=True) | |
| with tempfile.NamedTemporaryFile(dir=reference_dir, suffix=".wav", delete=False) as handle: | |
| staged_path = Path(handle.name) | |
| try: | |
| audio, sample_rate = sf.read(str(source), dtype="float32", always_2d=True) | |
| sf.write(str(staged_path), audio, sample_rate, subtype="PCM_16") | |
| except Exception: | |
| staged_path.unlink(missing_ok=True) | |
| raise | |
| return staged_path | |
| def save_reference_manifest(manifest, person_id="brian"): | |
| reference_dir = person_reference_dir(get_person(person_id)) | |
| manifest_path = person_reference_manifest(get_person(person_id)) | |
| reference_dir.mkdir(parents=True, exist_ok=True) | |
| with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=reference_dir, | |
| suffix=".tmp", delete=False) as handle: | |
| temp_path = Path(handle.name) | |
| json.dump(manifest, handle, indent=2) | |
| try: | |
| temp_path.replace(manifest_path) | |
| except Exception: | |
| temp_path.unlink(missing_ok=True) | |
| raise | |
| def active_reference_score(path, manifest, person_id="brian"): | |
| known = manifest.get("slots", {}) | |
| entry = known.get(path.name, {}) | |
| if entry.get("managed_path") == rel(path) or "score" in entry: | |
| return float(entry.get("score", 0)) | |
| from pick_voice_ref import analyze | |
| try: | |
| quality = analyze(path) | |
| except (OSError, RuntimeError, ValueError): | |
| return 0 | |
| if quality is None: | |
| return 0 | |
| length_bonus = 1.0 - min(abs(quality["seconds"] - 12.0) / 6.0, 1.0) | |
| return 1.5 + quality["speech_ratio"] + 0.5 * length_bonus | |
| def start_reference_review(requested_sources, person_id="brian"): | |
| person = get_person(person_id) | |
| candidates = reference_candidates(requested_sources) | |
| selected_sources = list(dict.fromkeys(requested_sources)) | |
| with REFERENCE_LOCK: | |
| manifest_path = person_reference_manifest(person) | |
| manifest = json.loads(manifest_path.read_text(encoding="utf-8")) \ | |
| if manifest_path.is_file() else {} | |
| manifest["review_sources"] = selected_sources | |
| manifest["review_pool"] = candidates | |
| manifest.setdefault("reviewed", {}) | |
| save_reference_manifest(manifest, person_id) | |
| return next_reference_candidate(manifest) | |
| def next_reference_candidate(manifest): | |
| reviewed = manifest.get("reviewed", {}) | |
| promoted = set(manifest.get("promoted", [])) | |
| active = {entry.get("path") for entry in manifest.get("slots", {}).values()} | |
| remaining = [ | |
| candidate for candidate in manifest.get("review_pool", []) | |
| if candidate.get("path") not in reviewed and candidate.get("path") not in promoted and | |
| candidate.get("path") not in active and Path(ROOT / candidate.get("path", "")).is_file() | |
| ] | |
| result = { | |
| "top_candidates": remaining[:6], | |
| "remaining": len(remaining), | |
| } | |
| if remaining: | |
| return {**result, "candidate": remaining[0]} | |
| return {**result, "candidate": None} | |
| def mark_dataset_clip_deleted(source_root, audio_path): | |
| report_path = source_root / "report.csv" | |
| if not report_path.is_file(): | |
| return | |
| with report_path.open(newline="", encoding="utf-8") as handle: | |
| reader = csv.DictReader(handle) | |
| fieldnames = reader.fieldnames | |
| rows = list(reader) | |
| if not fieldnames or not {"clip_id", "status", "reason"}.issubset(fieldnames): | |
| raise ValueError("The dataset report format is not recognized; the clip was not deleted.") | |
| clip_id = audio_path.stem | |
| matches = [row for row in rows if row.get("clip_id") == clip_id] | |
| if len(matches) != 1: | |
| raise ValueError("The dataset report does not identify this clip uniquely; it was not deleted.") | |
| matches[0]["status"] = "rejected" | |
| matches[0]["reason"] = "deleted_in_studio_review" | |
| with tempfile.NamedTemporaryFile("w", newline="", encoding="utf-8", | |
| dir=source_root, suffix=".tmp", delete=False) as handle: | |
| temp_path = Path(handle.name) | |
| writer = csv.DictWriter(handle, fieldnames=fieldnames) | |
| writer.writeheader() | |
| writer.writerows(rows) | |
| try: | |
| temp_path.replace(report_path) | |
| except Exception: | |
| temp_path.unlink(missing_ok=True) | |
| raise | |
| def review_reference(action, candidate_path, requested_sources, person_id="brian"): | |
| person = get_person(person_id) | |
| reference_dir = person_reference_dir(person) | |
| manifest_path = person_reference_manifest(person) | |
| if action not in {"keep", "delete", "promote"}: | |
| raise ValueError("Choose Keep, Delete, or Promote.") | |
| if not isinstance(candidate_path, str): | |
| raise ValueError("No review clip was selected.") | |
| if not isinstance(requested_sources, list) or not requested_sources: | |
| raise ValueError("Select at least one source folder.") | |
| if not manifest_path.is_file(): | |
| raise ValueError("Start a reference review before choosing an action.") | |
| with REFERENCE_LOCK: | |
| manifest = json.loads(manifest_path.read_text(encoding="utf-8")) | |
| sources = manifest.get("review_sources", []) | |
| if list(dict.fromkeys(requested_sources or [])) != sources: | |
| raise ValueError("The selected sources changed. Restart the reference review.") | |
| available = {item["path"]: path for item, path in profile_sources()} | |
| if not sources or any(source not in available for source in sources): | |
| raise ValueError("A selected source folder is no longer available.") | |
| candidate = next((item for item in manifest.get("review_pool", []) | |
| if item.get("path") == candidate_path), None) | |
| if candidate is None: | |
| raise ValueError("That clip is not in the current review queue.") | |
| allowed_dirs = [available[source] for source in sources] | |
| audio_path = safe_path(candidate_path, allowed_dirs) | |
| if candidate_path in manifest.get("reviewed", {}): | |
| raise ValueError("This clip has already been reviewed.") | |
| if action == "delete": | |
| source_root = max( | |
| (folder for folder in allowed_dirs | |
| if audio_path == folder or folder in audio_path.parents), | |
| key=lambda folder: len(folder.parts), default=None, | |
| ) | |
| if source_root is None: | |
| raise ValueError("The clip is outside the selected source folders.") | |
| report_path = source_root / "report.csv" | |
| report_backup = report_path.read_bytes() if report_path.is_file() else None | |
| if report_path.is_file(): | |
| mark_dataset_clip_deleted(source_root, audio_path) | |
| try: | |
| audio_path.unlink() | |
| except OSError: | |
| if report_backup is not None: | |
| report_path.write_bytes(report_backup) | |
| raise | |
| manifest.setdefault("rejected", []).append(candidate_path) | |
| result_message = "Source clip deleted." | |
| elif action == "keep": | |
| manifest.setdefault("reviewed", {})[candidate_path] = "kept" | |
| result_message = "Clip kept for later; active references unchanged." | |
| else: | |
| slots = manifest.setdefault("slots", {}) | |
| active_items = reference_items(person_id) | |
| occupied = {item["slot"] for item in active_items} | |
| if len(occupied) < REFERENCE_COUNT: | |
| slot = next(f"ref_{index:02d}.wav" for index in range(1, REFERENCE_COUNT + 1) | |
| if f"ref_{index:02d}.wav" not in occupied) | |
| else: | |
| weakest = min(active_items, key=lambda item: ( | |
| active_reference_score(reference_dir / item["name"], manifest, person_id), item["slot"])) | |
| slot = weakest["slot"] | |
| target = reference_dir / slot | |
| backup = target.read_bytes() if target.is_file() else None | |
| staged_path = stage_reference_copy(audio_path, person_id) | |
| try: | |
| staged_path.replace(target) | |
| slots[slot] = {**candidate, "managed_path": rel(target)} | |
| manifest.setdefault("promoted", []).append(candidate_path) | |
| manifest.setdefault("reviewed", {})[candidate_path] = "promoted" | |
| save_reference_manifest(manifest, person_id) | |
| except Exception: | |
| staged_path.unlink(missing_ok=True) | |
| if backup is not None: | |
| target.write_bytes(backup) | |
| else: | |
| target.unlink(missing_ok=True) | |
| raise | |
| result_message = f"Promoted into {slot}, replacing the weakest active reference." | |
| promoted_slot = slot | |
| if action != "promote": | |
| manifest.setdefault("reviewed", {})[candidate_path] = action | |
| save_reference_manifest(manifest, person_id) | |
| next_item = next_reference_candidate(manifest) | |
| result = {"message": result_message, **next_item, "references": reference_items(person_id)} | |
| if action == "promote": | |
| result["promoted_slot"] = promoted_slot | |
| return result | |
| def history(limit=60, approved_only=False): | |
| items = [] | |
| if OUTPUT.exists(): | |
| for meta_path in OUTPUT.glob("*.json"): | |
| wav = meta_path.with_suffix(".wav") | |
| if not wav.exists(): | |
| continue | |
| try: | |
| meta = json.loads(meta_path.read_text(encoding="utf-8")) | |
| except Exception: | |
| continue | |
| if approved_only and not meta.get("approved"): | |
| continue | |
| items.append((wav.stat().st_mtime, entry_for(wav, meta))) | |
| items.sort(key=lambda x: -x[0]) | |
| return [e for _, e in items[:limit]] | |
| def set_approved(file, approved): | |
| wav = safe_path(file, [OUTPUT]) | |
| meta_path = wav.with_suffix(".json") | |
| meta = json.loads(meta_path.read_text(encoding="utf-8")) if meta_path.exists() else {} | |
| meta["approved"] = bool(approved) | |
| meta["approved_at"] = datetime.now().isoformat(timespec="seconds") if approved else None | |
| meta_path.write_text(json.dumps(meta, indent=2), encoding="utf-8") | |
| return entry_for(wav, meta) | |
| def discard_take(file): | |
| if not isinstance(file, str) or not file.lower().endswith(".wav"): | |
| raise ValueError("Choose a Studio audio take to discard.") | |
| wav = safe_path(file, [OUTPUT]) | |
| meta_path = wav.with_suffix(".json") | |
| if not meta_path.is_file(): | |
| raise ValueError("This take has no Studio metadata and cannot be discarded here.") | |
| try: | |
| meta = json.loads(meta_path.read_text(encoding="utf-8")) | |
| except (OSError, json.JSONDecodeError) as e: | |
| raise ValueError("This take's metadata can't be read, so it was not deleted.") from e | |
| if not isinstance(meta, dict) or ( | |
| meta.get("source") != "studio" and | |
| not str(meta.get("note", "")).startswith("AI-generated voice (Chatterbox") | |
| ): | |
| raise ValueError("Only Chatterbox audio takes with matching project metadata can be discarded here.") | |
| wav.unlink() | |
| meta_path.unlink() | |
| return {"deleted": True, "file": rel(wav)} | |
| # ---------------------------------------------------------------- web server | |
| class Handler(BaseHTTPRequestHandler): | |
| engine: Engine = None | |
| server_version = "VoiceStudio/1.0" | |
| def log_message(self, fmt, *args): # keep the terminal quiet except for errors | |
| if args and str(args[1]).startswith(("4", "5")) and "favicon" not in str(args[0]): | |
| super().log_message(fmt, *args) | |
| def _json(self, obj, code=200): | |
| body = json.dumps(obj).encode() | |
| self.send_response(code) | |
| self.send_header("Content-Type", "application/json") | |
| self.send_header("Content-Length", str(len(body))) | |
| self.send_header("Cache-Control", "no-store") | |
| self.end_headers() | |
| self.wfile.write(body) | |
| def _error(self, msg, code=400): | |
| self._json({"error": msg}, code) | |
| def _body(self): | |
| n = int(self.headers.get("Content-Length") or 0) | |
| return json.loads(self.rfile.read(n) or b"{}") if n else {} | |
| def do_GET(self): | |
| url = urlparse(self.path) | |
| path = url.path | |
| try: | |
| if path in ("/", "/index.html", "/mobile", "/mobile/"): | |
| body = HTML.read_bytes() | |
| if path in ("/mobile", "/mobile/"): | |
| body = body.replace( | |
| b'href="/manifest.webmanifest"', | |
| b'href="/mobile-manifest.webmanifest"', | |
| 1, | |
| ) | |
| self.send_response(200) | |
| self.send_header("Content-Type", "text/html; charset=utf-8") | |
| self.send_header("Content-Length", str(len(body))) | |
| self.send_header("Cache-Control", "no-store") | |
| self.end_headers() | |
| self.wfile.write(body) | |
| elif path in APP_ASSETS: | |
| asset, content_type = APP_ASSETS[path] | |
| body = asset.read_bytes() | |
| self.send_response(200) | |
| self.send_header("Content-Type", content_type) | |
| self.send_header("Content-Length", str(len(body))) | |
| self.send_header("Cache-Control", "no-cache") | |
| self.end_headers() | |
| self.wfile.write(body) | |
| elif path == "/api/status": | |
| self._json(self.engine.status()) | |
| elif path == "/api/voices": | |
| self._json(list_voices()) | |
| elif path == "/api/people": | |
| registry = people_registry() | |
| self._json({"people": available_people(), "sources": registry.get("sources", {})}) | |
| elif path == "/api/person/media": | |
| person_id = parse_qs(url.query).get("person", ["brian"])[0] | |
| self._json(person_media_status(person_id)) | |
| elif path == "/api/person/reference-audio": | |
| query = parse_qs(url.query) | |
| person_id = query.get("person", ["brian"])[0] | |
| name = query.get("name", [""])[0] | |
| person = get_person(person_id) | |
| sample = next( | |
| (path for path in person_reference_audio(person) if path.name == name and | |
| re.fullmatch(r"voice_ref[1-5]\.wav", name)), | |
| None, | |
| ) | |
| if sample is None: | |
| return self._error("That voice sample is not available.", 404) | |
| self._file(sample) | |
| elif path == "/api/rebuild/options": | |
| query = parse_qs(url.query) | |
| person = get_person(query.get("person", ["brian"])[0]) | |
| paths = person_profile_paths(person, self.engine.kind) | |
| profile_path, candidate_path = paths["active"], paths["candidate"] | |
| registry = people_registry() | |
| source_items = [item for item, _ in profile_sources()] | |
| selected_sources = registry.get("sources", {}).get(person["id"]) | |
| if selected_sources is None and person["id"] == "brian": | |
| selected_sources = [item["path"] for item in source_items] | |
| elif selected_sources is None: | |
| person_sources = [ | |
| item["path"] for item in source_items | |
| if Path(item["path"]).resolve() == person_reference_dir(person).resolve() | |
| ] | |
| selected_sources = person_sources | |
| meta_path = profile_path.with_suffix(".json") | |
| candidate_meta_path = candidate_path.with_suffix(".json") | |
| profile_meta = json.loads(meta_path.read_text(encoding="utf-8")) if meta_path.is_file() else {} | |
| candidate_meta = json.loads(candidate_meta_path.read_text(encoding="utf-8")) \ | |
| if candidate_meta_path.is_file() else {} | |
| profile_anchor = "" | |
| if profile_meta.get("anchor"): | |
| try: | |
| profile_anchor = Path(profile_meta["anchor"]).resolve().relative_to(ROOT.resolve()).as_posix() | |
| except ValueError: | |
| pass | |
| self._json({ | |
| "person": person, | |
| "sources": [ | |
| {**item, "selected": item["path"] in (selected_sources or [])} | |
| for item in source_items | |
| ], | |
| "anchors": [item for item, _ in profile_anchors(selected_sources or [])], | |
| "profile": {"path": rel(profile_path), "exists": profile_path.is_file(), | |
| "created": profile_meta.get("created"), "clips_used": profile_meta.get("clips_used"), | |
| "anchor": profile_anchor}, | |
| "candidate": {"path": rel(candidate_path), "exists": candidate_path.is_file(), | |
| "created": candidate_meta.get("created"), | |
| "clips_used": candidate_meta.get("clips_used"), | |
| "anchor_agreement": candidate_meta.get("anchor_agreement")}, | |
| }) | |
| elif path == "/api/references": | |
| query = parse_qs(url.query) | |
| person_id = query.get("person", ["brian"])[0] | |
| self._json({"items": reference_items(person_id), "count": REFERENCE_COUNT}) | |
| elif path == "/api/reference-review/audio": | |
| requested = parse_qs(url.query).get("path", [""])[0] | |
| person_id = parse_qs(url.query).get("person", ["brian"])[0] | |
| manifest_path = person_reference_manifest(get_person(person_id)) | |
| if not manifest_path.is_file(): | |
| return self._error("Start a reference review before previewing clips.", 404) | |
| manifest = json.loads(manifest_path.read_text(encoding="utf-8")) | |
| candidate = next((item for item in manifest.get("review_pool", []) | |
| if item.get("path") == requested), None) | |
| sources = manifest.get("review_sources", []) | |
| available = {item["path"]: folder for item, folder in profile_sources()} | |
| if candidate is None or any(source not in available for source in sources): | |
| return self._error("That review clip is not available.", 404) | |
| self._file(safe_path(requested, [available[source] for source in sources])) | |
| elif path.startswith("/api/compare/"): | |
| job = self.engine.compare_jobs.get(path.rsplit("/", 1)[-1]) | |
| self._json(self.engine.public_compare(job)) if job else self._error("No such comparison.", 404) | |
| elif path.startswith("/api/rebuild/"): | |
| job = self.engine.rebuild_jobs.get(path.rsplit("/", 1)[-1]) | |
| self._json(self.engine.public_rebuild(job)) if job else self._error("No such rebuild job.", 404) | |
| elif path.startswith("/api/person/dataset/"): | |
| job = self.engine.dataset_jobs.get(path.rsplit("/", 1)[-1]) | |
| self._json(self.engine.public_dataset(job)) if job else self._error("No such dataset job.", 404) | |
| elif path == "/api/history": | |
| q = parse_qs(url.query) | |
| self._json({"items": history(approved_only=q.get("approved") == ["1"])}) | |
| elif path == "/api/anchor-audio": | |
| query = parse_qs(url.query) | |
| requested = query.get("path", [""])[0] | |
| person = get_person(query.get("person", ["brian"])[0]) | |
| registry = people_registry() | |
| selected_sources = registry.get("sources", {}).get(person["id"]) | |
| if selected_sources is None and person["id"] == "brian": | |
| selected_sources = [item["path"] for item, _ in profile_sources()] | |
| anchors = {item["path"]: audio_path | |
| for item, audio_path in profile_anchors(selected_sources or [])} | |
| anchor_path = anchors.get(requested) | |
| if anchor_path is None: | |
| return self._error("Choose an available anchor clip.", 404) | |
| self._file(anchor_path) | |
| elif path.startswith("/api/jobs/"): | |
| job = self.engine.jobs.get(path.rsplit("/", 1)[-1]) | |
| self._json(self.engine.public(job)) if job else self._error("No such job.", 404) | |
| elif path.startswith("/files/"): | |
| self._file(safe_path(path[len("/files/"):], [OUTPUT, *clip_dirs()])) | |
| elif path == "/favicon.ico": | |
| self.send_response(204) | |
| self.end_headers() | |
| else: | |
| self._error("Not found.", 404) | |
| except ValueError as e: | |
| self._error(str(e), 404) | |
| except (BrokenPipeError, ConnectionResetError): | |
| pass | |
| def do_POST(self): | |
| try: | |
| parsed = urlparse(self.path) | |
| if parsed.path == "/api/person/upload": | |
| self._upload_person_file(parsed) | |
| return | |
| data = self._body() | |
| if self.path == "/api/generate": | |
| voice = data.get("voice") or {} | |
| if voice.get("type") not in ("profile", "clip") or not voice.get("path"): | |
| return self._error("Choose a voice first.") | |
| job = self.engine.submit(data) | |
| self._json(self.engine.public(job)) | |
| elif self.path == "/api/people": | |
| self._json({"person": create_person(data.get("name")), "people": available_people()}, 201) | |
| elif self.path == "/api/person/reference/delete": | |
| self._json(delete_uploaded_voice_sample(data.get("person", "brian"), data.get("name"))) | |
| elif self.path == "/api/person/sources": | |
| person_id = data.get("person", "brian") | |
| self._json({"sources": save_person_sources(person_id, data.get("sources"))}) | |
| elif self.path == "/api/person/prepare-dataset": | |
| self._json(self.engine.submit_dataset(data.get("person", "brian")), 202) | |
| elif self.path == "/api/approve": | |
| self._json(set_approved(data.get("file", ""), data.get("approved", True))) | |
| elif self.path == "/api/discard": | |
| self._json(discard_take(data.get("file", ""))) | |
| elif self.path == "/api/rebuild": | |
| self._json(self.engine.submit_rebuild(data), 202) | |
| elif self.path == "/api/reference-review/start": | |
| self._json(start_reference_review(data.get("sources"), data.get("person", "brian"))) | |
| elif self.path == "/api/reference-review/action": | |
| self._json(review_reference(data.get("action"), data.get("path"), data.get("sources"), | |
| data.get("person", "brian"))) | |
| elif self.path == "/api/compare": | |
| self._json(self.engine.submit_compare(data), 202) | |
| elif self.path == "/api/rebuild/adopt": | |
| self._json(self.engine.adopt_candidate(data.get("person", "brian"))) | |
| else: | |
| self._error("Not found.", 404) | |
| except ValueError as e: | |
| self._error(str(e)) | |
| except Exception as e: | |
| traceback.print_exc() | |
| self._error(f"Something went wrong: {e}", 500) | |
| def _upload_person_file(self, parsed): | |
| query = parse_qs(parsed.query) | |
| person = get_person(query.get("person", [""])[0]) | |
| kind = query.get("kind", [""])[0] | |
| if any( | |
| job.get("person") == person["id"] and job["state"] in ("queued", "running") | |
| for job in self.engine.dataset_jobs.values() | |
| ): | |
| return self._error("Wait for video clip preparation to finish before uploading more files.") | |
| filename = query.get("name", [""])[0].replace("\\", "/").rsplit("/", 1)[-1] | |
| suffix = Path(filename).suffix.lower() | |
| stem = re.sub(r"[^A-Za-z0-9_-]+", "_", Path(filename).stem).strip("_") | |
| if not stem or kind not in {"video", "reference"}: | |
| return self._error("Choose a valid file and upload type.") | |
| if kind == "video" and suffix not in VIDEO_UPLOAD_EXTS: | |
| return self._error("Upload a supported video or audio file.") | |
| if kind == "reference" and suffix != ".wav": | |
| return self._error("Voice samples must be WAV files.") | |
| try: | |
| length = int(self.headers.get("Content-Length", "")) | |
| except ValueError: | |
| return self._error("The upload size is missing or invalid.") | |
| limit = MAX_VIDEO_UPLOAD_BYTES if kind == "video" else MAX_REFERENCE_UPLOAD_BYTES | |
| if length <= 0: | |
| return self._error("The selected file is empty.") | |
| if length > limit: | |
| return self._error( | |
| f"This file exceeds the {limit // (1024 ** 2)} MB upload limit.", 413 | |
| ) | |
| directory = person_video_dir(person) if kind == "video" else person_reference_dir(person) | |
| directory.mkdir(parents=True, exist_ok=True) | |
| temporary = None | |
| try: | |
| with tempfile.NamedTemporaryFile(dir=directory, suffix=".upload", delete=False) as handle: | |
| temporary = Path(handle.name) | |
| remaining = length | |
| while remaining: | |
| chunk = self.rfile.read(min(1024 * 1024, remaining)) | |
| if not chunk: | |
| raise ValueError("The upload ended before the whole file was received.") | |
| handle.write(chunk) | |
| remaining -= len(chunk) | |
| if kind == "reference": | |
| try: | |
| duration = sf.info(str(temporary)).duration | |
| except (OSError, RuntimeError) as e: | |
| raise ValueError("The voice sample is not a readable WAV file.") from e | |
| if not 10 <= duration <= 30: | |
| raise ValueError("Voice samples must be 10-30 seconds long.") | |
| with REFERENCE_LOCK: | |
| occupied = { | |
| path.name for path in directory.iterdir() | |
| if path.is_file() and re.fullmatch( | |
| r"(?:voice_ref[1-5]|ref_0[1-6])\.wav", path.name | |
| ) | |
| } | |
| if len(occupied) >= 5: | |
| raise ValueError("This person already has five voice samples.") | |
| slot = next((f"voice_ref{index}.wav" for index in range(1, 6) | |
| if f"voice_ref{index}.wav" not in occupied), None) | |
| if slot is None: | |
| raise ValueError("This person already has five voice samples.") | |
| target = directory / slot | |
| temporary.replace(target) | |
| registry = people_registry() | |
| sources = registry.setdefault("sources", {}).setdefault(person["id"], []) | |
| source = rel(directory) | |
| if source not in sources: | |
| sources.append(source) | |
| save_people_registry(registry) | |
| else: | |
| target = directory / f"{uuid.uuid4().hex[:8]}_{stem}{suffix}" | |
| temporary.replace(target) | |
| temporary = None | |
| self._json({"file": rel(target), "media": person_media_status(person["id"])}, 201) | |
| finally: | |
| if temporary is not None: | |
| temporary.unlink(missing_ok=True) | |
| def _file(self, p): | |
| """Serve a file with byte-range support, so audio players can seek.""" | |
| size = p.stat().st_size | |
| ctype = mimetypes.guess_type(p.name)[0] or "application/octet-stream" | |
| rng = self.headers.get("Range") | |
| start, end = 0, size - 1 | |
| if rng and (m := re.match(r"bytes=(\d*)-(\d*)", rng)): | |
| if m.group(1): | |
| start = int(m.group(1)) | |
| end = int(m.group(2)) if m.group(2) else end | |
| elif m.group(2): | |
| start = max(0, size - int(m.group(2))) | |
| end = min(end, size - 1) | |
| self.send_response(206) | |
| self.send_header("Content-Range", f"bytes {start}-{end}/{size}") | |
| else: | |
| self.send_response(200) | |
| length = end - start + 1 | |
| self.send_header("Content-Type", ctype) | |
| self.send_header("Accept-Ranges", "bytes") | |
| self.send_header("Content-Length", str(length)) | |
| self.send_header("Cache-Control", "no-store") | |
| if parse_qs(urlparse(self.path).query).get("download"): | |
| self.send_header("Content-Disposition", f'attachment; filename="{p.name}"') | |
| self.end_headers() | |
| with open(p, "rb") as f: | |
| f.seek(start) | |
| remaining = length | |
| while remaining > 0: | |
| chunk = f.read(min(65536, remaining)) | |
| if not chunk: | |
| break | |
| self.wfile.write(chunk) | |
| remaining -= len(chunk) | |
| def main(): | |
| p = argparse.ArgumentParser(description="Local voice cloning and generation studio.") | |
| p.add_argument("--model", choices=["standard", "turbo"], default="standard") | |
| p.add_argument("--device", default="auto") | |
| p.add_argument("--host", default="127.0.0.1") | |
| p.add_argument("--port", type=int, default=7860) | |
| p.add_argument("--no-browser", action="store_true") | |
| args = p.parse_args() | |
| if not HTML.exists(): | |
| raise SystemExit(f"studio.html is missing. Put it next to studio.py ({HTML.parent}).") | |
| device = speak.pick_device(args.device) | |
| Handler.engine = Engine(args.model, device) | |
| server = ThreadingHTTPServer((args.host, args.port), Handler) | |
| url = f"http://{'127.0.0.1' if args.host in ('0.0.0.0', '') else args.host}:{args.port}" | |
| print(f"\nVoice Studio running at {url}") | |
| if args.host not in ("127.0.0.1", "localhost"): | |
| print("Others on your network can open it at http://<this computer's IP>:" | |
| f"{args.port}. Anyone who can reach it can generate audio, so only do this on a trusted network.") | |
| print("The voice model is loading in the background. Press Ctrl+C to stop.\n") | |
| if not args.no_browser: | |
| threading.Timer(1.0, lambda: webbrowser.open(url)).start() | |
| try: | |
| server.serve_forever() | |
| except KeyboardInterrupt: | |
| print("\nStopped.") | |
| if __name__ == "__main__": | |
| main() |