"""Causal v1-mini evaluation using LiveKit Agents' official local transport."""
import os
for k in ('OMP_NUM_THREADS','OPENBLAS_NUM_THREADS','MKL_NUM_THREADS','EOT_PARALLELISM'): os.environ[k]='1'
import argparse,asyncio,csv,hashlib,json,math,platform,sys,time,weakref
import importlib.metadata as metadata
from collections import Counter
from pathlib import Path
import numpy as np
import soundfile as sf
from scipy.signal import resample_poly
from livekit.agents.inference import TurnDetector
from livekit.agents.inference.eot.transports import _LocalTransport
from livekit.agents.inference.eot.languages import LOCAL_LANGUAGES
from livekit.local_inference import _native
HERE=Path(__file__).resolve().parent
ROOT=HERE.parents[1]
OUT=HERE/'results'
LABELS=ROOT/'labels/EOT标注/工作数据/eot_labels.csv'
THRESHOLD=LOCAL_LANGUAGES['zh']
def sha(p): return hashlib.sha256(p.read_bytes()).hexdigest()
def pcm_at(user,sr,ms):
    end=min(len(user),ms*sr//1000)
    if end<=0: raise ValueError('No received audio')
    prefix=user[:end].astype(np.float32)/32768.
    g=math.gcd(sr,16000)
    audio=resample_poly(prefix,16000//g,sr//g) if sr!=16000 else prefix
    return np.ascontiguousarray((np.clip(audio[-19200:],-1,1)*32767).astype(np.int16))
class Receiver:
    def _resolve_prediction(self,request_id,prob,**kwargs): self.value=prob
class CheckedNative:
    """Reject the framework's error fallback instead of counting it as a model score."""
    def __init__(self,native): self.native=native;self.success=False
    def predict(self,pcm):
        self.success=False
        self.value=float(self.native.predict(pcm))
        if not np.isfinite(self.value) or not 0<=self.value<=1: raise ValueError('Invalid score')
        self.success=True
        return self.value
class Scorer:
    def __init__(self):
        self.detector=TurnDetector(version='v1-mini')
        self.transport=_LocalTransport(opts=self.detector._opts)
        self.checked=CheckedNative(self.transport._eot)
        self.transport._eot=self.checked
        self.receiver=Receiver()
        self.transport._stream_ref=weakref.ref(self.receiver)
    async def predict(self,pcm):
        self.checked.success=False;self.receiver.value=None
        await self.transport._predict('offline-evaluation',pcm)
        if not self.checked.success or self.receiver.value!=self.checked.value: raise RuntimeError('Inference fallback rejected')
        return self.receiver.value

def metrics(rows,threshold):
    c=Counter((int(r['can_reply_now']),int(r['probability']>=threshold)) for r in rows if r['can_reply_now'] in ('0','1'))
    tn,fp,fn,tp=c[0,0],c[0,1],c[1,0],c[1,1];n=tn+fp+fn+tp
    return dict(threshold=threshold,n=n,tn=tn,fp=fp,fn=fn,tp=tp,accuracy=(tn+tp)/n,
        false_reply_rate=fp/(tn+fp) if tn+fp else None,false_hold_rate=fn/(tp+fn) if tp+fn else None)
def provenance():
    return dict(machine=platform.platform(),hardware='Apple M4 Mac mini CPU, 16 GB',python=sys.version,
        dependencies={p:metadata.version(p) for p in ['livekit-local-inference','livekit-agents','numpy','scipy','soundfile']},
        native_binary_sha256=sha(Path(_native.__file__)),weight_identity='Weights embedded in pinned official binary; separate hash not exposed',
        wheel_sha256='fa4cb2d495c80be6121642dba4def3a88b22a68d52c1b0affccd66c563de6622',
        native_parallelism=_native._parallelism(),script_sha256=sha(Path(__file__)),annotation_sha256=sha(LABELS))
async def main(mode):
    OUT.mkdir(exist_ok=True)
    t=time.perf_counter();model=Scorer();load_ms=(time.perf_counter()-t)*1000
    rng=np.random.default_rng(42);x=(rng.normal(0,.005,19200)*32767).astype(np.int16)
    p=await model.predict(x);q=await model.predict(x);assert p==q
    print(f'WITNESS CPU seeded probability={p:.9f}, repeat={q:.9f}, parallelism={_native._parallelism()}',flush=True)
    if mode=='smoke':
        print(json.dumps(dict(load_ms=load_ms,provenance=provenance()),ensure_ascii=False));return
    with LABELS.open(encoding='utf-8-sig',newline='') as f: labels=list(csv.DictReader(f))
    calls,rows=[],[]
    for call in sorted({r['call_id'] for r in labels}):
        path=ROOT/'data/韩泰机器人抢话/音频'/f'{call}.mp3'
        pcm,sr=sf.read(path,dtype='int16',always_2d=True);user=pcm[:,0];duration=round(len(user)*1000/sr,3)
        exact={int(r['decision_ms']):r for r in labels if r['call_id']==call}
        if any(t>duration for t in exact): raise ValueError('Annotation exceeds audio')
        times=sorted(set(range(200,int(duration)+1,200))|set(exact)|{int(duration)}) if mode=='all' else sorted(exact)
        points=[];checkpoints=[]
        for ms in times:
            t=time.perf_counter();audio=pcm_at(user,sr,ms);prep=(time.perf_counter()-t)*1000
            t=time.perf_counter();score=await model.predict(audio);latency=(time.perf_counter()-t)*1000
            points.append([ms,round(score,7)])
            if ms in exact:
                assert score==await model.predict(audio)
                modified=user.copy();modified[ms*sr//1000:]=0
                assert np.array_equal(audio,pcm_at(modified,sr,ms))
                checkpoints.append([ms,round(score,7)]);row=exact[ms]
                rows.append(dict(call_id=call,decision_ms=ms,can_reply_now=row['can_reply_now'],probability=score,
                    prediction=int(score>=THRESHOLD),threshold=THRESHOLD,input_samples=len(audio),
                    available_end_ms=(ms*sr//1000)*1000/sr,preparation_ms=prep,inference_ms=latency))
        calls.append(dict(call_id=call,audio_path=f'韩泰机器人抢话/音频/{call}.mp3',audio_sha256=sha(path),
            duration_ms=duration,points=points,checkpoints=checkpoints))
        print(f'{call[:8]}: {len(points)} scores, {len(checkpoints)} checkpoints; repeat/future-suffix checks passed',flush=True)
    with (OUT/'predictions.csv').open('w',encoding='utf-8',newline='') as f:
        w=csv.DictWriter(f,fieldnames=list(rows[0]));w.writeheader();w.writerows(rows)
    method='Agents 1.8.8 local transport, native v1-mini 0.2.7. User channel 0; received int16 prefix /32768 cropped BEFORE scipy resample_poly to 16kHz; last 19200 samples converted with *32767. Native left-padding below 1.2s. No ASR, VAD gate, silence timeout or fallback scores. EOT_PARALLELISM=1. Prefix resampling differs from official librosa and online streaming resampling; not a full service replay.'
    summary=dict(metrics=metrics(rows,THRESHOLD),reference_threshold_05=metrics(rows,.5),model='LiveKit Turn Detector v1-mini',
        official_zh_threshold=THRESHOLD,load_ms=load_ms,method=method,provenance=provenance(),
        preparation_ms={k:float(np.percentile([r['preparation_ms'] for r in rows],v)) for k,v in [('p50',50),('p95',95)]},
        inference_ms={k:float(np.percentile([r['inference_ms'] for r in rows],v)) for k,v in [('p50',50),('p95',95)]},
        limitations=['44 selected points: exploratory, not held-out generalization.', 'Point false reply rate is not interval interruption rate; user channel may contain robot echo.', 'Raw zh threshold without production VAD/endpointing gates.', 'Wheel license restricts use to LiveKit Agents; do not use outputs to improve other models.'])
    (OUT/'summary.json').write_text(json.dumps(summary,ensure_ascii=False,indent=2)+'\n')
    if mode=='all':
        out=dict(schema_version=1,model='livekit_mini',name='LiveKit v1-mini',score='P(turn ended)',sample_interval_ms=200,
            method=method,provenance=provenance(),calls=calls)
        (OUT/'score-timeline.json').write_text(json.dumps(out,ensure_ascii=False,separators=(',',':'))+'\n')
    print(json.dumps(summary['metrics']),flush=True)
if __name__=='__main__':
    p=argparse.ArgumentParser();p.add_argument('mode',choices=['smoke','evaluate','all']);asyncio.run(main(p.parse_args().mode))