diff --git a/rs/distributed_reaction_system.py b/rs/distributed_reaction_system.py deleted file mode 100644 index 583f2cc..0000000 --- a/rs/distributed_reaction_system.py +++ /dev/null @@ -1,305 +0,0 @@ -#!/usr/bin/env python - -""" ---- Distributed Reaction Systems Manipulation -""" - -from sys import exit - -class DistributedReactionSystem(object): - - def __init__(self): - - self.__reactions = [] - self.__background_set = [] - self.__reactions_by_prod = [] - self.__states = [] - self.__transitions = [] - self.__init_state = None - - self.__context_sets_size = None - - @property - def background_set(self): - return self.__background_set - - @property - def set_of_background_ids(self): - return set(range(len(self.__background_set))) - - @property - def components_count(self): - return len(self.__reactions) - - def add_bg_set_entity(self, name): - if not self.is_in_background_set(name): - self.__background_set.append(name) - else: - print("The entity \"" + str(name) + "\" is already on the list") - exit(1) - - def add_bg_set_entities(self, names): - for name in names: - self.add_bg_set_entity(name) - - def is_in_background_set(self, entity): - """Checks if the given name is valid wrt the background set=""" - if entity in self.__background_set: - return True - else: - return False - - def get_entity_id(self, name): - try: - return self.__background_set.index(name) - except ValueError: - print("Undefined background set entity: " + repr(name)) - exit(1) - - def get_state_ids(self, state): - ids = [] - for entity in state: - ids.append(self.get_entity_id(entity)) - - return ids - - def get_entity_name(self, entity_id): - """Returns the string corresponding to the entity""" - return self.__background_set[entity_id] - - def ensure_reactions(self, k): - while len(self.__reactions) <= k: - self.__reactions.append([]) - - def add_reaction(self, k, R, I, P): - """Adds a reaction""" - - if R == [] or P == []: - print("No reactants of products defined") - raise - - reactants = [] - for entity in R: - reactants.append(self.get_entity_id(entity)) - - inhibitors = [] - for entity in I: - inhibitors.append(self.get_entity_id(entity)) - - products = [] - for entity in P: - products.append(self.get_entity_id(entity)) - - self.ensure_reactions(k) - self.__reactions[k].append((reactants, inhibitors, products)) - - def entities_names_set_to_str(self, entities): - s = "" - for entity in entities: - s += entity + ", " - s = s[:-2] - return s - - def entities_ids_set_to_str(self, entities): - s = "" - for entity in entities: - s += self.get_entity_name(entity) + ", " - s = s[:-2] - return s - - def show_reactions(self): - print("[*] Reactions:") - for i in range(len(self.__reactions)): - local_reactions = self.__reactions[i] - print(" agent = " + str(i)) - for rcts,inhib,prods in local_reactions: - print("\t - ( R={" + self.entities_ids_set_to_str(rcts) + "}, \tI={" + self.entities_ids_set_to_str(inhib) + "}, \tP={" + self.entities_ids_set_to_str(prods) + "} )") - - def show_background_set(self): - print("[*] Background set: {" + self.entities_names_set_to_str(self.__background_set) + "}") - - def show(self): - self.show_background_set() - self.show_reactions() - self.show_states() - self.show_transitions() - print() - - def reactions_by_product_are_cached(self, component_id): - if len(self.__reactions_by_prod) > component_id: - if self.__reactions_by_prod[component_id] != None: - return True - return False - - def get_reactions_by_product(self, component_id): - """Sorts reactions for the component given by component_id by their products and returns a dictionary of products""" - - if self.reactions_by_product_are_cached(component_id): - return self.__reactions_by_prod[component_id] - - producible_entities = set() - - for reactants,inhibitors,products in self.__reactions[component_id]: - producible_entities = producible_entities.union(set(products)) - - reactions_by_prod = dict() - - for prod_entity in producible_entities: - reactions_by_prod[prod_entity] = [] - for reactants,inhibitors,products in self.__reactions[component_id]: - if prod_entity in products: - reactions_by_prod[prod_entity].append([reactants,inhibitors]) - - # save in cache - while len(self.__reactions_by_prod) <= component_id: # ensure that the size of cache is right - self.__reactions_by_prod.append(None) - self.__reactions_by_prod[component_id] = reactions_by_prod - - return reactions_by_prod - - @property - def states(self): - return self.__states - - @property - def transitions(self): - return self.__transitions - - def add_state(self, name): - if name not in self.__states: - self.__states.append(name) - else: - print("\'%s\' already added. skipping..." % (name,)) - - def add_init_state(self, name): - self.add_state(name) - self.__init_state = self.__states.index(name) - - def get_init_state_name(self): - if self.__init_state == None: - return None - return self.__states[self.__init_state] - - def is_state(self, name): - if name in self.__states: - return True - else: - return False - - def get_state_id(self, name): - try: - return self.__states.index(name) - except ValueError: - print("Undefined context automaton state: " + repr(name)) - exit(1) - - def get_init_state_id(self): - return self.__init_state - - def print_states(self): - for state in self.__states: - print(state) - - def is_valid_context_sets(self, context_sets): - for c in context_sets: - if not self.is_valid_context(c): - return False - return True - - def is_valid_context(self, context): - if set(context).issubset(self.__background_set): - return True - else: - return False - - def add_transition(self, src, label, dst): - recipients = label[0] - context_sets = label[1] - if not type(context_sets) is list: - print("Context sets must be of type list") - exit(1) - for s in context_sets: - if not type(s) is set and not type(s) is list: - print("Each context set must be of type set or list") - exit(1) - - if self.__context_sets_size == None: - self.__context_sets_size = len(context_sets) - else: - if len(context_sets) != self.__context_sets_size: - print("Inconsistent size of the context sets: " + str(len(context_sets)) + " != " + str(self.__context_sets_size)) - exit(1) - - if not self.is_valid_context_sets(context_sets): - raise RuntimeError("one of the entities in the context set is unknown (undefined)!") - - if not self.is_state(src): - raise RuntimeError("\"" + src + "\" is an unknown (undefined) state") - - if not self.is_state(dst): - raise RuntimeError("\"" + dst + "\" is an unknown (undefined) state") - - new_context_sets = [] - for c in context_sets: - new_c_set = set() - for e in set(c): - new_c_set.add(self.get_entity_id(e)) - new_context_sets.append(new_c_set) - - recipients = set(recipients) - - self.__transitions.append((self.get_state_id(src),(recipients,new_context_sets),self.get_state_id(dst))) - - def context2str(self, context_sets): - """Converts the set of entities ids into the string with their names""" - s = "[" - for ctx in context_sets: - if len(ctx) == 0: - s += " 0" - else: - s += " {" - for c in ctx: - s += " " + self.get_entity_name(c) - s += " }" - s += " ]" - return s - - def show_transitions(self): - print("[*] Context automaton transitions:") - for transition in self.__transitions: - str_transition = str(transition[0]) + " --( " - str_transition += str(list(transition[1][0])) + "<=" + self.context2str(transition[1][1]) - str_transition += " )--> " + str(transition[2]) - print("\t- " + str_transition) - - def show_states(self): - init_state_name = self.get_init_state_name() - print("[*] Context automaton states:") - for state in self.__states: - print("\t- " + state, end="") - if state == init_state_name: - print(" [init]") - else: - print() - - def sanity_check(self): - """Performs a sanity check of the defined distributed reaction system""" - - print("[i] Performing sanity check of the DRS") - - if self.__reactions == []: - print("No reactions defined") - exit(1) - - if self.__background_set == []: - print("Empty background set") - exit(1) - - if self.__init_state == None: - print("Initial state not specified") - exit(1) - - if self.__context_sets_size != len(self.__reactions): - print("Inconsistent sizes of the context sets with respect to the number of components/agents/processes: " + str(self.__context_sets_size) + " != " + str(len(self.__reactions))) - exit(1) - diff --git a/rssmt.py b/rssmt.py index 287bf9e..c60ffa1 100755 --- a/rssmt.py +++ b/rssmt.py @@ -23,7 +23,7 @@ if profiling: ################################################################## -version = "2.5" +version = "2.99" rsmc_banner = """ Reaction Systems SMT-Based Model Checking diff --git a/smt/__init__.py b/smt/__init__.py index 8958447..d626ea0 100644 --- a/smt/__init__.py +++ b/smt/__init__.py @@ -2,5 +2,3 @@ from smt.smt_checker_rs import SmtCheckerRS from smt.smt_checker_rsc import SmtCheckerRSC from smt.smt_checker_rsc_param import SmtCheckerRSCParam -from smt.smt_checker_rs_na import SmtCheckerRSNA -from smt.smt_checker_distrib_rs import SmtCheckerDistribRS diff --git a/smt/legacy/smt_checker.py b/smt/legacy/smt_checker.py deleted file mode 100644 index 9939f2f..0000000 --- a/smt/legacy/smt_checker.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/smt/smt_checker_distrib_rs.py b/smt/smt_checker_distrib_rs.py deleted file mode 100644 index 53e84be..0000000 --- a/smt/smt_checker_distrib_rs.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/smt/smt_checker_rs_na.py b/smt/smt_checker_rs_na.py deleted file mode 100644 index 428a46f..0000000 --- a/smt/smt_checker_rs_na.py +++ /dev/null @@ -1,174 +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 - -from smt.smt_checker_rs import SmtCheckerRS - -class SmtCheckerRSNA(SmtCheckerRS): - """SMT-based Model Checking for Reaction Systems with Network of Automata""" - - def __init__(self, reaction_system_with_netaut): - - self.rs = reaction_system_with_netaut.rs - self.canet = reaction_system_with_netaut.cas - - self.number_of_automata = self.canet.number_of_automata - - self.v = [] - self.v_ctx = [] - self.v_canet_states = [] - self.v_canet_actions = [] - self.next_level_to_encode = 0 - - self.solver = Solver() - - self.verification_time = None - - def prepare_context_controller_variables(self): - """Encodes all the state variables""" - - level = self.next_level_to_encode - - ca_states = [] - for ca_id in range(self.number_of_automata): - ca_states.append(Int("CA"+str(level)+"_a"+str(ca_id)+"_state")) - self.v_canet_states.append(ca_states) - - # We do not encode actions when there are no transitions needed. - if level > 0: - ca_actions = [] - for ca in self.canet.automata: - for act_id in range(ca.number_of_actions): - ca_actions.append(Bool("CA"+str(level-1)+"_a"+str(self.canet.automata.index(ca))+"_act"+str(act_id))) - self.v_canet_actions.append(ca_actions) - - def enc_context_controller_init_state(self, level): - """Encodes the initial state for the network of automata""" - - canet_init_state_enc = True - for ca_idx in range(self.number_of_automata): - ca_init_state_enc = self.v_canet_states[level][ca_idx] == self.canet.automata[ca_idx].get_init_state_id() - canet_init_state_enc = simplify(And(canet_init_state_enc, ca_init_state_enc)) - - return canet_init_state_enc - - def enc_transition_relation(self, level): - return simplify(And(self.enc_rs_trans(level), self.enc_automaton_trans(level))) - - def enc_reactants_ids(self, level, set_of_ids): - """Encodes reactants given by their ids in RS""" - - enc_ents = True - for r_id in set_of_ids: - enc_ents = simplify(And(enc_ents, self.v[level][r_id])) - return enc_ents - - def enc_inhibitors_ids(self, level, set_of_ids): - """Encodes inhibitors given by their ids in RS""" - - enc_ents = True - for i_id in set_of_ids: - enc_ents = simplify(And(enc_ents, Not(self.v[level][i_id]))) - return enc_ents - - def enc_context_single_transition(self, level, automaton, transition): - """""" - - src_id, act_ids, (r_ids, i_ids, p_ids), dst_id = transition - - enc_transition = self.v_canet_states[level][automaton] == src_id - enc_transition = simplify(And(enc_transition, self.enc_reactants_ids(level, r_ids), self.enc_inhibitors_ids(level, i_ids))) - enc_transition = simplify(And(enc_transition, self.v_canet_states[level+1][automaton] == dst_id)) - # ACTIONS NOT ENCODED! - - print(enc_transition) - return enc_transition - - def enc_context_entity_production(self, level, entity): - """Encodes the automata transitions and the production for a given entity""" - - enc_production = False - - actions_producing_ent = self.canet.get_actions_producing_entity(entity) - enc_prod_action = False - - for act in actions_producing_ent: - - # automata_with_act = self.canet.get_automata_with_action(act) - enc_aut_with_act = True - - for aut_id in self.canet.automata_ids: - - aut = self.canet.automata[aut_id] - - t_producing_entity = aut.get_transitions_producing_entity(entity) - - enc_t_producing_entity = False - - if t_producing_entity: - # for the automata that produce the entity we take - # all the transitions that produce it: - - - for t in t_producing_entity: - enc_t_producing_entity = simplify(Or(enc_t_producing_entity, self.enc_context_single_transition(level, aut_id, t))) - - else: - # for all the automata that do not produce the entity - # we encode the transitions that synchronise with the action: - - pass - - enc_aut_with_act = simplify(And(enc_aut_with_act, enc_t_producing_entity)) - - enc_prod_action = simplify(Or(enc_prod_action, enc_aut_with_act)) - - # TODO - # for aut in remaining_automata_which_do_not_contain_act: - # encode no change - - enc_production = enc_prod_action - - return enc_production - - def enc_automaton_trans(self, level): - """Encodes the transition relation for the context automaton""" - - enc_trans = True - - prod_context = self.canet.prod_entities - never_produced_context = self.rs.set_of_bgset_ids - prod_context - - # (1) producible entities: - for ent in prod_context: - enc_trans = simplify(And(enc_trans, self.enc_context_entity_production(level, ent))) - - - # (2) entities that are never produced: - - # (3) TODO: no entity, empty set of entities: transitions with no entities - # simplify((enc_trans, self.enc_context_no_entity_production(level))) - # TRANSITIONS PRODUCING EMPTY SETS - - print(enc_trans) - - return enc_trans - - def check_reachability(self, state, print_witness=True, print_time=True, print_mem=True): - """The main method for checking reachability""" - - self.prepare_all_variables() - self.solver.add(self.enc_init_state(0)) - current_level = 0 - self.prepare_all_variables() - self.prepare_all_variables() - self.prepare_all_variables() - self.prepare_all_variables() - - print(self.enc_automaton_trans(0)) - diff --git a/smt/smt_checker_rsc.py b/smt/smt_checker_rsc.py index 678a680..24fd729 100644 --- a/smt/smt_checker_rsc.py +++ b/smt/smt_checker_rsc.py @@ -153,23 +153,6 @@ class SmtCheckerRSC(object): 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: @@ -234,17 +217,6 @@ class SmtCheckerRSC(object): 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))) @@ -299,21 +271,6 @@ class SmtCheckerRSC(object): 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""" @@ -322,11 +279,6 @@ class SmtCheckerRSC(object): 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): diff --git a/smt/smt_checker_rsc_param.py b/smt/smt_checker_rsc_param.py index 37f52e4..212f953 100644 --- a/smt/smt_checker_rsc_param.py +++ b/smt/smt_checker_rsc_param.py @@ -62,14 +62,13 @@ class SmtCheckerRSCParam(object): self.next_level_to_encode = 0 - # this is probably not needed anymore: - # self.producible_entities = self.rs.get_producible_entities() - # self.improducible_entities = set(self.rs.get_state_ids(self.rs.background_set)) - self.producible_entities - + # # WARNING: improd vs. improducible - # there is some confusion related to the variables naming: - # improd - intermediate products - # improducible - entities that are never produces (there is no reaction that produces that entity) + # there is some confusion related to the variable naming: + # + # * improd - intermediate products + # * improducible - entities that are never produces (there is no reaction that produces that entity) + # # TODO: number of loops == number of paths self.loop_position = None @@ -162,9 +161,6 @@ class SmtCheckerRSCParam(object): self.path_v_improd.setdefault(path_idx, []) self.path_v_improd_for_entities.setdefault(path_idx, []) - # assert len(self.path_v_improd[path_idx]) == level - # assert len(self.path_v_improd_for_entities[path_idx]) == level - if level < 1: # # If we are at level==0, we add a dummy "level" @@ -580,11 +576,6 @@ class SmtCheckerRSCParam(object): 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): @@ -629,7 +620,6 @@ class SmtCheckerRSCParam(object): 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: @@ -748,16 +738,10 @@ class SmtCheckerRSCParam(object): if not isinstance(formulae_list, (list, tuple)): print_error("Expected a list of formulae") - #print_info("Parameter constraint: {:s}".format(str(param_constr))) - #print_info("Parameter constraint defined") - - self.reset() num_of_paths = len(formulae_list) - # formula = formulae[0] - print_info("Running rsLTL bounded model checking") print_info("Tested formulae:") for form in formulae_list: @@ -780,8 +764,6 @@ class SmtCheckerRSCParam(object): self.current_level = 0 - # self.prepare_all_variables(num_of_paths) - # assertions for all the paths and parameters self.solver_add(self.enc_concentration_levels_assertions_for_paths(0, num_of_paths)) self.solver_add(self.enc_param_concentration_levels_assertion()) @@ -803,7 +785,6 @@ class SmtCheckerRSCParam(object): print( "\n{:-^70}".format("[ Working at level=" + str(self.current_level) + " ]")) - # stdout.flush() # reachability test: self.solver.push() @@ -817,9 +798,6 @@ class SmtCheckerRSCParam(object): print_info("Adding the encoding for the loops...") self.solver_add(self.get_loop_encodings()) - # if self.optimise: - # self.assert_param_optimisation() - print_info("Testing satisfiability...") result = self.solver.check() if result == sat: