import json,tempfile,unittest from pathlib import Path import numpy as np from peppa.schema import * from peppa.ternary import equilibrium,fit_cooperativity from peppa.ptm import fit,loss_gradient from peppa.metrics import * from peppa.engine import Engine,Tool from peppa.trace import Trace,replay from peppa.structure import af3_input,ternary_interface_features from peppa.data import normalize_snooppi,connected_splits from peppa.controllers import ScriptedController class ScienceTests(unittest.TestCase): def spec(self): return DesignSpec(episode_id='test',task='binding',objective='test',targets=[Target(id='T',accession='test',molecule=Molecule(sequence='ACY'),compartment='test')],requirements=[Requirement(endpoint='pKd',unit='logM',direction='ge',threshold=7,assay='test',scale=1)],administration='test') def decision(self,tool,arguments={}): return Decision(tool=tool,arguments=arguments,hypothesis='test',evidence_ids=[],decision_summary='test',expected_observation='test') def test_mass_balance_symmetry(self): for totals in [(10,50,100),(1,1,1e5),(100,1,1e-3)]: x=equilibrium(*totals,10,30,20) np.testing.assert_allclose([x['A']+x['AL']+x['ABL'],x['B']+x['BL']+x['ABL'],x['L']+x['AL']+x['BL']+x['ABL']],totals,rtol=1e-7) y=equilibrium(totals[1],totals[0],totals[2],30,10,20) self.assertAlmostEqual(x['ABL'],y['ABL'],places=6) def test_zero_component(self): x=equilibrium(10,0,20,5,5,10);self.assertEqual(x['ABL'],0);self.assertAlmostEqual(x['A']+x['AL'],10) def test_cooperativity_recovery(self): grid=np.array([[a,30,l] for a in [10,100] for l in np.logspace(-1,3,8)]) y=[equilibrium(*v,15,40,8)['ABL'] for v in grid] r=fit_cooperativity(grid,y,15,40);self.assertAlmostEqual(r['alpha'],8,places=4) def test_hook_effect(self): mid=equilibrium(100,100,100,10,10,20)['ABL'];high=equilibrium(100,100,1e8,10,10,20)['ABL'];self.assertGreater(mid,high) def test_ptm_gradient(self): rng=np.random.default_rng(7);b=rng.normal(size=(7,3));t=rng.normal(size=(7,2));theta=rng.normal(size=7);labels=np.array([1,0,1,np.nan,0,1,0]) f,g=loss_gradient(theta,b,t,labels,[(0,1),(2,3)]) numeric=[] for j in range(len(theta)): d=np.zeros_like(theta);d[j]=1e-6 numeric.append((loss_gradient(theta+d,b,t,labels,[(0,1),(2,3)])[0]-loss_gradient(theta-d,b,t,labels,[(0,1),(2,3)])[0])/2e-6) np.testing.assert_allclose(g,numeric,atol=1e-6) def test_ptm_fit(self): rng=np.random.default_rng(9);b=rng.normal(size=(80,3));t=rng.normal(size=(80,2));y=(b[:,0]*t[:,0]>0).astype(float) head,report=fit(b,t,y,l2=.001);self.assertGreater(np.mean((head.logits(b,t)>0)==y),.9) def test_chemistry_identity(self): x=Molecule(sequence='ASY');y=Molecule(sequence='ASY',modifications=[Modification(position=3,residue='Y',ccd='PTR')]);self.assertNotEqual(x.identity,y.identity) a=Molecule(sequence='ASY',bonds=[(1,'N',3,'C')]);b=Molecule(sequence='ASY',bonds=[(3,'C',1,'N')]);self.assertEqual(a.identity,b.identity) def test_invalid_modification(self): with self.assertRaises(ValueError):Molecule(sequence='ASY',modifications=[Modification(position=2,residue='Y',ccd='PTR')]) def test_af3_chemistry(self): s=self.spec();s.targets[0].molecule.modifications=[Modification(position=3,residue='Y',ccd='PTR')] x=af3_input('test',s.targets,Molecule(sequence='ACD',bonds=[(1,'N',3,'C')])) self.assertEqual(x['sequences'][0]['protein']['modifications'][0]['ptmType'],'PTR');self.assertEqual(x['bondedAtomPairs'][0],[['B',1,'N'],['B',3,'C']]) def test_weakest_interface(self): x=ternary_interface_features([[1,.9,.8],[.9,1,.2],[.8,.2,1]]);self.assertEqual(x['weakest_interface'],.2) def test_score_requires_calibration(self): m=Measurement(candidate_id='x',endpoint='pKd',value=8,unit='logM',source_id='x',kind='prediction',assay='test') with self.assertRaises(ValueError):candidate_score([m],self.spec().requirements) m.lower=7.5;m.upper=8.5;self.assertEqual(candidate_score([m],self.spec().requirements),.5) def test_selectivity_sign(self):self.assertAlmostEqual(selectivity(10,100),1) def test_unknown_remains_unknown(self): self.assertIsNone(normalize_snooppi({'SNOOPPI_final_label':'unknown','partner_A_sequence':'A','partner_B_sequence':'B'})['label']) def test_connected_split_no_leak(self): rows=[{'id':'a','target_clusters':['x'],'peptide_clusters':['p'],'source_ids':['1']},{'id':'b','target_clusters':['y'],'peptide_clusters':['q'],'source_ids':['1']},{'id':'c','target_clusters':['y'],'peptide_clusters':['r'],'source_ids':['2']}] self.assertEqual(len(set(connected_splits(rows))),1) def test_failed_worker_charged_and_replayed(self): with tempfile.TemporaryDirectory() as d: def fail(a,s):raise RuntimeError('deliberate failure') e=Engine(self.spec(),{'fail':Tool('fail',fail,{'tool_calls':1},'test')},Path(d)/'trace.jsonl') with self.assertRaises(RuntimeError):e.apply(self.decision('fail')) r=replay(e.trace.path);self.assertEqual(r['spent']['tool_calls'],1);self.assertEqual(len(r['errors']),1) def test_endpoint_immutable(self): with tempfile.TemporaryDirectory() as d: e=Engine(self.spec(),{},Path(d)/'trace.jsonl') with self.assertRaises(ValueError):e.apply(self.decision('revise_plan',{'requirements':[]})) self.assertEqual(e.state['spec']['requirements'][0]['threshold'],7) def test_trace_tampering(self): with tempfile.TemporaryDirectory() as d: p=Path(d)/'trace.jsonl';e=Engine(self.spec(),{},p);s=p.read_text().replace('"threshold":7.0','"threshold":8.0');p.write_text(s) with self.assertRaises(ValueError):Trace.read(p) def test_budget_stop(self): with tempfile.TemporaryDirectory() as d: s=self.spec();s.budgets['controller_calls']=0;e=Engine(s,{},Path(d)/'trace.jsonl');r=e.run(ScriptedController([]));self.assertTrue(r['stopped']) def test_redundancy_selection(self): r=diverse_select({'a':'AAAA','b':'AAAA','c':'CCCC'},{'a':2,'b':1.9,'c':1},2);self.assertEqual(r,['a','c']) def test_conformal_finite_sample(self): self.assertTrue(np.isinf(conformal_radius([1],[1],[1],.1))) self.assertEqual(conformal_radius(np.arange(20),np.arange(20),np.ones(20)),0) if __name__=='__main__':unittest.main()