File size: 5,887 Bytes
c49eca9 | 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 53 54 55 56 | """Observe the whole frozen eleven-case campaign through the native consumer."""
from pathlib import Path
import argparse,json,os,sys,time
ROOT=Path(__file__).resolve().parents[1];BASE=ROOT.parent
OBSERVER=BASE/'native-caption-observer-v1'
sys.path[:0]=[str(ROOT/'scripts'),str(OBSERVER/'scripts'),str(OBSERVER),str(ROOT)]
from silero_qualification_v2 import load_frozen,case_binding,sha,require
from probe_native_candidate_linux import capture_process,output_paths
from validate_caption_run_v2 import validate
from run_native_probe import parse_time
def main():
p=argparse.ArgumentParser();p.add_argument('--qualification-record',type=Path,required=True);p.add_argument('--qualification-sha256',required=True);p.add_argument('--transport',type=Path,required=True);p.add_argument('--output',type=Path,required=True);a=p.parse_args()
frozen,cases=load_frozen(a.qualification_record,a.qualification_sha256)
candidate=frozen['selected_candidate'];native=candidate['native'];profile=candidate['profile']['native']
require(profile==dict(workers=8,tasks=4,activation='bf16',startup='realtime',delay_ms=240,first_batch=1,batch=1,encoder_tiled_attention=True),'native profile differs')
require(sha(__file__)==frozen['source_sha256']['scripts/run_native_q3_qualification_v1.py'],'unfrozen native observer')
require(sorted(os.sched_getaffinity(0))==list(range(8)) and os.getpriority(os.PRIO_PROCESS,0)==15,'native CPU placement')
def verify_assets():
for item in native['assets']:
require(Path(item['path']).stat().st_size==item['bytes'] and sha(item['path'])==item['sha256'],'native asset changed: '+item['path'])
verify_assets()
transport=json.loads(a.transport.read_text());require(transport['qualification_sha256']==a.qualification_sha256,'transport freeze')
require([case_binding(c) for c in transport['cases']]==[case_binding(c) for c in cases],'transport order/labels')
prepared=[]
for original,moved in zip(cases,transport['cases'],strict=True):
source=(a.transport.parent/moved['path']).resolve();plan_path=(a.transport.parent/moved['segment_plan']).resolve()
require(source.is_relative_to(a.transport.parent.resolve()) and plan_path.is_relative_to(a.transport.parent.resolve()),'transport containment')
require(source.stat().st_size==original['samples']*4 and sha(source)==original['sha256'],'transport PCM')
require(sha(plan_path)==moved['segment_plan_sha256'],'transport boundary hash')
plan=json.loads(plan_path.read_text());require(plan['status']=='complete' and plan['qualification_sha256']==a.qualification_sha256 and plan['case']==case_binding(original),'boundary freeze/case')
require(plan['model_sha256']==frozen['silero_model_sha256'] and plan['silence_policy']==frozen['silence_policy'],'VAD policy')
prepared.append((dict(original,path=str(source)),plan_path,moved['segment_plan_sha256'],plan))
a.output.mkdir(exist_ok=False);report=dict(status='running',qualification_sha256=a.qualification_sha256,started_unix=time.time(),planned_cases=[c['id'] for c in cases],cases=[],release_qualification=False)
def save():(a.output/'campaign.json').write_text(json.dumps(report,indent=2)+'\n')
save()
for case,plan_path,plan_sha,plan in prepared:
folder=a.output/case['id'];folder.mkdir();paths=output_paths(folder/'run.json');prefix=native['prefixes'][case['language']]
command=[native['binary'],native['model'],native['frontend'],native['tokenizer'],'-',case['language'],'--workers','8','--tasks','4','--worker-qos','default','--activation','bf16','--startup','realtime','--delay-ms','240','--first-batch','1','--batch','1','--silero-vad',native['vad'],'--encoder-tiled-attention','--prefix-in',prefix['path']]
(folder/'invocation.json').write_text(json.dumps(dict(command=command,qualification_sha256=a.qualification_sha256,case=case_binding(case),boundary_sha256=plan_sha),indent=2)+'\n')
result=dict(status='running',case_id=case['id'],qualification_sha256=a.qualification_sha256)
try:
rows,observations,feed=capture_process(command,case['path'],case['samples'],paths,timeout_seconds=900,max_events=20000,ready_timeout=120,ready_kind='caption_stream_ready')
ready=rows[0]
expected=dict(backend='native_cpu',activation='bf16',activation_backend='avx512_bf16_q4_q3full',workers=8,encoder_attention='x86_four_query_tiled',constant_prefix_used=True,constant_prefix_body_sha256=prefix['body_sha256'],model_body_sha256=native['model_body_sha256'],startup='realtime18',delay_ms=240,batch=1,input_mode='stdin',silero_sha256=frozen['silero_model_sha256'],silence_policy=frozen['silence_policy'])
require(all(ready.get(k)==v for k,v in expected.items()),'actual native model/backend/profile')
require(sha(plan_path)==plan_sha,'boundary plan changed')
result.update(validate(rows,observations,case,native['tokenizer'],frozen_segments=plan['segments']),feeder=feed,memory=parse_time(paths['timing'].read_text(),'Linux'))
except Exception as error:result.update(status='failed',error=f'{type(error).__name__}: {error}')
result['logs']={k:dict(path=str(v),bytes=v.stat().st_size,sha256=sha(v)) for k,v in paths.items() if k!='report' and v.exists()}
paths['report'].write_text(json.dumps(result,indent=2,ensure_ascii=False)+'\n');report['cases'].append(dict(id=case['id'],status=result['status'],report=str(paths['report']),sha256=sha(paths['report'])));save()
verify_assets();load_frozen(a.qualification_record,a.qualification_sha256)
report.update(status='complete' if all(c['status']=='complete' for c in report['cases']) else 'failed',finished_unix=time.time(),all_eleven_cases_attempted=len(report['cases'])==11,all_assets_unchanged=True);save();print(json.dumps(report))
if __name__=='__main__':main()
|