diff --git a/rs_examples.py b/rs_examples.py index 82b1976..4bbd614 100755 --- a/rs_examples.py +++ b/rs_examples.py @@ -1,11 +1,7 @@ #!/usr/bin/env python -from rctsys import * -from distrib_rctsys import * -from smtchecker import SmtChecker -from smtcheckerpgrs import SmtCheckerPGRS -from smtcheckerdistribrs import SmtCheckerDistribRS -from smtcheckerrsc import SmtCheckerRSC +from rs import * +from smt import * import sys import resource diff --git a/rssmt.py b/rssmt.py index d9a1c97..bac3eab 100755 --- a/rssmt.py +++ b/rssmt.py @@ -6,11 +6,8 @@ """ -from rctsys import * -from smtchecker import SmtChecker -from smtcheckerpgrs import SmtCheckerPGRS -from smtcheckerdistribrs import SmtCheckerDistribRS -from smtcheckerrsc import SmtCheckerRSC +from rs import * +from smt import * import sys import rs_examples diff --git a/smtchecker.py b/smtchecker.py deleted file mode 100644 index 99ec664..0000000 --- a/smtchecker.py +++ /dev/null @@ -1,530 +0,0 @@ -""" -SMT-based Model Checking Module for RS -""" - -from z3 import * -from time import time -from sys import stdout - -class SmtChecker(object): - - def __init__(self, rs): - - ############################################################## - # Encoded RS - ############################################################## - rs.sanity_check() - self.reaction_system = rs - - ############################################################## - # SMT variables - ############################################################## - self.v = [] - self.v_init = [] - - #self.vSucc = [] - #self.vSuccInit = [] - - self.v_ctx = [] - - self.next_level_to_encode = 0 - - ############################################################## - # SMT solver instance - ############################################################## - self.solver = Solver() - - #def smtVar(self, level, entityID, primed=False): - # return "?" - - def prepare_context_variables(self): - """Encodes all the context variables""" - - level = self.next_level_to_encode - - variables = [] - for entity in self.reaction_system.background_set: - variables.append(Bool("C"+str(level)+"_"+entity)) - - self.v_ctx.append(variables) - - def prepare_state_variables(self): - """Encodes all the state variables (including successors)""" - - level = self.next_level_to_encode - - variables = [] - #variablesSucc = [] - for entity in self.reaction_system.background_set: - variables.append(Bool("L"+str(level)+"_"+entity)) - #variablesSucc.append(Bool("R"+str(level)+"_"+entity)) - - self.v.append(variables) - #self.vSucc.append(variablesSucc) - - self.v_init.append(Bool("L"+str(level)+"_Init")) - #self.vSuccInit.append(Bool("R"+str(level)+"_Init")) - - def prepare_all_variables(self): - """Encodes all the variables""" - - self.prepare_state_variables() - self.prepare_context_variables() - self.next_level_to_encode += 1 - - def enc_init_state(self, level): - """Encodes the initial state at the given level""" - - init_state_enc = self.v_init[level] - - for v in self.v[level]: - init_state_enc = simplify(And(init_state_enc, Not(v))) - - return init_state_enc - - def enc_init_contexts(self, level): - """Encodes the initial contexts set at the given level""" - - init_contexts_set_enc = False # Or - - for ctx in self.reaction_system.init_contexts: - single_ctx_enc = True # And - - not_ctx_entities = list(range(0, len(self.reaction_system.background_set))) - for entity in ctx: - single_ctx_enc = simplify(And(single_ctx_enc, self.v_ctx[level][entity])) - not_ctx_entities.remove(entity) - - for entity in not_ctx_entities: - single_ctx_enc = simplify(And(single_ctx_enc, Not(self.v_ctx[level][entity]))) - - init_contexts_set_enc = simplify(Or(init_contexts_set_enc, single_ctx_enc)) - - #print("initContextSetEnc: " + repr(initContextsSetEnc)) - return init_contexts_set_enc - - def enc_not_allowed_contexts(self, level): - """Encodes all the context entities that are not allowed in the context sets""" - - bg_set = set(range(0,len(self.reaction_system.background_set))) - ctx_ent_set = set(self.reaction_system.context_entities) - - not_ctx_ent_set = bg_set.difference(ctx_ent_set) - - enc = True - for entity in not_ctx_ent_set: - enc = simplify(And(enc, Not(self.v_ctx[level][entity]))) - - return 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.reaction_system.get_reactions_by_product()[prod_entity] - - if rcts_for_prod_entity == []: - return False - - #encInitEnab = simplify(Or(And(self.vInit[level], self.encInitContexts(level)), - # And(Not(self.vInit[level]), self.encNotAllowedContexts(level)))) - - enc_rct_prod = False - for ri_pair in rcts_for_prod_entity: # reactants-inhibitors pair - enc_reactants = True - enc_inhibitors = True - for reactant in ri_pair[0]: - enc_reactants = simplify(And(enc_reactants, - Or(self.v[level][reactant], self.v_ctx[level][reactant]))) - for inhibitor in ri_pair[1]: - 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))) - - #print("encEnabledness(" + repr(prodEntity) + "): " + repr(simplify(And(encInitEnab, encRctProd)))) - #return simplify(And(encInitEnab, encRctProd)) - 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 enc_ent_prod - - def enc_transition_relation(self, level): - """Encodes the transition relation""" - - unused_entities = list(range(0,len(self.reaction_system.background_set))) - - enc_trans = True - - for prod_entity in self.reaction_system.get_reactions_by_product(): - unused_entities.remove(prod_entity) - - enc_trans = simplify(And(enc_trans, self.enc_entity_production(level, prod_entity))) - - enc_trans = simplify(And(enc_trans, Not(self.v_init[level+1]))) - - for prod_entity in unused_entities: - enc_trans = simplify(And(enc_trans, Not(self.v[level+1][prod_entity]))) - - if level == 0: - enc_init_enab = self.enc_init_contexts(level) - else: # level > 0: - enc_init_enab = self.enc_not_allowed_contexts(level) - - #encInitEnab = simplify(Or(And(self.vInit[level], self.encInitContexts(level)), - # And(Not(self.vInit[level]), self.encNotAllowedContexts(level)))) - - enc_trans = simplify(And(enc_trans, enc_init_enab)) - - return enc_trans - - def enc_state(self, level, state): - """Encodes the state at the given level""" - - enc = Not(self.v_init[level]) - - state_ids = self.reaction_system.get_state_ids(state) - - for entity in state_ids: - enc = And(enc, self.v[level][entity]) - - not_in_state = set(range(0, len(self.reaction_system.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 decode_witness(self, max_level): - - m = self.solver.model() - - for level in range(0,max_level+1): - - print("\n[Level=" + repr(level) + "]") - - if repr(m[self.v_init[level]]) == "True": - print("** Initial state") - - print("State:\n{"), - for var_id in range(0, len(self.v[level])): - if repr(m[self.v[level][var_id]]) == "True": - print("\t" + self.reaction_system.get_entity_name(var_id)), - print("}") - - if level != max_level: - print("Context set:"), - print("{"), - for var_id in range(0, len(self.v[level])): - if repr(m[self.v_ctx[level][var_id]]) == "True": - print("\t" + self.reaction_system.get_entity_name(var_id)), - print("}") - - - def check_reachability(self, state, print_witness=True, print_time=False): - """Main testing function""" - - if print_time: - start = time() - - self.prepare_all_variables() - self.solver.add(self.enc_init_state(0)) - current_level = 0 - - while True: - #print("Level: " + str(current_level)) - print("\rLevel: " + str(current_level)), - stdout.flush() - - self.prepare_all_variables() - - # reachability test: - self.solver.push() - self.solver.add(self.enc_state(current_level,state)) - - result = self.solver.check() - print(result) - if result == sat: - print("\nSAT") - if print_witness: - self.decode_witness(current_level) - break - else: - self.solver.pop() - - self.solver.add(self.enc_transition_relation(current_level)) - - current_level += 1 - - if print_time: - stop = time() - print("Time: " + repr(stop-start)) - - -class SmtCheckerPGRS(object): - - def __init__(self, rs): - - ############################################################## - # Encoded RS - ############################################################## - rs.sanity_check() - self.reaction_system = rs - - ############################################################## - # SMT variables - ############################################################## - self.v = [] - self.v_init = [] - - #self.vSucc = [] - #self.vSuccInit = [] - - self.v_ctx = [] - - self.next_level_to_encode = 0 - - ############################################################## - # SMT solver instance - ############################################################## - self.solver = Solver() - - #def smtVar(self, level, entityID, primed=False): - # return "?" - - def prepare_context_variables(self): - """Encodes all the context variables""" - - level = self.next_level_to_encode - - variables = [] - for entity in self.reaction_system.background_set: - variables.append(Bool("C"+str(level)+"_"+entity)) - - self.v_ctx.append(variables) - - def prepare_state_variables(self): - """Encodes all the state variables (including successors)""" - - level = self.next_level_to_encode - - variables = [] - #variablesSucc = [] - for entity in self.reaction_system.background_set: - variables.append(Bool("L"+str(level)+"_"+entity)) - #variablesSucc.append(Bool("R"+str(level)+"_"+entity)) - - self.v.append(variables) - #self.vSucc.append(variablesSucc) - - self.v_init.append(Bool("L"+str(level)+"_Init")) - #self.vSuccInit.append(Bool("R"+str(level)+"_Init")) - - def prepare_all_variables(self): - """Encodes all the variables""" - - self.prepare_state_variables() - self.prepare_context_variables() - self.next_level_to_encode += 1 - - def enc_init_state(self, level): - """Encodes the initial state at the given level""" - - init_state_enc = self.v_init[level] - - for v in self.v[level]: - init_state_enc = simplify(And(init_state_enc, Not(v))) - - return init_state_enc - - def enc_init_contexts(self, level): - """Encodes the initial contexts set at the given level""" - - init_contexts_set_enc = False # Or - - for ctx in self.reaction_system.init_contexts: - single_ctx_enc = True # And - - not_ctx_entities = list(range(0, len(self.reaction_system.background_set))) - for entity in ctx: - single_ctx_enc = simplify(And(single_ctx_enc, self.v_ctx[level][entity])) - not_ctx_entities.remove(entity) - - for entity in not_ctx_entities: - single_ctx_enc = simplify(And(single_ctx_enc, Not(self.v_ctx[level][entity]))) - - init_contexts_set_enc = simplify(Or(init_contexts_set_enc, single_ctx_enc)) - - #print("initContextSetEnc: " + repr(initContextsSetEnc)) - return init_contexts_set_enc - - def enc_not_allowed_contexts(self, level): - """Encodes all the context entities that are not allowed in the context sets""" - - bg_set = set(range(0,len(self.reaction_system.background_set))) - ctx_ent_set = set(self.reaction_system.context_entities) - - not_ctx_ent_set = bg_set.difference(ctx_ent_set) - - enc = True - for entity in not_ctx_ent_set: - enc = simplify(And(enc, Not(self.v_ctx[level][entity]))) - - return 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.reaction_system.get_reactions_by_product()[prod_entity] - - if rcts_for_prod_entity == []: - return False - - #encInitEnab = simplify(Or(And(self.vInit[level], self.encInitContexts(level)), - # And(Not(self.vInit[level]), self.encNotAllowedContexts(level)))) - - enc_rct_prod = False - for ri_pair in rcts_for_prod_entity: # reactants-inhibitors pair - enc_reactants = True - enc_inhibitors = True - for reactant in ri_pair[0]: - enc_reactants = simplify(And(enc_reactants, - Or(self.v[level][reactant], self.v_ctx[level][reactant]))) - for inhibitor in ri_pair[1]: - 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))) - - #print("encEnabledness(" + repr(prodEntity) + "): " + repr(simplify(And(encInitEnab, encRctProd)))) - #return simplify(And(encInitEnab, encRctProd)) - 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 enc_ent_prod - - def enc_transition_relation(self, level): - """Encodes the transition relation""" - - unused_entities = list(range(0,len(self.reaction_system.background_set))) - - enc_trans = True - - for prod_entity in self.reaction_system.get_reactions_by_product(): - unused_entities.remove(prod_entity) - - enc_trans = simplify(And(enc_trans, self.enc_entity_production(level, prod_entity))) - - enc_trans = simplify(And(enc_trans, Not(self.v_init[level+1]))) - - for prod_entity in unused_entities: - enc_trans = simplify(And(enc_trans, Not(self.v[level+1][prod_entity]))) - - if level == 0: - enc_init_enab = self.enc_init_contexts(level) - else: # level > 0: - enc_init_enab = self.enc_not_allowed_contexts(level) - - #encInitEnab = simplify(Or(And(self.vInit[level], self.encInitContexts(level)), - # And(Not(self.vInit[level]), self.encNotAllowedContexts(level)))) - - enc_trans = simplify(And(enc_trans, enc_init_enab)) - - return enc_trans - - def enc_state(self, level, state): - """Encodes the state at the given level""" - - enc = Not(self.v_init[level]) - - state_ids = self.reaction_system.get_state_ids(state) - - for entity in state_ids: - enc = And(enc, self.v[level][entity]) - - not_in_state = set(range(0, len(self.reaction_system.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 decode_witness(self, max_level): - - m = self.solver.model() - - for level in range(0,max_level+1): - - print("\n[Level=" + repr(level) + "]") - - if repr(m[self.v_init[level]]) == "True": - print("** Initial state") - - print("State:\n{"), - for var_id in range(0, len(self.v[level])): - if repr(m[self.v[level][var_id]]) == "True": - print("\t" + self.reaction_system.get_entity_name(var_id)), - print("}") - - if level != max_level: - print("Context set:"), - print("{"), - for var_id in range(0, len(self.v[level])): - if repr(m[self.v_ctx[level][var_id]]) == "True": - print("\t" + self.reaction_system.get_entity_name(var_id)), - print("}") - - - def check_reachability(self, state, print_witness=True, print_time=False): - """Main testing function""" - - if print_time: - start = time() - - self.prepare_all_variables() - self.solver.add(self.enc_init_state(0)) - current_level = 0 - - while True: - #print("Level: " + str(current_level)) - print("\rLevel: " + str(current_level)), - stdout.flush() - - self.prepare_all_variables() - - # reachability test: - self.solver.push() - self.solver.add(self.enc_state(current_level,state)) - - result = self.solver.check() - print(result) - if result == sat: - print("\nSAT") - if print_witness: - self.decode_witness(current_level) - break - else: - self.solver.pop() - - self.solver.add(self.enc_transition_relation(current_level)) - - current_level += 1 - - if print_time: - stop = time() - print("Time: " + repr(stop-start)) \ No newline at end of file diff --git a/smtcheckerdistribrs.py b/smtcheckerdistribrs.py deleted file mode 100644 index 53e84be..0000000 --- a/smtcheckerdistribrs.py +++ /dev/null @@ -1,432 +0,0 @@ -""" -SMT-based Model Checking Module for RS with Context Automaton -""" - -from z3 import * -from time import time,sleep -from sys import stdout -import resource - -# def simplify(x): -# return x - -class SmtCheckerDistribRS(object): - - def __init__(self, drs, debug_level=1): - - print("[i] Initialising the SMT module") - - drs.sanity_check() - - self.solver = Solver() - - self.debug_level = debug_level - self.drs = drs - self.n_components = self.drs.components_count - - # encoding variables - self.v = [] # level -> component -> variables - self.v_ctx = [] - self.v_act = [] # indicators of which component is active - self.ca_state = [] - - self.level_to_encode = 0 - - def prepare_all_variables(self): - """Encodes the required variables""" - - self.prepare_state_variables() - self.prepare_context_variables() - self.prepare_activity_variables() - - self.level_to_encode += 1 # prepare for the next invocation - - def prepare_context_variables(self): - """Encodes the context variables""" - - level = self.level_to_encode - - if self.debug_level > 1: - print("[ii] Preparing context variables for level=" + str(level)) - - level_variables = [] - - for i in range(self.n_components): - - comp_variables = [] - - for entity in self.drs.background_set: - comp_variables.append(Bool("Ctx"+str(level)+"V"+str(i)+"_"+entity)) - - level_variables.append(comp_variables) - - self.v_ctx.append(level_variables) - - def prepare_activity_variables(self): - """Encodes the activity variables""" - - level = self.level_to_encode - - if self.debug_level > 1: - print("[ii] Preparing activity variables for level=" + str(level)) - - level_variables = [] - - for i in range(self.n_components): - # L - level, A - activity indicator - level_variables.append(Bool("L"+str(level)+"A"+str(i))) - - self.v_act.append(level_variables) - - def prepare_state_variables(self): - """Encodes all the state variables""" - - level = self.level_to_encode - - if self.debug_level > 1: - print("[ii] Preparing state variables for level=" + str(level)) - - level_variables = [] # level vars - - for i in range(self.n_components): - - comp_variables = [] - - for entity in self.drs.background_set: - # L - level, V - component - comp_variables.append(Bool("L"+str(level)+"V"+str(i)+"_"+entity)) - - level_variables.append(comp_variables) - - self.v.append(level_variables) - - # single state variable for CA - self.ca_state.append(Int("CA"+str(level)+"_state")) - - def enc_init_state(self, level): - """Encodes the initial state at the given level""" - - if self.debug_level > 1: - print("[ii] Encoding the initial state for level=" + str(level)) - - rs_init_state_enc = True - - for i in range(self.n_components): - for v in self.v[level][i]: - 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.drs.get_init_state_id() - init_state_enc = simplify(And(rs_init_state_enc, ca_init_state_enc)) - - # print ("init_state_enc:\n", init_state_enc) - - return init_state_enc - - def enc_enabledness(self, level, prod_entity, component_id): - """Encodes the enabledness condition for a given level and a given entity""" - - rcts_for_prod_entity = self.drs.get_reactions_by_product(component_id)[prod_entity] - - if rcts_for_prod_entity == []: - return False - - enc_rct_prod = False - for reactants,inhibitors in rcts_for_prod_entity: - - - enc_reactants = True - for reactant in reactants: - - enc_active_reactants = False - for i in range(self.n_components): - enc_active_reactants = simplify(Or(enc_active_reactants, - And( Or(self.v[level][i][reactant], self.v_ctx[level][i][reactant]), self.v_act[level][i] ))) - - enc_reactants = And(enc_reactants, enc_active_reactants) - - - enc_inhibitors = True - for inhibitor in inhibitors: - - enc_active_inhibitors = True - for i in range(self.n_components): - enc_active_inhibitors = simplify(And(enc_active_inhibitors, - And( - Or( And(Not(self.v[level][i][inhibitor]), Not(self.v_ctx[level][i][inhibitor])), Not(self.v_act[level][i]) ) - ))) - - enc_inhibitors = simplify(And(enc_inhibitors, enc_active_inhibitors)) - - # print("--> enc_inhibitors\n", enc_inhibitors) - enc_rct_prod = Or(enc_rct_prod, And(enc_reactants, enc_inhibitors)) - - # print("enc_rct_prod:\n", enc_rct_prod) - - enc_rct_prod = simplify(enc_rct_prod) - - return enc_rct_prod - - def enc_entity_production(self, level, prod_entity, component_id): - """Encodes the production of a given entity at level+1 from a given level""" - - enc_enab_cond = self.enc_enabledness(level, prod_entity, component_id) - - enc_base_ent_prod = simplify(Or(And(enc_enab_cond, self.v[level+1][component_id][prod_entity]), - And(Not(enc_enab_cond), Not(self.v[level+1][component_id][prod_entity])))) - - enc_active_ent_prod = simplify(And(self.v_act[level][component_id], enc_base_ent_prod)) - enc_inactive_ent_prod = simplify(And( - Not(self.v_act[level][component_id]), - self.v[level][component_id][prod_entity] == self.v[level+1][component_id][prod_entity])) - - enc_ent_prod = Or(enc_active_ent_prod, enc_inactive_ent_prod) - - # print("enc_ent_prod:\n", enc_ent_prod) - - return simplify(enc_ent_prod) - - def enc_rs_trans(self, level): - """Encodes the transition relation""" - - enc_trans = True - - for component_id in range(self.n_components): - - print("\rEncoding for reactions: %d/%d" % (component_id,self.n_components-1), flush=True, end="") - - unused_entities = list(range(len(self.drs.background_set))) - - for prod_entity in self.drs.get_reactions_by_product(component_id): - unused_entities.remove(prod_entity) - - enc_trans = simplify(And(enc_trans, self.enc_entity_production(level, prod_entity, component_id))) - - for prod_entity in unused_entities: - enc_trans = simplify(And(enc_trans, Not(self.v[level+1][component_id][prod_entity]))) - - print() - # print("enc_rs_trans:\n", enc_trans) - - enc_trans = simplify(enc_trans) - - return enc_trans - - def enc_automaton_trans(self, level): - """Encodes the transition relation for the context automaton""" - - enc_trans = False - - i = 0 - for src,(components,ctx_set),dst in self.drs.transitions: - src_enc = self.ca_state[level] == src - dst_enc = self.ca_state[level+1] == dst - - print("\rEncoding for context automaton: %d/%d" % (i,len(self.drs.transitions)-1), flush=True, end="") - - i = i + 1 - - # contexts { - ctx_set_enc = True - for comp_id in range(self.n_components): - - all_ent = set(range(len(self.drs.background_set))) - incl_ctx = ctx_set[comp_id] - excl_ctx = all_ent - incl_ctx - - ctx_enc = True - - for c in incl_ctx: - ctx_enc = And(ctx_enc, self.v_ctx[level][comp_id][c]) - for c in excl_ctx: - ctx_enc = And(ctx_enc, Not(self.v_ctx[level][comp_id][c])) - - ctx_set_enc = And(ctx_set_enc, ctx_enc) - # } contexts - - # active components { - all_active = set(range(self.n_components)) - incl_comp = components - excl_comp = all_active - incl_comp - - active_components_enc = True - for comp_id in incl_comp: - active_components_enc = And(active_components_enc, self.v_act[level][comp_id]) - for comp_id in excl_comp: - active_components_enc = And(active_components_enc, Not(self.v_act[level][comp_id])) - # } active components - - cur_trans = And(src_enc, ctx_set_enc, active_components_enc, dst_enc) - - enc_trans = Or(enc_trans, cur_trans) - - print() - - enc_trans = simplify(enc_trans) - - # print("enc_automaton_trans:\n", enc_trans) - return enc_trans - - def enc_transition_relation(self, level): - - rs_enc = self.enc_rs_trans(level) - aut_enc = self.enc_automaton_trans(level) - - print("Conjunction...", flush=True, end="") - - c = simplify(And(rs_enc, aut_enc)) - - print("done.") - - return c - - def enc_state(self, level, global_state): - """Encodes the state at the given level""" - - if len(global_state) != self.n_components: - print("EEE: Wrong size of the global state! " + "(is " + str(len(global_state)) + ", should be " + str(self.n_components) + ")") - exit(1) - - enc = True - - if self.debug_level > 2: - print("[iii] Encoding exclusive/exact global state " + str(global_state) + " for level=" + str(level)) - - for i in range(self.n_components): - local_state = global_state[i] - - local_state_ids = self.drs.get_state_ids(local_state) - - for entity in local_state_ids: - enc = And(enc, self.v[level][i][entity]) - - not_in_state = self.drs.set_of_background_ids - set(local_state_ids) - - for e_id in not_in_state: - enc = simplify(And(enc, Not(self.v[level][i][e_id]))) - - simplify(enc) - # print("state:\n", enc) - return enc - - def enc_inclusive_state(self, level, global_state): - """Encodes the state at the given level""" - - if len(global_state) != self.n_components: - print("EEE: Wrong size of the global state! " + "(is " + str(len(global_state)) + ", should be " + str(self.n_components) + ")") - exit(1) - - enc = True - - if self.debug_level > 2: - print("[iii] Encoding inclusive/general global state " + str(global_state) + " for level=" + str(level)) - - for i in range(self.n_components): - local_state = global_state[i] - - local_state_ids = self.drs.get_state_ids(local_state) - - for entity in local_state_ids: - enc = And(enc, self.v[level][i][entity]) - - simplify(enc) - return enc - - def decode_witness(self, max_level, print_model=False): - - m = self.solver.model() - - print("\nWitness:") - - if print_model: - print(m) - - for level in range(max_level+1): - - print("\n[Level=" + repr(level) + "]") - - #print(m) - #print(self.v[level][0][2]) - #print(m[self.v[level][0][2]]) - - print(" State: {", end=""), - for c in range(self.n_components): - print(" <", end="") - for var_id in range(len(self.v[level][c])): - if repr(m[self.v[level][c][var_id]]) == "True": - print(" " + self.drs.get_entity_name(var_id), end="") - print(" >", end="") - print(" }") - - if level != max_level: - print(" Context set: ", end="") - print("{", end="") - for c in range(self.n_components): - print(" <", end="") - for var_id in range(len(self.v[level][c])): - if repr(m[self.v_ctx[level][c][var_id]]) == "True": - print(" " + self.drs.get_entity_name(var_id), end="") - print(" >", end="") - print(" }") - - print(" Active components:", end="") - for c in range(self.n_components): - if repr(m[self.v_act[level][c]]) == "True": - print(" " + str(c), end="") - print() - - - - def check_reachability(self, state, exclusive_state=False, print_witness=True, print_time=False, print_mem=False, max_level=100): - """Reachability checking""" - - print("[i] Checking reachability...") - - if print_time: - start = time() - - self.prepare_all_variables() - self.solver.add(self.enc_init_state(0)) - current_level = 0 - - while True: - - self.prepare_all_variables() - - print("-----[ Working at level=" + str(current_level) + " ]-----") - stdout.flush() - - # reachability test: - self.solver.push() - print("[i] Adding the reachability test...") - if exclusive_state == False: - self.solver.add(self.enc_inclusive_state(current_level,state)) - else: - self.solver.add(self.enc_state(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 current_level > max_level: - print("Stopping at level=" + str(max_level)) - break - - if print_time: - stop = time() - print() - print("==== Time: " + repr(stop-start)) - - if print_mem: - usage=resource.getrusage(resource.RUSAGE_SELF) - print("MEM: usertime=%s systime=%s mem=%sMB" % (usage[0],usage[1], (resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024**2))) diff --git a/smtcheckerpgrs.py b/smtcheckerpgrs.py deleted file mode 100644 index b642ea3..0000000 --- a/smtcheckerpgrs.py +++ /dev/null @@ -1,275 +0,0 @@ -""" -SMT-based Model Checking Module for RS with Context Automaton -""" - -from z3 import * -from time import time -from sys import stdout -import resource - -class SmtCheckerPGRS(object): - - 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 diff --git a/smtcheckerrsc.py b/smtcheckerrsc.py deleted file mode 100644 index c9fae4c..0000000 --- a/smtcheckerrsc.py +++ /dev/null @@ -1,480 +0,0 @@ -""" -SMT-based Model Checking Module for RS with Concentrations and Context Automaton -""" - -from z3 import * -from time import time -from sys import stdout -from itertools import chain -import resource -from colour import * - -# def simplify(x): -# return x - -class SmtCheckerRSC(object): - - def __init__(self, rsca): - - rsca.sanity_check() - - if not rsca.is_with_concentrations(): - raise RuntimeError("RS and CA with concentrations expected") - - 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(Int("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(Int("L"+str(level)+"_"+entity)) - self.v.append(variables) - - self.ca_state.append(Int("CA"+str(level)+"_state")) - - def enc_concentration_levels_assertion(self, level): - """Encodes assertions that (some) variables need to be >0 - - We do not need to actually control all the variables, - only those that can possibly go below 0. - """ - - enc_nz = True - - for e_i in range(len(self.rs.background_set)): - v = self.v[level][e_i] - v_ctx = self.v_ctx[level][e_i] - e_max = self.rs.get_max_concentration_level(e_i) - enc_nz = simplify(And(enc_nz, v >= 0, v_ctx >= 0, v <= e_max, v_ctx <= e_max)) - - return enc_nz - - 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, v == 0)) # the initial concentration levels are zeroed - - 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_produced_concentration(self, level, prod_entity): - """Encodes the produced concentrations for the given level and entity""" - - rcts_for_prod_entity = [] - if prod_entity in self.rs.get_reactions_by_product(): - rcts_for_prod_entity = self.rs.get_reactions_by_product()[prod_entity] - - meta_reactions = [] - if prod_entity in self.rs.meta_reactions: - meta_reactions = self.rs.meta_reactions[prod_entity] - - permanency_inhibition = None - if prod_entity in self.rs.permanent_entities: - permanency_inhibition = self.rs.permanent_entities[prod_entity] - - if rcts_for_prod_entity == [] and meta_reactions == []: - return simplify(self.v[level+1][prod_entity] == 0) # this should never happen - - enc_enabledness = False - - # ----------- ordinary reactions -------------------------------------------- - - enc_rct_prod = False - - enc_ordinary_reactions_enabledness = False - - for reactants,inhibitors,products in rcts_for_prod_entity: - - enc_reactants = True - for reactant,concentration in reactants: - enc_reactants = simplify(And(enc_reactants, - Or(self.v[level][reactant] >= concentration, self.v_ctx[level][reactant] >= concentration))) - - enc_inhibitors = True - for inhibitor,concentration in inhibitors: - enc_inhibitors = simplify(And(enc_inhibitors, - And(self.v[level][inhibitor] < concentration, self.v_ctx[level][inhibitor] < concentration))) - - enc_rct_enabled = And(enc_reactants, enc_inhibitors) - enc_products = self.v[level+1][products[0][0]] == products[0][1] - enc_rct_prod = simplify(If(enc_rct_enabled, enc_products, enc_rct_prod)) - enc_enabledness = simplify(Or(enc_enabledness, enc_rct_enabled)) - - enc_ordinary_reactions_enabledness = simplify(Or(enc_ordinary_reactions_enabledness,enc_rct_enabled)) - - # for reactants,inhibitors,products in rcts_for_prod_entity: - # enc_reactants = True - # enc_inhibitors = True - # # enc_products -- below - # - # for reactant,concentration in reactants: - # enc_reactants = simplify(And(enc_reactants, - # Or(self.v[level][reactant] >= concentration, self.v_ctx[level][reactant] >= concentration))) - # for inhibitor,concentration in inhibitors: - # enc_inhibitors = simplify(And(enc_inhibitors, - # And(self.v[level][inhibitor] < concentration, self.v_ctx[level][inhibitor] < concentration))) - # - # enc_products = self.v[level+1][products[0][0]] == products[0][1] - # - # enc_enabledness = simplify(Or(enc_enabledness, And(enc_reactants, enc_inhibitors))) - # enc_rct_prod = simplify(Or(enc_rct_prod, And(enc_reactants, enc_inhibitors, enc_products))) - - # -------- meta reactions --------------------------------------------------- - - for r_type,command_entity,reactants,inhibitors in meta_reactions: - - # command entity is e.g. 'inc' for incrementation operation - # (inc,W) gives us the value W by which the given entity's value should be incremented - - enc_reactants = True - enc_inhibitors = True - - for reactant,concentration in reactants: - enc_reactants = simplify(And(enc_reactants, - Or(self.v[level][reactant] >= concentration, self.v_ctx[level][reactant] >= concentration))) - - # command entity needs to be present (with concentration level > 0) in order to perform the operation - enc_reactants = simplify(And(enc_reactants, - Or(self.v[level][command_entity] > 0, self.v_ctx[level][command_entity] > 0))) - - for inhibitor,concentration in inhibitors: - enc_inhibitors = simplify(And(enc_inhibitors, - And(self.v[level][inhibitor] < concentration, self.v_ctx[level][inhibitor] < concentration))) - - if r_type == "inc": - value_after_inc = If(self.v[level][prod_entity]>self.v_ctx[level][prod_entity],self.v[level][prod_entity],self.v_ctx[level][prod_entity]) + \ - If(self.v[level][command_entity]>self.v_ctx[level][command_entity],self.v[level][command_entity],self.v_ctx[level][command_entity]) - enc_products = self.v[level+1][prod_entity] == value_after_inc - - elif r_type == "dec": - value_after_dec = simplify(If(self.v[level][prod_entity]>self.v_ctx[level][prod_entity],self.v[level][prod_entity],self.v_ctx[level][prod_entity]) - \ - If(self.v[level][command_entity]>self.v_ctx[level][command_entity],self.v[level][command_entity],self.v_ctx[level][command_entity])) - enc_products = self.v[level+1][prod_entity] == If(value_after_dec < 0, 0, value_after_dec) - - else: - raise RuntimeError("Unknown meta-reaction type: " + repr(r_type)) - - enc_meta_reaction_enabledness = And(enc_reactants, enc_inhibitors, Not(enc_ordinary_reactions_enabledness)) - enc_enabledness = simplify(Or(enc_enabledness, enc_meta_reaction_enabledness)) - enc_rct_prod = simplify(Or(enc_rct_prod, And(enc_meta_reaction_enabledness, enc_products))) - - # ----------------------------------------------------------------------------- - - if not permanency_inhibition == None: - - enc_reactants = Or(self.v[level][prod_entity] >= concentration, self.v_ctx[level][prod_entity] >= concentration) - - enc_inhibitors = True - for inhibitor,concentration in permanency_inhibition: - enc_inhibitors = simplify(And(enc_inhibitors, - And(self.v[level][inhibitor] < concentration, self.v_ctx[level][inhibitor] < concentration))) - enc_products = simplify(self.v[level+1][prod_entity] == \ - If(self.v[level][prod_entity] > self.v_ctx[level][prod_entity],self.v[level][prod_entity],self.v_ctx[level][prod_entity])) - - enc_permanency_enabledness = And(enc_reactants, enc_inhibitors, Not(enc_ordinary_reactions_enabledness)) - enc_enabledness = simplify(Or(enc_enabledness, enc_permanency_enabledness)) - enc_permanency = And(enc_permanency_enabledness, enc_products) - enc_rct_prod = simplify(Or(enc_rct_prod, enc_permanency)) - - # ----------------------------------------------------------------------------- - - enc_when_to_produce_zero_conc = simplify(And(Not(enc_enabledness), self.v[level+1][prod_entity] == 0)) - - enc_rct_prod = Or(enc_rct_prod, enc_when_to_produce_zero_conc) - 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 = set(range(len(self.rs.background_set))) - - enc_trans = True - - reactions = self.rs.get_reactions_by_product() - meta_reactions = self.rs.meta_reactions - - for prod_entity in chain(reactions, meta_reactions): - unused_entities.discard(prod_entity) - enc_trans = simplify(And(enc_trans, self.enc_produced_concentration(level, prod_entity))) - - for prod_entity in unused_entities: - enc_trans = simplify(And(enc_trans, self.v[level+1][prod_entity] == 0)) - - 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 = set([e for e,c in ctx]) - excl_ctx = all_ent - incl_ctx - - ctx_enc = True - - for e,c in ctx: - ctx_enc = simplify(And(ctx_enc, self.v_ctx[level][e] == c)) - - for e in excl_ctx: - ctx_enc = simplify(And(ctx_enc, self.v_ctx[level][e] == 0)) - - cur_trans = simplify(And(src_enc, ctx_enc, dst_enc)) - enc_trans = simplify(Or(enc_trans, cur_trans)) - - return enc_trans - - def enc_exact_state(self, level, state): - """Encodes the state at the given level with the exact concentration values""" - - raise RuntimeError("Should not be used with RSC") - - # enc = True - # used_entities_ids = self.rs.get_state_ids(state) - # - # for ent,conc in state: - # e_id = self.rs.get_entity_id(ent) - # enc = And(enc, self.v[level][e_id] == conc) - # - # not_in_state = set(range(len(self.rs.background_set))) - # not_in_state = not_in_state.difference(set(used_entities_ids)) - # - # for entity in not_in_state: - # enc = And(enc, self.v[level][entity] == 0) - # - # return simplify(enc) - - def enc_min_state(self, level, state): - """Encodes the state at the given level with the minimal required concentration levels""" - - enc = True - for ent,conc in state: - e_id = self.rs.get_entity_id(ent) - enc = And(enc, self.v[level][e_id] >= conc) - - # state_ids = self.rs.get_state_ids(state) - # - # for entity in state_ids: - # enc = And(enc, self.v[level][entity]) - - return simplify(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 - for ent,conc in required: - e_id = self.rs.get_entity_id(ent) - enc = And(enc, self.v[level][e_id] >= conc) - - for ent,conc in blocked: - e_id = self.rs.get_entity_id(ent) - enc = And(enc, self.v[level][e_id] < conc) - - 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{: >70}".format("[ level=" + repr(level) + " ]")) - - print(" State: {", end=""), - for var_id in range(len(self.v[level])): - var_rep = repr(m[self.v[level][var_id]]) - if not var_rep.isdigit(): - raise RuntimeError("unexpected: representation is not a positive integer") - if int(var_rep) > 0: - print(" " + self.rs.get_entity_name(var_id) + "=" + var_rep, end="") - # print(" " + repr(m[self.v[level][var_id]]), end="") - print(" }") - - if level != max_level: - print(" Context set: ", end="") - print("{", end="") - for var_id in range(len(self.v[level])): - var_rep = repr(m[self.v_ctx[level][var_id]]) - if not var_rep.isdigit(): - raise RuntimeError("unexpected: representation is not a positive integer") - if int(var_rep) > 0: - print(" " + self.rs.get_entity_name(var_id) + "=" + var_rep, end="") - print(" }") - - def check_reachability(self, state, print_witness=True, - print_time=True, print_mem=True, max_level=100): - """Main testing function""" - - 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 - - self.prepare_all_variables() - - self.solver.add(self.enc_concentration_levels_assertion(0)) - - while True: - self.prepare_all_variables() - self.solver.add(self.enc_concentration_levels_assertion(current_level+1)) - - print("\n{:-^70}".format("[ Working at level=" + str(current_level) + " ]")) - stdout.flush() - - # reachability test: - print("[" + colour_str(C_BOLD, "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("[" + colour_str(C_BOLD, "+") + "] " + colour_str(C_GREEN, "SAT at level=" + str(current_level))) - if print_witness: - print("\n{:=^70}".format("[ WITNESS ]")) - self.decode_witness(current_level) - break - else: - self.solver.pop() - - print("[" + colour_str(C_BOLD, "i") + "] Unrolling the transition relation") - self.solver.add(self.enc_transition_relation(current_level)) - - print("{:->70}".format("[ level=" + str(current_level) + " done ]")) - current_level += 1 - - if current_level > max_level: - print("Stopping at level=" + str(max_level)) - break - - if print_time: - # stop = time() - stop = resource.getrusage(resource.RUSAGE_SELF).ru_utime - self.verification_time = stop-start - print() - print("\n[i] {: >60}".format(" Time: " + repr(self.verification_time) + " s")) - - if print_mem: - print("[i] {: >60}".format(" Memory: " + repr(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss/(1024*1024)) + " MB")) - - def get_verification_time(self): - return self.verification_time - - def show_encoding(self, state, print_witness=True, - print_time=False, print_mem=False, max_level=100): - """Encoding debug function""" - - self.prepare_all_variables() - init_s = self.enc_init_state(0) - print(init_s) - self.solver.add(init_s) - current_level = 0 - - self.prepare_all_variables() - - while True: - self.prepare_all_variables() - - print("-----[ Working at level=" + str(current_level) + " ]-----") - stdout.flush() - - # reachability test: - print("[i] Adding the reachability test...") - self.solver.push() - - s = self.enc_min_state(current_level,state) - print("Test: ", s) - - self.solver.add(s) - - result = self.solver.check() - if result == sat: - print("\n[+] " + colour_str(C_RED, "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") - t = self.enc_transition_relation(current_level) - print(t) - self.solver.add(t) - - print("-----[ level=" + str(current_level) + " done ]") - current_level += 1 - - if current_level > max_level: - print("Stopping at level=" + str(max_level)) - break - else: - x=input("Next level? ") - x=x.lower() - if not (x == "y" or x == "yes"): - break -