From 5e8417baf25d7d59c91c4de917bf5ef95e10c9a2 Mon Sep 17 00:00:00 2001 From: Artur Meski Date: Wed, 28 Dec 2016 22:13:16 +0100 Subject: [PATCH] SmtCheckerPGRS -> ..RS; new: SmtCheckerRSNA --- rs/__init__.py | 5 +- rs_examples.py | 4 +- rs_testing.py | 13 +- smt/__init__.py | 5 +- ...{smt_checker_pgrs.py => smt_checker_rs.py} | 2 +- smt/smt_checker_rs_na.py | 276 ++++++++++++++++++ 6 files changed, 292 insertions(+), 13 deletions(-) mode change 100644 => 100755 rs_examples.py rename smt/{smt_checker_pgrs.py => smt_checker_rs.py} (99%) create mode 100644 smt/smt_checker_rs_na.py diff --git a/rs/__init__.py b/rs/__init__.py index e64783c..54061b4 100644 --- a/rs/__init__.py +++ b/rs/__init__.py @@ -4,8 +4,7 @@ from rs.context_automaton import ContextAutomaton from rs.reaction_system_with_concentrations import ReactionSystemWithConcentrations from rs.context_automaton_with_concentrations import ContextAutomatonWithConcentrations -from rs.reaction_system_with_automaton import ReactionSystemWithAutomaton - from rs.extended_context_automaton import ExtendedContextAutomaton -from rs.network_of_context_automata import NetworkOfContextAutomata \ No newline at end of file +from rs.network_of_context_automata import NetworkOfContextAutomata +from rs.reaction_system_with_automaton import ReactionSystemWithAutomaton diff --git a/rs_examples.py b/rs_examples.py old mode 100644 new mode 100755 index 4bbd614..bbe9478 --- a/rs_examples.py +++ b/rs_examples.py @@ -67,7 +67,7 @@ def chain_reaction(print_system=False): if print_system: print("\nTranslated:") orc.show() - smt_tr_rs = SmtCheckerPGRS(orc) + smt_tr_rs = SmtCheckerRS(orc) smt_tr_rs.check_reachability(['e_'+str(chainLen)+"#"+str(maxConc)]) # print("Reaction System with Concentrations:", smt_rsc.get_verification_time()) @@ -181,7 +181,7 @@ def heat_shock_response(print_system=True,verify_rsc=True): if print_system: print("\nTranslated:") orc.show() - smt_tr_rs = SmtCheckerPGRS(orc) + smt_tr_rs = SmtCheckerRS(orc) smt_tr_rs.check_reachability(rs_prop) def state_translate_rsc2rs(p): diff --git a/rs_testing.py b/rs_testing.py index 8f11eed..62e6e95 100644 --- a/rs_testing.py +++ b/rs_testing.py @@ -32,9 +32,12 @@ def test_extended_automaton(): na = NetworkOfContextAutomata([c1,c2]) - rc = ReactionSystemWithAutomaton(r,na) + rna = ReactionSystemWithAutomaton(r,na) - rc.show() + rna.show() + + checker = SmtCheckerRS(rna) + def process(): @@ -44,15 +47,15 @@ def process(): -# PGRS: +# RS: # rsca = rs_examples.ca_toy_ex1() # rsca.show() -# smt = SmtCheckerPGRS(rsca) +# smt = SmtCheckerRS(rsca) # smt.check_reachability(rs_examples.ca_toy_ex1_property1(), print_time=True) # rsca = rs_examples.ca_bitctr(N) # rsca.show(True) -# smt = SmtCheckerPGRS(rsca) +# smt = SmtCheckerRS(rsca) # smt.check_reachability(rs_examples.ca_bitctr_property(N), print_time=True) # Distributed RS: diff --git a/smt/__init__.py b/smt/__init__.py index 7cf18cb..a8f6e2e 100644 --- a/smt/__init__.py +++ b/smt/__init__.py @@ -1,4 +1,5 @@ from smt.smt_checker import SmtChecker -from smt.smt_checker_distrib_rs import SmtCheckerDistribRS -from smt.smt_checker_pgrs import SmtCheckerPGRS +from smt.smt_checker_rs import SmtCheckerRS from smt.smt_checker_rsc import SmtCheckerRSC +from smt.smt_checker_rs_na import SmtCheckerRSNA +from smt.smt_checker_distrib_rs import SmtCheckerDistribRS diff --git a/smt/smt_checker_pgrs.py b/smt/smt_checker_rs.py similarity index 99% rename from smt/smt_checker_pgrs.py rename to smt/smt_checker_rs.py index b642ea3..e4d634a 100644 --- a/smt/smt_checker_pgrs.py +++ b/smt/smt_checker_rs.py @@ -7,7 +7,7 @@ from time import time from sys import stdout import resource -class SmtCheckerPGRS(object): +class SmtCheckerRS(object): def __init__(self, rsca): diff --git a/smt/smt_checker_rs_na.py b/smt/smt_checker_rs_na.py new file mode 100644 index 0000000..58ce3fb --- /dev/null +++ b/smt/smt_checker_rs_na.py @@ -0,0 +1,276 @@ +""" +SMT-based Model Checking Module for RS with Context Automaton +""" + +from z3 import * +from time import time +from sys import stdout +import resource + +class SmtCheckerRSNA(object): + """SMT-based Model Checking for Reaction Systems with Network of Automata""" + + def __init__(self, rsca): + + rsca.sanity_check() + + self.rs = rsca.rs + self.ca = rsca.ca + + self.v = [] + self.v_ctx = [] + self.ca_state = [] + self.next_level_to_encode = 0 + + self.solver = Solver() + + self.verification_time = None + + def prepare_all_variables(self): + """Encodes all the variables""" + + self.prepare_state_variables() + self.prepare_context_variables() + self.next_level_to_encode += 1 + + def prepare_context_variables(self): + """Encodes all the context variables""" + + level = self.next_level_to_encode + + variables = [] + for entity in self.rs.background_set: + variables.append(Bool("C"+str(level)+"_"+entity)) + + self.v_ctx.append(variables) + + def prepare_state_variables(self): + """Encodes all the state variables""" + + level = self.next_level_to_encode + + variables = [] + for entity in self.rs.background_set: + variables.append(Bool("L"+str(level)+"_"+entity)) + self.v.append(variables) + + self.ca_state.append(Int("CA"+str(level)+"_state")) + + def enc_init_state(self, level): + """Encodes the initial state at the given level""" + + rs_init_state_enc = True + + for v in self.v[level]: + rs_init_state_enc = simplify(And(rs_init_state_enc, Not(v))) # the initial state is empty + + ca_init_state_enc = self.ca_state[level] == self.ca.get_init_state_id() + + init_state_enc = simplify(And(rs_init_state_enc, ca_init_state_enc)) + + return init_state_enc + + def enc_enabledness(self, level, prod_entity): + """Encodes the enabledness condition for a given level and a given entity""" + + rcts_for_prod_entity = self.rs.get_reactions_by_product()[prod_entity] + + if rcts_for_prod_entity == []: + return False + + enc_rct_prod = False + for reactants,inhibitors in rcts_for_prod_entity: + enc_reactants = True + enc_inhibitors = True + for reactant in reactants: + enc_reactants = simplify(And(enc_reactants, + Or(self.v[level][reactant], self.v_ctx[level][reactant]))) + for inhibitor in inhibitors: + enc_inhibitors = simplify(And(enc_inhibitors, + Not(Or(self.v[level][inhibitor], self.v_ctx[level][inhibitor])))) + + enc_rct_prod = simplify(Or(enc_rct_prod, And(enc_reactants, enc_inhibitors))) + + return enc_rct_prod + + def enc_entity_production(self, level, prod_entity): + """Encodes the production of a given entity from a given level at level+1""" + + enc_enab_cond = self.enc_enabledness(level, prod_entity) + + enc_ent_prod = Or(And(enc_enab_cond, self.v[level+1][prod_entity]), + And(Not(enc_enab_cond), Not(self.v[level+1][prod_entity]))) + + return simplify(enc_ent_prod) + + def enc_transition_relation(self, level): + return simplify(And(self.enc_rs_trans(level), self.enc_automaton_trans(level))) + + def enc_rs_trans(self, level): + """Encodes the transition relation""" + + unused_entities = list(range(len(self.rs.background_set))) + + enc_trans = True + + for prod_entity in self.rs.get_reactions_by_product(): + unused_entities.remove(prod_entity) + + enc_trans = simplify(And(enc_trans, self.enc_entity_production(level, prod_entity))) + + for prod_entity in unused_entities: + enc_trans = simplify(And(enc_trans, Not(self.v[level+1][prod_entity]))) + + return enc_trans + + def enc_automaton_trans(self, level): + """Encodes the transition relation for the context automaton""" + + enc_trans = False + + for src,ctx,dst in self.ca.transitions: + src_enc = self.ca_state[level] == src + dst_enc = self.ca_state[level+1] == dst + + all_ent = set(range(len(self.rs.background_set))) + incl_ctx = ctx + excl_ctx = all_ent - incl_ctx + + ctx_enc = True + + for c in incl_ctx: + ctx_enc = simplify(And(ctx_enc, self.v_ctx[level][c])) + for c in excl_ctx: + ctx_enc = simplify(And(ctx_enc, Not(self.v_ctx[level][c]))) + + cur_trans = simplify(And(src_enc, ctx_enc, dst_enc)) + + enc_trans = simplify(Or(enc_trans, cur_trans)) + + return enc_trans + + def enc_state(self, level, state): + """Encodes the state at the given level""" + + enc = True + + state_ids = self.rs.get_state_ids(state) + + for entity in state_ids: + enc = And(enc, self.v[level][entity]) + + not_in_state = set(range(len(self.rs.background_set))) + not_in_state = not_in_state.difference(set(state_ids)) + + for entity in not_in_state: + enc = And(enc, Not(self.v[level][entity])) + + return enc + + def enc_non_exclusive_state(self, level, state): + """Encodes the state at the given level""" + + enc = True + + state_ids = self.rs.get_state_ids(state) + + for entity in state_ids: + enc = And(enc, self.v[level][entity]) + + return enc + + def enc_state_with_blocking(self, level, prop): + """Encodes the state at the given level with blocking certain concentrations""" + + required,blocked = prop + + enc = True + + required_ids = self.rs.get_state_ids(required) + blocked_ids = self.rs.get_state_ids(blocked) + + for e in required_ids: + enc = And(enc, self.v[level][e]) + for e in blocked_ids: + enc = And(enc, Not(self.v[level][e])) + + return simplify(enc) + + + def decode_witness(self, max_level, print_model=False): + + m = self.solver.model() + + if print_model: + print(m) + + for level in range(max_level+1): + + print("\n[Level=" + repr(level) + "]") + + print(" State: {", end=""), + for var_id in range(len(self.v[level])): + if repr(m[self.v[level][var_id]]) == "True": + print(" " + self.rs.get_entity_name(var_id), end="") + print(" }") + + if level != max_level: + print(" Context set: ", end="") + print("{", end="") + for var_id in range(len(self.v[level])): + if repr(m[self.v_ctx[level][var_id]]) == "True": + print(" " + self.rs.get_entity_name(var_id), end="") + print(" }") + + def check_reachability(self, state, print_witness=True, print_time=True, print_mem=True): + """Main testing function""" + + if not type(state) is tuple: + state = (state,[]) + + if print_time: + # start = time() + start = resource.getrusage(resource.RUSAGE_SELF).ru_utime + + self.prepare_all_variables() + self.solver.add(self.enc_init_state(0)) + current_level = 0 + + while True: + print("-----[ Working at level=" + str(current_level) + " ]-----") + stdout.flush() + + self.prepare_all_variables() + + # reachability test: + print("[i] Adding the reachability test...") + self.solver.push() + self.solver.add(self.enc_state_with_blocking(current_level,state)) + + result = self.solver.check() + if result == sat: + print("\n[+] SAT at level=" + str(current_level)) + if print_witness: + self.decode_witness(current_level) + break + else: + self.solver.pop() + + print("[i] Unrolling the transition relation") + self.solver.add(self.enc_transition_relation(current_level)) + + print("-----[ level=" + str(current_level) + " done ]") + current_level += 1 + + if print_time: + # stop = time() + stop = resource.getrusage(resource.RUSAGE_SELF).ru_utime + self.verification_time = stop-start + print() + print("[i] Time: " + repr(self.verification_time)) + + if print_mem: + print("[i] Memory: " + repr(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss/(1024*1024)) + " MB") + + def get_verification_time(self): + return self.verification_time