"""Semantic counterexamples and exact source-level regression certificates.""" import itertools import json from pathlib import Path from dataclasses import replace import unittest import tempfile from example import (CatalystFormula, Reaction, ReactionSystem, RAFAnalysis, SiphonCertificate, InterventionStudy, PINNED_SOURCE_SHA256, SMALL_DIRECTIONS, DELETED_ACTIONS, POOLING_DIRECTION, TARGET, PRESERVATION_TARGET) BASE=Path(__file__).resolve().parent class ScientificChecks(unittest.TestCase): @classmethod def setUpClass(cls): cls.system=ReactionSystem.read(BASE/'prokaryotic-network.txt',PINNED_SOURCE_SHA256) cls.inputs=json.loads((BASE/'certificate_inputs.json').read_text()) def test_formula_semantics_and_literal_parser(self): f=CatalystFormula.parse('(A,B)&C,D&E') for flags in itertools.product((False,True),repeat=5): pool={s for s,b in zip('ABCDE',flags) if b} self.assertEqual(f.holds(pool),((flags[0] or flags[1]) and flags[2]) or (flags[3] and flags[4])) self.assertEqual(CatalystFormula.parse('(A&B),(A&C)').mandatory_atoms(),frozenset(('A',))) for malformed in ('A&','(A,B','A,,B','A)'): with self.assertRaises(ValueError): CatalystFormula.parse(malformed) self.assertEqual(len(self.system.reactions),9231);self.assertEqual(len(self.system.food),68) self.assertEqual(len({r.action for r in self.system.reactions}),6039) self.assertNotIn('NADs',self.system.food);self.assertIn('Pooling',self.system.food) forward=self.system.by_id['R01210::f'];reverse=self.system.by_id['R01210::r'] self.assertEqual(forward.reactants,reverse.products);self.assertEqual(forward.products,reverse.reactants) self.assertEqual(forward.catalyst,reverse.catalyst) with tempfile.TemporaryDirectory() as folder: path=Path(folder)/'weighted.txt';path.write_text('Food: A\nr: 2 A [A] <=> 3 B\n') weighted=ReactionSystem.read(path) self.assertEqual(weighted.by_id['r::f'].reactants,(('A',2),)) self.assertEqual(weighted.by_id['r::r'].reactants,(('B',3),)) with self.assertRaises(ValueError): ReactionSystem.read(BASE/'prokaryotic-network.txt','0'*64) def test_raf_is_not_constructive_activation(self): reaction=Reaction('seed','seed',(('food',1),),(('B',1),),CatalystFormula.parse('B')) analysis=RAFAnalysis(ReactionSystem((reaction,),frozenset(('food',)))) self.assertTrue(analysis.is_raf(('seed',)));self.assertEqual(analysis.constructive()[0],frozenset()) self.assertEqual(analysis.closed_uniqueness_certificate()['status'],'unresolved_bounds') self.assertEqual(analysis.closed_extension(('seed',)),frozenset(('seed',))) # A supplied target alone does not meet Cap when no nonempty RAF exists. empty=RAFAnalysis(ReactionSystem((),frozenset(('B',)))) self.assertFalse(empty.capability('B'));self.assertFalse(empty.is_raf(())) def test_small_catalogue_necessary_producers_and_ambient_closedness(self): analysis=RAFAnalysis(self.system.subset(SMALL_DIRECTIONS)); result=analysis.catalogue(TARGET) H=frozenset(SMALL_DIRECTIONS);L=frozenset(SMALL_DIRECTIONS[i] for i in (0,1,4,7,8)) self.assertEqual(result['raf_count'],22) self.assertEqual({frozenset(s) for s in result['irreducible']},{frozenset((SMALL_DIRECTIONS[i],)) for i in (0,7)}) self.assertEqual({frozenset(s) for s in result['closed']},{L,H}) self.assertEqual(result['target_rafs'],[sorted(H)]) self.assertEqual(set(analysis.necessary_producers(TARGET)['forced_reactions']),set(H)) self.assertFalse(RAFAnalysis(self.system.subset(L)).capability(TARGET)) augmented=RAFAnalysis(self.system.subset((*H,POOLING_DIRECTION))); catalogue=augmented.catalogue(TARGET) self.assertEqual(catalogue['closed'],[sorted((*H,POOLING_DIRECTION))]) self.assertFalse(augmented.is_closed(L));self.assertFalse(augmented.is_closed(H)) self.assertEqual(augmented.closed_extension(L),augmented.universe) self.assertEqual(analysis.catalogue(TARGET,budget=1)['status'],'unresolved_budget') def test_selective_action_intervention_and_ranked_restoration(self): barrier=SiphonCertificate(frozenset(self.inputs['barrier'])) study=InterventionStudy(self.system,DELETED_ACTIONS,TARGET,PRESERVATION_TARGET,barrier) result=study.evaluate(self.inputs['restoration']) self.assertTrue(result['inclusion_minimal_cut_certified']);self.assertTrue(result['preservation_capability']) self.assertFalse(result['target_capability_by_maxraf']);self.assertEqual(len(result['deleted_directions']),4) self.assertEqual(sorted(w['size'] for w in result['restoration'].values()),[9,13,13]) # Check synchronous reactant ranks independently from the reported final closure. for witness in result['restoration'].values(): ranks=witness['reactant_ranks'] for k,rank in ranks.items(): previous=set(self.system.food) for earlier,r in ranks.items(): if r>i&1] capable=RAFAnalysis(self.system.delete_actions(actions)).capability(TARGET) self.assertEqual(capable,mask!=7) def test_siphon_annotation_independence_and_medium_boundary(self): barrier=SiphonCertificate(frozenset(self.inputs['barrier']));cut=self.system.delete_actions(DELETED_ACTIONS) audit=barrier.check(cut) self.assertEqual(len(barrier.species),18);self.assertEqual(len(audit['surviving_producers']),58) self.assertTrue(audit['absence_certified']);self.assertFalse(barrier.check(self.system)['valid_siphon']) self.assertTrue(barrier.check(cut.add_food(('external_unrelated_food',)))['absence_certified']) rescue=cut.add_food(('C00141',));self.assertFalse(barrier.check(rescue)['food_disjoint']) ids=[SMALL_DIRECTIONS[i] for i in (7,8,4,5)];analysis=RAFAnalysis(rescue) self.assertTrue(analysis.is_raf(ids));self.assertIn(TARGET,analysis.closure(ids)[0]) weak=replace(self.system,reactions=tuple(replace(r,catalyst=CatalystFormula('true')) for r in self.system.reactions)) result=InterventionStudy(weak,DELETED_ACTIONS,TARGET,PRESERVATION_TARGET,barrier).evaluate(self.inputs['restoration']) self.assertTrue(result['inclusion_minimal_cut_certified']);self.assertTrue(result['preservation_capability']) strong=replace(self.system,reactions=tuple(replace(r,catalyst=CatalystFormula.parse('absent_catalyst')) for r in self.system.reactions)) self.assertTrue(barrier.check(strong.delete_actions(DELETED_ACTIONS))['absence_certified']) self.assertFalse(RAFAnalysis(strong).is_raf(self.inputs['restoration']['R01209'])) def test_parent_set_equality_and_closed_extensions(self): L=frozenset(SMALL_DIRECTIONS[i] for i in (0,1,4,7,8)) for system,counts in [(self.system,[9231,2823,2148,2148]),(self.system.delete_actions(DELETED_ACTIONS),[9227,2584,2085,2085])]: analysis=RAFAnalysis(system); certificate=analysis.closed_uniqueness_certificate() self.assertEqual(certificate['status'],'unique_closed_raf');self.assertEqual(certificate['pruning_counts'],counts) self.assertEqual(len(certificate['activation_layers']),23) self.assertEqual(certificate['maximum'],certificate['activated']) pool=set(system.food) for layer in certificate['activation_layers']: self.assertTrue(all(analysis.supported(k,pool) for k in layer)) for k in layer: pool.update(system.by_id[k].outputs) self.assertTrue(analysis.is_closed(certificate['maximum'])) parent=RAFAnalysis(self.system) self.assertEqual(parent.closed_extension(L),parent.closed_extension(SMALL_DIRECTIONS)) self.assertEqual(parent.closed_extension(L),parent.maximum()[0]) def test_closure_and_maximum_against_independent_exhaustion(self): # Small supplied/unsupplied catalytic networks: maximum equals union of all RAFs. for cats in itertools.product(('food','A','B','A&B'),repeat=2): rows=(Reaction('r0','a0',(('food',1),),(('A',1),),CatalystFormula.parse(cats[0])), Reaction('r1','a1',(('A',1),),(('B',1),),CatalystFormula.parse(cats[1]))) analysis=RAFAnalysis(ReactionSystem(rows,frozenset(('food',)))) accepted=[] for subset in (('r0',),('r1',),('r0','r1')): pool={'food'};changed=True while changed: previous=set(pool) for k in subset: row=analysis.rows[k] if row.inputs<=pool:pool.update(row.outputs) changed=pool!=previous is_raf=all(analysis.rows[k].inputs<=pool and analysis.rows[k].catalyst.holds(pool) for k in subset) self.assertEqual(analysis.is_raf(subset),is_raf) if is_raf:accepted.extend(subset) self.assertEqual(analysis.maximum()[0],frozenset(accepted)) if __name__=='__main__':unittest.main()