emirhanyildirim34555's picture
Publish reviewed research staging to main
6d5d7e5 verified
Raw History Blame Contribute Delete
2.83 kB
from pathlib import Path
import sys,json,numpy as np
R=Path(__file__).resolve().parent;sys.path.insert(0,str(R))
from inference import run,PRODUCT,PIN
from humanization_research.parameters import ParameterPack,predict,predict_proba,transform,isolation_score
from humanization_research.predictors import MRIPredictor
p=ParameterPack(R/'parameters',PIN);checks=[]
with np.load(R/'parameters/reference_fixtures.npz',allow_pickle=False) as f:
if PRODUCT=='ncl':
x=f['inputs'][:1]
for key,model in p.data['models'].items():
np.testing.assert_allclose(predict(model,x),f[key][:1],rtol=0,atol=2e-12) if key!='progression_agent' else np.testing.assert_array_equal(predict(model,x),f[key][:1]);checks.append(key)
np.testing.assert_allclose(predict_proba(p.data['models']['progression_agent'],x),f['progression_agent_proba'][:1],rtol=0,atol=2e-12);checks.append('probabilities')
elif PRODUCT=='mri':
for space in ['baseline','huret']:
values=MRIPredictor(p).scores(space,f[space+'_inputs'][:1])
for value,key in zip(values,['raw','pct','cluster']):np.testing.assert_allclose(value,f[space+'_'+key][:1],rtol=0,atol=2e-10);checks.append(space+'_'+key)
else:
x=f['inputs'][:1];z=transform(p.data['scaler'],x);np.testing.assert_array_equal(z,f['scaled'][:1]);checks.append('scaler')
np.testing.assert_allclose(predict(p.data['state_encoder'],z),f['state'][:1],rtol=0,atol=2e-6);checks.append('state')
np.testing.assert_allclose(isolation_score(p.data['ood_model'],z),f['ood'][:1],rtol=0,atol=2e-12);checks.append('ood')
for name in ['fog_guard','balance_guard']:
g=p.data[name];np.testing.assert_allclose(predict_proba(g['model'],transform(g['scaler'],x)),f[name][:1],rtol=0,atol=2e-12);checks.append(name)
xx=f['response_input'][:1];np.testing.assert_allclose(predict(p.data['response_full_model'],xx),f['response_full'][:1],rtol=0,atol=2e-6);checks.append('response_full')
for i,m in enumerate(p.data['response_ensemble']):np.testing.assert_allclose(predict(m,xx),f['response_'+str(i)][:1],rtol=0,atol=2e-6);checks.append('response_'+str(i))
request=json.loads((R/'example.json').read_text());answer=run(request);assert answer['clinical_claim'] is False;checks.append('clinical_disabled')
bad=dict(request);bad['synthetic']=False
try:run(bad)
except (ValueError,TypeError):checks.append('patient_input_rejected')
else:raise AssertionError('Patient input accepted')
if PRODUCT=='rhythm':
bad=dict(request);bad.pop('acknowledge_export_version_mismatch')
try:run(bad)
except (ValueError,TypeError):checks.append('version_mismatch_default_rejection')
else:raise AssertionError('Version warning bypassed')
print(json.dumps({'status':'PASS','checks':checks,'n_checks':len(checks),'fixture_rows_per_space':1,'scope':'First saved synthetic export fixture; no new clinical or training evaluation'},indent=2))