258 lines
9.3 KiB
Python
258 lines
9.3 KiB
Python
"""Saved speech audio.
|
|
|
|
Every line the cloud voice synthesises is written here as an ordinary `.wav`
|
|
file with a readable name, plus an `index.json` describing it. That gives three
|
|
things a hidden cache could not:
|
|
|
|
* **Instant replay.** A saved line plays with no synthesis, no network, no
|
|
wait - just read the file and play it.
|
|
* **Files you can actually use.** Open the folder, double-click a `.wav`, drop
|
|
one into a video edit or a stand's playlist. Nothing is locked in.
|
|
* **Works offline.** Once a line is saved, the demo no longer needs internet
|
|
for that line.
|
|
|
|
Layout:
|
|
|
|
audio_library/
|
|
index.json
|
|
good-afternoon-and-welcome-to-our-a1b2c3d4.wav
|
|
please-follow-me-to-the-first-9f8e7d6c.wav
|
|
|
|
The id is a hash of (text, voice, model, style), so the same line spoken by a
|
|
different voice is a different entry, and re-saving an identical line is free.
|
|
|
|
NOTE: only the cloud voice produces files. The built-in OS voices play straight
|
|
to the sound card and never hand us audio, so utterances spoken by them are not
|
|
saved. `MOCK_VOICE_ENGINE=gemini` is what fills this library.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import struct
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
INDEX_FILE = "index.json"
|
|
_SLUG_STRIP = re.compile(r"[^a-z0-9]+")
|
|
|
|
|
|
def slugify(text: str, limit: int = 40) -> str:
|
|
"""A short, filesystem-safe, human-readable stem for a filename."""
|
|
ascii_text = text.encode("ascii", "ignore").decode("ascii").lower()
|
|
slug = _SLUG_STRIP.sub("-", ascii_text).strip("-")
|
|
if len(slug) > limit:
|
|
slug = slug[:limit].rsplit("-", 1)[0] or slug[:limit]
|
|
return slug or "utterance"
|
|
|
|
|
|
def wav_duration(path: Path) -> float:
|
|
"""Duration of a 16-bit PCM WAV, from its header."""
|
|
try:
|
|
with open(path, "rb") as handle:
|
|
header = handle.read(44)
|
|
if len(header) < 44:
|
|
return 0.0
|
|
channels, sample_rate = struct.unpack("<HI", header[22:28])
|
|
data_bytes = struct.unpack("<I", header[40:44])[0]
|
|
return round(data_bytes / float(max(1, sample_rate * channels * 2)), 2)
|
|
except Exception: # pragma: no cover
|
|
return 0.0
|
|
|
|
|
|
class AudioLibrary:
|
|
"""Saved `.wav` files plus their index. Safe to share across threads."""
|
|
|
|
def __init__(self, root: Path) -> None:
|
|
self.root = Path(root)
|
|
self.root.mkdir(parents=True, exist_ok=True)
|
|
self._lock = threading.RLock()
|
|
self._index: Dict[str, Dict[str, Any]] = {}
|
|
self._stamp: Optional[tuple] = None
|
|
self._load_index()
|
|
|
|
# -- identity ------------------------------------------------------------ #
|
|
@staticmethod
|
|
def make_id(text: str, voice: str, model: str, style: str = "") -> str:
|
|
raw = "\x00".join([model or "", voice or "", style or "", text or ""])
|
|
return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]
|
|
|
|
# -- index --------------------------------------------------------------- #
|
|
@property
|
|
def index_path(self) -> Path:
|
|
return self.root / INDEX_FILE
|
|
|
|
def _stamp_of_index(self) -> Optional[tuple]:
|
|
try:
|
|
stat = self.index_path.stat()
|
|
return (stat.st_mtime_ns, stat.st_size)
|
|
except OSError:
|
|
return None
|
|
|
|
def _refresh(self) -> None:
|
|
"""Re-read the index if another process changed it.
|
|
|
|
`warm_voice.py` and the server both open this library. Without this the
|
|
server would keep a stale copy in memory and write it back over the
|
|
other process's changes - resurrecting entries whose files were deleted.
|
|
"""
|
|
if self._stamp_of_index() != self._stamp:
|
|
self._load_index()
|
|
|
|
def _load_index(self) -> None:
|
|
self._stamp = self._stamp_of_index()
|
|
if not self.index_path.exists():
|
|
self._index = {}
|
|
return
|
|
try:
|
|
data = json.loads(self.index_path.read_text(encoding="utf-8"))
|
|
entries = data.get("entries", data) if isinstance(data, dict) else {}
|
|
if isinstance(entries, dict):
|
|
# Drop entries whose file was deleted by hand.
|
|
self._index = {
|
|
key: value for key, value in entries.items()
|
|
if (self.root / value.get("file", "")).exists()
|
|
}
|
|
except (json.JSONDecodeError, OSError) as exc:
|
|
logger.warning("audio index unreadable (%s); starting a fresh one", exc)
|
|
self._index = {}
|
|
|
|
def _save_index(self) -> None:
|
|
payload = {"version": 1, "entries": self._index}
|
|
temporary = self.index_path.with_suffix(".tmp")
|
|
try:
|
|
temporary.write_text(
|
|
json.dumps(payload, indent=1, ensure_ascii=False), encoding="utf-8"
|
|
)
|
|
os.replace(temporary, self.index_path)
|
|
self._stamp = self._stamp_of_index()
|
|
except OSError as exc: # pragma: no cover
|
|
logger.warning("could not write audio index: %s", exc)
|
|
|
|
# -- lookup -------------------------------------------------------------- #
|
|
def get(self, audio_id: str) -> Optional[Dict[str, Any]]:
|
|
with self._lock:
|
|
self._refresh()
|
|
entry = self._index.get(audio_id)
|
|
if entry is None:
|
|
return None
|
|
if not (self.root / entry["file"]).exists():
|
|
self._index.pop(audio_id, None)
|
|
self._save_index()
|
|
return None
|
|
return dict(entry)
|
|
|
|
def find(self, text: str, voice: str, model: str, style: str = "") -> Optional[Dict[str, Any]]:
|
|
return self.get(self.make_id(text, voice, model, style))
|
|
|
|
def path_of(self, audio_id: str) -> Optional[Path]:
|
|
entry = self.get(audio_id)
|
|
return self.root / entry["file"] if entry else None
|
|
|
|
def list(self, newest_first: bool = True) -> List[Dict[str, Any]]:
|
|
with self._lock:
|
|
self._refresh()
|
|
# Only list clips whose file is really there - the panel offers a
|
|
# play button per row, and a row that 404s is worse than no row.
|
|
alive, missing = [], []
|
|
for key, value in self._index.items():
|
|
(alive if (self.root / value.get("file", "")).exists() else missing).append(
|
|
(key, value)
|
|
)
|
|
if missing:
|
|
for key, _ in missing:
|
|
self._index.pop(key, None)
|
|
self._save_index()
|
|
items = [dict(value) for _, value in alive]
|
|
items.sort(key=lambda item: item.get("createdAt", 0), reverse=newest_first)
|
|
return items
|
|
|
|
# -- writing ------------------------------------------------------------- #
|
|
def save(self, text: str, wav_bytes: bytes, voice: str, model: str,
|
|
style: str = "") -> Dict[str, Any]:
|
|
audio_id = self.make_id(text, voice, model, style)
|
|
with self._lock:
|
|
self._refresh()
|
|
existing = self.get(audio_id)
|
|
if existing is not None:
|
|
return existing
|
|
|
|
filename = "{0}-{1}.wav".format(slugify(text), audio_id[:8])
|
|
destination = self.root / filename
|
|
temporary = destination.with_suffix(".part")
|
|
temporary.write_bytes(wav_bytes)
|
|
os.replace(temporary, destination) # never index a half-written file
|
|
|
|
entry = {
|
|
"id": audio_id,
|
|
"text": text,
|
|
"file": filename,
|
|
"voice": voice,
|
|
"model": model,
|
|
"style": style,
|
|
"createdAt": time.time(),
|
|
"durationSeconds": wav_duration(destination),
|
|
"bytes": destination.stat().st_size,
|
|
}
|
|
with self._lock:
|
|
self._index[audio_id] = entry
|
|
self._save_index()
|
|
return dict(entry)
|
|
|
|
# -- housekeeping -------------------------------------------------------- #
|
|
def delete(self, audio_id: str) -> bool:
|
|
with self._lock:
|
|
self._refresh()
|
|
entry = self._index.pop(audio_id, None)
|
|
if entry is None:
|
|
return False
|
|
self._save_index()
|
|
try:
|
|
(self.root / entry["file"]).unlink()
|
|
except OSError: # pragma: no cover
|
|
pass
|
|
return True
|
|
|
|
def clear(self) -> int:
|
|
with self._lock:
|
|
self._refresh()
|
|
entries = list(self._index.values())
|
|
self._index = {}
|
|
self._save_index()
|
|
for entry in entries:
|
|
try:
|
|
(self.root / entry["file"]).unlink()
|
|
except OSError: # pragma: no cover
|
|
pass
|
|
return len(entries)
|
|
|
|
def stats(self) -> Dict[str, Any]:
|
|
items = self.list()
|
|
return {
|
|
"count": len(items),
|
|
"bytes": sum(item.get("bytes", 0) for item in items),
|
|
"seconds": round(sum(item.get("durationSeconds", 0) for item in items), 1),
|
|
"dir": str(self.root),
|
|
}
|
|
|
|
|
|
_library: Optional[AudioLibrary] = None
|
|
|
|
|
|
def get_audio_library(root: Optional[Path] = None) -> AudioLibrary:
|
|
"""Process-wide library, so the API and the voice share one index."""
|
|
global _library
|
|
if _library is None:
|
|
if root is None:
|
|
root = Path(__file__).resolve().parents[2] / "audio_library"
|
|
_library = AudioLibrary(root)
|
|
return _library
|