66 lines
3.4 KiB
Python
66 lines
3.4 KiB
Python
"""Export real Smart Turn scores across complete recordings for EOT Studio."""
|
|
import csv
|
|
import hashlib
|
|
import json
|
|
|
|
import numpy as np
|
|
import soundfile as sf
|
|
from loguru import logger
|
|
from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
|
|
from evaluate_can_reply import HERE, ROOT, block_origin
|
|
|
|
|
|
def read(p):
|
|
with p.open(encoding='utf-8-sig') as f:
|
|
return list(csv.DictReader(f))
|
|
|
|
|
|
def main():
|
|
logger.remove()
|
|
checkpoints = read(HERE/'results/work-data-can-reply/predictions.csv')
|
|
history = read(ROOT/'labels/EOT标注/工作数据/asr_history.csv')
|
|
events = read(HERE/'results/work-data-pipecat/events.csv')
|
|
weight = HERE/'smart-turn-v3.2-cpu.onnx'
|
|
analyzer = LocalSmartTurnAnalyzerV3(smart_turn_model_path=str(weight),cpu_count=4)
|
|
calls=[]
|
|
for call in sorted({r['call_id'] for r in checkpoints}):
|
|
path=ROOT/'data/韩泰机器人抢话/音频'/f'{call}.mp3'
|
|
pcm,sr=sf.read(path,dtype='int16',always_2d=True)
|
|
duration=round(len(pcm)*1000/sr,3)
|
|
call_labels=[r for r in checkpoints if r['call_id']==call]
|
|
exact={int(r['decision_ms']):float(r['probability']) for r in call_labels}
|
|
times=sorted(set(range(200,int(duration)+1,200))|set(exact)|{int(duration)})
|
|
call_history=[r for r in history if r['call_id']==call]
|
|
starts=sorted(float(e['event_ms']) for e in events if e['call_id']==call and e['event']=='vad_start')
|
|
analyzer.set_sample_rate(sr)
|
|
points=[]
|
|
for decision in times:
|
|
origin=block_origin(call_history,decision)
|
|
if origin is None:
|
|
past=[t for t in starts if t<=decision]
|
|
origin=max(0,past[-1]+32-200-500) if past else 0
|
|
end=decision*sr//1000
|
|
start=max(round(origin*sr/1000),end-sr*8)
|
|
assert 0<=start<end<=len(pcm)
|
|
# Endpoint resampling can only read the already cropped prefix.
|
|
samples=np.ascontiguousarray(pcm[start:end,0].astype(np.float32)/32768.)
|
|
score=float(analyzer._predict_endpoint(samples)['probability'])
|
|
assert np.isfinite(score) and 0<=score<=1
|
|
if decision in exact:
|
|
assert abs(score-exact[decision])<1e-4,(call,decision,score,exact[decision])
|
|
points.append([decision,round(score,7)])
|
|
calls.append(dict(call_id=call,audio_path=f'韩泰机器人抢话/音频/{call}.mp3',
|
|
audio_sha256=hashlib.sha256(path.read_bytes()).hexdigest(),duration_ms=duration,
|
|
points=points,checkpoints=[[t,round(p,7)] for t,p in sorted(exact.items())]))
|
|
print(f'{call[:8]}: {len(points)} timeline scores; checkpoints verified',flush=True)
|
|
output=dict(schema_version=1,model='smart_turn',name='Smart Turn',sample_interval_ms=200,
|
|
score='completion probability',
|
|
method='Direct endpoint call every 200ms and at every original annotation. Same past-ASR block origin, 500ms pre-speech, <=3s continuation and last 8s user-channel audio as the point baseline; past VAD fallback when ASR missing. No EOT/timeout gates. Crop before resampling.',
|
|
model_sha256=hashlib.sha256(weight.read_bytes()).hexdigest(),
|
|
script_sha256=hashlib.sha256((HERE/'export_timeline.py').read_bytes()).hexdigest(),calls=calls)
|
|
(HERE/'results/score-timeline.json').write_text(json.dumps(output,ensure_ascii=False,separators=(',',':'))+'\n')
|
|
|
|
|
|
if __name__=='__main__':
|
|
main()
|