| import hashlib,json,subprocess |
| from pathlib import Path |
| root=Path.cwd(); bundle=root/'artifacts/bundle' |
| def sha(p):return hashlib.sha256(p.read_bytes()).hexdigest() |
| freeze=json.loads((root/'artifacts/pipeline/evaluation-freeze.json').read_text())['sha256'] |
| assert all(sha(root/p)==v for p,v in freeze.items()),'freeze changed' |
| checks=json.loads((bundle/'checksums.json').read_text()) |
| assert all(sha(bundle/p)==v for p,v in checks.items()),'bundle checksum mismatch' |
| cal=json.loads((bundle/'calibration.json').read_text()) |
| import bisect |
| def confidence(x): |
| xs,ys=cal['xs'],cal['ys'] |
| if x<=xs[0]:return ys[0] |
| if x>=xs[-1]:return ys[-1] |
| i=bisect.bisect_right(xs,x) |
| return ys[i-1]+(x-xs[i-1])*(ys[i]-ys[i-1])/(xs[i]-xs[i-1]) |
| n=tp=fp=fn=0 |
| with open('data/prepared/test.jsonl') as gold,open('artifacts/predictions/test.jsonl') as pred: |
| import itertools |
| for g,p in itertools.zip_longest(gold,pred): |
| assert g and p |
| g,p=json.loads(g),json.loads(p) |
| assert g['id']==p['id'] and g['text']==p['text'] |
| text=g['text']; offsets=[0] |
| for c in text:offsets.append(offsets[-1]+len(c.encode())) |
| targets={(s['label'],offsets[s['start']],offsets[s['end']]) for s in g['spans']} |
| findings={(s['label'],s['start'],s['end']) for s in p['findings'] if confidence(s['raw_score'])>=cal['threshold']} |
| tp+=len(targets&findings);fp+=len(findings-targets);fn+=len(targets-findings);n+=1 |
| expected=json.loads(Path('artifacts/evaluation/test.json').read_text()) |
| assert [tp,fp,fn]==[expected['overall'][k] for k in ('tp','fp','fn')] |
| assert n==90018 |
| proc=subprocess.run([str(bundle/'datafog-pii'),str(bundle)],input=json.dumps({'text':'🙂 Email me at alice@example.com.'})+'\n'+json.dumps({'text':'The screen is blank.'})+'\n',text=True,capture_output=True,check=True,env={'PATH':'/usr/bin:/bin'}) |
| outputs=[json.loads(x) for x in proc.stdout.splitlines()] |
| assert len(outputs)==2 and all('error' not in x for x in outputs) |
| report={'frozen_inputs_unchanged':True,'bundle_checksums_verified':len(checks),'independent_metric_recount':{'documents':n,'tp':tp,'fp':fp,'fn':fn},'packaged_native_smoke_without_python_path':outputs} |
| Path('artifacts/final-audit.json').write_text(json.dumps(report,indent=2)+'\n') |
| print(json.dumps(report,indent=2)) |
|
|
