artur/reactics
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Clean-up

Browse Source
This commit is contained in:
Artur Meski
2019-02-23 12:59:20 +00:00
parent 28821c78de
commit 493e4c5fc6
8 changed files with 7 additions and 1520 deletions
Download Patch File Download Diff File Expand all files Collapse all files

305
rs/distributed_reaction_system.py
View File

@@ -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)

2
rssmt.py
View File

@@ -23,7 +23,7 @@ if profiling:
################################################################## ##################################################################
version = "2.5" version = "2.99"
rsmc_banner = """ rsmc_banner = """
Reaction Systems SMT-Based Model Checking Reaction Systems SMT-Based Model Checking

2
smt/__init__.py
View File

@@ -2,5 +2,3 @@
from smt.smt_checker_rs import SmtCheckerRS from smt.smt_checker_rs import SmtCheckerRS
from smt.smt_checker_rsc import SmtCheckerRSC from smt.smt_checker_rsc import SmtCheckerRSC
from smt.smt_checker_rsc_param import SmtCheckerRSCParam from smt.smt_checker_rsc_param import SmtCheckerRSCParam
from smt.smt_checker_rs_na import SmtCheckerRSNA
from smt.smt_checker_distrib_rs import SmtCheckerDistribRS

530
smt/legacy/smt_checker.py
View File

@@ -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))

432
smt/smt_checker_distrib_rs.py
View File

@@ -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)))

174
smt/smt_checker_rs_na.py
View File

@@ -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))

48
smt/smt_checker_rsc.py
View File

@@ -153,23 +153,6 @@ class SmtCheckerRSC(object):
enc_ordinary_reactions_enabledness = simplify(Or(enc_ordinary_reactions_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 --------------------------------------------------- # -------- meta reactions ---------------------------------------------------
for r_type,command_entity,reactants,inhibitors in 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) enc_rct_prod = Or(enc_rct_prod, enc_when_to_produce_zero_conc)
return enc_rct_prod 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): def enc_transition_relation(self, level):
return simplify(And(self.enc_rs_trans(level), self.enc_automaton_trans(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") 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): def enc_min_state(self, level, state):
"""Encodes the state at the given level with the minimal required concentration levels""" """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) e_id = self.rs.get_entity_id(ent)
enc = And(enc, self.v[level][e_id] >= conc) 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) return simplify(enc)
def enc_state_with_blocking(self, level, prop): def enc_state_with_blocking(self, level, prop):

34
smt/smt_checker_rsc_param.py
View File

@@ -62,14 +62,13 @@ class SmtCheckerRSCParam(object):
self.next_level_to_encode = 0 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 # WARNING: improd vs. improducible
# there is some confusion related to the variables naming: # 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) # * improd - intermediate products
# * improducible - entities that are never produces (there is no reaction that produces that entity)
#
# TODO: number of loops == number of paths # TODO: number of loops == number of paths
self.loop_position = None self.loop_position = None
@@ -162,9 +161,6 @@ class SmtCheckerRSCParam(object):
self.path_v_improd.setdefault(path_idx, []) self.path_v_improd.setdefault(path_idx, [])
self.path_v_improd_for_entities.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 level < 1:
# #
# If we are at level==0, we add a dummy "level" # 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) e_id = self.rs.get_entity_id(ent)
enc = And(enc, self.v[level][e_id] >= conc) 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) return simplify(enc)
def enc_state_with_blocking(self, level, prop): def enc_state_with_blocking(self, level, prop):
@@ -629,7 +620,6 @@ class SmtCheckerRSCParam(object):
print( print(
" " + self.rs.get_entity_name(var_id) + "=" + var_rep, " " + self.rs.get_entity_name(var_id) + "=" + var_rep,
end="") end="")
# print(" " + repr(m[self.v[level][var_id]]), end="")
print(" }") print(" }")
if level != max_level: if level != max_level:
@@ -748,16 +738,10 @@ class SmtCheckerRSCParam(object):
if not isinstance(formulae_list, (list, tuple)): if not isinstance(formulae_list, (list, tuple)):
print_error("Expected a list of formulae") print_error("Expected a list of formulae")
#print_info("Parameter constraint: {:s}".format(str(param_constr)))
#print_info("Parameter constraint defined")
self.reset() self.reset()
num_of_paths = len(formulae_list) num_of_paths = len(formulae_list)
# formula = formulae[0]
print_info("Running rsLTL bounded model checking") print_info("Running rsLTL bounded model checking")
print_info("Tested formulae:") print_info("Tested formulae:")
for form in formulae_list: for form in formulae_list:
@@ -780,8 +764,6 @@ class SmtCheckerRSCParam(object):
self.current_level = 0 self.current_level = 0
# self.prepare_all_variables(num_of_paths)
# assertions for all the paths and parameters # 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_concentration_levels_assertions_for_paths(0, num_of_paths))
self.solver_add(self.enc_param_concentration_levels_assertion()) self.solver_add(self.enc_param_concentration_levels_assertion())
@@ -803,7 +785,6 @@ class SmtCheckerRSCParam(object):
print( print(
"\n{:-^70}".format("[ Working at level=" + str(self.current_level) + " ]")) "\n{:-^70}".format("[ Working at level=" + str(self.current_level) + " ]"))
# stdout.flush()
# reachability test: # reachability test:
self.solver.push() self.solver.push()
@@ -817,9 +798,6 @@ class SmtCheckerRSCParam(object):
print_info("Adding the encoding for the loops...") print_info("Adding the encoding for the loops...")
self.solver_add(self.get_loop_encodings()) self.solver_add(self.get_loop_encodings())
# if self.optimise:
# self.assert_param_optimisation()
print_info("Testing satisfiability...") print_info("Testing satisfiability...")
result = self.solver.check() result = self.solver.check()
if result == sat: if result == sat: