Pipeline: env overrides for endpoints and dirs; voice health check, retries, load-on-demand; refuse speed_factor

_pipeline/settings.py

@@ -11,9 +11,35 @@ def load(path=None):
     with open(path, encoding="utf-8") as f:
         cfg = yaml.safe_load(f)
     cfg["_path"] = path
+    apply_env(cfg)
     return cfg
 
 
+# Environment variables that override settings.yaml (used by the CI job; nothing is hard-coded to a host).
+ENV_OVERRIDES = [
+    ("AUDIOBOOK_BOOKS_DIR", ("books_dir",)),
+    ("AUDIOBOOK_WORK_DIR", ("work_dir",)),
+    ("AUDIOBOOK_ARCHIVE_DIR", ("archive_dir",)),
+    ("AUDIOBOOK_VOICE_ENDPOINT", ("voice", "endpoint")),
+    ("AUDIOBOOK_LLM_ENDPOINT", ("llm", "endpoint")),
+    ("AUDIOBOOK_LLM_MODEL", ("llm", "model")),
+]
+
+
+def apply_env(cfg, environ=None):
+    environ = os.environ if environ is None else environ
+    for var, keys in ENV_OVERRIDES:
+        value = environ.get(var, "").strip()
+        if value:
+            node = cfg
+            for k in keys[:-1]:
+                node = node[k]
+            node[keys[-1]] = value
+    flag = environ.get("AUDIOBOOK_REQUIRE_OCR", "").strip().lower()
+    if flag:
+        cfg["ocr"]["required"] = flag in ("1", "true", "yes")
+
+
 def resolve(cfg, key):
     """A path setting as an absolute path (relative values are relative to the repo root)."""
     value = cfg[key]

_pipeline/settings.yaml

@@ -27,7 +27,7 @@ voice:
   # LOCKED by Ty 2026-10-17 for MVP ("clip B"): Chatterbox, stock voice Michael, native settings only.
   # Never set speed_factor: the server applies it as a time-stretch after generation and ruins the sound.
   api: chatterbox_tts        # chatterbox_tts = native /tts route; anything else = OpenAI-style /v1/audio/speech
-  endpoint: http://voice:8004/tts
+  endpoint: http://voice:8004/tts   # override with env AUDIOBOOK_VOICE_ENDPOINT (CI uses the LAN address)
   model: chatterbox
   voice: Michael.wav
   extra:                     # extra JSON fields sent with every request
@@ -38,12 +38,16 @@ voice:
     split_text: true
     chunk_size: 240          # the server splits long text at sentence ends into chunks of about this size
   timeout_s: 600
+  retries: 3                 # a failed request is retried this many times (5 s, then 15 s, then 45 s apart) before the run fails
+  load_if_needed: true       # if the server is up but its model is unloaded, ask it to load (POST /restart_server) before rendering
+  unload_when_done: false    # true = free the GPU after a render (POST /api/unload); the next run loads it again
   max_chunk_chars: 100000    # one request per paragraph; the server does the sentence splitting (chunk_size above)
   gap_sentence_s: 0.25       # pause between pieces inside a paragraph
   gap_paragraph_s: 0.7       # pause between paragraphs and after cues and labels
 
 ocr:
   enabled: true              # run ocrmypdf only when a page has no text layer
+  required: true             # pages without a text layer and no working OCR = the run fails (env AUDIOBOOK_REQUIRE_OCR=0 to allow)
   language: eng
   min_chars_per_page: 10     # fewer characters than this = no text layer
 

_pipeline/voice.py

@@ -3,6 +3,9 @@ Pieces are sentence-sized. Output goes to scratch (work_dir), with a manifest th
 import json
 import os
 import re
+import time
+import urllib.error
+import urllib.parse
 import urllib.request
 
 import bookio
@@ -53,6 +56,63 @@ def plan(cfg, segs):
     return items
 
 
+class VoiceError(RuntimeError):
+    """The voice endpoint cannot be reached or refused the request. The message says what to check."""
+
+
+def base_url(cfg):
+    u = urllib.parse.urlsplit(cfg["voice"]["endpoint"])
+    return "%s://%s" % (u.scheme, u.netloc)
+
+
+def _get_json(url, timeout=15):
+    with urllib.request.urlopen(url, timeout=timeout) as r:
+        return json.load(r)
+
+
+def health(cfg):
+    """Prove the voice server is reachable without rendering anything. Uses the Chatterbox server's own
+    read-only routes: GET /api/model-info (is a model loaded) and GET /get_predefined_voices (is our voice there).
+    Returns a dict. Raises VoiceError when the server cannot be reached or the voice is missing."""
+    base, name = base_url(cfg), cfg["voice"]["voice"]
+    try:
+        info = _get_json(base + "/api/model-info")
+        voices = _get_json(base + "/get_predefined_voices")
+    except (urllib.error.URLError, OSError, ValueError) as e:
+        raise VoiceError("voice server not reachable at %s (%s). Check that the voice container is running and that "
+                         "AUDIOBOOK_VOICE_ENDPOINT / voice.endpoint is right." % (base, e))
+    names = [v.get("filename") for v in voices if isinstance(v, dict)]
+    if name not in names:
+        raise VoiceError("voice server at %s has no voice file '%s' (it has: %s)" % (base, name, ", ".join(map(str, names)) or "none"))
+    return {"base": base, "model_loaded": bool(info.get("loaded")), "model": info.get("class_name"),
+            "device": info.get("device"), "voice": name}
+
+
+def ensure_loaded(cfg):
+    """Health check, then load the model if the server is up but the model is unloaded (setting voice.load_if_needed)."""
+    h = health(cfg)
+    if not h["model_loaded"]:
+        if not cfg["voice"].get("load_if_needed", True):
+            raise VoiceError("voice server is up but its model is not loaded, and voice.load_if_needed is false")
+        print("voice model not loaded; asking the server to load it (POST /restart_server)")
+        req = urllib.request.Request(h["base"] + "/restart_server", b"{}", {"Content-Type": "application/json"}, method="POST")
+        try:
+            with urllib.request.urlopen(req, timeout=600) as r:
+                r.read()
+        except (urllib.error.URLError, OSError) as e:
+            raise VoiceError("voice server could not load its model: %s" % e)
+        h = health(cfg)
+        if not h["model_loaded"]:
+            raise VoiceError("voice server still reports no model loaded after /restart_server")
+    return h
+
+
+def unload(cfg):
+    req = urllib.request.Request(base_url(cfg) + "/api/unload", b"", method="POST")
+    with urllib.request.urlopen(req, timeout=120) as r:
+        r.read()
+
+
 def speak(cfg, text):
     v = cfg["voice"]
     if v.get("api") == "chatterbox_tts":
@@ -61,23 +121,40 @@ def speak(cfg, text):
     else:
         body = {"model": v["model"], "input": text, "voice": v["voice"], "response_format": "wav"}
     body.update(v.get("extra") or {})
+    if "speed_factor" in body:
+        raise VoiceError("speed_factor must never be set (it time-stretches after generation and ruins the sound)")
+    last = None
+    for attempt in range(int(v.get("retries", 3)) + 1):
         req = urllib.request.Request(v["endpoint"], json.dumps(body).encode(), {"Content-Type": "application/json"})
+        try:
             with urllib.request.urlopen(req, timeout=v["timeout_s"]) as r:
                 data = r.read()
             if data[:4] != b"RIFF":
-        raise RuntimeError("voice endpoint did not return a wav file")
+                raise VoiceError("voice endpoint did not return a wav file")
             return data
+        except urllib.error.HTTPError as e:
+            last = "HTTP %d %s" % (e.code, e.read()[:200].decode("utf-8", "replace"))
+            if e.code < 500:
+                break
+        except (urllib.error.URLError, OSError) as e:
+            last = str(e)
+        if attempt < int(v.get("retries", 3)):
+            time.sleep(5 * 3 ** attempt)
+    raise VoiceError("voice request to %s failed after %d attempts: %s" % (v["endpoint"], int(v.get("retries", 3)) + 1, last))
 
 
 def work_dir_for(cfg, book_dir, name):
     return os.path.join(settings.resolve(cfg, "work_dir"), os.path.basename(book_dir), "audio", name)
 
 
-def run(cfg, book_dir, only=None):
+def run(cfg, book_dir, only=None, names=None):
+    """names: render only these script names (None = all, or the one in `only`)."""
     book = bookio.read_book(book_dir)
     for name in book["audio_files"]:
         if only and name != only:
             continue
+        if names is not None and name not in names:
+            continue
         with open(os.path.join(book_dir, "script", name + ".json"), encoding="utf-8") as f:
             items = plan(cfg, json.load(f))
         out = work_dir_for(cfg, book_dir, name)