| """Raw-text convenience layer; no Torch import, jobs, installs, or downloads. | |
| Supply a G2P callable returning (full_phonemes, tokens), or an explicit FULL | |
| phoneme_override. No per-word dictionary, chunking, truncation, cached fallback, | |
| or phoneme cleanup is performed. Optional offline Misaki uses preinstalled | |
| American small-model assets; those external assets are NOT in the TTS budget. | |
| This optional module is not yet part of the frozen export manifest. | |
| """ | |
| from contextlib import contextmanager | |
| import hashlib | |
| import importlib | |
| import importlib.metadata | |
| import importlib.util | |
| from pathlib import Path | |
| import socket | |
| import threading | |
| VERSIONS = {"misaki":"0.9.4", "spacy":"3.8.16", "en_core_web_sm":"3.8.0"} | |
| _OFFLINE_LOCK = threading.RLock() | |
| class FrontendUnavailable(RuntimeError): | |
| pass | |
| def _require(ok, message): | |
| if not ok: raise ValueError(message) | |
| def prepare_phonemes(text, vocab, g2p=None, phoneme_override=None, max_phonemes=510): | |
| """Same full-override semantics as native27_phoneme_frontend; no fallback. | |
| Automatic source is user-g2p unless produced by our qualified Misaki wrapper. | |
| Display text is returned unchanged, including whitespace/punctuation. | |
| """ | |
| _require(isinstance(text,str) and bool(text.strip()) and len(text)<=20000, | |
| "Nonempty display text required, at most20000 characters") | |
| _require(type(max_phonemes) is int and 1<=max_phonemes<=510, "Phoneme cap must be1..510") | |
| _require(isinstance(vocab,dict) and bool(vocab) and all(isinstance(c,str) and len(c)==1 | |
| and type(i) is int and 0<i<178 for c,i in vocab.items()), "Native phoneme vocabulary required") | |
| if phoneme_override is not None: | |
| _require(isinstance(phoneme_override,str), "Phoneme override must be a string") | |
| phones,source = phoneme_override,"explicit-override" | |
| else: | |
| if not callable(g2p): | |
| raise FrontendUnavailable("Supply a G2P callable or explicit full phoneme_override; no automatic installation") | |
| result = g2p(text) | |
| _require(isinstance(result,tuple) and len(result)==2 and isinstance(result[0],str), | |
| "G2P must return (full_phoneme_string, tokens)") | |
| phones = result[0] | |
| source = "misaki" if isinstance(g2p,OfflineMisaki) else "user-g2p" | |
| _require(bool(phones) and len(phones)<=max_phonemes, "Phoneme length outside supported bounds; no truncation") | |
| missing = sorted(set(phones)-set(vocab)) | |
| _require(not missing, "Unsupported phonemes: "+repr(missing)) | |
| return dict(text=text,ps=phones,phones=phones,phoneme_source=source, | |
| ids=[0,*[vocab[c] for c in phones],0]) | |
| def synthesize_text(model, text, voice="hope", *, g2p=None, phoneme_override=None, | |
| speed=1., seed=None, max_phonemes=510): | |
| prepared = prepare_phonemes(text,getattr(model,"vocab",None),g2p,phoneme_override,max_phonemes) | |
| _require(callable(getattr(model,"synthesize_phonemes",None)), "Phoneme-input deployment model required") | |
| rendered = model.synthesize_phonemes(prepared["ps"],voice,speed=speed,seed=seed) | |
| _require(isinstance(rendered,dict) and "audio" in rendered, "Deployment must return an audio result dict") | |
| result = dict(rendered,**prepared) | |
| metadata = rendered.get("metadata",{}) | |
| _require(isinstance(metadata,dict), "Deployment metadata must be an object") | |
| result["metadata"] = dict(metadata,input="raw_text" if phoneme_override is None else "text-with-explicit-phonemes", | |
| raw_text_g2p=phoneme_override is None,phoneme_source=prepared["phoneme_source"], | |
| full_phoneme_override=phoneme_override is not None, | |
| external_g2p_assets_counted_in_tts_parameters=False) | |
| return result | |
| def _deny_network(): | |
| """Scoped process-wide socket guard; our G2P calls serialize and restore it.""" | |
| with _OFFLINE_LOCK: | |
| targets=[(socket.socket,name) for name in ("connect","connect_ex","sendto")]+[(socket,"getaddrinfo")] | |
| saved=[(obj,name,getattr(obj,name)) for obj,name in targets] | |
| def denied(*args,**kwargs): | |
| raise FrontendUnavailable("Offline G2P network/download attempt blocked") | |
| try: | |
| for obj,name,_ in saved: setattr(obj,name,denied) | |
| yield | |
| finally: | |
| for obj,name,value in saved: setattr(obj,name,value) | |
| def _installed_runtime(): | |
| packages = {} | |
| for name,expected in VERSIONS.items(): | |
| try: | |
| spec = importlib.util.find_spec(name) | |
| version = importlib.metadata.version(name.replace("_","-")) | |
| except (ImportError,ValueError,importlib.metadata.PackageNotFoundError) as exc: | |
| raise FrontendUnavailable("Preinstalled offline package required: "+name) from exc | |
| if spec is None or not spec.origin or version!=expected: | |
| raise FrontendUnavailable("Require preinstalled "+name+"=="+expected+"; no download/version fallback") | |
| path = Path(spec.origin) | |
| packages[name] = dict(version=version,init_path=str(path),init_sha256=hashlib.sha256(path.read_bytes()).hexdigest()) | |
| if name=="misaki": | |
| en = path.parent/"en.py" | |
| if not en.is_file(): raise FrontendUnavailable("Installed Misaki en.py is missing") | |
| packages[name]["en_sha256"] = hashlib.sha256(en.read_bytes()).hexdigest() | |
| return dict(packages=packages,trf=False,british=False,model="en_core_web_sm", | |
| network_allowed=False,external_dependency_assets=True,included_in_tts_parameter_budget=False) | |
| class OfflineMisaki: | |
| """Callable Misaki wrapper. Construct with make_offline_misaki(), not directly.""" | |
| def __init__(self, frontend, runtime): | |
| self._frontend,self.runtime = frontend,runtime | |
| def __call__(self, text): | |
| with _deny_network(): | |
| return self._frontend(text) | |
| def make_offline_misaki(): | |
| """Preinstalled Misaki0.9.4/spaCy3.8.16/en_core_web_sm3.8.0 ONLY. | |
| Does not load Torch/TTS, install packages, download spaCy models, or silently | |
| change frontend versions. Parent still qualifies actual offline outputs. | |
| """ | |
| runtime = _installed_runtime() | |
| with _deny_network(): | |
| spacy = importlib.import_module("spacy") | |
| if not spacy.util.is_package("en_core_web_sm"): | |
| raise FrontendUnavailable("Preinstalled en_core_web_sm package required; download forbidden") | |
| original = spacy.cli.download | |
| def no_download(*args,**kwargs): | |
| raise FrontendUnavailable("Automatic spaCy model download forbidden") | |
| try: | |
| spacy.cli.download = no_download | |
| misaki = importlib.import_module("misaki.en") | |
| frontend = misaki.G2P(trf=False,british=False) | |
| finally: | |
| spacy.cli.download = original | |
| return OfflineMisaki(frontend,runtime) | |