File size: 2,735 Bytes
d801da2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 | """Explicit real-model unknown-symbol pause + short-first-batch acceptance."""
import argparse
from datetime import datetime, timezone
import json
from pathlib import Path
import sys
import threading
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from tts.engine import SpeechEngine
from tts.text import SpeechError
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--assets", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
args.output.mkdir(parents=True, exist_ok=True)
engine = SpeechEngine(args.assets, ROOT / "assets.lock.json")
from mutagen.mp3 import MP3
report = {"at": datetime.now(timezone.utc).isoformat(), "engine": engine.info, "cases": [], "scope": "Real local CPU encoded MP3 and complete source ranges; not manual audible/production latency acceptance"}
cases = (("mixed", "校长读长文。𠮷※★😀αЖ，CPU和API继续。"), ("pause-only", "𠮷※★😀αЖ。"),
("control", "甲\f乙，后文继续。"), ("short-first", "这是一篇长文，校长文章写得很好，我得亲自处理。"))
for name, text in cases:
file = args.output / (name + ".mp3")
published = []
result = engine.synthesize(text, "zf_001", 1, file, threading.Event(), progress=published.append, unsupported_policy="pause")
assert result["segments"][0]["start"] == 0
assert result["segments"][-1]["end"] == len(text.encode("utf-16-le")) // 2
assert all(a["end"] == b["start"] for a, b in zip(result["segments"], result["segments"][1:]))
assert len(published) == len(result["segments"])
assert abs(MP3(file).info.length - result["audio_seconds"]) < .25
for part in published:
clip = file.with_name(f"{file.stem}.{part['index']}.mp3")
assert abs(MP3(clip).info.length - part["audio_seconds"]) < .25
if name == "mixed": assert result["unsupported_phoneme_count"] >= 6 and result["batches"] > 0
if name == "pause-only": assert result["batches"] == 0
report["cases"].append({"name": name, "result": result})
print(name, "PASS", "duration", result["audio_seconds"], "first-part", result["first_audio_seconds"], flush=True)
try:
engine.synthesize(cases[0][1], "zf_001", 1, args.output / "strict.mp3", threading.Event())
except SpeechError as error:
assert error.code == "unsupported_text"
else:
raise AssertionError("Default strict API must remain strict")
(args.output / "acceptance.json").write_text(json.dumps(report, ensure_ascii=False, indent=2))
if __name__ == "__main__": main()
|