From a1fb5836a571bfcff811559ca9b0bce88fdf3aa2 Mon Sep 17 00:00:00 2001 From: Artur Meski Date: Tue, 11 Jul 2023 19:48:37 +0100 Subject: [PATCH] black formatting --- reactics-smt/colour.py | 17 +- reactics-smt/doc/overview.py | 21 +- reactics-smt/logics/bags.py | 24 +- reactics-smt/logics/param_constr.py | 49 +- reactics-smt/logics/param_constr_encoder.py | 47 +- reactics-smt/logics/rsltl.py | 59 ++- reactics-smt/logics/rsltl_encoder.py | 188 ++++---- reactics-smt/rs/__init__.py | 6 +- reactics-smt/rs/context_automaton.py | 17 +- .../context_automaton_with_concentrations.py | 38 +- reactics-smt/rs/extended_context_automaton.py | 32 +- .../rs/network_of_context_automata.py | 15 +- reactics-smt/rs/reaction_system.py | 46 +- .../rs/reaction_system_with_autnet.py | 3 +- .../rs/reaction_system_with_automaton.py | 8 +- .../rs/reaction_system_with_concentrations.py | 114 ++--- ...action_system_with_concentrations_param.py | 90 ++-- reactics-smt/rsltl_shortcuts.py | 5 +- reactics-smt/rssmt.py | 38 +- reactics-smt/scripts/avg.py | 8 +- reactics-smt/smt/__init__.py | 2 +- reactics-smt/smt/smt_checker_rs.py | 74 ++-- reactics-smt/smt/smt_checker_rsc.py | 372 +++++++++------- reactics-smt/smt/smt_checker_rsc_param.py | 419 ++++++++++-------- 24 files changed, 965 insertions(+), 727 deletions(-) diff --git a/reactics-smt/colour.py b/reactics-smt/colour.py index 4000cb0..d615459 100644 --- a/reactics-smt/colour.py +++ b/reactics-smt/colour.py @@ -1,11 +1,11 @@ -C_HEADER = '\033[95m' -C_BLUE = '\033[94m' -C_GREEN = '\033[92m' -C_SOMEOTHER = '\033[93m' -C_RED = '\033[91m' -C_ENDC = '\033[0m' -C_BOLD = '\033[1m' -C_UNDERLINE = '\033[4m' +C_HEADER = "\033[95m" +C_BLUE = "\033[94m" +C_GREEN = "\033[92m" +C_SOMEOTHER = "\033[93m" +C_RED = "\033[91m" +C_ENDC = "\033[0m" +C_BOLD = "\033[1m" +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 @@ -30,4 +30,5 @@ def print_info(s): def print_positive(s): print("[" + colour_str(C_BOLD, "+") + "] {:s}".format(s)) + # EOF diff --git a/reactics-smt/doc/overview.py b/reactics-smt/doc/overview.py index 5fe094b..c4ef86c 100755 --- a/reactics-smt/doc/overview.py +++ b/reactics-smt/doc/overview.py @@ -5,26 +5,26 @@ from smt import * from logics import * from rsltl_shortcuts import * + def ex(): - prs = ReactionSystemWithConcentrationsParam() - - ent_with_conc = [("a", 3), ("b",2), ("c",1), ("h",1)] + + ent_with_conc = [("a", 3), ("b", 2), ("c", 1), ("h", 1)] for ec in ent_with_conc: prs.add_bg_set_entity(ec) lda = prs.get_param("lda") - - prs.add_reaction([("a",1)],[("h",1)],[("b", 2)]) - prs.add_reaction(lda,[("h",1)],[("c",1)]) - + + prs.add_reaction([("a", 1)], [("h", 1)], [("b", 2)]) + prs.add_reaction(lda, [("h", 1)], [("c", 1)]) + ## ca = ContextAutomatonWithConcentrations(prs) ca.add_init_state("0") ca.add_state("1") - + ca.add_transition("0", [("a", 3)], "1") ca.add_transition("1", [], "1") ca.add_transition("1", [("h", 1)], "1") @@ -37,8 +37,9 @@ def ex(): checker = SmtCheckerRSCParam(crprs, optimise=True) checker.check_rsltl(formulae_list=[f], param_constr=pc) - + + if __name__ == "__main__": - ex() + ex() # EOF diff --git a/reactics-smt/logics/bags.py b/reactics-smt/logics/bags.py index fd386e7..c689c43 100644 --- a/reactics-smt/logics/bags.py +++ b/reactics-smt/logics/bags.py @@ -1,10 +1,9 @@ from enum import Enum -BagDesc_oper = Enum( - 'BagDesc_oper', 'entity true l_and l_or l_not lt le eq ge gt') + +BagDesc_oper = Enum("BagDesc_oper", "entity true l_and l_or l_not lt le eq ge gt") class BagDescription(object): - def __init__(self, f_type, L_oper=None, R_oper=None, entity=""): self.f_type = f_type self.left_operand = L_oper @@ -19,9 +18,13 @@ class BagDescription(object): if self.f_type == BagDesc_oper.true: return "TRUE" if self.f_type == BagDesc_oper.l_and: - return "(" + repr(self.left_operand) + " & " + repr(self.right_operand) + ")" + return ( + "(" + repr(self.left_operand) + " & " + repr(self.right_operand) + ")" + ) if self.f_type == BagDesc_oper.l_or: - return "(" + repr(self.left_operand) + " | " + repr(self.right_operand) + ")" + return ( + "(" + repr(self.left_operand) + " | " + repr(self.right_operand) + ")" + ) if self.f_type == BagDesc_oper.l_not: return "~(" + repr(self.left_operand) + ")" if self.f_type == BagDesc_oper.lt: @@ -39,12 +42,14 @@ class BagDescription(object): """Sanity checks""" for operand in (self.left_operand, self.right_operand): if operand: - if not( - isinstance(operand, BagDescription) - or isinstance(operand, int)): + if not ( + isinstance(operand, BagDescription) or isinstance(operand, int) + ): raise RuntimeError( "Unexpected operand type for a bag: {:s} (type: {:s})".format( - str(operand), str(type(operand)))) + str(operand), str(type(operand)) + ) + ) @classmethod def f_entity(cls, entity_name): @@ -86,4 +91,5 @@ class BagDescription(object): def __invert__(self): return BagDescription(BagDesc_oper.l_not, L_oper=self) + # EOF diff --git a/reactics-smt/logics/param_constr.py b/reactics-smt/logics/param_constr.py index a9972b9..d49cce6 100644 --- a/reactics-smt/logics/param_constr.py +++ b/reactics-smt/logics/param_constr.py @@ -1,11 +1,11 @@ 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): - def __init__(self, f_type, L_oper=None, R_oper=None, param=None, entity=""): self.f_type = f_type self.left_operand = L_oper @@ -21,9 +21,13 @@ class ParamConstraint(object): if self.f_type == ParamConstraint_oper.true: return "TRUE" if self.f_type == ParamConstraint_oper.l_and: - return "(" + repr(self.left_operand) + " & " + repr(self.right_operand) + ")" + return ( + "(" + repr(self.left_operand) + " & " + repr(self.right_operand) + ")" + ) if self.f_type == ParamConstraint_oper.l_or: - return "(" + repr(self.left_operand) + " | " + repr(self.right_operand) + ")" + return ( + "(" + repr(self.left_operand) + " | " + repr(self.right_operand) + ")" + ) if self.f_type == ParamConstraint_oper.l_not: return "~(" + repr(self.left_operand) + ")" if self.f_type == ParamConstraint_oper.lt: @@ -41,17 +45,18 @@ class ParamConstraint(object): """Sanity checks""" for operand in (self.left_operand, self.right_operand): if operand: - if not( - isinstance(operand, ParamConstraint) - or isinstance(operand, int)): + if not ( + isinstance(operand, ParamConstraint) or isinstance(operand, int) + ): raise RuntimeError( "Unexpected operand type for a bag: {:s} (type: {:s})".format( - str(operand), str(type(operand)))) + str(operand), str(type(operand)) + ) + ) @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): @@ -66,34 +71,28 @@ 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) + # EOF diff --git a/reactics-smt/logics/param_constr_encoder.py b/reactics-smt/logics/param_constr_encoder.py index d74b9b7..c0580a3 100644 --- a/reactics-smt/logics/param_constr_encoder.py +++ b/reactics-smt/logics/param_constr_encoder.py @@ -14,57 +14,60 @@ class ParamConstr_Encoder(object): # 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) + 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/reactics-smt/logics/rsltl.py b/reactics-smt/logics/rsltl.py index d1735ba..bd6a04d 100644 --- a/reactics-smt/logics/rsltl.py +++ b/reactics-smt/logics/rsltl.py @@ -4,14 +4,13 @@ from enum import Enum from logics.bags import * rsLTL_form_type = Enum( - 'rsLTL_form_type', - 'bag l_and l_or l_not l_implies t_globally t_finally t_next t_until t_release') + "rsLTL_form_type", + "bag l_and l_or l_not l_implies t_globally t_finally t_next t_until t_release", +) class Formula_rsLTL(object): - - def __init__( - self, f_type, L_oper=None, R_oper=None, sub_oper=None, bag=None): + def __init__(self, f_type, L_oper=None, R_oper=None, sub_oper=None, bag=None): self.f_type = f_type self.left_operand = L_oper self.right_operand = R_oper @@ -21,7 +20,10 @@ class Formula_rsLTL(object): # it's silly, but it's a frequent mistake if callable(sub_oper): raise RuntimeError( - "sub_oper should not be a function. Missing () for {0}?".format(sub_oper)) + "sub_oper should not be a function. Missing () for {0}?".format( + sub_oper + ) + ) if isinstance(self.left_operand, BagDescription): self.left_operand = Formula_rsLTL.f_bag(self.left_operand) @@ -43,21 +45,37 @@ class Formula_rsLTL(object): if self.f_type == rsLTL_form_type.t_next: return "X[" + repr(self.sub_operand) + "](" + repr(self.left_operand) + ")" if self.f_type == rsLTL_form_type.l_and: - return "(" + repr(self.left_operand) + " & " + repr(self.right_operand) + ")" + return ( + "(" + repr(self.left_operand) + " & " + repr(self.right_operand) + ")" + ) if self.f_type == rsLTL_form_type.l_or: - return "(" + repr(self.left_operand) + " | " + repr(self.right_operand) + ")" + return ( + "(" + repr(self.left_operand) + " | " + repr(self.right_operand) + ")" + ) if self.f_type == rsLTL_form_type.l_implies: - return "(" + repr(self.left_operand) + " => " + repr(self.right_operand) + ")" + return ( + "(" + repr(self.left_operand) + " => " + repr(self.right_operand) + ")" + ) if self.f_type == rsLTL_form_type.t_until: - return "(" + repr( - self.left_operand) + " U[" + repr( - self.sub_operand) + "] " + repr( - self.right_operand) + ")" + return ( + "(" + + repr(self.left_operand) + + " U[" + + repr(self.sub_operand) + + "] " + + repr(self.right_operand) + + ")" + ) if self.f_type == rsLTL_form_type.t_release: - return "(" + repr( - self.left_operand) + " R[" + repr( - self.sub_operand) + "] " + repr( - self.right_operand) + ")" + return ( + "(" + + repr(self.left_operand) + + " R[" + + repr(self.sub_operand) + + "] " + + repr(self.right_operand) + + ")" + ) @property def is_bag(self): @@ -93,8 +111,7 @@ class Formula_rsLTL(object): @classmethod def f_U(cls, sub, arg_L, arg_R): - return cls( - rsLTL_form_type.t_until, L_oper=arg_L, R_oper=arg_R, sub_oper=sub) + return cls(rsLTL_form_type.t_until, L_oper=arg_L, R_oper=arg_R, sub_oper=sub) @classmethod def f_F(cls, sub, arg_L): @@ -102,9 +119,7 @@ class Formula_rsLTL(object): @classmethod def f_R(cls, sub, arg_L, arg_R): - return cls( - rsLTL_form_type.t_release, L_oper=arg_L, R_oper=arg_R, - sub_oper=sub) + return cls(rsLTL_form_type.t_release, L_oper=arg_L, R_oper=arg_R, sub_oper=sub) def __and__(self, other): return Formula_rsLTL(rsLTL_form_type.l_and, L_oper=self, R_oper=other) diff --git a/reactics-smt/logics/rsltl_encoder.py b/reactics-smt/logics/rsltl_encoder.py index f9e60b8..8ad9ded 100644 --- a/reactics-smt/logics/rsltl_encoder.py +++ b/reactics-smt/logics/rsltl_encoder.py @@ -24,13 +24,11 @@ class rsLTL_Encoder(object): 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 assert self.v_ctx is not None assert self.loop_position is not None @@ -49,8 +47,8 @@ class rsLTL_Encoder(object): 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 @@ -83,7 +81,6 @@ class rsLTL_Encoder(object): 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") @@ -100,41 +97,42 @@ class rsLTL_Encoder(object): if bag_formula.f_type == BagDesc_oper.l_and: return And( self.encode_bag(bag_formula.left_operand, level, context), - self.encode_bag(bag_formula.right_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), - self.encode_bag(bag_formula.right_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) @@ -149,7 +147,6 @@ class rsLTL_Encoder(object): return res def encode(self, formula, level, bound): - self.ncalls_encode += 1 from_cache = self.cache_query(formula, level) @@ -159,12 +156,12 @@ class rsLTL_Encoder(object): 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.") + "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) @@ -179,33 +176,40 @@ class rsLTL_Encoder(object): 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.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.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.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_bag_ctx(formula.sub_operand, level), ) 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)))) + 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 = And(enc, self.encode_bag_ctx(formula.sub_operand, level)) enc = simplify(enc) @@ -214,22 +218,25 @@ class rsLTL_Encoder(object): 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, 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), - ) - )) + 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 = And( self.encode(formula.left_operand, bound, bound), enc_loops, - self.encode_bag_ctx(formula.sub_operand, level) + self.encode_bag_ctx(formula.sub_operand, level), ) enc = simplify(enc) @@ -239,25 +246,26 @@ class rsLTL_Encoder(object): 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, 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), - ) - )) + 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 = Or( + self.encode(formula.left_operand, bound, bound), + And(enc_loops, self.encode_bag_ctx(formula.sub_operand, level)), ) enc = simplify(enc) @@ -268,21 +276,24 @@ class rsLTL_Encoder(object): 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) - ) - )) + 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 = Or( 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) - ) + self.encode_bag_ctx(formula.sub_operand, level), + ), ) elif formula.f_type == rsLTL_form_type.t_release: @@ -291,23 +302,23 @@ class rsLTL_Encoder(object): 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) - ) - )) + 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) - ) - ) + And(inner_enc, self.encode_bag_ctx(formula.sub_operand, level)), + ), ) else: @@ -341,8 +352,8 @@ class rsLTL_Encoder(object): And( self.encode(formula.left_operand, level, bound), self.encode_approx(formula, level + 1, bound), - self.encode_bag_ctx(formula.sub_operand, level) - ) + self.encode_bag_ctx(formula.sub_operand, level), + ), ) else: # level == bound @@ -356,9 +367,9 @@ class rsLTL_Encoder(object): self.encode(formula.left_operand, level, bound), And( self.encode_approx(formula, level + 1, bound), - self.encode_bag_ctx(formula.sub_operand, level) - ) - ) + self.encode_bag_ctx(formula.sub_operand, level), + ), + ), ) else: # level == bound @@ -369,7 +380,7 @@ class rsLTL_Encoder(object): 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_approx(formula, level + 1, bound), ) else: # level == bound @@ -381,16 +392,15 @@ class rsLTL_Encoder(object): 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_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") diff --git a/reactics-smt/rs/__init__.py b/reactics-smt/rs/__init__.py index a5d9f4b..3f20854 100644 --- a/reactics-smt/rs/__init__.py +++ b/reactics-smt/rs/__init__.py @@ -2,7 +2,11 @@ from rs.reaction_system import ReactionSystem from rs.context_automaton import ContextAutomaton from rs.reaction_system_with_concentrations import ReactionSystemWithConcentrations -from rs.reaction_system_with_concentrations_param import ReactionSystemWithConcentrationsParam, ParameterObj, is_param +from rs.reaction_system_with_concentrations_param import ( + ReactionSystemWithConcentrationsParam, + ParameterObj, + is_param, +) from rs.context_automaton_with_concentrations import ContextAutomatonWithConcentrations from rs.extended_context_automaton import ExtendedContextAutomaton diff --git a/reactics-smt/rs/context_automaton.py b/reactics-smt/rs/context_automaton.py index 8822fcc..6d0a084 100644 --- a/reactics-smt/rs/context_automaton.py +++ b/reactics-smt/rs/context_automaton.py @@ -3,7 +3,6 @@ from colour import * class ContextAutomaton(object): - def __init__(self, reaction_system): self._states = [] self._transitions = [] @@ -40,7 +39,7 @@ class ContextAutomaton(object): if name not in self._states: self._states.append(name) else: - print("\'%s\' already added. skipping..." % (name,)) + print("'%s' already added. skipping..." % (name,)) def add_states(self, states_set): for st in states_set: @@ -101,15 +100,14 @@ class ContextAutomaton(object): if not self.is_valid_context(context_set): raise RuntimeError( - "one of the entities in the context set is unknown (undefined)!") + "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): @@ -118,8 +116,8 @@ class ContextAutomaton(object): self._prod_entities |= new_context_set self._transitions.append( - (self.get_state_id(src), - new_context_set, self.get_state_id(dst))) + (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""" @@ -168,4 +166,5 @@ class ContextAutomaton(object): self.show_transitions() self.show_prod_entities() + # EOF diff --git a/reactics-smt/rs/context_automaton_with_concentrations.py b/reactics-smt/rs/context_automaton_with_concentrations.py index 18d81b8..14b0271 100644 --- a/reactics-smt/rs/context_automaton_with_concentrations.py +++ b/reactics-smt/rs/context_automaton_with_concentrations.py @@ -5,15 +5,13 @@ 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 @@ -33,45 +31,38 @@ class ContextAutomatonWithConcentrations(ContextAutomaton): if not self.is_valid_context(context_set): raise RuntimeError( - "one of the entities in the context set is unknown (undefined)!") + "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)) + 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))) + (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: - 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 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 @@ -80,4 +71,5 @@ class ContextAutomatonWithConcentrations(ContextAutomaton): self.show_states() self.show_transitions() + # EOF diff --git a/reactics-smt/rs/extended_context_automaton.py b/reactics-smt/rs/extended_context_automaton.py index c2bfef2..bae9ce2 100644 --- a/reactics-smt/rs/extended_context_automaton.py +++ b/reactics-smt/rs/extended_context_automaton.py @@ -75,23 +75,24 @@ class ExtendedContextAutomaton(ContextAutomaton): if not self.is_valid_rs_set(ctx_reactants): raise RuntimeError( - "one of the entities in the reactants set is unknown (undefined)!") + "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)!") + "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)!") + "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) @@ -119,8 +120,15 @@ class ExtendedContextAutomaton(ContextAutomaton): for src_id, act_id, reaction, dst_id in self._transitions: 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) @@ -130,7 +138,7 @@ class ExtendedContextAutomaton(ContextAutomaton): if action_name not in self._actions: self._actions.append(action_name) else: - print("\'%s\' already added. skipping..." % (action_name,)) + print("'%s' already added. skipping..." % (action_name,)) def get_action_id(self, action_name): """For an action name returns its id""" @@ -138,8 +146,7 @@ class ExtendedContextAutomaton(ContextAutomaton): 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): @@ -173,4 +180,5 @@ class ExtendedContextAutomaton(ContextAutomaton): super(ExtendedContextAutomaton, self).show() self.show_actions() + # EOF diff --git a/reactics-smt/rs/network_of_context_automata.py b/reactics-smt/rs/network_of_context_automata.py index efdee19..bcc42b2 100644 --- a/reactics-smt/rs/network_of_context_automata.py +++ b/reactics-smt/rs/network_of_context_automata.py @@ -3,7 +3,6 @@ from colour import * class NetworkOfContextAutomata(object): - def __init__(self, reaction_system, context_automata): self.automata = [] self._reaction_system = reaction_system @@ -44,14 +43,14 @@ class NetworkOfContextAutomata(object): for automaton in self.automata: if automaton.reaction_system != self._reaction_system: print_error( - "Mismatching reaction system used in \"" + - str(automaton.name) + "\"!!!") + '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)) @@ -96,7 +95,8 @@ class NetworkOfContextAutomata(object): for entity in aut.prod_entities: self._actions_for_products.setdefault(entity, set()) self._actions_for_products[entity] |= aut.get_actions_producing_entity( - entity) + entity + ) def show(self): print() @@ -108,4 +108,5 @@ class NetworkOfContextAutomata(object): self.show_prod_entities() self.show_actions() + # EOF diff --git a/reactics-smt/rs/reaction_system.py b/reactics-smt/rs/reaction_system.py index 908e1e7..ad10a9f 100644 --- a/reactics-smt/rs/reaction_system.py +++ b/reactics-smt/rs/reaction_system.py @@ -3,9 +3,7 @@ from colour import * class ReactionSystem(object): - def __init__(self): - self.reactions = [] self.background_set = [] @@ -30,8 +28,7 @@ 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) @@ -106,7 +103,6 @@ class ReactionSystem(object): self.init_contexts.append(integers) def set_context_entities(self, entities): - for entity in entities: entity_id = self.get_entity_id(entity) self.context_entities.append(entity_id) @@ -131,20 +127,36 @@ class ReactionSystem(object): 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")) + " " * 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]) + " }")) + 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]) + " }", + ) + ) def show_background_set(self): print( - C_MARK_INFO + " Background set: {" + self.entities_names_set_to_str(self.background_set) + "}") + 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: @@ -155,10 +167,12 @@ class ReactionSystem(object): 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)) + C_MARK_INFO + + " Context entities: " + + self.entities_ids_set_to_str(self.context_entities) + ) def show(self, soft=False): - self.show_background_set() self.show_initial_contexts() self.show_reactions(soft) @@ -181,8 +195,7 @@ class ReactionSystem(object): 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 @@ -200,4 +213,5 @@ class ReactionSystem(object): print("Empty background set") exit(1) + # EOF diff --git a/reactics-smt/rs/reaction_system_with_autnet.py b/reactics-smt/rs/reaction_system_with_autnet.py index 0054052..d7dd24f 100644 --- a/reactics-smt/rs/reaction_system_with_autnet.py +++ b/reactics-smt/rs/reaction_system_with_autnet.py @@ -1,6 +1,4 @@ - class ReactionSystemWithNetworkOfAutomata(object): - def __init__(self, reaction_system, context_automata): self.rs = reaction_system self.cas = context_automata @@ -9,4 +7,5 @@ class ReactionSystemWithNetworkOfAutomata(object): self.rs.show(soft) self.cas.show() + # EOF diff --git a/reactics-smt/rs/reaction_system_with_automaton.py b/reactics-smt/rs/reaction_system_with_automaton.py index 0fcceaa..2073bbe 100644 --- a/reactics-smt/rs/reaction_system_with_automaton.py +++ b/reactics-smt/rs/reaction_system_with_automaton.py @@ -1,10 +1,11 @@ from rs.reaction_system_with_concentrations import ReactionSystemWithConcentrations -from rs.reaction_system_with_concentrations_param import ReactionSystemWithConcentrationsParam +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 @@ -15,7 +16,7 @@ class ReactionSystemWithAutomaton(object): def is_concentr_and_param_compatible(self): """ - Checks if the underlying RS/CA are compatible + Checks if the underlying RS/CA are compatible with parameters and concentrations """ if not isinstance(self.rs, ReactionSystemWithConcentrationsParam): @@ -35,7 +36,6 @@ class ReactionSystemWithAutomaton(object): pass def get_ordinary_reaction_system_with_automaton(self): - if not self.is_with_concentrations(): raise RuntimeError("Not RS/CA with concentrations") diff --git a/reactics-smt/rs/reaction_system_with_concentrations.py b/reactics-smt/rs/reaction_system_with_concentrations.py index 314716b..9bdeb28 100644 --- a/reactics-smt/rs/reaction_system_with_concentrations.py +++ b/reactics-smt/rs/reaction_system_with_concentrations.py @@ -5,9 +5,7 @@ from rs.reaction_system import ReactionSystem class ReactionSystemWithConcentrations(ReactionSystem): - def __init__(self): - self.reactions = [] self.meta_reactions = dict() self.permanent_entities = dict() @@ -26,8 +24,7 @@ class ReactionSystemWithConcentrations(ReactionSystem): 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) @@ -41,7 +38,6 @@ 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: @@ -69,8 +65,7 @@ class ReactionSystemWithConcentrations(ReactionSystem): 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""" @@ -123,21 +118,25 @@ class ReactionSystemWithConcentrations(ReactionSystem): """Adds a macro/meta reaction for increasing the value of incr_entity""" reactants, inhibitors, products = self.process_rip( - R, I, [], ignore_empty_R=True) + 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)) + ("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) + 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)) + ("dec", self.get_entity_id(decrementer), reactants, inhibitors) + ) def add_permanency(self, ent, I): """Sets entity to be permanent unless it is inhibited""" @@ -145,8 +144,7 @@ class ReactionSystemWithConcentrations(ReactionSystem): ent_id = self.get_entity_id(ent) if ent_id in self.permanent_entities: - raise RuntimeError( - "Permanency for {0} already defined.".format(ent)) + 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 @@ -177,31 +175,50 @@ class ReactionSystemWithConcentrations(ReactionSystem): def show_background_set(self): print( - C_MARK_INFO + " Background set: {" + self.entities_names_set_to_str(self.background_set) + "}") + 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: 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):") + 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) + "}")) + 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() @@ -220,8 +237,7 @@ class ReactionSystemWithConcentrations(ReactionSystem): for reaction in self.reactions: product_entities = [e for e, c in reaction[2]] - producible_entities = producible_entities.union( - set(product_entities)) + producible_entities = producible_entities.union(set(product_entities)) reactions_by_prod = {} @@ -242,18 +258,20 @@ class ReactionSystemWithConcentrations(ReactionSystem): # 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)): - 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 + 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)) + insert_place, (reactants, inhibitors, products) + ) # save in cache self.reactions_by_prod = reactions_by_prod @@ -261,11 +279,9 @@ class ReactionSystemWithConcentrations(ReactionSystem): return reactions_by_prod def get_reaction_system(self): - rs = ReactionSystem() for reactants, inhibitors, products in self.reactions: - new_reactants = [] new_inhibitors = [] new_products = [] @@ -281,7 +297,7 @@ class ReactionSystemWithConcentrations(ReactionSystem): new_inhibitors.append(n) for ent, conc in products: - for i in range(1, conc+1): + 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) @@ -290,7 +306,6 @@ class ReactionSystemWithConcentrations(ReactionSystem): 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 = [] @@ -312,16 +327,15 @@ class ReactionSystemWithConcentrations(ReactionSystem): else: print( "WARNING:\n\tThere is no maximal concentration level defined for " - + self.get_entity_name(command)) + + 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 @@ -329,8 +343,8 @@ class ReactionSystemWithConcentrations(ReactionSystem): 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): + succ_value = i + l + for j in range(1, succ_value + 1): if j > self.max_concentration: break new_p = param_ent_name + "#" + str(j) @@ -340,15 +354,16 @@ class ReactionSystemWithConcentrations(ReactionSystem): rs.add_reaction( set(new_reactants + [pre_conc, cmd_ent]), set(new_inhibitors), - set(new_products)) + 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): + succ_value = i - l + for j in range(1, succ_value + 1): if j > self.max_concentration: break new_p = param_ent_name + "#" + str(j) @@ -358,28 +373,29 @@ class ReactionSystemWithConcentrations(ReactionSystem): rs.add_reaction( set(new_reactants + [pre_conc, cmd_ent]), set(new_inhibitors), - set(new_products)) + set(new_products), + ) else: raise RuntimeError( - "Unknown meta-reaction type: " + repr(r_type)) + "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)) + + 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 = [] @@ -391,7 +407,7 @@ class ReactionSystemWithConcentrations(ReactionSystem): rs.ensure_bg_set_entity(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) @@ -400,7 +416,6 @@ class ReactionSystemWithConcentrations(ReactionSystem): class ReactionSystemWithAutomaton(object): - def __init__(self, reaction_system, context_automaton): self.rs = reaction_system self.ca = context_automaton @@ -420,7 +435,6 @@ class ReactionSystemWithAutomaton(object): pass def get_ordinary_reaction_system_with_automaton(self): - if not self.is_with_concentrations(): raise RuntimeError("Not RS/CA with concentrations") diff --git a/reactics-smt/rs/reaction_system_with_concentrations_param.py b/reactics-smt/rs/reaction_system_with_concentrations_param.py index 54a8695..ab628dc 100644 --- a/reactics-smt/rs/reaction_system_with_concentrations_param.py +++ b/reactics-smt/rs/reaction_system_with_concentrations_param.py @@ -5,7 +5,6 @@ from rs.reaction_system import ReactionSystem class ParameterObj(object): - def __init__(self, name): self.name = name @@ -21,9 +20,7 @@ def is_param(some_object): class ReactionSystemWithConcentrationsParam(ReactionSystem): - def __init__(self): - self.reactions = [] self.parameters = dict() self.meta_reactions = dict() @@ -43,8 +40,7 @@ class ReactionSystemWithConcentrationsParam(ReactionSystem): 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) @@ -76,7 +72,6 @@ class ReactionSystemWithConcentrationsParam(ReactionSystem): self.parameters[param_key].idx = len(self.parameters) def get_max_concentration_level(self, e): - if e in self.max_conc_per_ent: return self.max_conc_per_ent[e] else: @@ -113,17 +108,18 @@ class ReactionSystemWithConcentrationsParam(ReactionSystem): def terminate_on_invalid_concentration(self, entity, level): if not self.is_valid_concentration_for_entity(entity, level): - print("FATAL. Invalid entity concentration: {:s}={:d}".format(entity, level)) + print( + "FATAL. Invalid entity concentration: {:s}={:d}".format(entity, level) + ) exit(1) - + def get_state_ids(self, state): """Returns entities of the given state without levels""" return [self.get_entity_id(e) for e 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""" @@ -207,21 +203,25 @@ class ReactionSystemWithConcentrationsParam(ReactionSystem): """Adds a macro/meta reaction for increasing the value of incr_entity""" reactants, inhibitors, products = self.process_rip( - R, I, [], ignore_empty_R=True) + 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)) + ("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) + 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)) + ("dec", self.get_entity_id(decrementer), reactants, inhibitors) + ) def add_permanency(self, ent, I): """Sets entity to be permanent unless it is inhibited""" @@ -229,8 +229,7 @@ class ReactionSystemWithConcentrationsParam(ReactionSystem): ent_id = self.get_entity_id(ent) if ent_id in self.permanent_entities: - raise RuntimeError( - "Permanency for {0} already defined.".format(ent)) + 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 @@ -269,18 +268,32 @@ class ReactionSystemWithConcentrationsParam(ReactionSystem): def show_background_set(self): print( - C_MARK_INFO + " Background set: {" + self.entities_names_set_to_str(self.background_set) + "}") + 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: 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:") @@ -290,8 +303,12 @@ class ReactionSystemWithConcentrationsParam(ReactionSystem): 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) + "}")) + 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() @@ -303,7 +320,7 @@ class ReactionSystemWithConcentrationsParam(ReactionSystem): def get_producible_entities(self): """ - Returns the set of entities that appear as products of + Returns the set of entities that appear as products of reactions. """ @@ -311,8 +328,7 @@ class ReactionSystemWithConcentrationsParam(ReactionSystem): 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)) + producible_entities = producible_entities.union(set(product_entities)) return producible_entities @@ -324,7 +340,6 @@ class ReactionSystemWithConcentrationsParam(ReactionSystem): rs = ReactionSystem() for reactants, inhibitors, products in self.reactions: - new_reactants = [] new_inhibitors = [] new_products = [] @@ -349,7 +364,6 @@ class ReactionSystemWithConcentrationsParam(ReactionSystem): 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 = [] @@ -371,16 +385,15 @@ class ReactionSystemWithConcentrationsParam(ReactionSystem): else: print( "WARNING:\n\tThere is no maximal concentration level defined for " - + self.get_entity_name(command)) + + 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): - 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 @@ -399,7 +412,8 @@ class ReactionSystemWithConcentrationsParam(ReactionSystem): rs.add_reaction( set(new_reactants + [pre_conc, cmd_ent]), set(new_inhibitors), - set(new_products)) + set(new_products), + ) elif r_type == "dec": for i in range(1, self.max_concentration + 1): @@ -417,28 +431,29 @@ class ReactionSystemWithConcentrationsParam(ReactionSystem): rs.add_reaction( set(new_reactants + [pre_conc, cmd_ent]), set(new_inhibitors), - set(new_products)) + set(new_products), + ) else: raise RuntimeError( - "Unknown meta-reaction type: " + repr(r_type)) + "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)) + + 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): - new_reactants = [] new_inhibitors = [] new_products = [] @@ -457,4 +472,5 @@ class ReactionSystemWithConcentrationsParam(ReactionSystem): return rs + # EOF diff --git a/reactics-smt/rsltl_shortcuts.py b/reactics-smt/rsltl_shortcuts.py index 479cc3b..34bc862 100644 --- a/reactics-smt/rsltl_shortcuts.py +++ b/reactics-smt/rsltl_shortcuts.py @@ -27,8 +27,7 @@ def bag_And(*args): assert len(args) > 1 last = get_bag_if_str(args[0]) for arg in args[1:]: - last = BagDescription.f_And( - last, get_bag_if_str(arg)) + last = BagDescription.f_And(last, get_bag_if_str(arg)) return last @@ -46,7 +45,6 @@ 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) @@ -107,5 +105,6 @@ def param_And(*args): last = ParamConstraint.f_And(last, arg) return last + def param_True(): return ParamConstraint.f_TRUE() diff --git a/reactics-smt/rssmt.py b/reactics-smt/rssmt.py index eaa26b8..4ebf902 100755 --- a/reactics-smt/rssmt.py +++ b/reactics-smt/rssmt.py @@ -24,12 +24,16 @@ if profiling: ################################################################## version = "2.99" -rsmc_banner = """ +rsmc_banner = ( + """ Reaction Systems SMT-Based Model Checking -Version: """ + version + """ +Version: """ + + version + + """ Author: Artur Meski """ +) ################################################################## @@ -40,6 +44,7 @@ def print_banner(): print(colour_str(C_GREEN, " " + 3 * "-" + " "), line) print() + ################################################################## @@ -48,22 +53,30 @@ 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", - help="scaling parameter value (used in some benchmarks)") - parser.add_argument("-s", "--special_mode", - help="special mode (used in some benchmarks)") + "-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)" + ) args = parser.parse_args() print_banner() rs_testing.run_tests(args) + ################################################################## @@ -71,7 +84,8 @@ if __name__ == "__main__": try: if profiling: import profile - profile.run('main()') + + profile.run("main()") else: main() except KeyboardInterrupt: diff --git a/reactics-smt/scripts/avg.py b/reactics-smt/scripts/avg.py index 1e55980..4d6cafa 100755 --- a/reactics-smt/scripts/avg.py +++ b/reactics-smt/scripts/avg.py @@ -5,7 +5,6 @@ from os import listdir def proc_files(files): - fs = [] for f in files: try: @@ -17,7 +16,6 @@ def proc_files(files): done = False result = [] while not done: - index = "NONE" val_sum = 0 @@ -30,7 +28,7 @@ def proc_files(files): sline = line.split() - assert(len(sline) == 2) + assert len(sline) == 2 index, value = sline value = float(value) @@ -40,7 +38,7 @@ def proc_files(files): # print(sline) else: - val_avg = val_sum/len(fs) + val_avg = val_sum / len(fs) result.append("{:s}\t{:f}".format(index, val_avg)) result.append("") @@ -48,7 +46,6 @@ def proc_files(files): def proc_dirs(dirs): - first_dir = dirs[0] for f in listdir(first_dir): @@ -59,7 +56,6 @@ def proc_dirs(dirs): def main(): - proc_dirs(sys.argv[1:]) # proc_files(sys.argv[1:]) diff --git a/reactics-smt/smt/__init__.py b/reactics-smt/smt/__init__.py index d626ea0..dbce176 100644 --- a/reactics-smt/smt/__init__.py +++ b/reactics-smt/smt/__init__.py @@ -1,4 +1,4 @@ -#from smt.smt_checker import SmtChecker +# from smt.smt_checker import SmtChecker from smt.smt_checker_rs import SmtCheckerRS from smt.smt_checker_rsc import SmtCheckerRSC from smt.smt_checker_rsc_param import SmtCheckerRSCParam diff --git a/reactics-smt/smt/smt_checker_rs.py b/reactics-smt/smt/smt_checker_rs.py index 193ff93..88242a4 100644 --- a/reactics-smt/smt/smt_checker_rs.py +++ b/reactics-smt/smt/smt_checker_rs.py @@ -9,9 +9,7 @@ import resource class SmtCheckerRS(object): - def __init__(self, rsca): - rsca.sanity_check() self.rs = rsca.rs @@ -40,7 +38,7 @@ class SmtCheckerRS(object): variables = [] for entity in self.rs.background_set: - variables.append(Bool("C"+str(level)+"_"+entity)) + variables.append(Bool("C" + str(level) + "_" + entity)) self.v_ctx.append(variables) @@ -51,7 +49,7 @@ class SmtCheckerRS(object): variables = [] for entity in self.rs.background_set: - variables.append(Bool("L"+str(level)+"_"+entity)) + variables.append(Bool("L" + str(level) + "_" + entity)) self.v.append(variables) def prepare_context_controller_variables(self): @@ -59,7 +57,7 @@ class SmtCheckerRS(object): level = self.next_level_to_encode - self.ca_state.append(Int("CA"+str(level)+"_state")) + self.ca_state.append(Int("CA" + str(level) + "_state")) def prepare_state_variables(self): """Encodes all the state variables""" @@ -84,8 +82,12 @@ class SmtCheckerRS(object): 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 @@ -102,14 +104,23 @@ class SmtCheckerRS(object): 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))) + Or(enc_rct_prod, And(enc_reactants, enc_inhibitors)) + ) return enc_rct_prod @@ -120,17 +131,15 @@ class SmtCheckerRS(object): 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]))) + 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""" @@ -143,20 +152,19 @@ class SmtCheckerRS(object): unused_entities.remove(prod_entity) enc_trans = simplify( - And(enc_trans, self.enc_entity_production(level, prod_entity))) + 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_enc = self.ca_state[level] == src - dst_enc = self.ca_state[level+1] == dst + dst_enc = self.ca_state[level + 1] == dst all_ent = set(range(len(self.rs.background_set))) incl_ctx = ctx @@ -179,7 +187,8 @@ class SmtCheckerRS(object): enc_trans = False for transition in self.ca.transitions: enc_trans = simplify( - Or(enc_trans, self.enc_automaton_single_trans(level, transition))) + Or(enc_trans, self.enc_automaton_single_trans(level, transition)) + ) return enc_trans @@ -231,14 +240,12 @@ class SmtCheckerRS(object): return simplify(enc) def decode_witness(self, max_level, print_model=False): - m = self.solver.model() if print_model: print(m) - for level in range(max_level+1): - + for level in range(max_level + 1): print("\n[Level=" + repr(level) + "]") print(" State: {", end=""), @@ -256,7 +263,8 @@ class SmtCheckerRS(object): print(" }") def check_reachability( - self, state, print_witness=True, print_time=True, print_mem=True): + self, state, print_witness=True, print_time=True, print_mem=True + ): """Main testing function""" if not type(state) is tuple: @@ -299,16 +307,18 @@ class SmtCheckerRS(object): if print_time: # stop = time() stop = resource.getrusage(resource.RUSAGE_SELF).ru_utime - self.verification_time = stop-start + 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") + "[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/reactics-smt/smt/smt_checker_rsc.py b/reactics-smt/smt/smt_checker_rsc.py index d98d572..9b4a795 100644 --- a/reactics-smt/smt/smt_checker_rsc.py +++ b/reactics-smt/smt/smt_checker_rsc.py @@ -16,9 +16,7 @@ from logics import rsLTL_Encoder class SmtCheckerRSC(object): - def __init__(self, rsca): - rsca.sanity_check() if not rsca.is_with_concentrations(): @@ -62,7 +60,7 @@ class SmtCheckerRSC(object): variables = [] for entity in self.rs.background_set: - variables.append(Int("C"+str(level)+"_"+entity)) + variables.append(Int("C" + str(level) + "_" + entity)) self.v_ctx.append(variables) @@ -73,10 +71,10 @@ class SmtCheckerRSC(object): variables = [] for entity in self.rs.background_set: - variables.append(Int("L"+str(level)+"_"+entity)) + variables.append(Int("L" + str(level) + "_" + entity)) self.v.append(variables) - self.ca_state.append(Int("CA"+str(level)+"_state")) + 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 @@ -92,7 +90,8 @@ class SmtCheckerRSC(object): 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)) + And(enc_nz, v >= 0, v_ctx >= 0, v <= e_max, v_ctx <= e_max) + ) return enc_nz @@ -116,8 +115,7 @@ 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: @@ -129,7 +127,7 @@ class SmtCheckerRSC(object): if rcts_for_prod_entity == [] and meta_reactions == []: # this should never happen - return simplify(self.v[level+1][prod_entity] == 0) + return simplify(self.v[level + 1][prod_entity] == 0) enc_enabledness = False @@ -140,30 +138,42 @@ class SmtCheckerRSC(object): enc_ordinary_reactions_enabledness = False for reactants, inhibitors, products in rcts_for_prod_entity: - enc_reactants = True for reactant, concentration in reactants: - enc_reactants = simplify(And(enc_reactants, Or( - self.v[level][reactant] >= concentration, self.v_ctx[level][reactant] >= concentration))) + enc_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 = 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_products = self.v[level + 1][products[0][0]] == products[0][1] + enc_rct_prod = simplify(If(enc_rct_enabled, enc_products, enc_rct_prod)) enc_enabledness = simplify(Or(enc_enabledness, enc_rct_enabled)) enc_ordinary_reactions_enabledness = simplify( - Or(enc_ordinary_reactions_enabledness, enc_rct_enabled)) + Or(enc_ordinary_reactions_enabledness, enc_rct_enabled) + ) # -------- 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 @@ -171,100 +181,127 @@ class SmtCheckerRSC(object): 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 = 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))) + 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], 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_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 + 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], 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) + 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_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))) + 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_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 + 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])) + 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_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)) + 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""" @@ -279,11 +316,11 @@ 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))) + 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 @@ -294,7 +331,7 @@ class SmtCheckerRSC(object): for src, ctx, dst in self.ca.transitions: src_enc = self.ca_state[level] == src - dst_enc = self.ca_state[level+1] == dst + dst_enc = self.ca_state[level + 1] == dst all_ent = set(range(len(self.rs.background_set))) @@ -346,14 +383,12 @@ class SmtCheckerRSC(object): return simplify(enc) def decode_witness(self, max_level, print_model=False): - m = self.solver.model() if print_model: print(m) - for level in range(max_level+1): - + for level in range(max_level + 1): print("\n{: >70}".format("[ level=" + repr(level) + " ]")) print(" State: {", end=""), @@ -361,11 +396,10 @@ class SmtCheckerRSC(object): var_rep = repr(m[self.v[level][var_id]]) if not var_rep.isdigit(): raise RuntimeError( - "unexpected: representation is not a positive integer") + "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(" }") @@ -376,22 +410,28 @@ class SmtCheckerRSC(object): var_rep = repr(m[self.v_ctx[level][var_id]]) if not var_rep.isdigit(): raise RuntimeError( - "unexpected: representation is not a positive integer") + "unexpected: representation is not a positive integer" + ) if int(var_rep) > 0: print( " " + self.rs.get_entity_name(var_id) + "=" + var_rep, - end="") + end="", + ) print(" }") def check_rsltl( - self, formula, print_witness=True, print_time=True, print_mem=True, - max_level=None): + 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: @@ -408,48 +448,61 @@ class SmtCheckerRSC(object): encoder = rsLTL_Encoder(self) encoder.load_variables( - var_rs=self.v, - var_ctx=self.v_ctx, - var_loop_pos=self.loop_position) + 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)) + self.enc_concentration_levels_assertion(self.current_level + 1) + ) print( - "\n{:-^70}".format("[ Working at level=" + str(self.current_level) + " ]")) + "\n{:-^70}".format( + "[ Working at level=" + str(self.current_level) + " ]" + ) + ) stdout.flush() # reachability test: 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))) + "[" + + 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) @@ -457,12 +510,10 @@ 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: @@ -472,25 +523,29 @@ class SmtCheckerRSC(object): if print_time: # stop = time() stop = resource.getrusage(resource.RUSAGE_SELF).ru_utime - self.verification_time = stop-start + self.verification_time = stop - start print() print( - "\n[i] {: >60}".format( - " Time: " + repr(self.verification_time) + " s")) + "\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")) + " 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 - for i in range(levels+1): + for i in range(levels + 1): self.prepare_all_variables() self.current_level += 1 @@ -511,7 +566,6 @@ class SmtCheckerRSC(object): return eq_enc def get_loop_encodings(self): - k = self.current_level loop_var = self.loop_position @@ -523,14 +577,16 @@ class SmtCheckerRSC(object): 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() @@ -550,27 +606,30 @@ class SmtCheckerRSC(object): while True: self.prepare_all_variables() self.solver.add( - self.enc_concentration_levels_assertion( - self.current_level + 1)) + self.enc_concentration_levels_assertion(self.current_level + 1) + ) print( - "\n{:-^70}".format("[ Working at level=" + str(self.current_level) + " ]")) + "\n{:-^70}".format( + "[ Working at level=" + str(self.current_level) + " ]" + ) + ) stdout.flush() # reachability test: - 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))) + "[" + + 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) @@ -578,12 +637,10 @@ 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: @@ -593,25 +650,35 @@ class SmtCheckerRSC(object): if print_time: # stop = time() stop = resource.getrusage(resource.RUSAGE_SELF).ru_utime - self.verification_time = stop-start + self.verification_time = stop - start print() print( - "\n[i] {: >60}".format( - " Time: " + repr(self.verification_time) + " s")) + "\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")) + " 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() @@ -627,8 +694,7 @@ class SmtCheckerRSC(object): 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: @@ -643,9 +709,9 @@ class SmtCheckerRSC(object): result = self.solver.check() if result == sat: print( - "\n[+] " + - colour_str( - C_RED, "SAT at level=" + str(self.current_level))) + "\n[+] " + + colour_str(C_RED, "SAT at level=" + str(self.current_level)) + ) if print_witness: self.decode_witness(self.current_level) break diff --git a/reactics-smt/smt/smt_checker_rsc_param.py b/reactics-smt/smt/smt_checker_rsc_param.py index 0a6df1d..2d48fcf 100644 --- a/reactics-smt/smt/smt_checker_rsc_param.py +++ b/reactics-smt/smt/smt_checker_rsc_param.py @@ -23,14 +23,13 @@ def z3_max(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") + "RS and CA with concentrations (and parameters) expected" + ) self.rs = rsca.rs self.ca = rsca.ca @@ -92,7 +91,6 @@ class SmtCheckerRSCParam(object): 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 @@ -100,8 +98,11 @@ class SmtCheckerRSCParam(object): 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) @@ -111,8 +112,7 @@ class SmtCheckerRSCParam(object): 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""" @@ -188,22 +188,26 @@ class SmtCheckerRSCParam(object): 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, []) @@ -224,14 +228,13 @@ class SmtCheckerRSCParam(object): """ 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))) + "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 @@ -266,23 +269,17 @@ class SmtCheckerRSCParam(object): 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_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_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) @@ -296,8 +293,11 @@ class SmtCheckerRSCParam(object): 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 @@ -305,15 +305,14 @@ class SmtCheckerRSCParam(object): 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)) + 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] + 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 @@ -326,7 +325,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)) @@ -335,8 +335,11 @@ class SmtCheckerRSCParam(object): def enc_transition_relation(self, level, path_idx): return simplify( - And(self.enc_rs_trans(level, path_idx), - self.enc_automaton_trans(level, path_idx))) + And( + self.enc_rs_trans(level, path_idx), + self.enc_automaton_trans(level, path_idx), + ) + ) def enc_param_sanity_for_reactions(self): """R < I constraint (R n I = 0)""" @@ -344,23 +347,21 @@ class SmtCheckerRSCParam(object): rct_inh_constr = True 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 @@ -370,8 +371,10 @@ class SmtCheckerRSCParam(object): rct_inh_constr = And( rct_inh_constr, Implies( - self.v_param[inh_param_name][entity] > 0, conc < - self.v_param[inh_param_name][entity])) + 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 @@ -379,7 +382,8 @@ class SmtCheckerRSCParam(object): 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, self.v_param[rct_param_name][entity] < conc + ) return rct_inh_constr @@ -405,31 +409,52 @@ class SmtCheckerRSCParam(object): if is_param(reactants): param_name = reactants.name 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_ctx[path_idx][level][entity] >= self.v_param[param_name][entity])) + 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_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_ctx[path_idx][level][entity] >= conc)) + enc_reactants = And( + enc_reactants, + Or( + self.path_v[path_idx][level][entity] >= conc, + self.path_v_ctx[path_idx][level][entity] >= conc, + ), + ) # ** INHIBITORS ****************************************** enc_inhibitors = True if is_param(inhibitors): param_name = inhibitors.name for entity in self.rs.set_of_bgset_ids: - 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_ctx[path_idx][level][entity] < self.v_param[param_name][entity]))) + 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_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_ctx[path_idx][level][entity] < conc)) + enc_inhibitors = And( + enc_inhibitors, + And( + self.path_v[path_idx][level][entity] < conc, + self.path_v_ctx[path_idx][level][entity] < conc, + ), + ) # ** PRODUCTS ******************************************* enc_products = True @@ -438,26 +463,38 @@ class SmtCheckerRSCParam(object): 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, + 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, + 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 @@ -467,8 +504,7 @@ class SmtCheckerRSCParam(object): # # ~(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) @@ -491,7 +527,12 @@ class SmtCheckerRSCParam(object): 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)) + Or( + enc_cond, + self.path_v[path_idx][level][entity] > 0, + self.path_v_ctx[path_idx][level][entity] > 0, + ) + ) return enc_cond @@ -533,25 +574,29 @@ class SmtCheckerRSCParam(object): # => # {products}[level+1] # - current_v_improd_for_entities = self.path_v_improd_for_entities[path_idx][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))) + 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_general_cond = self.enc_general_reaction_enabledness(level, path_idx) - enc_trans_with_max = simplify( - And(enc_general_cond, enc_max_prod, enc_trans)) + enc_trans_with_max = simplify(And(enc_general_cond, enc_max_prod, enc_trans)) # print(enc_trans_with_max) return enc_trans_with_max def enc_max(self, elements): - enc = None if elements == []: @@ -561,7 +606,6 @@ class SmtCheckerRSCParam(object): enc = z3_max(0, elements[0]) elif len(elements) > 1: - enc = 0 for i in range(len(elements) - 1): enc = z3_max(enc, z3_max(elements[i], elements[i + 1])) @@ -575,7 +619,7 @@ class SmtCheckerRSCParam(object): 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 + dst_enc = self.path_ca_state[path_idx][level + 1] == dst all_ent = set(range(len(self.rs.background_set))) @@ -586,11 +630,13 @@ class SmtCheckerRSCParam(object): for e, c in ctx: ctx_enc = simplify( - And(ctx_enc, self.path_v_ctx[path_idx][level][e] == c)) + 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)) + 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)) @@ -641,7 +687,6 @@ class SmtCheckerRSCParam(object): print(m) for level in range(max_level + 1): - print("\n{: >70}".format("[ level=" + repr(level) + " ]")) print(" State: {", end=""), @@ -649,11 +694,10 @@ class SmtCheckerRSCParam(object): var_rep = repr(m[self.path_v[path_idx][level][var_id]]) if not var_rep.isdigit(): raise RuntimeError( - "unexpected: representation is not a positive integer") + "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(" }") if level != max_level: @@ -663,11 +707,13 @@ class SmtCheckerRSCParam(object): var_rep = repr(m[self.path_v_ctx[path_idx][level][var_id]]) if not var_rep.isdigit(): raise RuntimeError( - "unexpected: representation is not a positive integer") + "unexpected: representation is not a positive integer" + ) if int(var_rep) > 0: print( " " + self.rs.get_entity_name(var_id) + "=" + var_rep, - end="") + end="", + ) print(" }") print() @@ -675,29 +721,34 @@ class SmtCheckerRSCParam(object): 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))) + print_info( + "Generating the encoding for {:s} ({:d} of {:d})".format( + str(formula), path_idx + 1, len(formulae_list) + ) + ) encoder.load_variables( var_rs=self.path_v[path_idx], var_ctx=self.path_v_ctx[path_idx], - var_loop_pos=self.path_loop_position[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])) + print_info( + "Cache hits: {:d}, encode calls: {:d} (approx: {:d})".format( + 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 ]")) @@ -707,7 +758,6 @@ class SmtCheckerRSCParam(object): self.print_parameter_valuations() def print_parameter_valuations(self): - m = self.solver.model() print("\n Parameters:\n") @@ -721,25 +771,24 @@ class SmtCheckerRSCParam(object): var_rep = repr(m[params[entity]]) if not var_rep.isdigit(): raise RuntimeError( - "unexpected: representation is not a positive integer") + "unexpected: representation is not a positive integer" + ) if int(var_rep) > 0: print( - " " + str(self.rs.get_entity_name(entity)) + "=" + - str(var_rep), - end="") + " " + 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()) + self.enc_concentration_levels_assertion(level, path_idx) + ) + additional_assertions.append(self.enc_param_concentration_levels_assertion()) return additional_assertions @@ -750,22 +799,25 @@ class SmtCheckerRSCParam(object): return enc_trans def print_level(self): - print( - "{:->70}".format("[ level=" + str(self.current_level) + " done ]")) + print("{:->70}".format("[ level=" + str(self.current_level) + " done ]")) def check_rsltl( - self, formulae_list, - print_witness=True, - print_time=True, print_mem=True, - max_level=None, cont_if_sat=False, - param_constr=None): + 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 - * max_level -- if not None, the methods + * max_level -- if not None, the methods stops at the specified level * cont_if_sat -- if True, then the method continues up until max_level is @@ -783,7 +835,7 @@ class SmtCheckerRSCParam(object): print_info("Running rsLTL bounded model checking") print_info("Tested formulae:") for form in formulae_list: - print_info(" "*4 + str(form)) + print_info(" " * 4 + str(form)) print_info("INITIALISING...") @@ -804,8 +856,8 @@ class SmtCheckerRSCParam(object): # assertions for all the paths and parameters self.solver_add( - self.enc_concentration_levels_assertions_for_paths( - 0, num_of_paths)) + 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()) @@ -822,9 +874,11 @@ class SmtCheckerRSCParam(object): print_info("STARTING TO ITERATE...") while True: - print( - "\n{:-^70}".format("[ Working at level=" + str(self.current_level) + " ]")) + "\n{:-^70}".format( + "[ Working at level=" + str(self.current_level) + " ]" + ) + ) # reachability test: self.solver.push() @@ -841,8 +895,9 @@ class SmtCheckerRSCParam(object): print_info("Testing satisfiability...") result = self.solver.check() if result == sat: - print_positive(green_str( - "SAT at level={:d}".format(self.current_level))) + print_positive( + green_str("SAT at level={:d}".format(self.current_level)) + ) # print(self.solver.model()) if print_witness: self.print_witness(formulae_list) @@ -861,11 +916,14 @@ class SmtCheckerRSCParam(object): # assertions for all the paths self.solver_add( self.enc_concentration_levels_assertions_for_paths( - self.current_level + 1, num_of_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() @@ -880,22 +938,24 @@ class SmtCheckerRSCParam(object): stop = resource.getrusage(resource.RUSAGE_SELF).ru_utime self.verification_time = stop - start print() - print_info("{: >60}".format( - " Time: " + repr(self.verification_time) + " s")) + print_info("{: >60}".format(" Time: " + repr(self.verification_time) + " s")) def print_mem(self): print_info( "{: >60}".format( - " Memory: " + - repr( - resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / - (1024 * 1024)) + " MB")) + " 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 - for i in range(levels+1): + for i in range(levels + 1): self.prepare_all_variables() self.current_level += 1 @@ -916,7 +976,6 @@ class SmtCheckerRSCParam(object): return eq_enc def get_loop_encodings(self): - k = self.current_level loop_var = self.loop_position @@ -928,9 +987,10 @@ class SmtCheckerRSCParam(object): 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 @@ -947,8 +1007,9 @@ class SmtCheckerRSCParam(object): 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() @@ -968,27 +1029,30 @@ class SmtCheckerRSCParam(object): while True: self.prepare_all_variables() self.solver_add( - self.enc_concentration_levels_assertion( - self.current_level + 1)) + self.enc_concentration_levels_assertion(self.current_level + 1) + ) print( - "\n{:-^70}".format("[ Working at level=" + str(self.current_level) + " ]")) + "\n{:-^70}".format( + "[ Working at level=" + str(self.current_level) + " ]" + ) + ) stdout.flush() # reachability test: - 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))) + "[" + + 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) @@ -996,12 +1060,10 @@ 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: @@ -1011,25 +1073,35 @@ class SmtCheckerRSCParam(object): if print_time: # stop = time() stop = resource.getrusage(resource.RUSAGE_SELF).ru_utime - self.verification_time = stop-start + self.verification_time = stop - start print() print( - "\n[i] {: >60}".format( - " Time: " + repr(self.verification_time) + " s")) + "\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")) + " 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() @@ -1045,8 +1117,7 @@ class SmtCheckerRSCParam(object): 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: @@ -1061,9 +1132,9 @@ class SmtCheckerRSCParam(object): result = self.solver.check() if result == sat: print( - "\n[+] " + - colour_str( - C_RED, "SAT at level=" + str(self.current_level))) + "\n[+] " + + colour_str(C_RED, "SAT at level=" + str(self.current_level)) + ) if print_witness: self.decode_witness(self.current_level) break