diff --git a/colour.py b/colour.py index 80b7c65..4000cb0 100644 --- a/colour.py +++ b/colour.py @@ -10,18 +10,23 @@ C_UNDERLINE = '\033[4m' C_MARK_INFO = C_BOLD + "[" + C_GREEN + "*" + C_ENDC + C_BOLD + "]" + C_ENDC C_MARK_ERROR = C_BOLD + "[" + C_RED + "!" + C_ENDC + C_BOLD + "]" + C_ENDC + def colour_str(col, s): return col + s + C_ENDC - + + def green_str(s): return C_GREEN + s + C_ENDC - + + def print_error(s): print(C_MARK_ERROR + " " + C_RED + s + C_ENDC) + def print_info(s): print("[" + colour_str(C_BOLD, "i") + "] {:s}".format(s)) + def print_positive(s): print("[" + colour_str(C_BOLD, "+") + "] {:s}".format(s)) diff --git a/logics/__init__.py b/logics/__init__.py index 3d0d860..a8089f9 100644 --- a/logics/__init__.py +++ b/logics/__init__.py @@ -2,4 +2,4 @@ from logics.rsltl import Formula_rsLTL from logics.bags import BagDescription from logics.param_constr import ParamConstraint from logics.rsltl_encoder import rsLTL_Encoder -from logics.param_constr_encoder import ParamConstr_Encoder \ No newline at end of file +from logics.param_constr_encoder import ParamConstr_Encoder diff --git a/logics/param_constr.py b/logics/param_constr.py index eb8fc50..a9972b9 100644 --- a/logics/param_constr.py +++ b/logics/param_constr.py @@ -1,6 +1,7 @@ from enum import Enum ParamConstraint_oper = Enum( - 'ParamConstraint_oper', 'param_entity true l_and l_or l_not lt le eq ge gt') + 'ParamConstraint_oper', + 'param_entity true l_and l_or l_not lt le eq ge gt') class ParamConstraint(object): @@ -49,7 +50,8 @@ class ParamConstraint(object): @classmethod def f_param_ent(cls, param, entity_name): - return cls(ParamConstraint_oper.param_entity, param=param, entity=entity_name) + return cls(ParamConstraint_oper.param_entity, param=param, + entity=entity_name) @classmethod def f_TRUE(cls): @@ -64,25 +66,32 @@ class ParamConstraint(object): return cls(ParamConstraint_oper.l_not, L_oper=arg) def __lt__(self, other): - return ParamConstraint(ParamConstraint_oper.lt, L_oper=self, R_oper=other) + return ParamConstraint( + ParamConstraint_oper.lt, L_oper=self, R_oper=other) def __le__(self, other): - return ParamConstraint(ParamConstraint_oper.le, L_oper=self, R_oper=other) + return ParamConstraint( + ParamConstraint_oper.le, L_oper=self, R_oper=other) def __eq__(self, other): - return ParamConstraint(ParamConstraint_oper.eq, L_oper=self, R_oper=other) + return ParamConstraint( + ParamConstraint_oper.eq, L_oper=self, R_oper=other) def __ge__(self, other): - return ParamConstraint(ParamConstraint_oper.ge, L_oper=self, R_oper=other) + return ParamConstraint( + ParamConstraint_oper.ge, L_oper=self, R_oper=other) def __gt__(self, other): - return ParamConstraint(ParamConstraint_oper.gt, L_oper=self, R_oper=other) + return ParamConstraint( + ParamConstraint_oper.gt, L_oper=self, R_oper=other) def __and__(self, other): - return ParamConstraint(ParamConstraint_oper.l_and, L_oper=self, R_oper=other) + return ParamConstraint( + ParamConstraint_oper.l_and, L_oper=self, R_oper=other) def __or__(self, other): - return ParamConstraint(ParamConstraint_oper.l_or, L_oper=self, R_oper=other) + return ParamConstraint( + ParamConstraint_oper.l_or, L_oper=self, R_oper=other) def __invert__(self): return ParamConstraint(ParamConstraint_oper.l_not, L_oper=self) diff --git a/logics/param_constr_encoder.py b/logics/param_constr_encoder.py index 6fb6032..d74b9b7 100644 --- a/logics/param_constr_encoder.py +++ b/logics/param_constr_encoder.py @@ -1,9 +1,10 @@ from logics.param_constr import * from z3 import And, Not, Or + class ParamConstr_Encoder(object): """Class for encoding parameter constraints""" - + def __init__(self, smt_checker): self.smt_checker = smt_checker self.rs = smt_checker.rs @@ -11,49 +12,59 @@ class ParamConstr_Encoder(object): # self.v = None # self.v_ctx = None # self.loop_position = None - + def load_variables(self, var_rs, var_ctx, var_loop_pos): - + self.v = var_rs self.v_ctx = var_ctx self.loop_position = var_loop_pos - + def encode(self, param_constr): - + if not param_constr: raise RuntimeError("param_constr is None") if param_constr.f_type == ParamConstraint_oper.param_entity: - return self.smt_checker.get_enc_param(param_constr.param.name, param_constr.entity) - + return self.smt_checker.get_enc_param( + param_constr.param.name, param_constr.entity) + if param_constr.f_type == ParamConstraint_oper.true: return True - + if param_constr.f_type == ParamConstraint_oper.l_and: - return And(self.encode(param_constr.left_operand), - self.encode(param_constr.right_operand)) - + return And(self.encode(param_constr.left_operand), + self.encode(param_constr.right_operand)) + if param_constr.f_type == ParamConstraint_oper.l_or: - return Or(self.encode(param_constr.left_operand), - self.encode(param_constr.right_operand)) - + return Or(self.encode(param_constr.left_operand), + self.encode(param_constr.right_operand)) + if param_constr.f_type == ParamConstraint_oper.l_not: return Not(self.encode(param_constr.left_operand)) - + if param_constr.f_type == ParamConstraint_oper.lt: - return self.encode(param_constr.left_operand) < int(param_constr.right_operand) - + return self.encode( + param_constr.left_operand) < int( + param_constr.right_operand) + if param_constr.f_type == ParamConstraint_oper.le: - return self.encode(param_constr.left_operand) <= int(param_constr.right_operand) - + return self.encode( + param_constr.left_operand) <= int( + param_constr.right_operand) + if param_constr.f_type == ParamConstraint_oper.eq: - return self.encode(param_constr.left_operand) == int(param_constr.right_operand) - + return self.encode( + param_constr.left_operand) == int( + param_constr.right_operand) + if param_constr.f_type == ParamConstraint_oper.ge: - return self.encode(param_constr.left_operand) >= int(param_constr.right_operand) - + return self.encode( + param_constr.left_operand) >= int( + param_constr.right_operand) + if param_constr.f_type == ParamConstraint_oper.gt: - return self.encode(param_constr.left_operand) > int(param_constr.right_operand) - + return self.encode( + param_constr.left_operand) > int( + param_constr.right_operand) + assert False, "Unsupported case {:s}".format(param_constr.f_type) - diff --git a/logics/rsltl.py b/logics/rsltl.py index 91fdcc1..d1735ba 100644 --- a/logics/rsltl.py +++ b/logics/rsltl.py @@ -27,7 +27,7 @@ class Formula_rsLTL(object): self.left_operand = Formula_rsLTL.f_bag(self.left_operand) if isinstance(self.right_operand, BagDescription): self.right_operand = Formula_rsLTL.f_bag(self.right_operand) - + if self.sub_operand is True: self.sub_operand = BagDescription.f_TRUE() diff --git a/logics/rsltl_encoder.py b/logics/rsltl_encoder.py index 9211e39..f9e60b8 100644 --- a/logics/rsltl_encoder.py +++ b/logics/rsltl_encoder.py @@ -1,11 +1,13 @@ from logics.rsltl import * + def simplify(x): - return x + return x + class rsLTL_Encoder(object): """Class for encoding rsLTL formulae for a given smt_checker instance""" - + def __init__(self, smt_checker): self.smt_checker = smt_checker self.rs = smt_checker.rs @@ -13,20 +15,20 @@ class rsLTL_Encoder(object): self.v = None self.v_ctx = None self.loop_position = None - + # self.load_variables( # var_rs=smt_checker.v, # var_ctx=smt_checker.v_ctx, # var_loop_pos=smt_checker.loop_position) - + self.init_ncalls() - + def load_variables(self, var_rs, var_ctx, var_loop_pos): - + self.v = var_rs self.v_ctx = var_ctx self.loop_position = var_loop_pos - + def get_encoding(self, formula, bound): assert self.v is not None @@ -35,27 +37,27 @@ class rsLTL_Encoder(object): self.cache_init(bound) self.init_ncalls() - + return self.encode(formula, 0, bound) - + def init_ncalls(self): self.ncalls_encode = 0 self.ncalls_encode_approx = 0 - + def cache_init(self, bound): """ Cache for formulae encodings """ self.cache_hits = 0 - self.enc_fcache = [{} for level in range(0,bound+1)] - self.enc_fcache_approx = [{} for level in range(0,bound+1)] - + self.enc_fcache = [{} for level in range(0, bound+1)] + self.enc_fcache_approx = [{} for level in range(0, bound+1)] + def cache_save(self, formula, level, formula_encoding): self.enc_fcache[level][formula] = formula_encoding - + def cache_approx_save(self, formula, level, formula_encoding): self.enc_fcache_approx[level][formula] = formula_encoding - + def cache_query(self, formula, level): if formula in self.enc_fcache[level]: self.cache_hits += 1 @@ -72,16 +74,16 @@ class rsLTL_Encoder(object): def get_cache_hits(self): return self.cache_hits - + def flush_cache(self): self.enc_fcache = None self.enc_fcache_approx = None - + def get_ncalls(self): return (self.ncalls_encode, self.ncalls_encode_approx) - + def encode_bag(self, bag_formula, level, context=False): - + if not bag_formula: raise RuntimeError("bag_formula is None") @@ -91,64 +93,79 @@ class rsLTL_Encoder(object): return self.v_ctx[level][entity_id] else: return self.v[level][entity_id] - + if bag_formula.f_type == BagDesc_oper.true: return True - + if bag_formula.f_type == BagDesc_oper.l_and: - return And(self.encode_bag(bag_formula.left_operand, level, context), + return And( + self.encode_bag(bag_formula.left_operand, level, context), self.encode_bag(bag_formula.right_operand, level, context)) - + if bag_formula.f_type == BagDesc_oper.l_or: - return Or(self.encode_bag(bag_formula.left_operand, level, context), + return Or( + self.encode_bag(bag_formula.left_operand, level, context), self.encode_bag(bag_formula.right_operand, level, context)) - + if bag_formula.f_type == BagDesc_oper.l_not: - return Not(self.encode_bag(bag_formula.left_operand, level, context)) - + return Not(self.encode_bag( + bag_formula.left_operand, level, context)) + if bag_formula.f_type == BagDesc_oper.lt: - return self.encode_bag(bag_formula.left_operand, level, context) < int(bag_formula.right_operand) - + return self.encode_bag( + bag_formula.left_operand, level, context) < int( + bag_formula.right_operand) + if bag_formula.f_type == BagDesc_oper.le: - return self.encode_bag(bag_formula.left_operand, level, context) <= int(bag_formula.right_operand) - + return self.encode_bag( + bag_formula.left_operand, level, context) <= int( + bag_formula.right_operand) + if bag_formula.f_type == BagDesc_oper.eq: - return self.encode_bag(bag_formula.left_operand, level, context) == int(bag_formula.right_operand) - + return self.encode_bag( + bag_formula.left_operand, level, context) == int( + bag_formula.right_operand) + if bag_formula.f_type == BagDesc_oper.ge: - return self.encode_bag(bag_formula.left_operand, level, context) >= int(bag_formula.right_operand) - + return self.encode_bag( + bag_formula.left_operand, level, context) >= int( + bag_formula.right_operand) + if bag_formula.f_type == BagDesc_oper.gt: - return self.encode_bag(bag_formula.left_operand, level, context) > int(bag_formula.right_operand) - + return self.encode_bag( + bag_formula.left_operand, level, context) > int( + bag_formula.right_operand) + assert False, "Unsupported case {:s}".format(bag_formula.f_type) - + def encode_bag_state(self, bag_formula, level): res = self.encode_bag(bag_formula, level) assert res is not None return res - + def encode_bag_ctx(self, bag_formula, level): res = self.encode_bag(bag_formula, level, context=True) assert res is not None return res - + def encode(self, formula, level, bound): - + self.ncalls_encode += 1 - + from_cache = self.cache_query(formula, level) if from_cache is not None: return from_cache - + enc = None - + if not isinstance(formula, Formula_rsLTL): - raise NotImplementedError("Unsupported formula type: " + str(type(formula))) - + raise NotImplementedError( + "Unsupported formula type: " + str(type(formula))) + if level > bound: - raise RuntimeError("level > bound. Unexpected behaviour. The encoding does not support levels higher than a bound.") - + raise RuntimeError( + "level > bound. Unexpected behaviour. The encoding does not support levels higher than a bound.") + if formula.f_type == rsLTL_form_type.bag: enc = self.encode_bag_state(formula.bag_descr, level) @@ -158,140 +175,140 @@ class rsLTL_Encoder(object): enc = Not(self.encode_bag_state(subform, level)) else: raise RuntimeError("Negation can be applied to bags only") - + elif formula.f_type == rsLTL_form_type.l_and: enc = And( - self.encode(formula.left_operand, level, bound), - self.encode(formula.right_operand, level, bound) + self.encode(formula.left_operand, level, bound), + self.encode(formula.right_operand, level, bound) ) elif formula.f_type == rsLTL_form_type.l_or: enc = Or( - self.encode(formula.left_operand, level, bound), - self.encode(formula.right_operand, level, bound) + self.encode(formula.left_operand, level, bound), + self.encode(formula.right_operand, level, bound) ) elif formula.f_type == rsLTL_form_type.l_implies: enc = Implies( - self.encode(formula.left_operand, level, bound), - self.encode(formula.right_operand, level, bound) + self.encode(formula.left_operand, level, bound), + self.encode(formula.right_operand, level, bound) ) - + elif formula.f_type == rsLTL_form_type.t_next: if level < bound: enc = And( - self.encode(formula.left_operand, level + 1, bound), - self.encode_bag_ctx(formula.sub_operand, level) + self.encode(formula.left_operand, level + 1, bound), + self.encode_bag_ctx(formula.sub_operand, level) ) - else: + else: # level == bound enc = False for loop_level in range(1, bound+1): - enc = simplify(Or(enc, And(self.loop_position == loop_level, - self.encode(formula.left_operand, loop_level, bound)))) + enc = simplify(Or(enc, And(self.loop_position == loop_level, self.encode( + formula.left_operand, loop_level, bound)))) enc = And(enc, self.encode_bag_ctx(formula.sub_operand, level)) enc = simplify(enc) - + elif formula.f_type == rsLTL_form_type.t_globally: if level < bound: enc = And( - self.encode(formula.left_operand, level, bound), - self.encode_bag_ctx(formula.sub_operand, level), - self.encode(formula, level + 1, bound) + self.encode(formula.left_operand, level, bound), + self.encode_bag_ctx(formula.sub_operand, level), + self.encode(formula, level + 1, bound) ) else: # level == bound enc_loops = False for loop_level in range(1, bound+1): - enc_loops = simplify(Or(enc_loops, - And( - self.loop_position == loop_level, - self.encode_approx(formula, loop_level, bound), - ) - )) + enc_loops = simplify(Or(enc_loops, + And( + self.loop_position == loop_level, + self.encode_approx(formula, loop_level, bound), + ) + )) enc = And( self.encode(formula.left_operand, bound, bound), - enc_loops, + enc_loops, self.encode_bag_ctx(formula.sub_operand, level) ) - enc = simplify(enc) + enc = simplify(enc) elif formula.f_type == rsLTL_form_type.t_finally: if level < bound: enc = Or( - self.encode(formula.left_operand, level, bound), - And( - self.encode_bag_ctx(formula.sub_operand, level), - self.encode(formula, level + 1, bound) - ) + self.encode(formula.left_operand, level, bound), + And( + self.encode_bag_ctx(formula.sub_operand, level), + self.encode(formula, level + 1, bound) ) + ) else: # level == bound enc_loops = False for loop_level in range(1, bound+1): - enc_loops = simplify(Or(enc_loops, - And( - self.loop_position == loop_level, - self.encode_approx(formula, loop_level, bound), - ) - )) - #print(enc) - enc = Or(self.encode(formula.left_operand, bound, bound), - And( - enc_loops, - self.encode_bag_ctx(formula.sub_operand, level) - ) - ) - + enc_loops = simplify(Or(enc_loops, + And( + self.loop_position == loop_level, + self.encode_approx(formula, loop_level, bound), + ) + )) + # print(enc) + enc = Or(self.encode(formula.left_operand, bound, bound), + And( + enc_loops, + self.encode_bag_ctx(formula.sub_operand, level) + ) + ) + enc = simplify(enc) elif formula.f_type == rsLTL_form_type.t_until: if level < bound: inner_enc = self.encode(formula, level + 1, bound) - else: + else: # level == bound inner_enc = False for loop_level in range(1, bound+1): - inner_enc = simplify(Or(inner_enc, - And( - self.loop_position == loop_level, - self.encode_approx(formula, loop_level, bound) - ) - )) + inner_enc = simplify(Or(inner_enc, + And( + self.loop_position == loop_level, + self.encode_approx(formula, loop_level, bound) + ) + )) enc = Or( - self.encode(formula.right_operand, level, bound), + self.encode(formula.right_operand, level, bound), + And( + self.encode(formula.left_operand, level, bound), + inner_enc, + self.encode_bag_ctx(formula.sub_operand, level) + ) + ) + + elif formula.f_type == rsLTL_form_type.t_release: + if level < bound: + inner_enc = self.encode(formula, level + 1, bound) + else: + # level == bound + inner_enc = False + for loop_level in range(1, bound+1): + inner_enc = simplify(Or(inner_enc, + And( + self.loop_position == loop_level, + self.encode_approx(formula, loop_level, bound) + ) + )) + + enc = And( + self.encode(formula.right_operand, level, bound), + Or( + self.encode(formula.left_operand, level, bound), And( - self.encode(formula.left_operand, level, bound), inner_enc, self.encode_bag_ctx(formula.sub_operand, level) ) ) - - elif formula.f_type == rsLTL_form_type.t_release: - if level < bound: - inner_enc = self.encode(formula, level + 1, bound) - else: - # level == bound - inner_enc = False - for loop_level in range(1, bound+1): - inner_enc = simplify(Or(inner_enc, - And( - self.loop_position == loop_level, - self.encode_approx(formula, loop_level, bound) - ) - )) - - enc = And( - self.encode(formula.right_operand, level, bound), - Or( - self.encode(formula.left_operand, level, bound), - And( - inner_enc, - self.encode_bag_ctx(formula.sub_operand, level) - ) - ) - ) + ) else: raise NotImplementedError("Unsupported operator") @@ -305,55 +322,55 @@ class rsLTL_Encoder(object): def encode_approx(self, formula, level, bound): """Provides the approximation-encoding - + Used by encode() """ - + self.ncalls_encode_approx += 1 - + enc = None - + from_cache = self.cache_query_approx(formula, level) if from_cache is not None: return from_cache - + if formula.f_type == rsLTL_form_type.t_until: if level < bound: enc = Or( - self.encode(formula.right_operand, level, bound), + self.encode(formula.right_operand, level, bound), + And( + self.encode(formula.left_operand, level, bound), + self.encode_approx(formula, level + 1, bound), + self.encode_bag_ctx(formula.sub_operand, level) + ) + ) + else: + # level == bound + enc = self.encode(formula.right_operand, bound, bound) + + elif formula.f_type == rsLTL_form_type.t_release: + if level < bound: + enc = And( + self.encode(formula.right_operand, level, bound), + Or( + self.encode(formula.left_operand, level, bound), And( - self.encode(formula.left_operand, level, bound), self.encode_approx(formula, level + 1, bound), self.encode_bag_ctx(formula.sub_operand, level) ) ) - else: - # level == bound - enc = self.encode(formula.right_operand, bound, bound) - - elif formula.f_type == rsLTL_form_type.t_release: - if level < bound: - enc = And( - self.encode(formula.right_operand, level, bound), - Or( - self.encode(formula.left_operand, level, bound), - And( - self.encode_approx(formula, level + 1, bound), - self.encode_bag_ctx(formula.sub_operand, level) - ) - ) - ) - else: + ) + else: # level == bound enc = self.encode(formula.right_operand, bound, bound) elif formula.f_type == rsLTL_form_type.t_globally: if level < bound: enc = And( - self.encode(formula.left_operand, level, bound), - self.encode_bag_ctx(formula.sub_operand, level), - self.encode_approx(formula, level + 1, bound) - ) + self.encode(formula.left_operand, level, bound), + self.encode_bag_ctx(formula.sub_operand, level), + self.encode_approx(formula, level + 1, bound) + ) else: # level == bound enc = self.encode(formula.left_operand, bound, bound) @@ -361,22 +378,23 @@ class rsLTL_Encoder(object): elif formula.f_type == rsLTL_form_type.t_finally: if level < bound: enc = Or( - self.encode(formula.left_operand, level, bound), - And( - self.encode_bag_ctx(formula.sub_operand, level), - self.encode_approx(formula, level + 1, bound) - ) + self.encode(formula.left_operand, level, bound), + And( + self.encode_bag_ctx(formula.sub_operand, level), + self.encode_approx(formula, level + 1, bound) ) + ) else: # level == bound enc = self.encode(formula.left_operand, bound, bound) else: - raise NotImplementedError("Unsupported operator in approximation encoding") + raise NotImplementedError( + "Unsupported operator in approximation encoding") if enc is None: raise RuntimeError("Encoding is NONE. Should never happen") self.cache_approx_save(formula, level, enc) - + return enc diff --git a/rs/context_automaton.py b/rs/context_automaton.py index 22a74b8..8822fcc 100644 --- a/rs/context_automaton.py +++ b/rs/context_automaton.py @@ -1,8 +1,9 @@ from sys import exit from colour import * - + + class ContextAutomaton(object): - + def __init__(self, reaction_system): self._states = [] self._transitions = [] @@ -10,41 +11,41 @@ class ContextAutomaton(object): self._reaction_system = reaction_system self._name = "" self._prod_entities = set() - + @property def states(self): return self._states - + @property def transitions(self): return self._transitions - + @property def prod_entities(self): return self._prod_entities - + @property def reaction_system(self): return self._reaction_system - + @property def name(self): return self._name - + @name.setter def name(self, automaton_name): self._name = automaton_name - + def add_state(self, name): if name not in self._states: self._states.append(name) else: print("\'%s\' already added. skipping..." % (name,)) - + def add_states(self, states_set): for st in states_set: self.add_state(st) - + def add_init_state(self, name): self.add_state(name) self._init_state = self._states.index(name) @@ -72,48 +73,53 @@ class ContextAutomaton(object): def get_init_state_id(self): return self._init_state - + def print_states(self): - for state in self._states: + for state in self._states: print(state) - + def is_valid_rs_set(self, elements): if set(elements).issubset(self._reaction_system.background_set): return True else: return False - + def is_valid_context(self, context): return self.is_valid_rs_set(context) - + def get_set_of_ids(self, elements): """Converts a set/list/tuple of entities into a set of their ids""" - + new_set = set() for e in set(elements): new_set.add(self._reaction_system.get_entity_id(e)) return new_set - + def add_transition(self, src, context_set, dst): if not type(context_set) is set and not type(context_set) is list: print("Contexts set must be of type set or list") - + if not self.is_valid_context(context_set): - raise RuntimeError("one of the entities in the context set is unknown (undefined)!") - + 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") + raise RuntimeError( + "\"" + src + "\" is an unknown (undefined) state") if not self.is_state(dst): - raise RuntimeError("\"" + dst + "\" is an unknown (undefined) state") + raise RuntimeError( + "\"" + dst + "\" is an unknown (undefined) state") new_context_set = set() for e in set(context_set): new_context_set.add(self._reaction_system.get_entity_id(e)) - + self._prod_entities |= new_context_set - - self._transitions.append((self.get_state_id(src),new_context_set,self.get_state_id(dst))) + + self._transitions.append( + (self.get_state_id(src), + new_context_set, self.get_state_id(dst))) def rsset2str(self, elements): """Converts the set of entities ids into the string with their names""" @@ -121,13 +127,13 @@ class ContextAutomaton(object): return "0" s = "{" for c in elements: - s += " " + self._reaction_system.get_entity_name(c) + s += " " + self._reaction_system.get_entity_name(c) s += " }" return s - + def context2str(self, ctx): return self.rsset2str(ctx) - + def show_transitions(self): print(C_MARK_INFO + " Context automaton transitions:") for transition in self._transitions: @@ -135,7 +141,7 @@ class ContextAutomaton(object): str_transition += self.context2str(transition[1]) str_transition += " )--> " + str(transition[2]) print(" - " + str_transition) - + def show_states(self): init_state_name = self.get_init_state_name() print(C_MARK_INFO + " Context automaton states:") @@ -149,13 +155,13 @@ class ContextAutomaton(object): def show_header(self): if self.name: name_string = ": " + colour_str(C_BOLD, self.name) - print(C_MARK_INFO + " Context automaton" + name_string) + print(C_MARK_INFO + " Context automaton" + name_string) def show_prod_entities(self): print(C_MARK_INFO + " Context automaton possible products:") for entity in self._prod_entities: print(" - " + self._reaction_system.get_entity_name(entity)) - + def show(self): self.show_header() self.show_states() diff --git a/rs/context_automaton_with_concentrations.py b/rs/context_automaton_with_concentrations.py index 7369ffa..18d81b8 100644 --- a/rs/context_automaton_with_concentrations.py +++ b/rs/context_automaton_with_concentrations.py @@ -2,66 +2,79 @@ from sys import exit from colour import * from rs.context_automaton import ContextAutomaton - + + class ContextAutomatonWithConcentrations(ContextAutomaton): def __init__(self, reaction_system): - super(ContextAutomatonWithConcentrations, self).__init__(reaction_system) - + super(ContextAutomatonWithConcentrations, + self).__init__(reaction_system) + def is_valid_context(self, context): - if set([e for e,lvl in context]).issubset(self._reaction_system.background_set): + if set( + [e for e, lvl in context]).issubset( + self._reaction_system.background_set): return True else: return False - + def context2str(self, ctx): if len(ctx) == 0: return "0" s = "{" - for ent,lvl in ctx: - s += " " + str((self._reaction_system.get_entity_name(ent),lvl)) + for ent, lvl in ctx: + s += " " + str((self._reaction_system.get_entity_name(ent), lvl)) s += " }" return s - + def add_transition(self, src, context_set, dst): if not type(context_set) is set and not type(context_set) is list: print("Contexts set must be of type set or list") - + if not self.is_valid_context(context_set): - raise RuntimeError("one of the entities in the context set is unknown (undefined)!") - + 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") + raise RuntimeError( + "\"" + src + "\" is an unknown (undefined) state") if not self.is_state(dst): - raise RuntimeError("\"" + dst + "\" is an unknown (undefined) state") + raise RuntimeError( + "\"" + dst + "\" is an unknown (undefined) state") new_context_set = set() - for ent,lvl in set(context_set): - new_context_set.add((self._reaction_system.get_entity_id(ent),lvl)) - - self._transitions.append((self.get_state_id(src),new_context_set,self.get_state_id(dst))) + for ent, lvl in set(context_set): + new_context_set.add( + (self._reaction_system.get_entity_id(ent), lvl)) + + self._transitions.append( + (self.get_state_id(src), + new_context_set, self.get_state_id(dst))) def get_automaton_with_flat_contexts(self, ordinary_reaction_system): - + ca = ContextAutomaton(ordinary_reaction_system) ca._states = self._states ca._init_state = self._init_state - - for src,ctx,dst in self._transitions: - + + for src, ctx, dst in self._transitions: + new_ctx = set() - - for ent,conc in ctx: - for i in range(1,conc+1): - n = self._reaction_system.get_entity_name(ent) + "#" + str(i) + + for ent, conc in ctx: + for i in range(1, conc+1): + n = self._reaction_system.get_entity_name( + ent) + "#" + str(i) ca._reaction_system.ensure_bg_set_entity(n) new_ctx.add(n) - - ca.add_transition(ca.get_state_name(src),new_ctx,ca.get_state_name(dst)) - + + ca.add_transition( + ca.get_state_name(src), + new_ctx, ca.get_state_name(dst)) + return ca - + def show(self): self.show_header() self.show_states() diff --git a/rs/extended_context_automaton.py b/rs/extended_context_automaton.py index bed3dfc..c2bfef2 100644 --- a/rs/extended_context_automaton.py +++ b/rs/extended_context_automaton.py @@ -3,14 +3,15 @@ from colour import * from rs.context_automaton import ContextAutomaton + class ExtendedContextAutomaton(ContextAutomaton): """Extended Context Automaton - + Supports transitions with actions. - + Each transitions is additionally guarded with reactants and inhibitors. - + The provided context entities are the products of the reactions labelling the transition taken. """ @@ -20,36 +21,36 @@ class ExtendedContextAutomaton(ContextAutomaton): self._actions = [] self._transitions_for_products = dict() self._actions_for_products = dict() - + @property def number_of_actions(self): return len(self._actions) - + @property def actions(self): return self._actions - + def has_action(self, action): """Checks if the automaton supports a given action""" - + return action in self._actions - + def get_transitions_producing_entity(self, entity): """Returns the transitions that produce a given entity""" - + if entity in self._transitions_for_products: return self._transitions_for_products[entity] else: return [] - + def get_actions_producing_entity(self, entity): """Returns the actions that produce a given entity""" - + if entity in self._actions_for_products: return self._actions_for_products[entity] else: return set() - + def can_produce_entity(self, entity): """Check if the automaton can produce an entity""" @@ -57,35 +58,40 @@ class ExtendedContextAutomaton(ContextAutomaton): return True else: return False - + def add_transition(self, src, actions, ctx_reaction, dst): """Adds a transition - + src: is the source state name dst: is the destination state name actions: is the set of actions with which the transitions is synchronised ctx_reaction: is the context reaction associated with the transition """ - - ctx_reactants, ctx_inhibitors, ctx_products = ctx_reaction - + + ctx_reactants, ctx_inhibitors, ctx_products = ctx_reaction + if not type(ctx_products) is set and not type(ctx_products) is list: print("Contexts set (context products) must be of type set or list") - + if not self.is_valid_rs_set(ctx_reactants): - raise RuntimeError("one of the entities in the reactants set is unknown (undefined)!") - + raise RuntimeError( + "one of the entities in the reactants set is unknown (undefined)!") + if not self.is_valid_rs_set(ctx_inhibitors): - raise RuntimeError("one of the entities in the inhibitors set is unknown (undefined)!") - + raise RuntimeError( + "one of the entities in the inhibitors set is unknown (undefined)!") + if not self.is_valid_rs_set(ctx_products): - raise RuntimeError("one of the entities in the context set is unknown (undefined)!") - + 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") + raise RuntimeError( + "\"" + src + "\" is an unknown (undefined) state") if not self.is_state(dst): - raise RuntimeError("\"" + dst + "\" is an unknown (undefined) state") + raise RuntimeError( + "\"" + dst + "\" is an unknown (undefined) state") src_id = self.get_state_id(src) dst_id = self.get_state_id(dst) @@ -93,76 +99,78 @@ class ExtendedContextAutomaton(ContextAutomaton): r_ids = self.get_set_of_ids(ctx_reactants) i_ids = self.get_set_of_ids(ctx_inhibitors) p_ids = self.get_set_of_ids(ctx_products) - + new_transition = (src_id, act_ids, (r_ids, i_ids, p_ids), dst_id) - + for product_id in p_ids: self._transitions_for_products.setdefault(product_id, []) self._transitions_for_products[product_id].append(new_transition) self._actions_for_products.setdefault(product_id, set()) self._actions_for_products[product_id] |= set(actions) - + self._prod_entities |= p_ids - + self._transitions.append(new_transition) - + def show_transitions(self): """Prints the set of registered transitions""" - + print(C_MARK_INFO + " Context automaton transitions:") for src_id, act_id, reaction, dst_id in self._transitions: - str_transition = self.get_state_name(src_id) + " --( " + str_transition = self.get_state_name(src_id) + " --( " str_transition += "<" + self.get_actions_str(act_id) + "> | " - str_transition += "( " + self.rsset2str(reaction[0]) + "," + self.rsset2str(reaction[1]) + "," + self.rsset2str(reaction[2]) + " )" + str_transition += "( " + self.rsset2str(reaction[0]) + "," + self.rsset2str( + reaction[1]) + "," + self.rsset2str(reaction[2]) + " )" str_transition += " )--> " + self.get_state_name(dst_id) print(" - " + str_transition) - + def add_action(self, action_name): """Registers an action""" - + if action_name not in self._actions: self._actions.append(action_name) else: print("\'%s\' already added. skipping..." % (action_name,)) - + def get_action_id(self, action_name): """For an action name returns its id""" - + try: return self._actions.index(action_name) except ValueError: - print_error("Undefined context automaton action: " + repr(action_name)) + print_error("Undefined context automaton action: " + + repr(action_name)) exit(1) - + def get_action_name(self, action_id): return self._actions[action_id] - + def get_set_of_action_ids(self, actions): """Converts a set of actions into the set of their ids""" - + act_ids = set() for act in actions: act_ids.add(self.get_action_id(act)) return act_ids - + def get_actions_str(self, actions): """Returns the string for the set of action ids given by actions""" - + s = "" for act in actions: s += self.get_action_name(act) + ", " s = s[:-2] return s - + def show_actions(self): """Prints all the actions""" - + print(C_MARK_INFO + " Context automaton actions:") for act in self._actions: print(" - " + act + " (id=" + str(self.get_action_id(act)) + ")") - + def show(self): super(ExtendedContextAutomaton, self).show() self.show_actions() -# EOF +# EOF diff --git a/rs/network_of_context_automata.py b/rs/network_of_context_automata.py index e470552..efdee19 100644 --- a/rs/network_of_context_automata.py +++ b/rs/network_of_context_automata.py @@ -1,8 +1,9 @@ from sys import exit from colour import * + class NetworkOfContextAutomata(object): - + def __init__(self, reaction_system, context_automata): self.automata = [] self._reaction_system = reaction_system @@ -14,7 +15,7 @@ class NetworkOfContextAutomata(object): if len(context_automata) < 1: print("Context automata network must contain at least one automaton!") exit(1) - + for automaton in context_automata: self.add(automaton) @@ -39,14 +40,18 @@ class NetworkOfContextAutomata(object): def sanity_check(self): """Performs a sanity check of the network of automata""" - + for automaton in self.automata: if automaton.reaction_system != self._reaction_system: - print_error("Mismatching reaction system used in \"" + str(automaton.name) + "\"!!!") + print_error( + "Mismatching reaction system used in \"" + + str(automaton.name) + "\"!!!") exit(1) def show_prod_entities(self): - print(C_MARK_INFO + " Possible context-products for the network of automata:") + print( + C_MARK_INFO + + " Possible context-products for the network of automata:") for entity in self.prod_entities: print(" - " + self._reaction_system.get_entity_name(entity)) @@ -57,22 +62,22 @@ class NetworkOfContextAutomata(object): def register_action(self, action, aut): """Associates an action with an automaton""" - + self._automata_for_actions.setdefault(action, set()) aut_index = self.automata.index(aut) self._automata_for_actions[action].add(aut_index) - + def get_actions_producing_entity(self, entity): """Returns the set of actions producing an entity""" - + if entity in self._actions_for_products: return self._actions_for_products[entity] else: return set() - + def get_automata_with_action(self, action): """Returns the set of automata that support an action""" - + if action in self._automata_for_actions: return self._automata_for_actions[action] else: @@ -80,17 +85,18 @@ class NetworkOfContextAutomata(object): def add(self, aut): """Adds an automaton to the network""" - + self.automata.append(aut) self._prod_entities |= aut.prod_entities self._actions |= set(aut.actions) - + for action in aut.actions: self.register_action(action, aut) for entity in aut.prod_entities: self._actions_for_products.setdefault(entity, set()) - self._actions_for_products[entity] |= aut.get_actions_producing_entity(entity) + self._actions_for_products[entity] |= aut.get_actions_producing_entity( + entity) def show(self): print() @@ -102,4 +108,4 @@ class NetworkOfContextAutomata(object): self.show_prod_entities() self.show_actions() -# EOF +# EOF diff --git a/rs/reaction_system.py b/rs/reaction_system.py index 5d27df5..908e1e7 100644 --- a/rs/reaction_system.py +++ b/rs/reaction_system.py @@ -1,6 +1,7 @@ from sys import exit from colour import * + class ReactionSystem(object): def __init__(self): @@ -8,17 +9,17 @@ class ReactionSystem(object): self.reactions = [] self.background_set = [] - #self.reactions_by_agents = [] # each element is 'reactions_by_prod' + # self.reactions_by_agents = [] # each element is 'reactions_by_prod' self.reactions_by_prod = None - ## legacy: + # legacy: self.init_contexts = [] self.context_entities = [] @property def background_set_size(self): return len(self.background_set) - + @property def set_of_bgset_ids(self): return set(range(self.background_set_size)) @@ -29,12 +30,13 @@ class ReactionSystem(object): def assume_not_in_bgset(self, name): if self.is_in_background_set(name): - raise RuntimeError("The entity " + name + " is already on the list") - + raise RuntimeError( + "The entity " + name + " is already on the list") + def add_bg_set_entity(self, name): self.assume_not_in_bgset(name) self.background_set.append(name) - + def ensure_bg_set_entity(self, name): if not self.is_in_background_set(name): self.background_set.append(name) @@ -70,7 +72,7 @@ class ReactionSystem(object): def add_reaction(self, R, I, P): """Adds a reaction""" - + if R == [] or P == []: raise RuntimeError("No reactants or products defined") @@ -122,34 +124,38 @@ class ReactionSystem(object): s += self.get_entity_name(entity) + ", " s = s[:-2] return s - + def state_to_str(self, state): return self.entities_ids_set_to_str(state) def show_reactions(self, soft=False): print(C_MARK_INFO + " Reactions:") if soft and len(self.reactions) > 50: - print(" -> there are more than 50 reactions (" + str(len(self.reactions)) + ")") + print(" -> there are more than 50 reactions (" + + str(len(self.reactions)) + ")") else: - print(" "*4 + "{0: ^35}{1: ^25}{2: ^15}".format("reactants"," inhibitors"," products")) + print( + " "*4 + "{0: ^35}{1: ^25}{2: ^15}".format("reactants", " inhibitors", " products")) for reaction in self.reactions: # print("\t( R={" + self.state_to_str(reaction[0]) + "}, I={" + self.state_to_str(reaction[1]) + "}, P={" + self.state_to_str(reaction[2]) + "} )") print(" " + "- {0: ^35}{1: ^25}{2: ^15}".format("{ " + self.state_to_str(reaction[0]) + " }", - " { " + self.state_to_str(reaction[1]) + " }", - " { " + self.state_to_str(reaction[2]) + " }")) + " { " + self.state_to_str(reaction[1]) + " }", + " { " + self.state_to_str(reaction[2]) + " }")) def show_background_set(self): - print(C_MARK_INFO + " Background set: {" + self.entities_names_set_to_str(self.background_set) + "}") + print( + C_MARK_INFO + " Background set: {" + self.entities_names_set_to_str(self.background_set) + "}") def show_initial_contexts(self): if len(self.init_contexts) > 0: print(C_MARK_INFO + " Initial context sets:") for ctx in self.init_contexts: - print(" - {" + self.entities_ids_set_to_str(ctx) + "}") + print(" - {" + self.entities_ids_set_to_str(ctx) + "}") def show_context_entities(self): if len(self.context_entities) > 0: - print(C_MARK_INFO + " Context entities: " + self.entities_ids_set_to_str(self.context_entities)) + print( + C_MARK_INFO + " Context entities: " + self.entities_ids_set_to_str(self.context_entities)) def show(self, soft=False): @@ -157,7 +163,7 @@ class ReactionSystem(object): self.show_initial_contexts() self.show_reactions(soft) self.show_context_entities() - + def get_reactions_by_product(self): """Sorts reactions by their products and returns a dictionary of products""" @@ -169,13 +175,14 @@ class ReactionSystem(object): for reaction in self.reactions: producible_entities = producible_entities.union(set(reaction[2])) - reactions_by_prod = {} + reactions_by_prod = {} for prod_entity in producible_entities: reactions_by_prod[prod_entity] = [] for reaction in self.reactions: if prod_entity in reaction[2]: - reactions_by_prod[prod_entity].append([reaction[0],reaction[1]]) + reactions_by_prod[prod_entity].append( + [reaction[0], reaction[1]]) # save in cache self.reactions_by_prod = reactions_by_prod diff --git a/rs/reaction_system_with_autnet.py b/rs/reaction_system_with_autnet.py index ca7472c..0054052 100644 --- a/rs/reaction_system_with_autnet.py +++ b/rs/reaction_system_with_autnet.py @@ -1,6 +1,6 @@ class ReactionSystemWithNetworkOfAutomata(object): - + def __init__(self, reaction_system, context_automata): self.rs = reaction_system self.cas = context_automata @@ -9,4 +9,4 @@ class ReactionSystemWithNetworkOfAutomata(object): self.rs.show(soft) self.cas.show() -# EOF \ No newline at end of file +# EOF diff --git a/rs/reaction_system_with_automaton.py b/rs/reaction_system_with_automaton.py index 500a883..0fcceaa 100644 --- a/rs/reaction_system_with_automaton.py +++ b/rs/reaction_system_with_automaton.py @@ -2,8 +2,9 @@ from rs.reaction_system_with_concentrations import ReactionSystemWithConcentrati from rs.reaction_system_with_concentrations_param import ReactionSystemWithConcentrationsParam from rs.context_automaton_with_concentrations import ContextAutomatonWithConcentrations + class ReactionSystemWithAutomaton(object): - + def __init__(self, reaction_system, context_automaton): self.rs = reaction_system self.ca = context_automaton @@ -11,7 +12,7 @@ class ReactionSystemWithAutomaton(object): def show(self, soft=False): self.rs.show(soft) self.ca.show() - + def is_concentr_and_param_compatible(self): """ Checks if the underlying RS/CA are compatible @@ -22,7 +23,7 @@ class ReactionSystemWithAutomaton(object): if not isinstance(self.ca, ContextAutomatonWithConcentrations): return False return True - + def is_with_concentrations(self): if not isinstance(self.rs, ReactionSystemWithConcentrations): return False @@ -32,14 +33,13 @@ class ReactionSystemWithAutomaton(object): def sanity_check(self): pass - + def get_ordinary_reaction_system_with_automaton(self): - + if not self.is_with_concentrations(): raise RuntimeError("Not RS/CA with concentrations") - + ors = self.rs.get_reaction_system() oca = self.ca.get_automaton_with_flat_contexts(ors) - + return ReactionSystemWithAutomaton(ors, oca) - diff --git a/rs/reaction_system_with_concentrations.py b/rs/reaction_system_with_concentrations.py index 0236e63..314716b 100644 --- a/rs/reaction_system_with_concentrations.py +++ b/rs/reaction_system_with_concentrations.py @@ -3,29 +3,31 @@ from colour import * from rs.reaction_system import ReactionSystem + class ReactionSystemWithConcentrations(ReactionSystem): def __init__(self): - self.reactions = [] - self.meta_reactions = dict() - self.permanent_entities = dict() - self.background_set = [] - self.context_entities = [] # legacy. to be removed - self.reactions_by_prod = None - self.max_concentration = 0 - self.max_conc_per_ent = dict() + self.reactions = [] + self.meta_reactions = dict() + self.permanent_entities = dict() + self.background_set = [] + self.context_entities = [] # legacy. to be removed + self.reactions_by_prod = None + self.max_concentration = 0 + self.max_conc_per_ent = dict() def add_bg_set_entity(self, e): name = "" def_max_conc = -1 if type(e) is tuple and len(e) == 2: - name,def_max_conc = e + name, def_max_conc = e elif type(e) is str: name = e print("\nWARNING: no maximal concentration level specified for:", e, "\n") else: - raise RuntimeError("Bad entity type when adding background set element") + raise RuntimeError( + "Bad entity type when adding background set element") self.assume_not_in_bgset(name) self.background_set.append(name) @@ -39,7 +41,7 @@ class ReactionSystemWithConcentrations(ReactionSystem): self.max_concentration = def_max_conc def get_max_concentration_level(self, e): - + if e in self.max_conc_per_ent: return self.max_conc_per_ent[e] else: @@ -47,102 +49,108 @@ class ReactionSystemWithConcentrations(ReactionSystem): def is_valid_entity_with_concentration(self, e): """Sanity check for entities with concentration""" - + if type(e) is tuple: if len(e) == 2 and type(e[1]) is int: return True - + if type(e) is list: if len(e) == 2 and type(e[1]) is int: return True print("FATAL. Invalid entity+concentration: {:s}".format(e)) exit(1) - + return False def get_state_ids(self, state): """Returns entities of the given state without levels""" - return [e for e,c in state] + return [e for e, c in state] def has_non_zero_concentration(self, elem): if elem[1] < 1: - raise RuntimeError("Unexpected concentration level in state: " + str(elem)) + raise RuntimeError( + "Unexpected concentration level in state: " + str(elem)) def process_rip(self, R, I, P, ignore_empty_R=False): """Chcecks concentration levels and converts entities names into their ids""" if R == [] and not ignore_empty_R: raise RuntimeError("No reactants defined") - + reactants = [] for r in R: self.is_valid_entity_with_concentration(r) self.has_non_zero_concentration(r) - entity,level = r - reactants.append((self.get_entity_id(entity),level)) + entity, level = r + reactants.append((self.get_entity_id(entity), level)) if self.max_concentration < level: self.max_concentration = level inhibitors = [] for i in I: self.is_valid_entity_with_concentration(i) self.has_non_zero_concentration(i) - entity,level = i - inhibitors.append((self.get_entity_id(entity),level)) + entity, level = i + inhibitors.append((self.get_entity_id(entity), level)) if self.max_concentration < level: self.max_concentration = level products = [] for p in P: self.is_valid_entity_with_concentration(p) self.has_non_zero_concentration(p) - entity,level = p - products.append((self.get_entity_id(entity),level)) - - return reactants,inhibitors,products + entity, level = p + products.append((self.get_entity_id(entity), level)) + + return reactants, inhibitors, products def add_reaction(self, R, I, P): """Adds a reaction""" - + if P == []: raise RuntimeError("No products defined") - reaction = self.process_rip(R,I,P) + reaction = self.process_rip(R, I, P) self.reactions.append(reaction) def add_reaction_without_reactants(self, R, I, P): """Adds a reaction""" - + if P == []: raise RuntimeError("No products defined") - reaction = self.process_rip(R,I,P,ignore_empty_R=True) + reaction = self.process_rip(R, I, P, ignore_empty_R=True) self.reactions.append(reaction) def add_reaction_inc(self, incr_entity, incrementer, R, I): """Adds a macro/meta reaction for increasing the value of incr_entity""" - reactants,inhibitors,products = self.process_rip(R,I,[],ignore_empty_R=True) + reactants, inhibitors, products = self.process_rip( + R, I, [], ignore_empty_R=True) incr_entity_id = self.get_entity_id(incr_entity) - self.meta_reactions.setdefault(incr_entity_id,[]) - self.meta_reactions[incr_entity_id].append(("inc", self.get_entity_id(incrementer), reactants, inhibitors)) + self.meta_reactions.setdefault(incr_entity_id, []) + self.meta_reactions[incr_entity_id].append( + ("inc", self.get_entity_id(incrementer), reactants, inhibitors)) def add_reaction_dec(self, decr_entity, decrementer, R, I): """Adds a macro/meta reaction for decreasing the value of incr_entity""" - reactants,inhibitors,products = self.process_rip(R,I,[],ignore_empty_R=True) + reactants, inhibitors, products = self.process_rip( + R, I, [], ignore_empty_R=True) decr_entity_id = self.get_entity_id(decr_entity) - self.meta_reactions.setdefault(decr_entity_id,[]) - self.meta_reactions[decr_entity_id].append(("dec", self.get_entity_id(decrementer), reactants, inhibitors)) + self.meta_reactions.setdefault(decr_entity_id, []) + self.meta_reactions[decr_entity_id].append( + ("dec", self.get_entity_id(decrementer), reactants, inhibitors)) def add_permanency(self, ent, I): """Sets entity to be permanent unless it is inhibited""" - + ent_id = self.get_entity_id(ent) - + if ent_id in self.permanent_entities: - raise RuntimeError("Permanency for {0} already defined.".format(ent)) - - inhibitors = self.process_rip([],I,[],ignore_empty_R=True)[1] + raise RuntimeError( + "Permanency for {0} already defined.".format(ent)) + + inhibitors = self.process_rip([], I, [], ignore_empty_R=True)[1] self.permanent_entities[ent_id] = inhibitors - + def set_context_entities(self, entities): raise NotImplementedError @@ -162,33 +170,38 @@ class ReactionSystemWithConcentrations(ReactionSystem): def state_to_str(self, state): s = "" - for ent,level in state: + for ent, level in state: s += self.get_entity_name(ent) + "=" + str(level) + ", " s = s[:-2] - return s + return s def show_background_set(self): - print(C_MARK_INFO + " Background set: {" + self.entities_names_set_to_str(self.background_set) + "}") + print( + C_MARK_INFO + " Background set: {" + self.entities_names_set_to_str(self.background_set) + "}") def show_meta_reactions(self): print(C_MARK_INFO + " Meta reactions:") - for param_ent,reactions in self.meta_reactions.items(): - for r_type,command,reactants,inhibitors in reactions: + for param_ent, reactions in self.meta_reactions.items(): + for r_type, command, reactants, inhibitors in reactions: if r_type == "inc" or r_type == "dec": - print(" - [ Type=" + repr(r_type) + " Operand=( " + self.get_entity_name(param_ent) + \ - " ) Command=( " + self.get_entity_name(command) + " ) ] -- ( R={" + self.state_to_str(reactants) + "}, I={" + self.state_to_str(inhibitors) + "} )") + print(" - [ Type=" + repr(r_type) + " Operand=( " + self.get_entity_name(param_ent) + " ) Command=( " + self.get_entity_name( + command) + " ) ] -- ( R={" + self.state_to_str(reactants) + "}, I={" + self.state_to_str(inhibitors) + "} )") else: - raise RuntimeError("Unknown meta-reaction type: " + repr(r_type)) + raise RuntimeError( + "Unknown meta-reaction type: " + repr(r_type)) def show_max_concentrations(self): - print(C_MARK_INFO + " Maximal allowed concentration levels (for optimized translation to RS):") - for e,max_conc in self.max_conc_per_ent.items(): - print(" - {0:^20} = {1:<6}".format(self.get_entity_name(e),max_conc)) + print( + C_MARK_INFO + + " Maximal allowed concentration levels (for optimized translation to RS):") + for e, max_conc in self.max_conc_per_ent.items(): + print(" - {0:^20} = {1:<6}".format(self.get_entity_name(e), max_conc)) def show_permanent_entities(self): print(C_MARK_INFO + " Permanent entities:") - for e,inhibitors in self.permanent_entities.items(): - print(" - {0:^20}{1:<6}".format(self.get_entity_name(e) + ": ","I={" + self.state_to_str(inhibitors) + "}")) + for e, inhibitors in self.permanent_entities.items(): + print(" - {0:^20}{1:<6}".format(self.get_entity_name(e) + ": ", + "I={" + self.state_to_str(inhibitors) + "}")) def show(self, soft=False): self.show_background_set() @@ -196,7 +209,7 @@ class ReactionSystemWithConcentrations(ReactionSystem): self.show_permanent_entities() self.show_meta_reactions() self.show_max_concentrations() - + def get_reactions_by_product(self): """Sorts reactions by their products and returns a dictionary of products""" @@ -206,8 +219,9 @@ class ReactionSystemWithConcentrations(ReactionSystem): producible_entities = set() for reaction in self.reactions: - product_entities = [e for e,c in reaction[2]] - producible_entities = producible_entities.union(set(product_entities)) + product_entities = [e for e, c in reaction[2]] + producible_entities = producible_entities.union( + set(product_entities)) reactions_by_prod = {} @@ -216,29 +230,30 @@ class ReactionSystemWithConcentrations(ReactionSystem): rcts_for_p_e = reactions_by_prod[p_e] for r in self.reactions: - product_entities = [e for e,c in r[2]] - + product_entities = [e for e, c in r[2]] + if p_e in product_entities: reactants = r[0] inhibitors = r[1] - products = [(e,c) for e,c in r[2] if e == p_e] - + products = [(e, c) for e, c in r[2] if e == p_e] + prod_conc = products[0][1] insert_place = None - + # we need to order the reactions w.r.t. the concentration levels produced (increasing order) - for i in range(0,len(rcts_for_p_e)): - + for i in range(0, len(rcts_for_p_e)): + checked_conc = rcts_for_p_e[i][2][0][1] if prod_conc <= checked_conc: insert_place = i break - if insert_place == None: # empty or the is only one element which is smaller than the element being added - rcts_for_p_e.append((reactants, inhibitors, products)) # we append (to the end) + if insert_place == None: # empty or the is only one element which is smaller than the element being added + # we append (to the end) + rcts_for_p_e.append((reactants, inhibitors, products)) else: - rcts_for_p_e.insert(insert_place,(reactants, inhibitors, products)) - + rcts_for_p_e.insert( + insert_place, (reactants, inhibitors, products)) # save in cache self.reactions_by_prod = reactions_by_prod @@ -246,47 +261,47 @@ class ReactionSystemWithConcentrations(ReactionSystem): return reactions_by_prod def get_reaction_system(self): - + rs = ReactionSystem() - - for reactants,inhibitors,products in self.reactions: + + for reactants, inhibitors, products in self.reactions: new_reactants = [] new_inhibitors = [] new_products = [] - for ent,conc in reactants: + for ent, conc in reactants: n = self.get_entity_name(ent) + "#" + str(conc) rs.ensure_bg_set_entity(n) new_reactants.append(n) - - for ent,conc in inhibitors: + + for ent, conc in inhibitors: n = self.get_entity_name(ent) + "#" + str(conc) rs.ensure_bg_set_entity(n) new_inhibitors.append(n) - for ent,conc in products: - for i in range(1,conc+1): + for ent, conc in products: + for i in range(1, conc+1): n = self.get_entity_name(ent) + "#" + str(i) rs.ensure_bg_set_entity(n) new_products.append(n) - - rs.add_reaction(new_reactants,new_inhibitors,new_products) - - for param_ent,reactions in self.meta_reactions.items(): - for r_type,command,reactants,inhibitors in reactions: - + + rs.add_reaction(new_reactants, new_inhibitors, new_products) + + for param_ent, reactions in self.meta_reactions.items(): + for r_type, command, reactants, inhibitors in reactions: + param_ent_name = self.get_entity_name(param_ent) - + new_reactants = [] new_inhibitors = [] - - for ent,conc in reactants: + + for ent, conc in reactants: n = self.get_entity_name(ent) + "#" + str(conc) rs.ensure_bg_set_entity(n) new_reactants.append(n) - - for ent,conc in inhibitors: + + for ent, conc in inhibitors: n = self.get_entity_name(ent) + "#" + str(conc) rs.ensure_bg_set_entity(n) new_inhibitors.append(n) @@ -295,84 +310,97 @@ class ReactionSystemWithConcentrations(ReactionSystem): if command in self.max_conc_per_ent: max_cmd_c = self.max_conc_per_ent[command] else: - print("WARNING:\n\tThere is no maximal concentration level defined for " + self.get_entity_name(command)) + print( + "WARNING:\n\tThere is no maximal concentration level defined for " + + self.get_entity_name(command)) print("\tThis is a very bad idea -- expect degraded performance\n") - - for l in range(1,max_cmd_c+1): + + for l in range(1, max_cmd_c+1): cmd_ent = self.get_entity_name(command) + "#" + str(l) rs.ensure_bg_set_entity(cmd_ent) - + if r_type == "inc": - + # pre_conc -- predecessor concentration # succ_conc -- successor concentration concentration - - for i in range(1,self.max_concentration): + + for i in range(1, self.max_concentration): pre_conc = param_ent_name + "#" + str(i) rs.ensure_bg_set_entity(pre_conc) new_products = [] succ_value = i+l - for j in range(1,succ_value+1): - if j > self.max_concentration: break + for j in range(1, succ_value+1): + if j > self.max_concentration: + break new_p = param_ent_name + "#" + str(j) rs.ensure_bg_set_entity(new_p) new_products.append(new_p) if new_products != []: - rs.add_reaction(set(new_reactants + [pre_conc,cmd_ent]), set(new_inhibitors), set(new_products)) + rs.add_reaction( + set(new_reactants + [pre_conc, cmd_ent]), + set(new_inhibitors), + set(new_products)) elif r_type == "dec": - for i in range(1,self.max_concentration+1): + for i in range(1, self.max_concentration+1): pre_conc = param_ent_name + "#" + str(i) rs.ensure_bg_set_entity(pre_conc) new_products = [] succ_value = i-l - for j in range(1,succ_value+1): - if j > self.max_concentration: break + for j in range(1, succ_value+1): + if j > self.max_concentration: + break new_p = param_ent_name + "#" + str(j) rs.ensure_bg_set_entity(new_p) new_products.append(new_p) if new_products != []: - rs.add_reaction(set(new_reactants + [pre_conc,cmd_ent]), set(new_inhibitors), set(new_products)) - + rs.add_reaction( + set(new_reactants + [pre_conc, cmd_ent]), + set(new_inhibitors), + set(new_products)) + else: - raise RuntimeError("Unknown meta-reaction type: " + repr(r_type)) - - for ent,inhibitors in self.permanent_entities.items(): - + raise RuntimeError( + "Unknown meta-reaction type: " + repr(r_type)) + + for ent, inhibitors in self.permanent_entities.items(): + max_c = self.max_concentration if ent in self.max_conc_per_ent: max_c = self.max_conc_per_ent[ent] else: - print("WARNING:\n\tThere is no maximal concentration level defined for " + self.get_entity_name(ent)) + print( + "WARNING:\n\tThere is no maximal concentration level defined for " + + self.get_entity_name(ent)) print("\tThis is a very bad idea -- expect degraded performance\n") - + def e_value(i): return self.get_entity_name(ent) + "#" + str(i) - - for value in range(1,max_c+1): - + + for value in range(1, max_c+1): + new_reactants = [] new_inhibitors = [] new_products = [] - + new_reactants = [e_value(value)] - - for e_inh,conc in inhibitors: + + for e_inh, conc in inhibitors: n = self.get_entity_name(e_inh) + "#" + str(conc) rs.ensure_bg_set_entity(n) - new_inhibitors.append(n) + new_inhibitors.append(n) - for i in range(1,value+1): + for i in range(1, value+1): new_products.append(e_value(i)) - - rs.add_reaction(new_reactants,new_inhibitors,new_products) - + + rs.add_reaction(new_reactants, new_inhibitors, new_products) + return rs - + class ReactionSystemWithAutomaton(object): - + def __init__(self, reaction_system, context_automaton): self.rs = reaction_system self.ca = context_automaton @@ -380,7 +408,7 @@ class ReactionSystemWithAutomaton(object): def show(self, soft=False): self.rs.show(soft) self.ca.show() - + def is_with_concentrations(self): if not isinstance(self.rs, ReactionSystemWithConcentrations): return False @@ -390,17 +418,16 @@ class ReactionSystemWithAutomaton(object): def sanity_check(self): pass - + def get_ordinary_reaction_system_with_automaton(self): - + if not self.is_with_concentrations(): raise RuntimeError("Not RS/CA with concentrations") - + ors = self.rs.get_reaction_system() oca = self.ca.get_automaton_with_flat_contexts(ors) - + return ReactionSystemWithAutomaton(ors, oca) - + # EOF - diff --git a/rs/reaction_system_with_concentrations_param.py b/rs/reaction_system_with_concentrations_param.py index 2bec6cb..29a318b 100644 --- a/rs/reaction_system_with_concentrations_param.py +++ b/rs/reaction_system_with_concentrations_param.py @@ -3,20 +3,23 @@ from colour import * from rs.reaction_system import ReactionSystem + class ParameterObj(object): def __init__(self, name): self.name = name - + def __repr__(self): return "@{0}".format(self.name) + def is_param(some_object): if isinstance(some_object, ParameterObj): return True else: return False + class ReactionSystemWithConcentrationsParam(ReactionSystem): def __init__(self): @@ -124,7 +127,7 @@ class ReactionSystemWithConcentrationsParam(ReactionSystem): reactants.append((self.get_entity_id(entity), level)) if self.max_concentration < level: self.max_concentration = level - + # # INHIBITORS # @@ -139,7 +142,7 @@ class ReactionSystemWithConcentrationsParam(ReactionSystem): inhibitors.append((self.get_entity_id(entity), level)) if self.max_concentration < level: self.max_concentration = level - + # # PRODUCTS # @@ -161,51 +164,56 @@ class ReactionSystemWithConcentrationsParam(ReactionSystem): def add_reaction(self, R, I, P): """Adds a reaction - + R, I, and P are sets of entities (not their IDs) """ - + if P == []: raise RuntimeError("No products defined") - reaction = self.process_rip(R,I,P) - + reaction = self.process_rip(R, I, P) + self.reactions.append(reaction) def add_reaction_without_reactants(self, R, I, P): """Adds a reaction""" - + if P == []: raise RuntimeError("No products defined") - reaction = self.process_rip(R,I,P,ignore_empty_R=True) + reaction = self.process_rip(R, I, P, ignore_empty_R=True) self.reactions.append(reaction) def add_reaction_inc(self, incr_entity, incrementer, R, I): """Adds a macro/meta reaction for increasing the value of incr_entity""" - reactants,inhibitors,products = self.process_rip(R,I,[],ignore_empty_R=True) + reactants, inhibitors, products = self.process_rip( + R, I, [], ignore_empty_R=True) incr_entity_id = self.get_entity_id(incr_entity) - self.meta_reactions.setdefault(incr_entity_id,[]) - self.meta_reactions[incr_entity_id].append(("inc", self.get_entity_id(incrementer), reactants, inhibitors)) + self.meta_reactions.setdefault(incr_entity_id, []) + self.meta_reactions[incr_entity_id].append( + ("inc", self.get_entity_id(incrementer), reactants, inhibitors)) def add_reaction_dec(self, decr_entity, decrementer, R, I): """Adds a macro/meta reaction for decreasing the value of incr_entity""" - reactants,inhibitors,products = self.process_rip(R,I,[],ignore_empty_R=True) + reactants, inhibitors, products = self.process_rip( + R, I, [], ignore_empty_R=True) decr_entity_id = self.get_entity_id(decr_entity) - self.meta_reactions.setdefault(decr_entity_id,[]) - self.meta_reactions[decr_entity_id].append(("dec", self.get_entity_id(decrementer), reactants, inhibitors)) + self.meta_reactions.setdefault(decr_entity_id, []) + self.meta_reactions[decr_entity_id].append( + ("dec", self.get_entity_id(decrementer), reactants, inhibitors)) def add_permanency(self, ent, I): """Sets entity to be permanent unless it is inhibited""" - + ent_id = self.get_entity_id(ent) - + if ent_id in self.permanent_entities: - raise RuntimeError("Permanency for {0} already defined.".format(ent)) - - inhibitors = self.process_rip([],I,[],ignore_empty_R=True)[1] + raise RuntimeError( + "Permanency for {0} already defined.".format(ent)) + + inhibitors = self.process_rip([], I, [], ignore_empty_R=True)[1] self.permanent_entities[ent_id] = inhibitors - + def set_context_entities(self, entities): raise NotImplementedError @@ -233,33 +241,36 @@ class ReactionSystemWithConcentrationsParam(ReactionSystem): return str(state) else: s = "" - for ent,level in state: + for ent, level in state: s += self.get_entity_name(ent) + "=" + str(level) + ", " s = s[:-2] - return s + return s def show_background_set(self): - print(C_MARK_INFO + " Background set: {" + self.entities_names_set_to_str(self.background_set) + "}") + print( + C_MARK_INFO + " Background set: {" + self.entities_names_set_to_str(self.background_set) + "}") def show_meta_reactions(self): print(C_MARK_INFO + " Meta reactions:") - for param_ent,reactions in self.meta_reactions.items(): - for r_type,command,reactants,inhibitors in reactions: + for param_ent, reactions in self.meta_reactions.items(): + for r_type, command, reactants, inhibitors in reactions: if r_type == "inc" or r_type == "dec": - print(" - [ Type=" + repr(r_type) + " Operand=( " + self.get_entity_name(param_ent) + \ - " ) Command=( " + self.get_entity_name(command) + " ) ] -- ( R={" + self.state_to_str(reactants) + "}, I={" + self.state_to_str(inhibitors) + "} )") + print(" - [ Type=" + repr(r_type) + " Operand=( " + self.get_entity_name(param_ent) + " ) Command=( " + self.get_entity_name( + command) + " ) ] -- ( R={" + self.state_to_str(reactants) + "}, I={" + self.state_to_str(inhibitors) + "} )") else: - raise RuntimeError("Unknown meta-reaction type: " + repr(r_type)) + raise RuntimeError( + "Unknown meta-reaction type: " + repr(r_type)) def show_max_concentrations(self): print(C_MARK_INFO + " Maximal allowed concentration levels:") - for e,max_conc in self.max_conc_per_ent.items(): - print(" - {0:^20} = {1:<6}".format(self.get_entity_name(e),max_conc)) + for e, max_conc in self.max_conc_per_ent.items(): + print(" - {0:^20} = {1:<6}".format(self.get_entity_name(e), max_conc)) def show_permanent_entities(self): print(C_MARK_INFO + " Permanent entities:") - for e,inhibitors in self.permanent_entities.items(): - print(" - {0:^20}{1:<6}".format(self.get_entity_name(e) + ": ","I={" + self.state_to_str(inhibitors) + "}")) + for e, inhibitors in self.permanent_entities.items(): + print(" - {0:^20}{1:<6}".format(self.get_entity_name(e) + ": ", + "I={" + self.state_to_str(inhibitors) + "}")) def show(self, soft=False): self.show_background_set() @@ -268,21 +279,22 @@ class ReactionSystemWithConcentrationsParam(ReactionSystem): self.show_permanent_entities() self.show_meta_reactions() self.show_max_concentrations() - + def get_producible_entities(self): """ Returns the set of entities that appear as products of reactions. """ - + producible_entities = set() for reaction in self.reactions: - product_entities = [e for e,c in reaction[2] if c > 0] - producible_entities = producible_entities.union(set(product_entities)) - + product_entities = [e for e, c in reaction[2] if c > 0] + producible_entities = producible_entities.union( + set(product_entities)) + return producible_entities - + def get_reaction_system(self): """ Translates RSC into RS diff --git a/rsltl_shortcuts.py b/rsltl_shortcuts.py index c9349ce..cdcf713 100644 --- a/rsltl_shortcuts.py +++ b/rsltl_shortcuts.py @@ -1,23 +1,28 @@ from logics import * -##### SHORTCUTS +# SHORTCUTS + def bag_True(): return BagDescription.f_TRUE() + def bag_entity(name): return BagDescription.f_entity(name) + def get_bag_if_str(arg): if isinstance(arg, str): return bag_entity(arg) > 0 else: return arg + def bag_Not(a0): a0 = get_bag_if_str(a0) return BagDescription.f_Not(a0) + def bag_And(*args): assert len(args) > 1 last = get_bag_if_str(args[0]) @@ -25,7 +30,8 @@ def bag_And(*args): last = BagDescription.f_And( last, get_bag_if_str(arg)) return last - + + def exact_state(contained_entities, all_entities): """ Assumes 0 concentration level for all the @@ -40,32 +46,37 @@ def exact_state(contained_entities, all_entities): expr.append(ent == 0) if len(expr) > 0: - + last = expr[0] for e in expr[1:]: last = BagDescription.f_And(last, e) return last - + else: assert False + def ltl_F(ctx_arg, a0): a0 = get_bag_if_str(a0) return Formula_rsLTL.f_F(ctx_arg, a0) + def ltl_G(ctx_arg, a0): a0 = get_bag_if_str(a0) return Formula_rsLTL.f_G(ctx_arg, a0) + def ltl_X(ctx_arg, a0): a0 = get_bag_if_str(a0) return Formula_rsLTL.f_X(ctx_arg, a0) + def ltl_U(ctx_arg, a0, a1): a0 = get_bag_if_str(a0) a1 = get_bag_if_str(a1) return Formula_rsLTL.f_U(ctx_arg, a0, a1) + def ltl_And(*args): assert len(args) > 1 last = get_bag_if_str(args[0]) @@ -73,22 +84,25 @@ def ltl_And(*args): last = Formula_rsLTL.f_And(last, get_bag_if_str(arg)) return last + def ltl_Not(a0): a0 = get_bag_if_str(a0) return Formula_rsLTL.f_Not(a0) - + + def ltl_Implies(a0, a1): a0 = get_bag_if_str(a0) a1 = get_bag_if_str(a1) return Formula_rsLTL.f_Implies(a0, a1) + def param_entity(param, entity_name): return ParamConstraint.f_param_ent(param, entity_name) + def param_And(*args): assert len(args) > 1 last = args[0] for arg in args[1:]: last = ParamConstraint.f_And(last, arg) return last - diff --git a/rssmt.py b/rssmt.py index c60ffa1..eaa26b8 100755 --- a/rssmt.py +++ b/rssmt.py @@ -48,14 +48,16 @@ def main(): parser = argparse.ArgumentParser() - parser.add_argument("-v", "--verbose", - help="turn verbosity on", action="store_true") - parser.add_argument("-o", "--optimise", - help="minimise the parametric computation result", action="store_true") - parser.add_argument("-n", "--scaling-parameter", + parser.add_argument("-v", "--verbose", + help="turn verbosity on", action="store_true") + parser.add_argument("-o", "--optimise", + help="minimise the parametric computation result", + action="store_true") + parser.add_argument( + "-n", "--scaling-parameter", help="scaling parameter value (used in some benchmarks)") parser.add_argument("-s", "--special_mode", - help="special mode (used in some benchmarks)") + help="special mode (used in some benchmarks)") args = parser.parse_args() @@ -64,6 +66,7 @@ def main(): ################################################################## + if __name__ == "__main__": try: if profiling: diff --git a/scripts/avg.py b/scripts/avg.py index 2b70c25..1e55980 100755 --- a/scripts/avg.py +++ b/scripts/avg.py @@ -3,8 +3,9 @@ import sys from os import listdir + def proc_files(files): - + fs = [] for f in files: try: @@ -12,14 +13,14 @@ def proc_files(files): except Exception as exct: print("Failed to read file {:s}".format(f)) sys.exit(1) - + done = False result = [] while not done: - + index = "NONE" val_sum = 0 - + for f in fs: try: line = f.__next__() @@ -28,16 +29,16 @@ def proc_files(files): break sline = line.split() - + assert(len(sline) == 2) - + index, value = sline value = float(value) - + val_sum += value - + # print(sline) - + else: val_avg = val_sum/len(fs) result.append("{:s}\t{:f}".format(index, val_avg)) @@ -45,23 +46,24 @@ def proc_files(files): result.append("") return result + def proc_dirs(dirs): - + first_dir = dirs[0] for f in listdir(first_dir): avg_out = proc_files(["{:s}/{:s}".format(d, f) for d in dirs]) - + with open(f, "w") as outfile: outfile.write("\n".join(avg_out)) + def main(): - + proc_dirs(sys.argv[1:]) - + # proc_files(sys.argv[1:]) if __name__ == "__main__": main() - diff --git a/smt/smt_checker_rs.py b/smt/smt_checker_rs.py index ad03bf6..193ff93 100644 --- a/smt/smt_checker_rs.py +++ b/smt/smt_checker_rs.py @@ -7,14 +7,15 @@ from time import time from sys import stdout import resource + class SmtCheckerRS(object): def __init__(self, rsca): rsca.sanity_check() - + self.rs = rsca.rs - self.ca = rsca.ca + self.ca = rsca.ca self.v = [] self.v_ctx = [] @@ -22,7 +23,7 @@ class SmtCheckerRS(object): self.next_level_to_encode = 0 self.solver = Solver() - + self.verification_time = None def prepare_all_variables(self): @@ -31,7 +32,7 @@ class SmtCheckerRS(object): self.prepare_state_variables() self.prepare_context_variables() self.next_level_to_encode += 1 - + def prepare_context_variables(self): """Encodes all the context variables""" @@ -45,44 +46,46 @@ class SmtCheckerRS(object): def prepare_rs_state_variables(self): """Encodes all the state variables of the reaction system""" - + level = self.next_level_to_encode variables = [] for entity in self.rs.background_set: variables.append(Bool("L"+str(level)+"_"+entity)) self.v.append(variables) - + def prepare_context_controller_variables(self): """Encodes all the variables required for controlling context sequences""" level = self.next_level_to_encode - + self.ca_state.append(Int("CA"+str(level)+"_state")) def prepare_state_variables(self): """Encodes all the state variables""" - + self.prepare_rs_state_variables() - self.prepare_context_controller_variables() + self.prepare_context_controller_variables() def enc_rs_init_state(self, level): """Encodes the initial state for the reaction system""" - + rs_init_state_enc = True for v in self.v[level]: - rs_init_state_enc = simplify(And(rs_init_state_enc, Not(v))) # the initial state is empty + # the initial state is empty + rs_init_state_enc = simplify(And(rs_init_state_enc, Not(v))) return rs_init_state_enc def enc_context_controller_init_state(self, level): """Encodes the initial state for controlling context sequences""" - + return self.ca_state[level] == self.ca.get_init_state_id() def enc_init_state(self, level): """Encodes the initial state at the given level""" - init_state_enc = simplify(And(self.enc_rs_init_state(level), self.enc_context_controller_init_state(level))) + init_state_enc = simplify(And(self.enc_rs_init_state( + level), self.enc_context_controller_init_state(level))) return init_state_enc @@ -95,17 +98,18 @@ class SmtCheckerRS(object): return False enc_rct_prod = False - for reactants,inhibitors in rcts_for_prod_entity: + for reactants, inhibitors in rcts_for_prod_entity: enc_reactants = True enc_inhibitors = True for reactant in reactants: - enc_reactants = simplify(And(enc_reactants, - Or(self.v[level][reactant], self.v_ctx[level][reactant]))) + enc_reactants = simplify(And(enc_reactants, Or( + self.v[level][reactant], self.v_ctx[level][reactant]))) for inhibitor in inhibitors: - enc_inhibitors = simplify(And(enc_inhibitors, - Not(Or(self.v[level][inhibitor], self.v_ctx[level][inhibitor])))) + enc_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))) + enc_rct_prod = simplify( + Or(enc_rct_prod, And(enc_reactants, enc_inhibitors))) return enc_rct_prod @@ -114,15 +118,19 @@ class SmtCheckerRS(object): 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]))) + 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): """Encodes the combined transition relation""" - - 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))) def enc_rs_trans(self, level): """Encodes the transition relation""" @@ -133,43 +141,46 @@ class SmtCheckerRS(object): for prod_entity in self.rs.get_reactions_by_product(): unused_entities.remove(prod_entity) - - enc_trans = simplify(And(enc_trans, self.enc_entity_production(level, prod_entity))) + + enc_trans = simplify( + And(enc_trans, self.enc_entity_production(level, prod_entity))) for prod_entity in unused_entities: - enc_trans = simplify(And(enc_trans, Not(self.v[level+1][prod_entity]))) + enc_trans = simplify( + And(enc_trans, Not(self.v[level+1][prod_entity]))) return enc_trans - + def enc_automaton_single_trans(self, level, transition): - - src,ctx,dst = transition - + + src, ctx, dst = transition + src_enc = self.ca_state[level] == src dst_enc = self.ca_state[level+1] == dst - + all_ent = set(range(len(self.rs.background_set))) incl_ctx = ctx excl_ctx = all_ent - incl_ctx - + ctx_enc = True - + for c in incl_ctx: ctx_enc = simplify(And(ctx_enc, self.v_ctx[level][c])) for c in excl_ctx: ctx_enc = simplify(And(ctx_enc, Not(self.v_ctx[level][c]))) - + enc_single_trans = simplify(And(src_enc, ctx_enc, dst_enc)) - + return enc_single_trans - + def enc_automaton_trans(self, level): """Encodes the transition relation for the context automaton""" - + enc_trans = False - for transition in self.ca.transitions: - enc_trans = simplify(Or(enc_trans, self.enc_automaton_single_trans(level, transition))) - + for transition in self.ca.transitions: + enc_trans = simplify( + Or(enc_trans, self.enc_automaton_single_trans(level, transition))) + return enc_trans def enc_state(self, level, state): @@ -201,14 +212,14 @@ class SmtCheckerRS(object): enc = And(enc, self.v[level][entity]) return enc - + def enc_state_with_blocking(self, level, prop): """Encodes the state at the given level with blocking certain concentrations""" - required,blocked = prop + required, blocked = prop enc = True - + required_ids = self.rs.get_state_ids(required) blocked_ids = self.rs.get_state_ids(blocked) @@ -219,7 +230,6 @@ class SmtCheckerRS(object): return simplify(enc) - def decode_witness(self, max_level, print_model=False): m = self.solver.model() @@ -245,11 +255,12 @@ class SmtCheckerRS(object): print(" " + self.rs.get_entity_name(var_id), end="") print(" }") - def check_reachability(self, state, print_witness=True, print_time=True, print_mem=True): + def check_reachability( + self, state, print_witness=True, print_time=True, print_mem=True): """Main testing function""" if not type(state) is tuple: - state = (state,[]) + state = (state, []) if print_time: # start = time() @@ -266,9 +277,9 @@ class SmtCheckerRS(object): self.prepare_all_variables() # reachability test: - print("[i] Adding the reachability test...") + print("[i] Adding the reachability test...") self.solver.push() - self.solver.add(self.enc_state_with_blocking(current_level,state)) + self.solver.add(self.enc_state_with_blocking(current_level, state)) result = self.solver.check() if result == sat: @@ -278,7 +289,7 @@ class SmtCheckerRS(object): break else: self.solver.pop() - + print("[i] Unrolling the transition relation") self.solver.add(self.enc_transition_relation(current_level)) @@ -291,9 +302,13 @@ class SmtCheckerRS(object): self.verification_time = stop-start print() print("[i] Time: " + repr(self.verification_time)) - + if print_mem: - print("[i] Memory: " + repr(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss/(1024*1024)) + " MB") - + print( + "[i] Memory: " + + repr( + resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / + (1024 * 1024)) + " MB") + def get_verification_time(self): return self.verification_time diff --git a/smt/smt_checker_rsc.py b/smt/smt_checker_rsc.py index 24fd729..d98d572 100644 --- a/smt/smt_checker_rsc.py +++ b/smt/smt_checker_rsc.py @@ -14,6 +14,7 @@ from logics import rsLTL_Encoder # def simplify(x): # return x + class SmtCheckerRSC(object): def __init__(self, rsca): @@ -22,38 +23,38 @@ class SmtCheckerRSC(object): if not rsca.is_with_concentrations(): raise RuntimeError("RS and CA with concentrations expected") - + self.rs = rsca.rs self.ca = rsca.ca self.initialise() - + def initialise(self): """Initialises all the variables used by the checker""" - + self.v = [] self.v_ctx = [] self.ca_state = [] self.next_level_to_encode = 0 - + self.loop_position = Int("loop_position") - - self.solver = Solver() #For("QF_FD") - - self.verification_time = None + + self.solver = Solver() # For("QF_FD") + + self.verification_time = None def reset(self): """Reinitialises the state of the checker""" - + self.initialise() - + def prepare_all_variables(self): """Encodes all the variables""" self.prepare_state_variables() self.prepare_context_variables() self.next_level_to_encode += 1 - + def prepare_context_variables(self): """Encodes all the context variables""" @@ -74,24 +75,25 @@ class SmtCheckerRSC(object): for entity in self.rs.background_set: variables.append(Int("L"+str(level)+"_"+entity)) self.v.append(variables) - + self.ca_state.append(Int("CA"+str(level)+"_state")) def enc_concentration_levels_assertion(self, level): """Encodes assertions that (some) variables need to be >0 - + We do not need to actually control all the variables, only those that can possibly go below 0. """ - + enc_nz = True - + for e_i in range(len(self.rs.background_set)): v = self.v[level][e_i] v_ctx = self.v_ctx[level][e_i] e_max = self.rs.get_max_concentration_level(e_i) - enc_nz = simplify(And(enc_nz, v >= 0, v_ctx >= 0, v <= e_max, v_ctx <= e_max)) - + enc_nz = simplify( + And(enc_nz, v >= 0, v_ctx >= 0, v <= e_max, v_ctx <= e_max)) + return enc_nz def enc_init_state(self, level): @@ -100,10 +102,11 @@ class SmtCheckerRSC(object): rs_init_state_enc = True for v in self.v[level]: - rs_init_state_enc = simplify(And(rs_init_state_enc, v == 0)) # the initial concentration levels are zeroed + # the initial concentration levels are zeroed + rs_init_state_enc = simplify(And(rs_init_state_enc, v == 0)) ca_init_state_enc = self.ca_state[level] == self.ca.get_init_state_id() - + init_state_enc = simplify(And(rs_init_state_enc, ca_init_state_enc)) return init_state_enc @@ -113,112 +116,155 @@ class SmtCheckerRSC(object): rcts_for_prod_entity = [] if prod_entity in self.rs.get_reactions_by_product(): - rcts_for_prod_entity = self.rs.get_reactions_by_product()[prod_entity] + rcts_for_prod_entity = self.rs.get_reactions_by_product()[ + prod_entity] meta_reactions = [] if prod_entity in self.rs.meta_reactions: meta_reactions = self.rs.meta_reactions[prod_entity] - permanency_inhibition = None + permanency_inhibition = None if prod_entity in self.rs.permanent_entities: permanency_inhibition = self.rs.permanent_entities[prod_entity] if rcts_for_prod_entity == [] and meta_reactions == []: - return simplify(self.v[level+1][prod_entity] == 0) # this should never happen - + # this should never happen + return simplify(self.v[level+1][prod_entity] == 0) + enc_enabledness = False # ----------- ordinary reactions -------------------------------------------- enc_rct_prod = False - + enc_ordinary_reactions_enabledness = False - - for reactants,inhibitors,products in rcts_for_prod_entity: - enc_reactants = True - for reactant,concentration in reactants: - enc_reactants = simplify(And(enc_reactants, - Or(self.v[level][reactant] >= concentration, self.v_ctx[level][reactant] >= concentration))) + for reactants, inhibitors, products in rcts_for_prod_entity: + + enc_reactants = True + for reactant, concentration in reactants: + enc_reactants = simplify(And(enc_reactants, Or( + self.v[level][reactant] >= concentration, self.v_ctx[level][reactant] >= concentration))) + + enc_inhibitors = True + for inhibitor, concentration in inhibitors: + enc_inhibitors = simplify(And(enc_inhibitors, And( + self.v[level][inhibitor] < concentration, self.v_ctx[level][inhibitor] < concentration))) - enc_inhibitors = True - for inhibitor,concentration in inhibitors: - enc_inhibitors = simplify(And(enc_inhibitors, - And(self.v[level][inhibitor] < concentration, self.v_ctx[level][inhibitor] < concentration))) - enc_rct_enabled = And(enc_reactants, enc_inhibitors) enc_products = self.v[level+1][products[0][0]] == products[0][1] - enc_rct_prod = simplify(If(enc_rct_enabled, enc_products, enc_rct_prod)) + enc_rct_prod = simplify( + If(enc_rct_enabled, enc_products, enc_rct_prod)) enc_enabledness = simplify(Or(enc_enabledness, enc_rct_enabled)) - - enc_ordinary_reactions_enabledness = simplify(Or(enc_ordinary_reactions_enabledness,enc_rct_enabled)) - + + enc_ordinary_reactions_enabledness = simplify( + Or(enc_ordinary_reactions_enabledness, enc_rct_enabled)) + # -------- meta reactions --------------------------------------------------- - - for r_type,command_entity,reactants,inhibitors in meta_reactions: + + for r_type, command_entity, reactants, inhibitors in meta_reactions: # command entity is e.g. 'inc' for incrementation operation # (inc,W) gives us the value W by which the given entity's value should be incremented - enc_reactants = True - enc_inhibitors = True - - for reactant,concentration in reactants: - enc_reactants = simplify(And(enc_reactants, - Or(self.v[level][reactant] >= concentration, self.v_ctx[level][reactant] >= concentration))) + enc_reactants = True + enc_inhibitors = True + + for reactant, concentration in reactants: + enc_reactants = simplify(And(enc_reactants, Or( + self.v[level][reactant] >= concentration, self.v_ctx[level][reactant] >= concentration))) # command entity needs to be present (with concentration level > 0) in order to perform the operation - enc_reactants = simplify(And(enc_reactants, - Or(self.v[level][command_entity] > 0, self.v_ctx[level][command_entity] > 0))) + enc_reactants = simplify(And(enc_reactants, Or( + self.v[level][command_entity] > 0, self.v_ctx[level][command_entity] > 0))) + + for inhibitor, concentration in inhibitors: + enc_inhibitors = simplify(And(enc_inhibitors, And( + self.v[level][inhibitor] < concentration, self.v_ctx[level][inhibitor] < concentration))) - for inhibitor,concentration in inhibitors: - enc_inhibitors = simplify(And(enc_inhibitors, - And(self.v[level][inhibitor] < concentration, self.v_ctx[level][inhibitor] < concentration))) - if r_type == "inc": - value_after_inc = If(self.v[level][prod_entity]>self.v_ctx[level][prod_entity],self.v[level][prod_entity],self.v_ctx[level][prod_entity]) + \ - If(self.v[level][command_entity]>self.v_ctx[level][command_entity],self.v[level][command_entity],self.v_ctx[level][command_entity]) + value_after_inc = If( + self.v[level][prod_entity] > self.v_ctx[level] + [prod_entity], + self.v[level][prod_entity], + self.v_ctx[level][prod_entity]) + If( + self.v[level][command_entity] > self. + v_ctx[level][command_entity], + self.v[level][command_entity], + self.v_ctx[level][command_entity]) enc_products = self.v[level+1][prod_entity] == value_after_inc elif r_type == "dec": - value_after_dec = simplify(If(self.v[level][prod_entity]>self.v_ctx[level][prod_entity],self.v[level][prod_entity],self.v_ctx[level][prod_entity]) - \ - If(self.v[level][command_entity]>self.v_ctx[level][command_entity],self.v[level][command_entity],self.v_ctx[level][command_entity])) - enc_products = self.v[level+1][prod_entity] == If(value_after_dec < 0, 0, value_after_dec) + value_after_dec = simplify( + If( + self.v[level][prod_entity] > + self.v_ctx[level] + [prod_entity], + self.v[level][prod_entity], + self.v_ctx[level] + [prod_entity]) - + If( + self.v[level] + [command_entity] > self. + v_ctx[level][command_entity], + self.v[level] + [command_entity], + self.v_ctx[level] + [command_entity])) + enc_products = self.v[level+1][prod_entity] == If( + value_after_dec < 0, 0, value_after_dec) else: - raise RuntimeError("Unknown meta-reaction type: " + repr(r_type)) + raise RuntimeError( + "Unknown meta-reaction type: " + repr(r_type)) - enc_meta_reaction_enabledness = And(enc_reactants, enc_inhibitors, Not(enc_ordinary_reactions_enabledness)) - enc_enabledness = simplify(Or(enc_enabledness, enc_meta_reaction_enabledness)) - enc_rct_prod = simplify(Or(enc_rct_prod, And(enc_meta_reaction_enabledness, enc_products))) + enc_meta_reaction_enabledness = And( + enc_reactants, enc_inhibitors, + Not(enc_ordinary_reactions_enabledness)) + enc_enabledness = simplify( + Or(enc_enabledness, enc_meta_reaction_enabledness)) + enc_rct_prod = simplify(Or(enc_rct_prod, And( + enc_meta_reaction_enabledness, enc_products))) # ----------------------------------------------------------------------------- - + if not permanency_inhibition == None: - - enc_reactants = Or(self.v[level][prod_entity] >= concentration, self.v_ctx[level][prod_entity] >= concentration) + + enc_reactants = Or(self.v[level][prod_entity] >= concentration, + self.v_ctx[level][prod_entity] >= concentration) enc_inhibitors = True - for inhibitor,concentration in permanency_inhibition: - enc_inhibitors = simplify(And(enc_inhibitors, - And(self.v[level][inhibitor] < concentration, self.v_ctx[level][inhibitor] < concentration))) - enc_products = simplify(self.v[level+1][prod_entity] == \ - If(self.v[level][prod_entity] > self.v_ctx[level][prod_entity],self.v[level][prod_entity],self.v_ctx[level][prod_entity])) + for inhibitor, concentration in permanency_inhibition: + enc_inhibitors = simplify(And(enc_inhibitors, And( + self.v[level][inhibitor] < concentration, self.v_ctx[level][inhibitor] < concentration))) + enc_products = simplify( + self.v[level + 1][prod_entity] == + If( + self.v[level][prod_entity] > self.v_ctx[level] + [prod_entity], + self.v[level][prod_entity], + self.v_ctx[level][prod_entity])) - enc_permanency_enabledness = And(enc_reactants, enc_inhibitors, Not(enc_ordinary_reactions_enabledness)) - enc_enabledness = simplify(Or(enc_enabledness, enc_permanency_enabledness)) + enc_permanency_enabledness = And( + enc_reactants, enc_inhibitors, + Not(enc_ordinary_reactions_enabledness)) + enc_enabledness = simplify( + Or(enc_enabledness, enc_permanency_enabledness)) enc_permanency = And(enc_permanency_enabledness, enc_products) enc_rct_prod = simplify(Or(enc_rct_prod, enc_permanency)) # ----------------------------------------------------------------------------- - - enc_when_to_produce_zero_conc = simplify(And(Not(enc_enabledness), self.v[level+1][prod_entity] == 0)) + + enc_when_to_produce_zero_conc = simplify( + And(Not(enc_enabledness), self.v[level+1][prod_entity] == 0)) enc_rct_prod = Or(enc_rct_prod, enc_when_to_produce_zero_conc) return enc_rct_prod def enc_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))) def enc_rs_trans(self, level): """Encodes the transition relation""" @@ -232,38 +278,40 @@ class SmtCheckerRSC(object): for prod_entity in chain(reactions, meta_reactions): unused_entities.discard(prod_entity) - enc_trans = simplify(And(enc_trans, self.enc_produced_concentration(level, prod_entity))) + enc_trans = simplify( + And(enc_trans, self.enc_produced_concentration(level, prod_entity))) for prod_entity in unused_entities: - enc_trans = simplify(And(enc_trans, self.v[level+1][prod_entity] == 0)) + enc_trans = simplify( + And(enc_trans, self.v[level+1][prod_entity] == 0)) return enc_trans - + def enc_automaton_trans(self, level): """Encodes the transition relation for the context automaton""" - + enc_trans = False - - for src,ctx,dst in self.ca.transitions: - src_enc = self.ca_state[level] == src - dst_enc = self.ca_state[level+1] == dst - - all_ent = set(range(len(self.rs.background_set))) - - incl_ctx = set([e for e,c in ctx]) - excl_ctx = all_ent - incl_ctx - - ctx_enc = True - - for e,c in ctx: - ctx_enc = simplify(And(ctx_enc, self.v_ctx[level][e] == c)) - - for e in excl_ctx: - ctx_enc = simplify(And(ctx_enc, self.v_ctx[level][e] == 0)) - - cur_trans = simplify(And(src_enc, ctx_enc, dst_enc)) - enc_trans = simplify(Or(enc_trans, cur_trans)) - + + for src, ctx, dst in self.ca.transitions: + src_enc = self.ca_state[level] == src + dst_enc = self.ca_state[level+1] == dst + + all_ent = set(range(len(self.rs.background_set))) + + incl_ctx = set([e for e, c in ctx]) + excl_ctx = all_ent - incl_ctx + + ctx_enc = True + + for e, c in ctx: + ctx_enc = simplify(And(ctx_enc, self.v_ctx[level][e] == c)) + + for e in excl_ctx: + ctx_enc = simplify(And(ctx_enc, self.v_ctx[level][e] == 0)) + + cur_trans = simplify(And(src_enc, ctx_enc, dst_enc)) + enc_trans = simplify(Or(enc_trans, cur_trans)) + return enc_trans def enc_exact_state(self, level, state): @@ -275,7 +323,7 @@ class SmtCheckerRSC(object): """Encodes the state at the given level with the minimal required concentration levels""" enc = True - for ent,conc in state: + for ent, conc in state: e_id = self.rs.get_entity_id(ent) enc = And(enc, self.v[level][e_id] >= conc) @@ -284,16 +332,16 @@ class SmtCheckerRSC(object): def enc_state_with_blocking(self, level, prop): """Encodes the state at the given level with blocking certain concentrations""" - required,blocked = prop + required, blocked = prop enc = True - for ent,conc in required: - e_id = self.rs.get_entity_id(ent) - enc = And(enc, self.v[level][e_id] >= conc) + for ent, conc in required: + e_id = self.rs.get_entity_id(ent) + enc = And(enc, self.v[level][e_id] >= conc) - for ent,conc in blocked: - e_id = self.rs.get_entity_id(ent) - enc = And(enc, self.v[level][e_id] < conc) + for ent, conc in blocked: + e_id = self.rs.get_entity_id(ent) + enc = And(enc, self.v[level][e_id] < conc) return simplify(enc) @@ -312,9 +360,12 @@ class SmtCheckerRSC(object): for var_id in range(len(self.v[level])): var_rep = repr(m[self.v[level][var_id]]) if not var_rep.isdigit(): - raise RuntimeError("unexpected: representation is not a positive integer") + raise RuntimeError( + "unexpected: representation is not a positive integer") if int(var_rep) > 0: - print(" " + self.rs.get_entity_name(var_id) + "=" + var_rep, end="") + print( + " " + self.rs.get_entity_name(var_id) + "=" + var_rep, + end="") # print(" " + repr(m[self.v[level][var_id]]), end="") print(" }") @@ -324,19 +375,25 @@ class SmtCheckerRSC(object): for var_id in range(len(self.v[level])): var_rep = repr(m[self.v_ctx[level][var_id]]) if not var_rep.isdigit(): - raise RuntimeError("unexpected: representation is not a positive integer") + raise RuntimeError( + "unexpected: representation is not a positive integer") if int(var_rep) > 0: - print(" " + self.rs.get_entity_name(var_id) + "=" + var_rep, end="") + print( + " " + self.rs.get_entity_name(var_id) + "=" + var_rep, + end="") print(" }") - - def check_rsltl(self, formula, print_witness=True, print_time=True, print_mem=True, max_level=None): + + def check_rsltl( + self, formula, print_witness=True, print_time=True, print_mem=True, + max_level=None): """Bounded Model Checking for rsLTL properties""" - + self.reset() - - print("[" + colour_str(C_BOLD, "i") + "] Running rsLTL bounded model checking") + + print("[" + colour_str(C_BOLD, "i") + + "] Running rsLTL bounded model checking") print("[" + colour_str(C_BOLD, "i") + "] Formula: " + str(formula)) - + if print_time: # start = time() start = resource.getrusage(resource.RUSAGE_SELF).ru_utime @@ -346,42 +403,53 @@ class SmtCheckerRSC(object): self.current_level = 0 self.prepare_all_variables() - + self.solver.add(self.enc_concentration_levels_assertion(0)) - + encoder = rsLTL_Encoder(self) encoder.load_variables( var_rs=self.v, var_ctx=self.v_ctx, var_loop_pos=self.loop_position) - + while True: self.prepare_all_variables() - self.solver.add(self.enc_concentration_levels_assertion(self.current_level+1)) - - print("\n{:-^70}".format("[ Working at level=" + str(self.current_level) + " ]")) + self.solver.add( + self.enc_concentration_levels_assertion( + self.current_level + 1)) + + print( + "\n{:-^70}".format("[ Working at level=" + str(self.current_level) + " ]")) stdout.flush() # reachability test: self.solver.push() - - print("[" + colour_str(C_BOLD, "i") + "] Generating the formula encoding...") - + + print("[" + colour_str(C_BOLD, "i") + + "] Generating the formula encoding...") + f = encoder.get_encoding(formula, self.current_level) ncalls = encoder.get_ncalls() - - print("[" + colour_str(C_BOLD, "i") + "] Cache hits: " + str(encoder.get_cache_hits()) + ", encode calls: " + str(ncalls[0]) + " (approx: " + str(ncalls[1]) + ")") - print("[" + colour_str(C_BOLD, "i") + "] Adding the formula to the solver...") - + + print("[" + colour_str(C_BOLD, "i") + "] Cache hits: " + + str(encoder.get_cache_hits()) + ", encode calls: " + + str(ncalls[0]) + " (approx: " + str(ncalls[1]) + ")") + print("[" + colour_str(C_BOLD, "i") + + "] Adding the formula to the solver...") + encoder.flush_cache() self.solver.add(f) - - print("[" + colour_str(C_BOLD, "i") + "] Adding the loops encoding...") + + print("[" + colour_str(C_BOLD, "i") + + "] Adding the loops encoding...") self.solver.add(self.get_loop_encodings()) - + result = self.solver.check() if result == sat: - print("[" + colour_str(C_BOLD, "+") + "] " + colour_str(C_GREEN, "SAT at level=" + str(self.current_level))) + print( + "[" + colour_str(C_BOLD, "+") + "] " + + colour_str( + C_GREEN, "SAT at level=" + str(self.current_level))) if print_witness: print("\n{:=^70}".format("[ WITNESS ]")) self.decode_witness(self.current_level) @@ -389,10 +457,12 @@ class SmtCheckerRSC(object): else: self.solver.pop() - print("[" + colour_str(C_BOLD, "i") + "] Unrolling the transition relation") + print("[" + colour_str(C_BOLD, "i") + + "] Unrolling the transition relation") self.solver.add(self.enc_transition_relation(self.current_level)) - print("{:->70}".format("[ level=" + str(self.current_level) + " done ]")) + print( + "{:->70}".format("[ level=" + str(self.current_level) + " done ]")) self.current_level += 1 if not max_level is None and self.current_level > max_level: @@ -404,26 +474,33 @@ class SmtCheckerRSC(object): stop = resource.getrusage(resource.RUSAGE_SELF).ru_utime self.verification_time = stop-start print() - print("\n[i] {: >60}".format(" Time: " + repr(self.verification_time) + " s")) - + print( + "\n[i] {: >60}".format( + " Time: " + repr(self.verification_time) + " s")) + if print_mem: - print("[i] {: >60}".format(" Memory: " + repr(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss/(1024*1024)) + " MB")) - + print( + "[i] {: >60}".format( + " Memory: " + + repr( + resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / + (1024 * 1024)) + " MB")) + def dummy_unroll(self, levels): """Unrolls the variables for testing purposes""" - self.current_level = -1 + self.current_level = -1 for i in range(levels+1): self.prepare_all_variables() self.current_level += 1 print(C_MARK_INFO + " Dummy Unrolling done.") - + def state_equality(self, level_A, level_B): """Encodes equality of two states at two different levels""" - + eq_enc = True - + for e_i in range(len(self.rs.background_set)): e_i_equality = self.v[level_A][e_i] == self.v[level_B][e_i] eq_enc = simplify(And(eq_enc, e_i_equality)) @@ -432,27 +509,28 @@ class SmtCheckerRSC(object): eq_enc = simplify(And(eq_enc, eq_enc_ctxaut)) return eq_enc - + def get_loop_encodings(self): - + k = self.current_level loop_var = self.loop_position - + loop_enc = True - + """ (loop_var == i) means that there is a loop taking back to the state (i-1) Therefore, the encoding starts at 1, not at 0. """ - - for i in range(1,k+1): - loop_enc = simplify(And(loop_enc, Implies( loop_var == i, self.state_equality(i-1, k) ))) - + + for i in range(1, k+1): + loop_enc = simplify(And(loop_enc, Implies( + loop_var == i, self.state_equality(i-1, k)))) + return loop_enc - - def check_reachability(self, state, print_witness=True, - print_time=True, print_mem=True, max_level=1000): + + def check_reachability(self, state, print_witness=True, + print_time=True, print_mem=True, max_level=1000): """Main testing function""" self.reset() @@ -466,25 +544,33 @@ class SmtCheckerRSC(object): self.current_level = 0 self.prepare_all_variables() - + self.solver.add(self.enc_concentration_levels_assertion(0)) - + while True: self.prepare_all_variables() - self.solver.add(self.enc_concentration_levels_assertion(self.current_level+1)) - - print("\n{:-^70}".format("[ Working at level=" + str(self.current_level) + " ]")) + self.solver.add( + self.enc_concentration_levels_assertion( + self.current_level + 1)) + + print( + "\n{:-^70}".format("[ Working at level=" + str(self.current_level) + " ]")) stdout.flush() # reachability test: - print("[" + colour_str(C_BOLD, "i") + "] Adding the reachability test...") + print("[" + colour_str(C_BOLD, "i") + + "] Adding the reachability test...") self.solver.push() - self.solver.add(self.enc_state_with_blocking(self.current_level,state)) - + self.solver.add(self.enc_state_with_blocking( + self.current_level, state)) + result = self.solver.check() if result == sat: - print("[" + colour_str(C_BOLD, "+") + "] " + colour_str(C_GREEN, "SAT at level=" + str(self.current_level))) + print( + "[" + colour_str(C_BOLD, "+") + "] " + + colour_str( + C_GREEN, "SAT at level=" + str(self.current_level))) if print_witness: print("\n{:=^70}".format("[ WITNESS ]")) self.decode_witness(self.current_level) @@ -492,10 +578,12 @@ class SmtCheckerRSC(object): else: self.solver.pop() - print("[" + colour_str(C_BOLD, "i") + "] Unrolling the transition relation") + print("[" + colour_str(C_BOLD, "i") + + "] Unrolling the transition relation") self.solver.add(self.enc_transition_relation(self.current_level)) - print("{:->70}".format("[ level=" + str(self.current_level) + " done ]")) + print( + "{:->70}".format("[ level=" + str(self.current_level) + " done ]")) self.current_level += 1 if self.current_level > max_level: @@ -507,16 +595,23 @@ class SmtCheckerRSC(object): stop = resource.getrusage(resource.RUSAGE_SELF).ru_utime self.verification_time = stop-start print() - print("\n[i] {: >60}".format(" Time: " + repr(self.verification_time) + " s")) - + print( + "\n[i] {: >60}".format( + " Time: " + repr(self.verification_time) + " s")) + if print_mem: - print("[i] {: >60}".format(" Memory: " + repr(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss/(1024*1024)) + " MB")) - + print( + "[i] {: >60}".format( + " Memory: " + + repr( + resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / + (1024 * 1024)) + " MB")) + def get_verification_time(self): return self.verification_time - - def show_encoding(self, state, print_witness=True, - print_time=False, print_mem=False, max_level=100): + + def show_encoding(self, state, print_witness=True, + print_time=False, print_mem=False, max_level=100): """Encoding debug function""" self.reset() @@ -528,25 +623,29 @@ class SmtCheckerRSC(object): self.current_level = 0 self.prepare_all_variables() - + while True: self.prepare_all_variables() - print("-----[ Working at level=" + str(self.current_level) + " ]-----") + print( + "-----[ Working at level=" + str(self.current_level) + " ]-----") stdout.flush() # reachability test: - print("[i] Adding the reachability test...") + print("[i] Adding the reachability test...") self.solver.push() - s = self.enc_min_state(self.current_level,state) + s = self.enc_min_state(self.current_level, state) print("Test: ", s) - + self.solver.add(s) - + result = self.solver.check() if result == sat: - print("\n[+] " + colour_str(C_RED, "SAT at level=" + str(self.current_level))) + print( + "\n[+] " + + colour_str( + C_RED, "SAT at level=" + str(self.current_level))) if print_witness: self.decode_witness(self.current_level) break @@ -565,8 +664,7 @@ class SmtCheckerRSC(object): print("Stopping at level=" + str(max_level)) break else: - x=input("Next level? ") - x=x.lower() + x = input("Next level? ") + x = x.lower() if not (x == "y" or x == "yes"): break - diff --git a/smt/smt_checker_rsc_param.py b/smt/smt_checker_rsc_param.py index 212f953..0a6df1d 100644 --- a/smt/smt_checker_rsc_param.py +++ b/smt/smt_checker_rsc_param.py @@ -17,46 +17,49 @@ from rs.reaction_system_with_concentrations_param import ParameterObj, is_param # def simplify(x): # return x + def z3_max(a, b): return If(a > b, a, b) + class SmtCheckerRSCParam(object): def __init__(self, rsca, optimise=False): - + rsca.sanity_check() if not rsca.is_concentr_and_param_compatible(): - raise RuntimeError("RS and CA with concentrations (and parameters) expected") - + raise RuntimeError( + "RS and CA with concentrations (and parameters) expected") + self.rs = rsca.rs self.ca = rsca.ca - + self.optimise = optimise self.initialise() - + def initialise(self): """Initialises all the variables used by the checker""" - ### "Currently" used variables (loaded from self.path_v...) + # "Currently" used variables (loaded from self.path_v...) self.v = None self.v_ctx = None self.ca_state = None - + # intermediate products: self.v_improd = None self.v_improd_for_entities = None - - ### Per-path variables + + # Per-path variables self.path_v = dict() self.path_v_ctx = dict() self.path_ca_state = dict() - + # intermediate products: self.path_v_improd = dict() self.path_v_improd_for_entities = dict() - + # parameters: self.v_param = dict() @@ -69,46 +72,48 @@ class SmtCheckerRSCParam(object): # * 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 self.path_loop_position = dict() - + if self.optimise: self.solver = Optimize() else: - self.solver = Solver() #For("QF_FD") + self.solver = Solver() # For("QF_FD") self.verification_time = None - + self.prepare_param_variables() def reset(self): """Reinitialises the state of the checker""" - + self.initialise() - + def prepare_all_variables(self, num_of_paths): for path_idx in range(num_of_paths): self.prepare_all_path_variables(path_idx) self.next_level_to_encode += 1 - + def prepare_all_path_variables(self, path_idx): """Prepares the variables for a given path index""" - print_info("Preparing variables for path={:d} (level={:d})".format(path_idx, self.next_level_to_encode)) - + print_info("Preparing variables for path={:d} (level={:d})".format( + path_idx, self.next_level_to_encode)) + self.prepare_state_variables(path_idx) self.prepare_context_variables(path_idx) self.prepare_intermediate_product_variables(path_idx) self.prepare_loop_position_variables(path_idx) - + def prepare_loop_position_variables(self, path_idx): """Prepares the variables for loop positions""" - - self.path_loop_position[path_idx] = Int("p{:d}_loop_pos".format(path_idx)) - + + self.path_loop_position[path_idx] = Int( + "p{:d}_loop_pos".format(path_idx)) + def prepare_context_variables(self, path_idx): """Prepares all the context variables""" @@ -116,7 +121,7 @@ class SmtCheckerRSCParam(object): self.path_v_ctx.setdefault(path_idx, []) assert len(self.path_v_ctx[path_idx]) == level - + variables = [] for entity in self.rs.background_set: new_var = Int("p{:d}C{:d}_{:s}".format(path_idx, level, entity)) @@ -128,7 +133,7 @@ class SmtCheckerRSCParam(object): """Prepares all the state variables""" level = self.next_level_to_encode - + # RS vars self.path_v.setdefault(path_idx, []) assert len(self.path_v[path_idx]) == level @@ -138,11 +143,11 @@ class SmtCheckerRSCParam(object): new_var = Int("p{:d}L{:d}_{:s}".format(path_idx, level, entity)) variables.append(new_var) self.path_v[path_idx].append(variables) - + # Context automaton states: self.path_ca_state.setdefault(path_idx, []) assert len(self.path_ca_state[path_idx]) == level - + ca_state_var = Int("p{:d}CA{:d}_state".format(path_idx, level)) self.path_ca_state[path_idx].append(ca_state_var) @@ -160,7 +165,7 @@ class SmtCheckerRSCParam(object): self.path_v_improd.setdefault(path_idx, []) self.path_v_improd_for_entities.setdefault(path_idx, []) - + if level < 1: # # If we are at level==0, we add a dummy "level" @@ -181,48 +186,52 @@ class SmtCheckerRSCParam(object): reaction_id = self.rs.reactions.index(reaction) entities_dict = dict() - + if is_param(products): for entity in self.rs.set_of_bgset_ids: entity_name = self.rs.get_entity_name(entity) - new_var = Int("p{:d}L{:d}_ImProd_r{:d}_{:s}".format(path_idx, level, reaction_id, entity_name)) + new_var = Int("p{:d}L{:d}_ImProd_r{:d}_{:s}".format( + path_idx, level, reaction_id, entity_name)) entities_dict[entity] = new_var all_entities_dict.setdefault(entity, []) all_entities_dict[entity].append(new_var) - + else: - + for entity, conc in products: entity_name = self.rs.get_entity_name(entity) - new_var = Int("p{:d}L{:d}_ImProd_r{:d}_{:s}".format(path_idx, level, reaction_id, entity_name)) + new_var = Int("p{:d}L{:d}_ImProd_r{:d}_{:s}".format( + path_idx, level, reaction_id, entity_name)) entities_dict[entity] = new_var all_entities_dict.setdefault(entity, []) all_entities_dict[entity].append(new_var) reactions_dict[reaction_id] = entities_dict - + self.path_v_improd[path_idx].append(reactions_dict) self.path_v_improd_for_entities[path_idx].append(all_entities_dict) - + def prepare_param_variables(self): """ Prepares variables for parameters - + A parameter (it's valuation) is a subset of the background set, therefore we need separate variables for each element of the background set. """ - + for param_name in self.rs.parameters.keys(): # we start collecting bg-related vars for the given param vars_for_param = [] for entity in self.rs.ordered_list_of_bgset_ids: - new_var = Int("Pm{:s}_{:s}".format(param_name, self.rs.get_entity_name(entity))) + new_var = Int( + "Pm{:s}_{:s}".format( + param_name, self.rs.get_entity_name(entity))) vars_for_param.append(new_var) self.v_param[param_name] = vars_for_param @@ -233,14 +242,14 @@ class SmtCheckerRSCParam(object): """ entity_id = self.rs.get_entity_id(entity_name) return self.v_param[param_name][entity_id] - + def load_varset_for_path(self, path_idx): """ Loads the the variables for the path with path_idx """ self.v = self.path_v[path_idx] self.v_ctx = self.path_v_ctx[path_idx] - self.ca_state = self.path_ca_state[path_idx] + self.ca_state = self.path_ca_state[path_idx] self.v_improd = self.path_v_improd[path_idx] self.v_improd_for_entities = self.path_v_improd_for_entities[path_idx] self.loop_position = self.path_loop_position[path_idx] @@ -249,58 +258,65 @@ class SmtCheckerRSCParam(object): """ Assertions for the parameter variables """ - + if len(self.v_param) == 0: return True - + enc_param_gz = True enc_non_empty = True - + for param_vars in self.v_param.values(): - + enc_param_at_least_one = False for pvar in param_vars: - + # TODO: fixed upper limit: 100 (have a per-param setting for that) - enc_param_gz = simplify(And(enc_param_gz, pvar >= 0, pvar < 100)) - enc_param_at_least_one = simplify(Or(enc_param_at_least_one, pvar > 0)) - - enc_non_empty = simplify(And(enc_non_empty, enc_param_at_least_one)) - + enc_param_gz = simplify( + And(enc_param_gz, pvar >= 0, pvar < 100)) + enc_param_at_least_one = simplify( + Or(enc_param_at_least_one, pvar > 0)) + + enc_non_empty = simplify( + And(enc_non_empty, enc_param_at_least_one)) + return simplify(And(enc_param_gz, enc_non_empty)) - + def assert_param_optimisation(self): for param_vars in self.v_param.values(): for pvar in param_vars: # self.solver.add_soft(pvar == 0) self.solver.minimize(pvar) - + def enc_concentration_levels_assertion(self, level, path_idx): """ Encodes assertions that (some) variables need to be >=0 - + We do not need to actually control all the variables, only those that can possibly go below 0. """ - print_info("Concentration level assertions for path={:d} (level={:d})".format(path_idx, level)) - + print_info("Concentration level assertions for path={:d} (level={:d})".format( + path_idx, level)) + enc_gz = True for e_i in self.rs.set_of_bgset_ids: var = self.path_v[path_idx][level][e_i] var_ctx = self.path_v_ctx[path_idx][level][e_i] e_max = self.rs.get_max_concentration_level(e_i) - enc_gz = simplify(And(enc_gz, var >= 0, var_ctx >= 0, var <= e_max, var_ctx <= e_max)) - - vars_per_reaction = self.path_v_improd_for_entities[path_idx][level + 1] + enc_gz = simplify(And(enc_gz, var >= 0, var_ctx >= + 0, var <= e_max, var_ctx <= e_max)) + + vars_per_reaction = self.path_v_improd_for_entities[path_idx][ + level + 1] if e_i in vars_per_reaction: for var_improd in vars_per_reaction[e_i]: - enc_gz = simplify(And(enc_gz, var_improd >= 0, var_improd <= e_max)) - + enc_gz = simplify( + And(enc_gz, var_improd >= 0, var_improd <= e_max)) + return enc_gz - + def enc_init_state(self, level, path_idx): """Encodes the initial state at the given level""" @@ -310,7 +326,8 @@ class SmtCheckerRSCParam(object): # the initial concentration levels are zeroed rs_init_state_enc = simplify(And(rs_init_state_enc, v == 0)) - ca_init_state_enc = self.path_ca_state[path_idx][level] == self.ca.get_init_state_id() + ca_init_state_enc = self.path_ca_state[path_idx][level] == self.ca.get_init_state_id( + ) init_state_enc = simplify(And(rs_init_state_enc, ca_init_state_enc)) @@ -329,35 +346,42 @@ class SmtCheckerRSCParam(object): for reactants, inhibitors, products in self.rs.reactions: if is_param(reactants) or is_param(inhibitors): - + # 1. R and I if is_param(reactants) and is_param(inhibitors): rct_param_name = reactants.name inh_param_name = inhibitors.name for entity in self.rs.set_of_bgset_ids: - rct_inh_constr = And(rct_inh_constr, - Implies(self.v_param[inh_param_name][entity] > 0, - self.v_param[rct_param_name][entity] < self.v_param[inh_param_name][entity])) + rct_inh_constr = And(rct_inh_constr, + Implies( + self.v_param + [inh_param_name][entity] > 0, + self.v_param + [rct_param_name][entity] < + self.v_param + [inh_param_name][entity])) elif (not is_param(reactants)) and is_param(inhibitors): inh_param_name = inhibitors.name - + for entity, conc in reactants: assert conc > 0, "Unexpected concentration level!" - rct_inh_constr = And(rct_inh_constr, - Implies(self.v_param[inh_param_name][entity] > 0, - conc < self.v_param[inh_param_name][entity])) - + rct_inh_constr = And( + rct_inh_constr, + Implies( + self.v_param[inh_param_name][entity] > 0, conc < + self.v_param[inh_param_name][entity])) + elif is_param(reactants) and (not is_param(inhibitors)): rct_param_name = reactants.name - + for entity, conc in inhibitors: assert conc > 0, "Unexpected concentration level!" - rct_inh_constr = And(rct_inh_constr, self.v_param[rct_param_name][entity] < conc) + rct_inh_constr = And( + rct_inh_constr, self.v_param[rct_param_name][entity] < conc) return rct_inh_constr - def enc_single_reaction(self, level, path_idx, reaction): """ @@ -383,12 +407,12 @@ class SmtCheckerRSCParam(object): for entity in self.rs.set_of_bgset_ids: enc_reactants = And(enc_reactants, Or( self.v_param[param_name][entity] == 0, - self.path_v[path_idx][level][entity] >= self.v_param[param_name][entity], + self.path_v[path_idx][level][entity] >= self.v_param[param_name][entity], self.path_v_ctx[path_idx][level][entity] >= self.v_param[param_name][entity])) else: for entity, conc in reactants: enc_reactants = And(enc_reactants, Or( - self.path_v[path_idx][level][entity] >= conc, + self.path_v[path_idx][level][entity] >= conc, self.path_v_ctx[path_idx][level][entity] >= conc)) # ** INHIBITORS ****************************************** @@ -399,12 +423,12 @@ class SmtCheckerRSCParam(object): enc_inhibitors = And(enc_inhibitors, Or( self.v_param[param_name][entity] == 0, And( - self.path_v[path_idx][level][entity] < self.v_param[param_name][entity], + self.path_v[path_idx][level][entity] < self.v_param[param_name][entity], self.path_v_ctx[path_idx][level][entity] < self.v_param[param_name][entity]))) else: for entity, conc in inhibitors: enc_inhibitors = And(enc_inhibitors, And( - self.path_v[path_idx][level][entity] < conc, + self.path_v[path_idx][level][entity] < conc, self.path_v_ctx[path_idx][level][entity] < conc)) # ** PRODUCTS ******************************************* @@ -412,35 +436,40 @@ class SmtCheckerRSCParam(object): if is_param(products): param_name = products.name for entity in self.rs.set_of_bgset_ids: - enc_products = simplify(And(enc_products, - self.path_v_improd[path_idx][level + 1][reaction_id][entity] == self.v_param[param_name][entity])) + enc_products = simplify( + And( + enc_products, self. + path_v_improd[path_idx] + [level + 1][reaction_id] + [entity] == self.v_param + [param_name][entity])) else: for entity, conc in products: - enc_products = simplify(And(enc_products, - self.path_v_improd[path_idx][level + 1][reaction_id][entity] == conc)) + enc_products = simplify(And( + enc_products, self.path_v_improd[path_idx][level + 1][reaction_id][entity] == conc)) # Nothing is produced (when the reaction is disabled) enc_no_prod = True if is_param(products): for entity in self.rs.set_of_bgset_ids: - enc_no_prod = And(enc_no_prod, - self.path_v_improd[path_idx][level + 1][reaction_id][entity] == 0) + enc_no_prod = And( + enc_no_prod, self.path_v_improd[path_idx][level + 1][reaction_id][entity] == 0) else: for entity, _ in products: - enc_no_prod = simplify(And(enc_no_prod, - self.path_v_improd[path_idx][level + 1][reaction_id][entity] == 0)) + enc_no_prod = simplify(And( + enc_no_prod, self.path_v_improd[path_idx][level + 1][reaction_id][entity] == 0)) - # # (R and I) iff P # enc_enabled = And(enc_reactants, enc_inhibitors) == enc_products - + # # ~(R and I) iff P_zero # - enc_not_enabled = Not(And(enc_reactants, enc_inhibitors)) == enc_no_prod - + enc_not_enabled = Not( + And(enc_reactants, enc_inhibitors)) == enc_no_prod + enc_reaction = And(enc_enabled, enc_not_enabled) return enc_reaction @@ -448,20 +477,21 @@ class SmtCheckerRSCParam(object): def enc_general_reaction_enabledness(self, level, path_idx): """ General enabledness condition for reactions - + The necessary condition for a reaction to be enabled is that the state is not empty, i.e., at least one entity is present in the current state. - + This condition must be used when there are parametric reactions because parameters could have all the entities set to zero and that immediately allows for all the conditions on the reactants to be fulfilled: (entity <= param) -> (0 <= 0) """ - + enc_cond = False for entity in self.rs.set_of_bgset_ids: - enc_cond = simplify(Or(enc_cond, self.path_v[path_idx][level][entity] > 0, self.path_v_ctx[path_idx][level][entity] > 0)) + enc_cond = simplify( + Or(enc_cond, self.path_v[path_idx][level][entity] > 0, self.path_v_ctx[path_idx][level][entity] > 0)) return enc_cond @@ -476,9 +506,9 @@ class SmtCheckerRSCParam(object): # # They should have concentration levels set to 0. # - # That needs to happen automatically (in the MAX encoding) -- for the parametric - # case it makes no sense to identify the entities that are never produced, unless - # we have no parameters as products (special case, so that could be an + # That needs to happen automatically (in the MAX encoding) -- for the parametric + # case it makes no sense to identify the entities that are never produced, unless + # we have no parameters as products (special case, so that could be an # optimisation) # @@ -488,7 +518,7 @@ class SmtCheckerRSCParam(object): enc_reaction = self.enc_single_reaction(level, path_idx, reaction) enc_trans = simplify(And(enc_trans, enc_reaction)) - # Next we encode the MAX concentration values: + # Next we encode the MAX concentration values: # we collect those from the intermediate product variables enc_max_prod = True @@ -497,22 +527,24 @@ class SmtCheckerRSCParam(object): # # - Intermediate products of (level+1) correspond to the next level # - # {reactants & inhibitors}[level] - # => - # {improd}[level+1] - # => + # {reactants & inhibitors}[level] + # => + # {improd}[level+1] + # => # {products}[level+1] # current_v_improd_for_entities = self.path_v_improd_for_entities[path_idx][level + 1] for entity in self.rs.set_of_bgset_ids: per_reaction_vars = current_v_improd_for_entities.get(entity, []) - enc_max_prod = simplify( - And(enc_max_prod, self.path_v[path_idx][level + 1][entity] == self.enc_max(per_reaction_vars))) - - # make sure at least one entity is >0 - enc_general_cond = self.enc_general_reaction_enabledness(level, path_idx) + enc_max_prod = simplify(And( + enc_max_prod, self.path_v[path_idx][level + 1][entity] == self.enc_max(per_reaction_vars))) - enc_trans_with_max = simplify(And(enc_general_cond, enc_max_prod, enc_trans)) + # make sure at least one entity is >0 + enc_general_cond = self.enc_general_reaction_enabledness( + level, path_idx) + + enc_trans_with_max = simplify( + And(enc_general_cond, enc_max_prod, enc_trans)) # print(enc_trans_with_max) @@ -521,7 +553,7 @@ class SmtCheckerRSCParam(object): def enc_max(self, elements): enc = None - + if elements == []: enc = 0 @@ -535,32 +567,34 @@ class SmtCheckerRSCParam(object): enc = z3_max(enc, z3_max(elements[i], elements[i + 1])) return enc - + def enc_automaton_trans(self, level, path_idx): """Encodes the transition relation for the context automaton""" - + enc_trans = False - - for src,ctx,dst in self.ca.transitions: - src_enc = self.path_ca_state[path_idx][level] == src - dst_enc = self.path_ca_state[path_idx][level+1] == dst - - all_ent = set(range(len(self.rs.background_set))) - - incl_ctx = set([e for e,c in ctx]) - excl_ctx = all_ent - incl_ctx - - ctx_enc = True - - for e,c in ctx: - ctx_enc = simplify(And(ctx_enc, self.path_v_ctx[path_idx][level][e] == c)) - - for e in excl_ctx: - ctx_enc = simplify(And(ctx_enc, self.path_v_ctx[path_idx][level][e] == 0)) - - cur_trans = simplify(And(src_enc, ctx_enc, dst_enc)) - enc_trans = simplify(Or(enc_trans, cur_trans)) - + + for src, ctx, dst in self.ca.transitions: + src_enc = self.path_ca_state[path_idx][level] == src + dst_enc = self.path_ca_state[path_idx][level+1] == dst + + all_ent = set(range(len(self.rs.background_set))) + + incl_ctx = set([e for e, c in ctx]) + excl_ctx = all_ent - incl_ctx + + ctx_enc = True + + for e, c in ctx: + ctx_enc = simplify( + And(ctx_enc, self.path_v_ctx[path_idx][level][e] == c)) + + for e in excl_ctx: + ctx_enc = simplify( + And(ctx_enc, self.path_v_ctx[path_idx][level][e] == 0)) + + cur_trans = simplify(And(src_enc, ctx_enc, dst_enc)) + enc_trans = simplify(Or(enc_trans, cur_trans)) + return enc_trans def enc_exact_state(self, level, state): @@ -572,32 +606,32 @@ class SmtCheckerRSCParam(object): """Encodes the state at the given level with the minimal required concentration levels""" enc = True - for ent,conc in state: + for ent, conc in state: e_id = self.rs.get_entity_id(ent) enc = And(enc, self.v[level][e_id] >= conc) - return simplify(enc) + return simplify(enc) def enc_state_with_blocking(self, level, prop): """Encodes the state at the given level with blocking certain concentrations""" - required,blocked = prop + required, blocked = prop enc = True - for ent,conc in required: - e_id = self.rs.get_entity_id(ent) - enc = And(enc, self.v[level][e_id] >= conc) + for ent, conc in required: + e_id = self.rs.get_entity_id(ent) + enc = And(enc, self.v[level][e_id] >= conc) - for ent,conc in blocked: - e_id = self.rs.get_entity_id(ent) - enc = And(enc, self.v[level][e_id] < conc) + for ent, conc in blocked: + e_id = self.rs.get_entity_id(ent) + enc = And(enc, self.v[level][e_id] < conc) return simplify(enc) def decode_witness(self, max_level, path_idx, print_model=False): """ Decodes the witness - + Also decodes the parameters """ @@ -635,15 +669,15 @@ class SmtCheckerRSCParam(object): " " + self.rs.get_entity_name(var_id) + "=" + var_rep, end="") print(" }") - + print() - + def get_enc_formulae(self, encoder, formulae_list): enc_form = [] for formula in formulae_list: - + path_idx = formulae_list.index(formula) - + print_info("Generating the encoding for {:s} ({:d} of {:d})".format( str(formula), path_idx+1, len(formulae_list))) @@ -651,27 +685,27 @@ class SmtCheckerRSCParam(object): var_rs=self.path_v[path_idx], var_ctx=self.path_v_ctx[path_idx], var_loop_pos=self.path_loop_position[path_idx]) - + enc_form.append(encoder.get_encoding(formula, self.current_level)) ncalls = encoder.get_ncalls() print_info("Cache hits: {:d}, encode calls: {:d} (approx: {:d})".format( - encoder.get_cache_hits(), ncalls[0], ncalls[1])) + encoder.get_cache_hits(), ncalls[0], ncalls[1])) encoder.flush_cache() - + return enc_form - + def print_witness(self, formulae_list): - + for formula in formulae_list: path_idx = formulae_list.index(formula) print("\n{:=^70}".format("[ WITNESS ]")) print("\n Witness for: {:s}".format(str(formula))) self.decode_witness(self.current_level, path_idx) - - self.print_parameter_valuations() - + + self.print_parameter_valuations() + def print_parameter_valuations(self): m = self.solver.model() @@ -680,9 +714,9 @@ class SmtCheckerRSCParam(object): for param_name in sorted(self.rs.parameters.keys()): print("{: >6}: ".format(param_name), end="") print("{", end="") - + params = self.v_param[param_name] - + for entity in self.rs.set_of_bgset_ids: var_rep = repr(m[params[entity]]) if not var_rep.isdigit(): @@ -690,40 +724,44 @@ class SmtCheckerRSCParam(object): "unexpected: representation is not a positive integer") if int(var_rep) > 0: print( - " " + str(self.rs.get_entity_name(entity)) + "=" + str(var_rep), + " " + str(self.rs.get_entity_name(entity)) + "=" + + str(var_rep), end="") print(" }") - + print() - - def enc_concentration_levels_assertions_for_paths(self, level, num_of_paths): - + + def enc_concentration_levels_assertions_for_paths( + self, level, num_of_paths): + additional_assertions = [] for path_idx in range(num_of_paths): - additional_assertions.append(self.enc_concentration_levels_assertion(level, path_idx)) - additional_assertions.append(self.enc_param_concentration_levels_assertion()) - + additional_assertions.append( + self.enc_concentration_levels_assertion(level, path_idx)) + additional_assertions.append( + self.enc_param_concentration_levels_assertion()) + return additional_assertions - + def enc_transition_relation_for_paths(self, level, num_of_paths): enc_trans = [] for path_idx in range(num_of_paths): enc_trans.append(self.enc_transition_relation(level, path_idx)) return enc_trans - + def print_level(self): print( "{:->70}".format("[ level=" + str(self.current_level) + " done ]")) - + def check_rsltl( - self, formulae_list, - print_witness=True, + self, formulae_list, + print_witness=True, print_time=True, print_mem=True, max_level=None, cont_if_sat=False, param_constr=None): """ Bounded Model Checking for rsLTL properties - + * print_witness -- prints the decoded witness * print_time -- prints the time consumed * print_mem -- prints the memory consumed @@ -734,14 +772,14 @@ class SmtCheckerRSCParam(object): reached (even if sat found) * param_constr -- constraints on parameters """ - + if not isinstance(formulae_list, (list, tuple)): print_error("Expected a list of formulae") - + self.reset() num_of_paths = len(formulae_list) - + print_info("Running rsLTL bounded model checking") print_info("Tested formulae:") for form in formulae_list: @@ -753,24 +791,26 @@ class SmtCheckerRSCParam(object): start_time = resource.getrusage(resource.RUSAGE_SELF).ru_utime self.prepare_all_variables(num_of_paths) - + self.load_varset_for_path(0) - + # initial states for all the paths initial_states = [] for path_idx in range(num_of_paths): initial_states.append(self.enc_init_state(0, path_idx)) self.solver_add(initial_states) - + self.current_level = 0 # 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_sanity_for_reactions()) encoder = rsLTL_Encoder(self) - + if param_constr: param_contr_encoder = ParamConstr_Encoder(self) enc_param_constr = param_contr_encoder.encode(param_constr) @@ -780,9 +820,9 @@ class SmtCheckerRSCParam(object): self.assert_param_optimisation() print_info("STARTING TO ITERATE...") - + while True: - + print( "\n{:-^70}".format("[ Working at level=" + str(self.current_level) + " ]")) @@ -802,7 +842,7 @@ class SmtCheckerRSCParam(object): result = self.solver.check() if result == sat: print_positive(green_str( - "SAT at level={:d}".format(self.current_level))) + "SAT at level={:d}".format(self.current_level))) # print(self.solver.model()) if print_witness: self.print_witness(formulae_list) @@ -817,12 +857,15 @@ class SmtCheckerRSCParam(object): break self.prepare_all_variables(num_of_paths) - + # assertions for all the paths - self.solver_add(self.enc_concentration_levels_assertions_for_paths(self.current_level + 1, num_of_paths)) + self.solver_add( + self.enc_concentration_levels_assertions_for_paths( + self.current_level + 1, num_of_paths)) print_info("Unrolling the transition relation") - self.solver_add(self.enc_transition_relation_for_paths(self.current_level, num_of_paths)) + self.solver_add(self.enc_transition_relation_for_paths( + self.current_level, num_of_paths)) self.print_level() @@ -838,8 +881,8 @@ class SmtCheckerRSCParam(object): self.verification_time = stop - start print() print_info("{: >60}".format( - " Time: " + repr(self.verification_time) + " s")) - + " Time: " + repr(self.verification_time) + " s")) + def print_mem(self): print_info( "{: >60}".format( @@ -847,22 +890,22 @@ class SmtCheckerRSCParam(object): repr( resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / (1024 * 1024)) + " MB")) - + def dummy_unroll(self, levels): """Unrolls the variables for testing purposes""" - self.current_level = -1 + self.current_level = -1 for i in range(levels+1): self.prepare_all_variables() self.current_level += 1 print(C_MARK_INFO + " Dummy Unrolling done.") - + def state_equality(self, level_A, level_B): """Encodes equality of two states at two different levels""" - + eq_enc = True - + for e_i in range(len(self.rs.background_set)): e_i_equality = self.v[level_A][e_i] == self.v[level_B][e_i] eq_enc = simplify(And(eq_enc, e_i_equality)) @@ -870,41 +913,42 @@ class SmtCheckerRSCParam(object): eq_enc_ctxaut = self.ca_state[level_A] == self.ca_state[level_B] eq_enc = simplify(And(eq_enc, eq_enc_ctxaut)) - return eq_enc - + return eq_enc + def get_loop_encodings(self): - + k = self.current_level loop_var = self.loop_position - + loop_enc = True - + """ (loop_var == i) means that there is a loop taking back to the state (i-1) Therefore, the encoding starts at 1, not at 0. """ - - for i in range(1,k+1): - loop_enc = simplify(And(loop_enc, Implies( loop_var == i, self.state_equality(i-1, k) ))) - + + for i in range(1, k+1): + loop_enc = simplify(And(loop_enc, Implies( + loop_var == i, self.state_equality(i-1, k)))) + return loop_enc - + def solver_add(self, expression): """ This is a solver.add() wrapper """ - + if expression == True: return - + if expression == False: raise RuntimeError("Trying to assert False.") - + self.solver.add(expression) - def check_reachability(self, state, print_witness=True, - print_time=True, print_mem=True, max_level=1000): + def check_reachability(self, state, print_witness=True, + print_time=True, print_mem=True, max_level=1000): """Main testing function""" self.reset() @@ -918,25 +962,33 @@ class SmtCheckerRSCParam(object): self.current_level = 0 self.prepare_all_variables() - + self.solver_add(self.enc_concentration_levels_assertion(0)) - + while True: self.prepare_all_variables() - self.solver_add(self.enc_concentration_levels_assertion(self.current_level+1)) - - print("\n{:-^70}".format("[ Working at level=" + str(self.current_level) + " ]")) + self.solver_add( + self.enc_concentration_levels_assertion( + self.current_level + 1)) + + print( + "\n{:-^70}".format("[ Working at level=" + str(self.current_level) + " ]")) stdout.flush() # reachability test: - print("[" + colour_str(C_BOLD, "i") + "] Adding the reachability test...") + print("[" + colour_str(C_BOLD, "i") + + "] Adding the reachability test...") self.solver.push() - self.solver_add(self.enc_state_with_blocking(self.current_level,state)) - + self.solver_add(self.enc_state_with_blocking( + self.current_level, state)) + result = self.solver.check() if result == sat: - print("[" + colour_str(C_BOLD, "+") + "] " + colour_str(C_GREEN, "SAT at level=" + str(self.current_level))) + print( + "[" + colour_str(C_BOLD, "+") + "] " + + colour_str( + C_GREEN, "SAT at level=" + str(self.current_level))) if print_witness: print("\n{:=^70}".format("[ WITNESS ]")) self.decode_witness(self.current_level) @@ -944,10 +996,12 @@ class SmtCheckerRSCParam(object): else: self.solver.pop() - print("[" + colour_str(C_BOLD, "i") + "] Unrolling the transition relation") + print("[" + colour_str(C_BOLD, "i") + + "] Unrolling the transition relation") self.solver_add(self.enc_transition_relation(self.current_level)) - print("{:->70}".format("[ level=" + str(self.current_level) + " done ]")) + print( + "{:->70}".format("[ level=" + str(self.current_level) + " done ]")) self.current_level += 1 if self.current_level > max_level: @@ -959,18 +1013,23 @@ class SmtCheckerRSCParam(object): stop = resource.getrusage(resource.RUSAGE_SELF).ru_utime self.verification_time = stop-start print() - print("\n[i] {: >60}".format(" Time: " + repr(self.verification_time) + " s")) - + print( + "\n[i] {: >60}".format( + " Time: " + repr(self.verification_time) + " s")) + if print_mem: - print("[i] {: >60}".format(" Memory: " + repr(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss/(1024*1024)) + " MB")) - - + print( + "[i] {: >60}".format( + " Memory: " + + repr( + resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / + (1024 * 1024)) + " MB")) + def get_verification_time(self): return self.verification_time - - - def show_encoding(self, state, print_witness=True, - print_time=False, print_mem=False, max_level=100): + + def show_encoding(self, state, print_witness=True, + print_time=False, print_mem=False, max_level=100): """Encoding debug function""" self.reset() @@ -982,25 +1041,29 @@ class SmtCheckerRSCParam(object): self.current_level = 0 self.prepare_all_variables() - + while True: self.prepare_all_variables() - print("-----[ Working at level=" + str(self.current_level) + " ]-----") + print( + "-----[ Working at level=" + str(self.current_level) + " ]-----") stdout.flush() # reachability test: - print("[i] Adding the reachability test...") + print("[i] Adding the reachability test...") self.solver.push() - s = self.enc_min_state(self.current_level,state) + s = self.enc_min_state(self.current_level, state) print("Test: ", s) - + self.solver_add(s) result = self.solver.check() if result == sat: - print("\n[+] " + colour_str(C_RED, "SAT at level=" + str(self.current_level))) + print( + "\n[+] " + + colour_str( + C_RED, "SAT at level=" + str(self.current_level))) if print_witness: self.decode_witness(self.current_level) break @@ -1019,8 +1082,8 @@ class SmtCheckerRSCParam(object): print("Stopping at level=" + str(max_level)) break else: - x=input("Next level? ") - x=x.lower() + x = input("Next level? ") + x = x.lower() if not (x == "y" or x == "yes"): break