"""Preset-voice MiMo WAV example. No request runs on import. prepare/save are local; synthesize requires --execute and a unique run directory. No retries. A saved WAV proves structural decoding, not voice quality. """ from __future__ import annotations import argparse import base64 import binascii import io import json import os import sys import time import wave from datetime import datetime, timezone from pathlib import Path from urllib.error import HTTPError, URLError from urllib.parse import urlsplit, quote from urllib.request import Request, HTTPRedirectHandler, build_opener from http.client import HTTPException class WorkflowError(Exception): pass class NoRedirect(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None def open_url(request, timeout): # Redirects are rejected; credentials never follow a redirect to another host. return build_opener(NoRedirect()).open(request, timeout=timeout) def exclusive(path: Path, data: bytes): path.parent.mkdir(parents=True, exist_ok=True) with path.open("xb") as f: f.write(data) def read_json(path): value = json.loads(Path(path).read_text(encoding="utf-8")) if not isinstance(value, dict): raise WorkflowError("Expected a JSON object") return value def json_bytes(value): return (json.dumps(value, indent=2, ensure_ascii=False) + "\n").encode("utf-8") def require_execute(args): if not args.execute: raise WorkflowError("Network command requires --execute; it may use a billable API") key = os.environ.get("TOKENHOT_API_KEY") if not key: raise WorkflowError("Set TOKENHOT_API_KEY in the environment") return key def api_request(method, url, payload, key, run_dir, stem="response"): # run_dir is reserved by the command BEFORE any request. Do not reuse it. body_path = run_dir / (stem + ".json") meta_path = run_dir / (stem + ".meta.json") if body_path.exists() or meta_path.exists(): raise WorkflowError("Request records already exist; refusing to send again") meta = {"method": method, "endpoint": url.split("?")[0], "started_at": datetime.now(timezone.utc).isoformat(), "http_status": None, "body_saved": False} exclusive(meta_path, json_bytes(meta)) request = Request(url, data=json_bytes(payload) if payload is not None else None, headers={"Authorization": "Bearer " + key, "Content-Type": "application/json"}, method=method) try: try: response = open_url(request, timeout=120) except HTTPError as error: response = error with response: meta["http_status"] = response.code meta["request_id"] = response.headers.get("x-request-id") raw = response.read(64 * 1024 * 1024 + 1) if len(raw) > 64 * 1024 * 1024: raise WorkflowError("Response exceeded the example's local 64 MiB limit") exclusive(body_path, raw) meta["body_saved"] = True except (OSError, URLError, HTTPException, WorkflowError) as error: meta["transport_failure"] = type(error).__name__ meta_path.write_bytes(json_bytes(meta)) raise WorkflowError("Transport/read failure; inspect local metadata. " "A POST outcome can be unknown. No automatic retry.") from None meta_path.write_bytes(json_bytes(meta)) if not (200 <= meta["http_status"] < 300): raise WorkflowError("HTTP rejection; raw body is saved for private inspection, not echoed") try: value = json.loads(raw) except (ValueError, UnicodeError): raise WorkflowError("Invalid JSON; raw response is preserved") from None if not isinstance(value, dict): raise WorkflowError("Non-object JSON; raw response is preserved") return value ENDPOINT = "https://api.tokenhot.ai/v1/chat/completions" MODEL = "mimo-v2.5-tts" VOICES = ("Mia", "Chloe", "Milo", "Dean", "冰糖", "茉莉", "苏打", "白桦") def make_payload(text, voice="Mia", style="Read clearly at a steady conversational pace."): if not isinstance(text, str) or not text.strip(): raise WorkflowError("Spoken text must not be empty") if voice not in VOICES: raise WorkflowError("Choose an explicit preset listed by this example") messages = [] if style: messages.append({"role": "user", "content": style}) messages.append({"role": "assistant", "content": text}) return {"model": MODEL, "messages": messages, "audio": {"format": "wav", "voice": voice}} def wav_bytes(result): if not isinstance(result, dict): raise WorkflowError("Expected an audio response object") try: choice = result["choices"][0] if not isinstance(choice, dict): raise WorkflowError("Expected a choice object") if choice.get("finish_reason") != "stop": raise WorkflowError("Audio response did not show normal completion") encoded = choice["message"]["audio"]["data"] except (KeyError, IndexError, TypeError): raise WorkflowError("Missing choices[0].message.audio.data") from None if not isinstance(encoded, str) or not encoded: raise WorkflowError("Audio data is empty or is not a Base64 string") try: raw = base64.b64decode(encoded, validate=True) except (ValueError, binascii.Error): raise WorkflowError("Invalid Base64; inspect the saved response") from None try: with wave.open(io.BytesIO(raw), "rb") as w: info = {"channels": w.getnchannels(), "sample_width_bytes": w.getsampwidth(), "sample_rate_hz": w.getframerate(), "frames": w.getnframes()} if w.getcomptype() != "NONE" or min(info.values()) <= 0: raise WorkflowError("Expected nonempty uncompressed PCM WAV") # getnframes() rounds down; read one extra frame to catch a partial tail. frames = w.readframes(info["frames"] + 1) expected = info["frames"] * info["channels"] * info["sample_width_bytes"] if len(frames) != expected: raise WorkflowError("Truncated or misaligned WAV frame data") except (wave.Error, EOFError): raise WorkflowError("Not a supported PCM WAV container; do not add a guessed header") from None return raw, info def save_wav(result, output): if output.exists(): raise WorkflowError("Output already exists; choose a fresh WAV path") raw, info = wav_bytes(result) exclusive(output, raw) return info def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) sub = parser.add_subparsers(dest="command", required=True) p = sub.add_parser("prepare"); p.add_argument("--text-file", required=True) p.add_argument("--voice", default="Mia", choices=VOICES) p.add_argument("--style", default="Read clearly at a steady conversational pace.") p.add_argument("--output", required=True) p = sub.add_parser("synthesize"); p.add_argument("--request", required=True) p.add_argument("--run", required=True); p.add_argument("--execute", action="store_true") p = sub.add_parser("save"); p.add_argument("--response", required=True) p.add_argument("--output", required=True) args = parser.parse_args(argv) try: if args.command == "prepare": body = make_payload(Path(args.text_file).read_text(encoding="utf-8"), args.voice, args.style) exclusive(Path(args.output), json_bytes(body)) print("Prepared JSON locally; no request sent") elif args.command == "save": info = save_wav(read_json(args.response), Path(args.output)) print(json.dumps({"wav_structure": info, "listening_quality": "not_checked"})) else: key = require_execute(args) payload = read_json(args.request) if payload.get("model") != MODEL or (not isinstance(payload.get("audio"), dict) or payload["audio"].get("format") != "wav"): raise WorkflowError("This example only submits the preset model in WAV mode") run = Path(args.run); run.mkdir(parents=True, exist_ok=False) exclusive(run / "request.json", json_bytes(payload)) result = api_request("POST", ENDPOINT, payload, key, run) info = save_wav(result, run / "speech.wav") print(json.dumps({"wav_structure": info, "listening_quality": "not_checked"})) return 0 except (WorkflowError, OSError, ValueError, TypeError, HTTPException) as error: # Never echo arbitrary transport messages, API bodies or credentials. text = str(error) if isinstance(error, WorkflowError) else type(error).__name__ print(text, file=sys.stderr); return 2 if __name__ == "__main__": raise SystemExit(main())