From daadf33d02da2bde6a3c2d162f1dcbd701d659f9 Mon Sep 17 00:00:00 2001 From: Artur Meski Date: Wed, 28 Dec 2016 15:55:03 +0100 Subject: [PATCH] smt package --- smt/__init__.py | 4 + smt/smt_checker.py | 530 ++++++++++++++++++++++++++++++++++ smt/smt_checker_distrib_rs.py | 432 +++++++++++++++++++++++++++ smt/smt_checker_pgrs.py | 275 ++++++++++++++++++ smt/smt_checker_rsc.py | 480 ++++++++++++++++++++++++++++++ 5 files changed, 1721 insertions(+) create mode 100644 smt/__init__.py create mode 100644 smt/smt_checker.py create mode 100644 smt/smt_checker_distrib_rs.py create mode 100644 smt/smt_checker_pgrs.py create mode 100644 smt/smt_checker_rsc.py diff --git a/smt/__init__.py b/smt/__init__.py new file mode 100644 index 0000000..7cf18cb --- /dev/null +++ b/smt/__init__.py @@ -0,0 +1,4 @@ +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_rsc import SmtCheckerRSC diff --git a/smt/smt_checker.py b/smt/smt_checker.py new file mode 100644 index 0000000..9939f2f --- /dev/null +++ b/smt/smt_checker.py @@ -0,0 +1,530 @@ +""" +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/smt/smt_checker_distrib_rs.py b/smt/smt_checker_distrib_rs.py new file mode 100644 index 0000000..53e84be --- /dev/null +++ b/smt/smt_checker_distrib_rs.py @@ -0,0 +1,432 @@ +""" +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/smt/smt_checker_pgrs.py b/smt/smt_checker_pgrs.py new file mode 100644 index 0000000..b642ea3 --- /dev/null +++ b/smt/smt_checker_pgrs.py @@ -0,0 +1,275 @@ +""" +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/smt/smt_checker_rsc.py b/smt/smt_checker_rsc.py new file mode 100644 index 0000000..c9fae4c --- /dev/null +++ b/smt/smt_checker_rsc.py @@ -0,0 +1,480 @@ +""" +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 +