"""Portable, hash-checked native LoRA+CFG inference from an agent route.

Install the dependencies documented by the pinned model repository. Place
bucket_native_cfg.py beside this script. This helper never merges the ancestor
rank4 twice. Benchmark replication retains a two-seed generation batch.
"""
import argparse,hashlib,importlib.util,json,urllib.request
from pathlib import Path
def module(name,path):
    s=importlib.util.spec_from_file_location(name,path);m=importlib.util.module_from_spec(s);s.loader.exec_module(m);return m
def read(path):
    if str(path).startswith('https://'):
        with urllib.request.urlopen(str(path),timeout=60) as response:return json.load(response)
    return json.loads(Path(path).read_text())
def compiler(sentences,durations,candidate,neutral=False):
    general='One adult speaker, clean high-quality recording, no music or sound effects.'
    if not neutral and candidate.get('recipe')!='P09':general+=' '+candidate['general']
    lines=[]
    for i,(text,seconds) in enumerate(zip(sentences,durations)):
        local='Natural clear speech.' if neutral or candidate.get('recipe')=='P08' else candidate['inline'][i]
        lines.append(f'({local}) [{seconds:.1f} seconds duration] {text}')
    return 'GENERAL: '+general+'\nSCRIPT:\n'+'\n'.join(lines)
def main():
    p=argparse.ArgumentParser();p.add_argument('--routes',required=True);p.add_argument('--target',required=True)
    p.add_argument('--sentences-json',type=Path,required=True);p.add_argument('--reference-wav',type=Path,required=True)
    p.add_argument('--language',choices=['en','de','fr','nl','es'],default='en');p.add_argument('--seed',type=int,default=1701)
    p.add_argument('--durations',default='3.2,3.8,3.2');p.add_argument('--output',type=Path,required=True)
    p.add_argument('--codec',default='OpenMOSS-Team/MOSS-Audio-Tokenizer-v2');a=p.parse_args()
    routes=read(a.routes);r=next(r for r in routes['routes'] if r['target']==a.target)
    if r['status']!='VALIDATED_HELDOUT':raise ValueError('No validated new route; inspect the manual and use its documented fallback')
    if a.output.exists():raise FileExistsError(a.output)
    sentences=read(a.sentences_json);durations=[float(x) for x in a.durations.split(',')]
    if len(sentences)!=3 or len(durations)!=3 or not all(isinstance(s,str) and s for s in sentences):raise ValueError('This validated helper expects three literal sentences and three durations')
    if not all(0<d<=30 for d in durations) or sum(durations)>60:raise ValueError('Invalid duration budget')
    from huggingface_hub import snapshot_download
    root=Path(snapshot_download(r['model_repo'],revision=r['model_revision'],allow_patterns=['base_model/*','native_adapters.py',r['adapter_subfolder']+'/*']))
    import numpy as np,torch,soundfile as sf
    from transformers import AutoModel,AutoTokenizer
    from transformers.dynamic_module_utils import get_class_from_dynamic_module
    from safetensors.torch import load_file
    base=root/'base_model';model=AutoModel.from_pretrained(base,trust_remote_code=True,dtype=torch.float32,attn_implementation='sdpa').cuda().eval()
    cls=get_class_from_dynamic_module('processing_moss_tts.MossTTSLocalProcessor',str(base));proc=cls(tokenizer=AutoTokenizer.from_pretrained(base,trust_remote_code=True),audio_tokenizer=None,model_config=model.config)
    api=module('tool_base_infer',base/'infer.py');api.load_codec(proc,a.codec)
    config=r['configuration'];scale=float(config['scale']);cfg=float(config['cfg'])
    if config['adapter']:
        weight=root/r['adapter_subfolder']/'adapter.safetensors'
        if hashlib.sha256(weight.read_bytes()).hexdigest()!=r['adapter_sha256']:raise ValueError('Adapter checksum mismatch')
        ad=module('tool_adapters',root/'native_adapters.py');ad.attach(model,rank=32);ad.restore(model,load_file(str(weight),device='cpu'))
        with torch.no_grad():
            for m in model.modules():
                if hasattr(m,'enabled') and hasattr(m,'a') and hasattr(m,'b'):m.b.mul_(scale)
    elif scale!=0:raise ValueError('No-adapter route must have strength0')
    model.eval().requires_grad_(False)
    wave,sr=sf.read(a.reference_wav,dtype='float32',always_2d=True)
    if len(wave)/sr<5.04 or not np.isfinite(wave).all():raise ValueError('Use a clean single-speaker reference of5.04–14.96seconds')
    wave=wave[:round(sr*14.96)].mean(axis=1)
    with torch.inference_mode():reference=proc.encode_audios_from_wav([torch.from_numpy(wave[None,:].copy())],sr,n_vq=12)[0][:187]
    conditional=compiler(sentences,durations,r['candidate']);neutral=compiler(sentences,durations,r['candidate'],True)
    args=dict(text=' '.join(sentences),frames=round(sum(durations)*12.5),language=a.language,reference=reference)
    user=api.build_user(proc,prompt=conditional,**args);neu=api.build_user(proc,prompt=neutral,**args)
    seeds=[1701,1702] if a.seed in [1701,1702] else [a.seed,a.seed+1]
    decoder=module('tool_native_cfg',Path(__file__).with_name('bucket_native_cfg.py'))
    codes,timing=decoder.generate(model,proc,user,neu,seeds,cfg,min(1000,args['frames']+60));ix=seeds.index(a.seed)
    audio,rate=api.decode(proc,codes[ix]);a.output.parent.mkdir(parents=True,exist_ok=True);sf.write(a.output,audio,rate,subtype='PCM_16')
    print(json.dumps(dict(output=str(a.output),target=a.target,scale=scale,cfg=cfg,seed=a.seed,ended=timing['ended'][ix],model_revision=r['model_revision'])))
if __name__=='__main__':main()