updated imports
This commit is contained in:
Artur Meski
@@ -1,11 +1,7 @@
|
|||||||
#!/usr/bin/env python
|
#!/usr/bin/env python
|
||||||
|
|
||||||
from rctsys import *
|
from rs import *
|
||||||
from distrib_rctsys import *
|
from smt import *
|
||||||
from smtchecker import SmtChecker
|
|
||||||
from smtcheckerpgrs import SmtCheckerPGRS
|
|
||||||
from smtcheckerdistribrs import SmtCheckerDistribRS
|
|
||||||
from smtcheckerrsc import SmtCheckerRSC
|
|
||||||
import sys
|
import sys
|
||||||
import resource
|
import resource
|
||||||
|
|
||||||
|
|||||||
7
rssmt.py
7
rssmt.py
@@ -6,11 +6,8 @@
|
|||||||
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from rctsys import *
|
from rs import *
|
||||||
from smtchecker import SmtChecker
|
from smt import *
|
||||||
from smtcheckerpgrs import SmtCheckerPGRS
|
|
||||||
from smtcheckerdistribrs import SmtCheckerDistribRS
|
|
||||||
from smtcheckerrsc import SmtCheckerRSC
|
|
||||||
import sys
|
import sys
|
||||||
import rs_examples
|
import rs_examples
|
||||||
|
|
||||||
|
|||||||
530
smtchecker.py
530
smtchecker.py
@@ -1,530 +0,0 @@
|
|||||||
"""
|
|
||||||
SMT-based Model Checking Module for RS
|
|
||||||
"""
|
|
||||||
|
|
||||||
from z3 import *
|
|
||||||
from time import time
|
|
||||||
from sys import stdout
|
|
||||||
|
|
||||||
class SmtChecker(object):
|
|
||||||
|
|
||||||
def __init__(self, rs):
|
|
||||||
|
|
||||||
##############################################################
|
|
||||||
# Encoded RS
|
|
||||||
##############################################################
|
|
||||||
rs.sanity_check()
|
|
||||||
self.reaction_system = rs
|
|
||||||
|
|
||||||
##############################################################
|
|
||||||
# SMT variables
|
|
||||||
##############################################################
|
|
||||||
self.v = []
|
|
||||||
self.v_init = []
|
|
||||||
|
|
||||||
#self.vSucc = []
|
|
||||||
#self.vSuccInit = []
|
|
||||||
|
|
||||||
self.v_ctx = []
|
|
||||||
|
|
||||||
self.next_level_to_encode = 0
|
|
||||||
|
|
||||||
##############################################################
|
|
||||||
# SMT solver instance
|
|
||||||
##############################################################
|
|
||||||
self.solver = Solver()
|
|
||||||
|
|
||||||
#def smtVar(self, level, entityID, primed=False):
|
|
||||||
# return "?"
|
|
||||||
|
|
||||||
def prepare_context_variables(self):
|
|
||||||
"""Encodes all the context variables"""
|
|
||||||
|
|
||||||
level = self.next_level_to_encode
|
|
||||||
|
|
||||||
variables = []
|
|
||||||
for entity in self.reaction_system.background_set:
|
|
||||||
variables.append(Bool("C"+str(level)+"_"+entity))
|
|
||||||
|
|
||||||
self.v_ctx.append(variables)
|
|
||||||
|
|
||||||
def prepare_state_variables(self):
|
|
||||||
"""Encodes all the state variables (including successors)"""
|
|
||||||
|
|
||||||
level = self.next_level_to_encode
|
|
||||||
|
|
||||||
variables = []
|
|
||||||
#variablesSucc = []
|
|
||||||
for entity in self.reaction_system.background_set:
|
|
||||||
variables.append(Bool("L"+str(level)+"_"+entity))
|
|
||||||
#variablesSucc.append(Bool("R"+str(level)+"_"+entity))
|
|
||||||
|
|
||||||
self.v.append(variables)
|
|
||||||
#self.vSucc.append(variablesSucc)
|
|
||||||
|
|
||||||
self.v_init.append(Bool("L"+str(level)+"_Init"))
|
|
||||||
#self.vSuccInit.append(Bool("R"+str(level)+"_Init"))
|
|
||||||
|
|
||||||
def prepare_all_variables(self):
|
|
||||||
"""Encodes all the variables"""
|
|
||||||
|
|
||||||
self.prepare_state_variables()
|
|
||||||
self.prepare_context_variables()
|
|
||||||
self.next_level_to_encode += 1
|
|
||||||
|
|
||||||
def enc_init_state(self, level):
|
|
||||||
"""Encodes the initial state at the given level"""
|
|
||||||
|
|
||||||
init_state_enc = self.v_init[level]
|
|
||||||
|
|
||||||
for v in self.v[level]:
|
|
||||||
init_state_enc = simplify(And(init_state_enc, Not(v)))
|
|
||||||
|
|
||||||
return init_state_enc
|
|
||||||
|
|
||||||
def enc_init_contexts(self, level):
|
|
||||||
"""Encodes the initial contexts set at the given level"""
|
|
||||||
|
|
||||||
init_contexts_set_enc = False # Or
|
|
||||||
|
|
||||||
for ctx in self.reaction_system.init_contexts:
|
|
||||||
single_ctx_enc = True # And
|
|
||||||
|
|
||||||
not_ctx_entities = list(range(0, len(self.reaction_system.background_set)))
|
|
||||||
for entity in ctx:
|
|
||||||
single_ctx_enc = simplify(And(single_ctx_enc, self.v_ctx[level][entity]))
|
|
||||||
not_ctx_entities.remove(entity)
|
|
||||||
|
|
||||||
for entity in not_ctx_entities:
|
|
||||||
single_ctx_enc = simplify(And(single_ctx_enc, Not(self.v_ctx[level][entity])))
|
|
||||||
|
|
||||||
init_contexts_set_enc = simplify(Or(init_contexts_set_enc, single_ctx_enc))
|
|
||||||
|
|
||||||
#print("initContextSetEnc: " + repr(initContextsSetEnc))
|
|
||||||
return init_contexts_set_enc
|
|
||||||
|
|
||||||
def enc_not_allowed_contexts(self, level):
|
|
||||||
"""Encodes all the context entities that are not allowed in the context sets"""
|
|
||||||
|
|
||||||
bg_set = set(range(0,len(self.reaction_system.background_set)))
|
|
||||||
ctx_ent_set = set(self.reaction_system.context_entities)
|
|
||||||
|
|
||||||
not_ctx_ent_set = bg_set.difference(ctx_ent_set)
|
|
||||||
|
|
||||||
enc = True
|
|
||||||
for entity in not_ctx_ent_set:
|
|
||||||
enc = simplify(And(enc, Not(self.v_ctx[level][entity])))
|
|
||||||
|
|
||||||
return enc
|
|
||||||
|
|
||||||
def enc_enabledness(self, level, prod_entity):
|
|
||||||
"""Encodes the enabledness condition for a given level and a given entity"""
|
|
||||||
|
|
||||||
rcts_for_prod_entity = self.reaction_system.get_reactions_by_product()[prod_entity]
|
|
||||||
|
|
||||||
if rcts_for_prod_entity == []:
|
|
||||||
return False
|
|
||||||
|
|
||||||
#encInitEnab = simplify(Or(And(self.vInit[level], self.encInitContexts(level)),
|
|
||||||
# And(Not(self.vInit[level]), self.encNotAllowedContexts(level))))
|
|
||||||
|
|
||||||
enc_rct_prod = False
|
|
||||||
for ri_pair in rcts_for_prod_entity: # reactants-inhibitors pair
|
|
||||||
enc_reactants = True
|
|
||||||
enc_inhibitors = True
|
|
||||||
for reactant in ri_pair[0]:
|
|
||||||
enc_reactants = simplify(And(enc_reactants,
|
|
||||||
Or(self.v[level][reactant], self.v_ctx[level][reactant])))
|
|
||||||
for inhibitor in ri_pair[1]:
|
|
||||||
enc_inhibitors = simplify(And(enc_inhibitors,
|
|
||||||
Not(Or(self.v[level][inhibitor], self.v_ctx[level][inhibitor]))))
|
|
||||||
|
|
||||||
enc_rct_prod = simplify(Or(enc_rct_prod, And(enc_reactants, enc_inhibitors)))
|
|
||||||
|
|
||||||
#print("encEnabledness(" + repr(prodEntity) + "): " + repr(simplify(And(encInitEnab, encRctProd))))
|
|
||||||
#return simplify(And(encInitEnab, encRctProd))
|
|
||||||
return enc_rct_prod
|
|
||||||
|
|
||||||
def enc_entity_production(self, level, prod_entity):
|
|
||||||
"""Encodes the production of a given entity from a given level at level+1"""
|
|
||||||
|
|
||||||
enc_enab_cond = self.enc_enabledness(level, prod_entity)
|
|
||||||
|
|
||||||
enc_ent_prod = Or(And(enc_enab_cond, self.v[level+1][prod_entity]),
|
|
||||||
And(Not(enc_enab_cond), Not(self.v[level+1][prod_entity])))
|
|
||||||
|
|
||||||
return enc_ent_prod
|
|
||||||
|
|
||||||
def enc_transition_relation(self, level):
|
|
||||||
"""Encodes the transition relation"""
|
|
||||||
|
|
||||||
unused_entities = list(range(0,len(self.reaction_system.background_set)))
|
|
||||||
|
|
||||||
enc_trans = True
|
|
||||||
|
|
||||||
for prod_entity in self.reaction_system.get_reactions_by_product():
|
|
||||||
unused_entities.remove(prod_entity)
|
|
||||||
|
|
||||||
enc_trans = simplify(And(enc_trans, self.enc_entity_production(level, prod_entity)))
|
|
||||||
|
|
||||||
enc_trans = simplify(And(enc_trans, Not(self.v_init[level+1])))
|
|
||||||
|
|
||||||
for prod_entity in unused_entities:
|
|
||||||
enc_trans = simplify(And(enc_trans, Not(self.v[level+1][prod_entity])))
|
|
||||||
|
|
||||||
if level == 0:
|
|
||||||
enc_init_enab = self.enc_init_contexts(level)
|
|
||||||
else: # level > 0:
|
|
||||||
enc_init_enab = self.enc_not_allowed_contexts(level)
|
|
||||||
|
|
||||||
#encInitEnab = simplify(Or(And(self.vInit[level], self.encInitContexts(level)),
|
|
||||||
# And(Not(self.vInit[level]), self.encNotAllowedContexts(level))))
|
|
||||||
|
|
||||||
enc_trans = simplify(And(enc_trans, enc_init_enab))
|
|
||||||
|
|
||||||
return enc_trans
|
|
||||||
|
|
||||||
def enc_state(self, level, state):
|
|
||||||
"""Encodes the state at the given level"""
|
|
||||||
|
|
||||||
enc = Not(self.v_init[level])
|
|
||||||
|
|
||||||
state_ids = self.reaction_system.get_state_ids(state)
|
|
||||||
|
|
||||||
for entity in state_ids:
|
|
||||||
enc = And(enc, self.v[level][entity])
|
|
||||||
|
|
||||||
not_in_state = set(range(0, len(self.reaction_system.background_set)))
|
|
||||||
not_in_state = not_in_state.difference(set(state_ids))
|
|
||||||
|
|
||||||
for entity in not_in_state:
|
|
||||||
enc = And(enc, Not(self.v[level][entity]))
|
|
||||||
|
|
||||||
return enc
|
|
||||||
|
|
||||||
def decode_witness(self, max_level):
|
|
||||||
|
|
||||||
m = self.solver.model()
|
|
||||||
|
|
||||||
for level in range(0,max_level+1):
|
|
||||||
|
|
||||||
print("\n[Level=" + repr(level) + "]")
|
|
||||||
|
|
||||||
if repr(m[self.v_init[level]]) == "True":
|
|
||||||
print("** Initial state")
|
|
||||||
|
|
||||||
print("State:\n{"),
|
|
||||||
for var_id in range(0, len(self.v[level])):
|
|
||||||
if repr(m[self.v[level][var_id]]) == "True":
|
|
||||||
print("\t" + self.reaction_system.get_entity_name(var_id)),
|
|
||||||
print("}")
|
|
||||||
|
|
||||||
if level != max_level:
|
|
||||||
print("Context set:"),
|
|
||||||
print("{"),
|
|
||||||
for var_id in range(0, len(self.v[level])):
|
|
||||||
if repr(m[self.v_ctx[level][var_id]]) == "True":
|
|
||||||
print("\t" + self.reaction_system.get_entity_name(var_id)),
|
|
||||||
print("}")
|
|
||||||
|
|
||||||
|
|
||||||
def check_reachability(self, state, print_witness=True, print_time=False):
|
|
||||||
"""Main testing function"""
|
|
||||||
|
|
||||||
if print_time:
|
|
||||||
start = time()
|
|
||||||
|
|
||||||
self.prepare_all_variables()
|
|
||||||
self.solver.add(self.enc_init_state(0))
|
|
||||||
current_level = 0
|
|
||||||
|
|
||||||
while True:
|
|
||||||
#print("Level: " + str(current_level))
|
|
||||||
print("\rLevel: " + str(current_level)),
|
|
||||||
stdout.flush()
|
|
||||||
|
|
||||||
self.prepare_all_variables()
|
|
||||||
|
|
||||||
# reachability test:
|
|
||||||
self.solver.push()
|
|
||||||
self.solver.add(self.enc_state(current_level,state))
|
|
||||||
|
|
||||||
result = self.solver.check()
|
|
||||||
print(result)
|
|
||||||
if result == sat:
|
|
||||||
print("\nSAT")
|
|
||||||
if print_witness:
|
|
||||||
self.decode_witness(current_level)
|
|
||||||
break
|
|
||||||
else:
|
|
||||||
self.solver.pop()
|
|
||||||
|
|
||||||
self.solver.add(self.enc_transition_relation(current_level))
|
|
||||||
|
|
||||||
current_level += 1
|
|
||||||
|
|
||||||
if print_time:
|
|
||||||
stop = time()
|
|
||||||
print("Time: " + repr(stop-start))
|
|
||||||
|
|
||||||
|
|
||||||
class SmtCheckerPGRS(object):
|
|
||||||
|
|
||||||
def __init__(self, rs):
|
|
||||||
|
|
||||||
##############################################################
|
|
||||||
# Encoded RS
|
|
||||||
##############################################################
|
|
||||||
rs.sanity_check()
|
|
||||||
self.reaction_system = rs
|
|
||||||
|
|
||||||
##############################################################
|
|
||||||
# SMT variables
|
|
||||||
##############################################################
|
|
||||||
self.v = []
|
|
||||||
self.v_init = []
|
|
||||||
|
|
||||||
#self.vSucc = []
|
|
||||||
#self.vSuccInit = []
|
|
||||||
|
|
||||||
self.v_ctx = []
|
|
||||||
|
|
||||||
self.next_level_to_encode = 0
|
|
||||||
|
|
||||||
##############################################################
|
|
||||||
# SMT solver instance
|
|
||||||
##############################################################
|
|
||||||
self.solver = Solver()
|
|
||||||
|
|
||||||
#def smtVar(self, level, entityID, primed=False):
|
|
||||||
# return "?"
|
|
||||||
|
|
||||||
def prepare_context_variables(self):
|
|
||||||
"""Encodes all the context variables"""
|
|
||||||
|
|
||||||
level = self.next_level_to_encode
|
|
||||||
|
|
||||||
variables = []
|
|
||||||
for entity in self.reaction_system.background_set:
|
|
||||||
variables.append(Bool("C"+str(level)+"_"+entity))
|
|
||||||
|
|
||||||
self.v_ctx.append(variables)
|
|
||||||
|
|
||||||
def prepare_state_variables(self):
|
|
||||||
"""Encodes all the state variables (including successors)"""
|
|
||||||
|
|
||||||
level = self.next_level_to_encode
|
|
||||||
|
|
||||||
variables = []
|
|
||||||
#variablesSucc = []
|
|
||||||
for entity in self.reaction_system.background_set:
|
|
||||||
variables.append(Bool("L"+str(level)+"_"+entity))
|
|
||||||
#variablesSucc.append(Bool("R"+str(level)+"_"+entity))
|
|
||||||
|
|
||||||
self.v.append(variables)
|
|
||||||
#self.vSucc.append(variablesSucc)
|
|
||||||
|
|
||||||
self.v_init.append(Bool("L"+str(level)+"_Init"))
|
|
||||||
#self.vSuccInit.append(Bool("R"+str(level)+"_Init"))
|
|
||||||
|
|
||||||
def prepare_all_variables(self):
|
|
||||||
"""Encodes all the variables"""
|
|
||||||
|
|
||||||
self.prepare_state_variables()
|
|
||||||
self.prepare_context_variables()
|
|
||||||
self.next_level_to_encode += 1
|
|
||||||
|
|
||||||
def enc_init_state(self, level):
|
|
||||||
"""Encodes the initial state at the given level"""
|
|
||||||
|
|
||||||
init_state_enc = self.v_init[level]
|
|
||||||
|
|
||||||
for v in self.v[level]:
|
|
||||||
init_state_enc = simplify(And(init_state_enc, Not(v)))
|
|
||||||
|
|
||||||
return init_state_enc
|
|
||||||
|
|
||||||
def enc_init_contexts(self, level):
|
|
||||||
"""Encodes the initial contexts set at the given level"""
|
|
||||||
|
|
||||||
init_contexts_set_enc = False # Or
|
|
||||||
|
|
||||||
for ctx in self.reaction_system.init_contexts:
|
|
||||||
single_ctx_enc = True # And
|
|
||||||
|
|
||||||
not_ctx_entities = list(range(0, len(self.reaction_system.background_set)))
|
|
||||||
for entity in ctx:
|
|
||||||
single_ctx_enc = simplify(And(single_ctx_enc, self.v_ctx[level][entity]))
|
|
||||||
not_ctx_entities.remove(entity)
|
|
||||||
|
|
||||||
for entity in not_ctx_entities:
|
|
||||||
single_ctx_enc = simplify(And(single_ctx_enc, Not(self.v_ctx[level][entity])))
|
|
||||||
|
|
||||||
init_contexts_set_enc = simplify(Or(init_contexts_set_enc, single_ctx_enc))
|
|
||||||
|
|
||||||
#print("initContextSetEnc: " + repr(initContextsSetEnc))
|
|
||||||
return init_contexts_set_enc
|
|
||||||
|
|
||||||
def enc_not_allowed_contexts(self, level):
|
|
||||||
"""Encodes all the context entities that are not allowed in the context sets"""
|
|
||||||
|
|
||||||
bg_set = set(range(0,len(self.reaction_system.background_set)))
|
|
||||||
ctx_ent_set = set(self.reaction_system.context_entities)
|
|
||||||
|
|
||||||
not_ctx_ent_set = bg_set.difference(ctx_ent_set)
|
|
||||||
|
|
||||||
enc = True
|
|
||||||
for entity in not_ctx_ent_set:
|
|
||||||
enc = simplify(And(enc, Not(self.v_ctx[level][entity])))
|
|
||||||
|
|
||||||
return enc
|
|
||||||
|
|
||||||
def enc_enabledness(self, level, prod_entity):
|
|
||||||
"""Encodes the enabledness condition for a given level and a given entity"""
|
|
||||||
|
|
||||||
rcts_for_prod_entity = self.reaction_system.get_reactions_by_product()[prod_entity]
|
|
||||||
|
|
||||||
if rcts_for_prod_entity == []:
|
|
||||||
return False
|
|
||||||
|
|
||||||
#encInitEnab = simplify(Or(And(self.vInit[level], self.encInitContexts(level)),
|
|
||||||
# And(Not(self.vInit[level]), self.encNotAllowedContexts(level))))
|
|
||||||
|
|
||||||
enc_rct_prod = False
|
|
||||||
for ri_pair in rcts_for_prod_entity: # reactants-inhibitors pair
|
|
||||||
enc_reactants = True
|
|
||||||
enc_inhibitors = True
|
|
||||||
for reactant in ri_pair[0]:
|
|
||||||
enc_reactants = simplify(And(enc_reactants,
|
|
||||||
Or(self.v[level][reactant], self.v_ctx[level][reactant])))
|
|
||||||
for inhibitor in ri_pair[1]:
|
|
||||||
enc_inhibitors = simplify(And(enc_inhibitors,
|
|
||||||
Not(Or(self.v[level][inhibitor], self.v_ctx[level][inhibitor]))))
|
|
||||||
|
|
||||||
enc_rct_prod = simplify(Or(enc_rct_prod, And(enc_reactants, enc_inhibitors)))
|
|
||||||
|
|
||||||
#print("encEnabledness(" + repr(prodEntity) + "): " + repr(simplify(And(encInitEnab, encRctProd))))
|
|
||||||
#return simplify(And(encInitEnab, encRctProd))
|
|
||||||
return enc_rct_prod
|
|
||||||
|
|
||||||
def enc_entity_production(self, level, prod_entity):
|
|
||||||
"""Encodes the production of a given entity from a given level at level+1"""
|
|
||||||
|
|
||||||
enc_enab_cond = self.enc_enabledness(level, prod_entity)
|
|
||||||
|
|
||||||
enc_ent_prod = Or(And(enc_enab_cond, self.v[level+1][prod_entity]),
|
|
||||||
And(Not(enc_enab_cond), Not(self.v[level+1][prod_entity])))
|
|
||||||
|
|
||||||
return enc_ent_prod
|
|
||||||
|
|
||||||
def enc_transition_relation(self, level):
|
|
||||||
"""Encodes the transition relation"""
|
|
||||||
|
|
||||||
unused_entities = list(range(0,len(self.reaction_system.background_set)))
|
|
||||||
|
|
||||||
enc_trans = True
|
|
||||||
|
|
||||||
for prod_entity in self.reaction_system.get_reactions_by_product():
|
|
||||||
unused_entities.remove(prod_entity)
|
|
||||||
|
|
||||||
enc_trans = simplify(And(enc_trans, self.enc_entity_production(level, prod_entity)))
|
|
||||||
|
|
||||||
enc_trans = simplify(And(enc_trans, Not(self.v_init[level+1])))
|
|
||||||
|
|
||||||
for prod_entity in unused_entities:
|
|
||||||
enc_trans = simplify(And(enc_trans, Not(self.v[level+1][prod_entity])))
|
|
||||||
|
|
||||||
if level == 0:
|
|
||||||
enc_init_enab = self.enc_init_contexts(level)
|
|
||||||
else: # level > 0:
|
|
||||||
enc_init_enab = self.enc_not_allowed_contexts(level)
|
|
||||||
|
|
||||||
#encInitEnab = simplify(Or(And(self.vInit[level], self.encInitContexts(level)),
|
|
||||||
# And(Not(self.vInit[level]), self.encNotAllowedContexts(level))))
|
|
||||||
|
|
||||||
enc_trans = simplify(And(enc_trans, enc_init_enab))
|
|
||||||
|
|
||||||
return enc_trans
|
|
||||||
|
|
||||||
def enc_state(self, level, state):
|
|
||||||
"""Encodes the state at the given level"""
|
|
||||||
|
|
||||||
enc = Not(self.v_init[level])
|
|
||||||
|
|
||||||
state_ids = self.reaction_system.get_state_ids(state)
|
|
||||||
|
|
||||||
for entity in state_ids:
|
|
||||||
enc = And(enc, self.v[level][entity])
|
|
||||||
|
|
||||||
not_in_state = set(range(0, len(self.reaction_system.background_set)))
|
|
||||||
not_in_state = not_in_state.difference(set(state_ids))
|
|
||||||
|
|
||||||
for entity in not_in_state:
|
|
||||||
enc = And(enc, Not(self.v[level][entity]))
|
|
||||||
|
|
||||||
return enc
|
|
||||||
|
|
||||||
def decode_witness(self, max_level):
|
|
||||||
|
|
||||||
m = self.solver.model()
|
|
||||||
|
|
||||||
for level in range(0,max_level+1):
|
|
||||||
|
|
||||||
print("\n[Level=" + repr(level) + "]")
|
|
||||||
|
|
||||||
if repr(m[self.v_init[level]]) == "True":
|
|
||||||
print("** Initial state")
|
|
||||||
|
|
||||||
print("State:\n{"),
|
|
||||||
for var_id in range(0, len(self.v[level])):
|
|
||||||
if repr(m[self.v[level][var_id]]) == "True":
|
|
||||||
print("\t" + self.reaction_system.get_entity_name(var_id)),
|
|
||||||
print("}")
|
|
||||||
|
|
||||||
if level != max_level:
|
|
||||||
print("Context set:"),
|
|
||||||
print("{"),
|
|
||||||
for var_id in range(0, len(self.v[level])):
|
|
||||||
if repr(m[self.v_ctx[level][var_id]]) == "True":
|
|
||||||
print("\t" + self.reaction_system.get_entity_name(var_id)),
|
|
||||||
print("}")
|
|
||||||
|
|
||||||
|
|
||||||
def check_reachability(self, state, print_witness=True, print_time=False):
|
|
||||||
"""Main testing function"""
|
|
||||||
|
|
||||||
if print_time:
|
|
||||||
start = time()
|
|
||||||
|
|
||||||
self.prepare_all_variables()
|
|
||||||
self.solver.add(self.enc_init_state(0))
|
|
||||||
current_level = 0
|
|
||||||
|
|
||||||
while True:
|
|
||||||
#print("Level: " + str(current_level))
|
|
||||||
print("\rLevel: " + str(current_level)),
|
|
||||||
stdout.flush()
|
|
||||||
|
|
||||||
self.prepare_all_variables()
|
|
||||||
|
|
||||||
# reachability test:
|
|
||||||
self.solver.push()
|
|
||||||
self.solver.add(self.enc_state(current_level,state))
|
|
||||||
|
|
||||||
result = self.solver.check()
|
|
||||||
print(result)
|
|
||||||
if result == sat:
|
|
||||||
print("\nSAT")
|
|
||||||
if print_witness:
|
|
||||||
self.decode_witness(current_level)
|
|
||||||
break
|
|
||||||
else:
|
|
||||||
self.solver.pop()
|
|
||||||
|
|
||||||
self.solver.add(self.enc_transition_relation(current_level))
|
|
||||||
|
|
||||||
current_level += 1
|
|
||||||
|
|
||||||
if print_time:
|
|
||||||
stop = time()
|
|
||||||
print("Time: " + repr(stop-start))
|
|
||||||
@@ -1,432 +0,0 @@
|
|||||||
"""
|
|
||||||
SMT-based Model Checking Module for RS with Context Automaton
|
|
||||||
"""
|
|
||||||
|
|
||||||
from z3 import *
|
|
||||||
from time import time,sleep
|
|
||||||
from sys import stdout
|
|
||||||
import resource
|
|
||||||
|
|
||||||
# def simplify(x):
|
|
||||||
# return x
|
|
||||||
|
|
||||||
class SmtCheckerDistribRS(object):
|
|
||||||
|
|
||||||
def __init__(self, drs, debug_level=1):
|
|
||||||
|
|
||||||
print("[i] Initialising the SMT module")
|
|
||||||
|
|
||||||
drs.sanity_check()
|
|
||||||
|
|
||||||
self.solver = Solver()
|
|
||||||
|
|
||||||
self.debug_level = debug_level
|
|
||||||
self.drs = drs
|
|
||||||
self.n_components = self.drs.components_count
|
|
||||||
|
|
||||||
# encoding variables
|
|
||||||
self.v = [] # level -> component -> variables
|
|
||||||
self.v_ctx = []
|
|
||||||
self.v_act = [] # indicators of which component is active
|
|
||||||
self.ca_state = []
|
|
||||||
|
|
||||||
self.level_to_encode = 0
|
|
||||||
|
|
||||||
def prepare_all_variables(self):
|
|
||||||
"""Encodes the required variables"""
|
|
||||||
|
|
||||||
self.prepare_state_variables()
|
|
||||||
self.prepare_context_variables()
|
|
||||||
self.prepare_activity_variables()
|
|
||||||
|
|
||||||
self.level_to_encode += 1 # prepare for the next invocation
|
|
||||||
|
|
||||||
def prepare_context_variables(self):
|
|
||||||
"""Encodes the context variables"""
|
|
||||||
|
|
||||||
level = self.level_to_encode
|
|
||||||
|
|
||||||
if self.debug_level > 1:
|
|
||||||
print("[ii] Preparing context variables for level=" + str(level))
|
|
||||||
|
|
||||||
level_variables = []
|
|
||||||
|
|
||||||
for i in range(self.n_components):
|
|
||||||
|
|
||||||
comp_variables = []
|
|
||||||
|
|
||||||
for entity in self.drs.background_set:
|
|
||||||
comp_variables.append(Bool("Ctx"+str(level)+"V"+str(i)+"_"+entity))
|
|
||||||
|
|
||||||
level_variables.append(comp_variables)
|
|
||||||
|
|
||||||
self.v_ctx.append(level_variables)
|
|
||||||
|
|
||||||
def prepare_activity_variables(self):
|
|
||||||
"""Encodes the activity variables"""
|
|
||||||
|
|
||||||
level = self.level_to_encode
|
|
||||||
|
|
||||||
if self.debug_level > 1:
|
|
||||||
print("[ii] Preparing activity variables for level=" + str(level))
|
|
||||||
|
|
||||||
level_variables = []
|
|
||||||
|
|
||||||
for i in range(self.n_components):
|
|
||||||
# L - level, A - activity indicator
|
|
||||||
level_variables.append(Bool("L"+str(level)+"A"+str(i)))
|
|
||||||
|
|
||||||
self.v_act.append(level_variables)
|
|
||||||
|
|
||||||
def prepare_state_variables(self):
|
|
||||||
"""Encodes all the state variables"""
|
|
||||||
|
|
||||||
level = self.level_to_encode
|
|
||||||
|
|
||||||
if self.debug_level > 1:
|
|
||||||
print("[ii] Preparing state variables for level=" + str(level))
|
|
||||||
|
|
||||||
level_variables = [] # level vars
|
|
||||||
|
|
||||||
for i in range(self.n_components):
|
|
||||||
|
|
||||||
comp_variables = []
|
|
||||||
|
|
||||||
for entity in self.drs.background_set:
|
|
||||||
# L - level, V - component
|
|
||||||
comp_variables.append(Bool("L"+str(level)+"V"+str(i)+"_"+entity))
|
|
||||||
|
|
||||||
level_variables.append(comp_variables)
|
|
||||||
|
|
||||||
self.v.append(level_variables)
|
|
||||||
|
|
||||||
# single state variable for CA
|
|
||||||
self.ca_state.append(Int("CA"+str(level)+"_state"))
|
|
||||||
|
|
||||||
def enc_init_state(self, level):
|
|
||||||
"""Encodes the initial state at the given level"""
|
|
||||||
|
|
||||||
if self.debug_level > 1:
|
|
||||||
print("[ii] Encoding the initial state for level=" + str(level))
|
|
||||||
|
|
||||||
rs_init_state_enc = True
|
|
||||||
|
|
||||||
for i in range(self.n_components):
|
|
||||||
for v in self.v[level][i]:
|
|
||||||
rs_init_state_enc = simplify(And(rs_init_state_enc, Not(v))) # the initial state is empty
|
|
||||||
|
|
||||||
ca_init_state_enc = self.ca_state[level] == self.drs.get_init_state_id()
|
|
||||||
init_state_enc = simplify(And(rs_init_state_enc, ca_init_state_enc))
|
|
||||||
|
|
||||||
# print ("init_state_enc:\n", init_state_enc)
|
|
||||||
|
|
||||||
return init_state_enc
|
|
||||||
|
|
||||||
def enc_enabledness(self, level, prod_entity, component_id):
|
|
||||||
"""Encodes the enabledness condition for a given level and a given entity"""
|
|
||||||
|
|
||||||
rcts_for_prod_entity = self.drs.get_reactions_by_product(component_id)[prod_entity]
|
|
||||||
|
|
||||||
if rcts_for_prod_entity == []:
|
|
||||||
return False
|
|
||||||
|
|
||||||
enc_rct_prod = False
|
|
||||||
for reactants,inhibitors in rcts_for_prod_entity:
|
|
||||||
|
|
||||||
|
|
||||||
enc_reactants = True
|
|
||||||
for reactant in reactants:
|
|
||||||
|
|
||||||
enc_active_reactants = False
|
|
||||||
for i in range(self.n_components):
|
|
||||||
enc_active_reactants = simplify(Or(enc_active_reactants,
|
|
||||||
And( Or(self.v[level][i][reactant], self.v_ctx[level][i][reactant]), self.v_act[level][i] )))
|
|
||||||
|
|
||||||
enc_reactants = And(enc_reactants, enc_active_reactants)
|
|
||||||
|
|
||||||
|
|
||||||
enc_inhibitors = True
|
|
||||||
for inhibitor in inhibitors:
|
|
||||||
|
|
||||||
enc_active_inhibitors = True
|
|
||||||
for i in range(self.n_components):
|
|
||||||
enc_active_inhibitors = simplify(And(enc_active_inhibitors,
|
|
||||||
And(
|
|
||||||
Or( And(Not(self.v[level][i][inhibitor]), Not(self.v_ctx[level][i][inhibitor])), Not(self.v_act[level][i]) )
|
|
||||||
)))
|
|
||||||
|
|
||||||
enc_inhibitors = simplify(And(enc_inhibitors, enc_active_inhibitors))
|
|
||||||
|
|
||||||
# print("--> enc_inhibitors\n", enc_inhibitors)
|
|
||||||
enc_rct_prod = Or(enc_rct_prod, And(enc_reactants, enc_inhibitors))
|
|
||||||
|
|
||||||
# print("enc_rct_prod:\n", enc_rct_prod)
|
|
||||||
|
|
||||||
enc_rct_prod = simplify(enc_rct_prod)
|
|
||||||
|
|
||||||
return enc_rct_prod
|
|
||||||
|
|
||||||
def enc_entity_production(self, level, prod_entity, component_id):
|
|
||||||
"""Encodes the production of a given entity at level+1 from a given level"""
|
|
||||||
|
|
||||||
enc_enab_cond = self.enc_enabledness(level, prod_entity, component_id)
|
|
||||||
|
|
||||||
enc_base_ent_prod = simplify(Or(And(enc_enab_cond, self.v[level+1][component_id][prod_entity]),
|
|
||||||
And(Not(enc_enab_cond), Not(self.v[level+1][component_id][prod_entity]))))
|
|
||||||
|
|
||||||
enc_active_ent_prod = simplify(And(self.v_act[level][component_id], enc_base_ent_prod))
|
|
||||||
enc_inactive_ent_prod = simplify(And(
|
|
||||||
Not(self.v_act[level][component_id]),
|
|
||||||
self.v[level][component_id][prod_entity] == self.v[level+1][component_id][prod_entity]))
|
|
||||||
|
|
||||||
enc_ent_prod = Or(enc_active_ent_prod, enc_inactive_ent_prod)
|
|
||||||
|
|
||||||
# print("enc_ent_prod:\n", enc_ent_prod)
|
|
||||||
|
|
||||||
return simplify(enc_ent_prod)
|
|
||||||
|
|
||||||
def enc_rs_trans(self, level):
|
|
||||||
"""Encodes the transition relation"""
|
|
||||||
|
|
||||||
enc_trans = True
|
|
||||||
|
|
||||||
for component_id in range(self.n_components):
|
|
||||||
|
|
||||||
print("\rEncoding for reactions: %d/%d" % (component_id,self.n_components-1), flush=True, end="")
|
|
||||||
|
|
||||||
unused_entities = list(range(len(self.drs.background_set)))
|
|
||||||
|
|
||||||
for prod_entity in self.drs.get_reactions_by_product(component_id):
|
|
||||||
unused_entities.remove(prod_entity)
|
|
||||||
|
|
||||||
enc_trans = simplify(And(enc_trans, self.enc_entity_production(level, prod_entity, component_id)))
|
|
||||||
|
|
||||||
for prod_entity in unused_entities:
|
|
||||||
enc_trans = simplify(And(enc_trans, Not(self.v[level+1][component_id][prod_entity])))
|
|
||||||
|
|
||||||
print()
|
|
||||||
# print("enc_rs_trans:\n", enc_trans)
|
|
||||||
|
|
||||||
enc_trans = simplify(enc_trans)
|
|
||||||
|
|
||||||
return enc_trans
|
|
||||||
|
|
||||||
def enc_automaton_trans(self, level):
|
|
||||||
"""Encodes the transition relation for the context automaton"""
|
|
||||||
|
|
||||||
enc_trans = False
|
|
||||||
|
|
||||||
i = 0
|
|
||||||
for src,(components,ctx_set),dst in self.drs.transitions:
|
|
||||||
src_enc = self.ca_state[level] == src
|
|
||||||
dst_enc = self.ca_state[level+1] == dst
|
|
||||||
|
|
||||||
print("\rEncoding for context automaton: %d/%d" % (i,len(self.drs.transitions)-1), flush=True, end="")
|
|
||||||
|
|
||||||
i = i + 1
|
|
||||||
|
|
||||||
# contexts {
|
|
||||||
ctx_set_enc = True
|
|
||||||
for comp_id in range(self.n_components):
|
|
||||||
|
|
||||||
all_ent = set(range(len(self.drs.background_set)))
|
|
||||||
incl_ctx = ctx_set[comp_id]
|
|
||||||
excl_ctx = all_ent - incl_ctx
|
|
||||||
|
|
||||||
ctx_enc = True
|
|
||||||
|
|
||||||
for c in incl_ctx:
|
|
||||||
ctx_enc = And(ctx_enc, self.v_ctx[level][comp_id][c])
|
|
||||||
for c in excl_ctx:
|
|
||||||
ctx_enc = And(ctx_enc, Not(self.v_ctx[level][comp_id][c]))
|
|
||||||
|
|
||||||
ctx_set_enc = And(ctx_set_enc, ctx_enc)
|
|
||||||
# } contexts
|
|
||||||
|
|
||||||
# active components {
|
|
||||||
all_active = set(range(self.n_components))
|
|
||||||
incl_comp = components
|
|
||||||
excl_comp = all_active - incl_comp
|
|
||||||
|
|
||||||
active_components_enc = True
|
|
||||||
for comp_id in incl_comp:
|
|
||||||
active_components_enc = And(active_components_enc, self.v_act[level][comp_id])
|
|
||||||
for comp_id in excl_comp:
|
|
||||||
active_components_enc = And(active_components_enc, Not(self.v_act[level][comp_id]))
|
|
||||||
# } active components
|
|
||||||
|
|
||||||
cur_trans = And(src_enc, ctx_set_enc, active_components_enc, dst_enc)
|
|
||||||
|
|
||||||
enc_trans = Or(enc_trans, cur_trans)
|
|
||||||
|
|
||||||
print()
|
|
||||||
|
|
||||||
enc_trans = simplify(enc_trans)
|
|
||||||
|
|
||||||
# print("enc_automaton_trans:\n", enc_trans)
|
|
||||||
return enc_trans
|
|
||||||
|
|
||||||
def enc_transition_relation(self, level):
|
|
||||||
|
|
||||||
rs_enc = self.enc_rs_trans(level)
|
|
||||||
aut_enc = self.enc_automaton_trans(level)
|
|
||||||
|
|
||||||
print("Conjunction...", flush=True, end="")
|
|
||||||
|
|
||||||
c = simplify(And(rs_enc, aut_enc))
|
|
||||||
|
|
||||||
print("done.")
|
|
||||||
|
|
||||||
return c
|
|
||||||
|
|
||||||
def enc_state(self, level, global_state):
|
|
||||||
"""Encodes the state at the given level"""
|
|
||||||
|
|
||||||
if len(global_state) != self.n_components:
|
|
||||||
print("EEE: Wrong size of the global state! " + "(is " + str(len(global_state)) + ", should be " + str(self.n_components) + ")")
|
|
||||||
exit(1)
|
|
||||||
|
|
||||||
enc = True
|
|
||||||
|
|
||||||
if self.debug_level > 2:
|
|
||||||
print("[iii] Encoding exclusive/exact global state " + str(global_state) + " for level=" + str(level))
|
|
||||||
|
|
||||||
for i in range(self.n_components):
|
|
||||||
local_state = global_state[i]
|
|
||||||
|
|
||||||
local_state_ids = self.drs.get_state_ids(local_state)
|
|
||||||
|
|
||||||
for entity in local_state_ids:
|
|
||||||
enc = And(enc, self.v[level][i][entity])
|
|
||||||
|
|
||||||
not_in_state = self.drs.set_of_background_ids - set(local_state_ids)
|
|
||||||
|
|
||||||
for e_id in not_in_state:
|
|
||||||
enc = simplify(And(enc, Not(self.v[level][i][e_id])))
|
|
||||||
|
|
||||||
simplify(enc)
|
|
||||||
# print("state:\n", enc)
|
|
||||||
return enc
|
|
||||||
|
|
||||||
def enc_inclusive_state(self, level, global_state):
|
|
||||||
"""Encodes the state at the given level"""
|
|
||||||
|
|
||||||
if len(global_state) != self.n_components:
|
|
||||||
print("EEE: Wrong size of the global state! " + "(is " + str(len(global_state)) + ", should be " + str(self.n_components) + ")")
|
|
||||||
exit(1)
|
|
||||||
|
|
||||||
enc = True
|
|
||||||
|
|
||||||
if self.debug_level > 2:
|
|
||||||
print("[iii] Encoding inclusive/general global state " + str(global_state) + " for level=" + str(level))
|
|
||||||
|
|
||||||
for i in range(self.n_components):
|
|
||||||
local_state = global_state[i]
|
|
||||||
|
|
||||||
local_state_ids = self.drs.get_state_ids(local_state)
|
|
||||||
|
|
||||||
for entity in local_state_ids:
|
|
||||||
enc = And(enc, self.v[level][i][entity])
|
|
||||||
|
|
||||||
simplify(enc)
|
|
||||||
return enc
|
|
||||||
|
|
||||||
def decode_witness(self, max_level, print_model=False):
|
|
||||||
|
|
||||||
m = self.solver.model()
|
|
||||||
|
|
||||||
print("\nWitness:")
|
|
||||||
|
|
||||||
if print_model:
|
|
||||||
print(m)
|
|
||||||
|
|
||||||
for level in range(max_level+1):
|
|
||||||
|
|
||||||
print("\n[Level=" + repr(level) + "]")
|
|
||||||
|
|
||||||
#print(m)
|
|
||||||
#print(self.v[level][0][2])
|
|
||||||
#print(m[self.v[level][0][2]])
|
|
||||||
|
|
||||||
print(" State: {", end=""),
|
|
||||||
for c in range(self.n_components):
|
|
||||||
print(" <", end="")
|
|
||||||
for var_id in range(len(self.v[level][c])):
|
|
||||||
if repr(m[self.v[level][c][var_id]]) == "True":
|
|
||||||
print(" " + self.drs.get_entity_name(var_id), end="")
|
|
||||||
print(" >", end="")
|
|
||||||
print(" }")
|
|
||||||
|
|
||||||
if level != max_level:
|
|
||||||
print(" Context set: ", end="")
|
|
||||||
print("{", end="")
|
|
||||||
for c in range(self.n_components):
|
|
||||||
print(" <", end="")
|
|
||||||
for var_id in range(len(self.v[level][c])):
|
|
||||||
if repr(m[self.v_ctx[level][c][var_id]]) == "True":
|
|
||||||
print(" " + self.drs.get_entity_name(var_id), end="")
|
|
||||||
print(" >", end="")
|
|
||||||
print(" }")
|
|
||||||
|
|
||||||
print(" Active components:", end="")
|
|
||||||
for c in range(self.n_components):
|
|
||||||
if repr(m[self.v_act[level][c]]) == "True":
|
|
||||||
print(" " + str(c), end="")
|
|
||||||
print()
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def check_reachability(self, state, exclusive_state=False, print_witness=True, print_time=False, print_mem=False, max_level=100):
|
|
||||||
"""Reachability checking"""
|
|
||||||
|
|
||||||
print("[i] Checking reachability...")
|
|
||||||
|
|
||||||
if print_time:
|
|
||||||
start = time()
|
|
||||||
|
|
||||||
self.prepare_all_variables()
|
|
||||||
self.solver.add(self.enc_init_state(0))
|
|
||||||
current_level = 0
|
|
||||||
|
|
||||||
while True:
|
|
||||||
|
|
||||||
self.prepare_all_variables()
|
|
||||||
|
|
||||||
print("-----[ Working at level=" + str(current_level) + " ]-----")
|
|
||||||
stdout.flush()
|
|
||||||
|
|
||||||
# reachability test:
|
|
||||||
self.solver.push()
|
|
||||||
print("[i] Adding the reachability test...")
|
|
||||||
if exclusive_state == False:
|
|
||||||
self.solver.add(self.enc_inclusive_state(current_level,state))
|
|
||||||
else:
|
|
||||||
self.solver.add(self.enc_state(current_level,state))
|
|
||||||
|
|
||||||
result = self.solver.check()
|
|
||||||
if result == sat:
|
|
||||||
print("\n[+] SAT at level=" + str(current_level))
|
|
||||||
if print_witness:
|
|
||||||
self.decode_witness(current_level)
|
|
||||||
break
|
|
||||||
else:
|
|
||||||
self.solver.pop()
|
|
||||||
|
|
||||||
print("[i] Unrolling the transition relation")
|
|
||||||
self.solver.add(self.enc_transition_relation(current_level))
|
|
||||||
|
|
||||||
print("-----[ level=" + str(current_level) + " done ]")
|
|
||||||
current_level += 1
|
|
||||||
|
|
||||||
if current_level > max_level:
|
|
||||||
print("Stopping at level=" + str(max_level))
|
|
||||||
break
|
|
||||||
|
|
||||||
if print_time:
|
|
||||||
stop = time()
|
|
||||||
print()
|
|
||||||
print("==== Time: " + repr(stop-start))
|
|
||||||
|
|
||||||
if print_mem:
|
|
||||||
usage=resource.getrusage(resource.RUSAGE_SELF)
|
|
||||||
print("MEM: usertime=%s systime=%s mem=%sMB" % (usage[0],usage[1], (resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024**2)))
|
|
||||||
@@ -1,275 +0,0 @@
|
|||||||
"""
|
|
||||||
SMT-based Model Checking Module for RS with Context Automaton
|
|
||||||
"""
|
|
||||||
|
|
||||||
from z3 import *
|
|
||||||
from time import time
|
|
||||||
from sys import stdout
|
|
||||||
import resource
|
|
||||||
|
|
||||||
class SmtCheckerPGRS(object):
|
|
||||||
|
|
||||||
def __init__(self, rsca):
|
|
||||||
|
|
||||||
rsca.sanity_check()
|
|
||||||
|
|
||||||
self.rs = rsca.rs
|
|
||||||
self.ca = rsca.ca
|
|
||||||
|
|
||||||
self.v = []
|
|
||||||
self.v_ctx = []
|
|
||||||
self.ca_state = []
|
|
||||||
self.next_level_to_encode = 0
|
|
||||||
|
|
||||||
self.solver = Solver()
|
|
||||||
|
|
||||||
self.verification_time = None
|
|
||||||
|
|
||||||
def prepare_all_variables(self):
|
|
||||||
"""Encodes all the variables"""
|
|
||||||
|
|
||||||
self.prepare_state_variables()
|
|
||||||
self.prepare_context_variables()
|
|
||||||
self.next_level_to_encode += 1
|
|
||||||
|
|
||||||
def prepare_context_variables(self):
|
|
||||||
"""Encodes all the context variables"""
|
|
||||||
|
|
||||||
level = self.next_level_to_encode
|
|
||||||
|
|
||||||
variables = []
|
|
||||||
for entity in self.rs.background_set:
|
|
||||||
variables.append(Bool("C"+str(level)+"_"+entity))
|
|
||||||
|
|
||||||
self.v_ctx.append(variables)
|
|
||||||
|
|
||||||
def prepare_state_variables(self):
|
|
||||||
"""Encodes all the state variables"""
|
|
||||||
|
|
||||||
level = self.next_level_to_encode
|
|
||||||
|
|
||||||
variables = []
|
|
||||||
for entity in self.rs.background_set:
|
|
||||||
variables.append(Bool("L"+str(level)+"_"+entity))
|
|
||||||
self.v.append(variables)
|
|
||||||
|
|
||||||
self.ca_state.append(Int("CA"+str(level)+"_state"))
|
|
||||||
|
|
||||||
def enc_init_state(self, level):
|
|
||||||
"""Encodes the initial state at the given level"""
|
|
||||||
|
|
||||||
rs_init_state_enc = True
|
|
||||||
|
|
||||||
for v in self.v[level]:
|
|
||||||
rs_init_state_enc = simplify(And(rs_init_state_enc, Not(v))) # the initial state is empty
|
|
||||||
|
|
||||||
ca_init_state_enc = self.ca_state[level] == self.ca.get_init_state_id()
|
|
||||||
|
|
||||||
init_state_enc = simplify(And(rs_init_state_enc, ca_init_state_enc))
|
|
||||||
|
|
||||||
return init_state_enc
|
|
||||||
|
|
||||||
def enc_enabledness(self, level, prod_entity):
|
|
||||||
"""Encodes the enabledness condition for a given level and a given entity"""
|
|
||||||
|
|
||||||
rcts_for_prod_entity = self.rs.get_reactions_by_product()[prod_entity]
|
|
||||||
|
|
||||||
if rcts_for_prod_entity == []:
|
|
||||||
return False
|
|
||||||
|
|
||||||
enc_rct_prod = False
|
|
||||||
for reactants,inhibitors in rcts_for_prod_entity:
|
|
||||||
enc_reactants = True
|
|
||||||
enc_inhibitors = True
|
|
||||||
for reactant in reactants:
|
|
||||||
enc_reactants = simplify(And(enc_reactants,
|
|
||||||
Or(self.v[level][reactant], self.v_ctx[level][reactant])))
|
|
||||||
for inhibitor in inhibitors:
|
|
||||||
enc_inhibitors = simplify(And(enc_inhibitors,
|
|
||||||
Not(Or(self.v[level][inhibitor], self.v_ctx[level][inhibitor]))))
|
|
||||||
|
|
||||||
enc_rct_prod = simplify(Or(enc_rct_prod, And(enc_reactants, enc_inhibitors)))
|
|
||||||
|
|
||||||
return enc_rct_prod
|
|
||||||
|
|
||||||
def enc_entity_production(self, level, prod_entity):
|
|
||||||
"""Encodes the production of a given entity from a given level at level+1"""
|
|
||||||
|
|
||||||
enc_enab_cond = self.enc_enabledness(level, prod_entity)
|
|
||||||
|
|
||||||
enc_ent_prod = Or(And(enc_enab_cond, self.v[level+1][prod_entity]),
|
|
||||||
And(Not(enc_enab_cond), Not(self.v[level+1][prod_entity])))
|
|
||||||
|
|
||||||
return simplify(enc_ent_prod)
|
|
||||||
|
|
||||||
def enc_transition_relation(self, level):
|
|
||||||
return simplify(And(self.enc_rs_trans(level), self.enc_automaton_trans(level)))
|
|
||||||
|
|
||||||
def enc_rs_trans(self, level):
|
|
||||||
"""Encodes the transition relation"""
|
|
||||||
|
|
||||||
unused_entities = list(range(len(self.rs.background_set)))
|
|
||||||
|
|
||||||
enc_trans = True
|
|
||||||
|
|
||||||
for prod_entity in self.rs.get_reactions_by_product():
|
|
||||||
unused_entities.remove(prod_entity)
|
|
||||||
|
|
||||||
enc_trans = simplify(And(enc_trans, self.enc_entity_production(level, prod_entity)))
|
|
||||||
|
|
||||||
for prod_entity in unused_entities:
|
|
||||||
enc_trans = simplify(And(enc_trans, Not(self.v[level+1][prod_entity])))
|
|
||||||
|
|
||||||
return enc_trans
|
|
||||||
|
|
||||||
def enc_automaton_trans(self, level):
|
|
||||||
"""Encodes the transition relation for the context automaton"""
|
|
||||||
|
|
||||||
enc_trans = False
|
|
||||||
|
|
||||||
for src,ctx,dst in self.ca.transitions:
|
|
||||||
src_enc = self.ca_state[level] == src
|
|
||||||
dst_enc = self.ca_state[level+1] == dst
|
|
||||||
|
|
||||||
all_ent = set(range(len(self.rs.background_set)))
|
|
||||||
incl_ctx = ctx
|
|
||||||
excl_ctx = all_ent - incl_ctx
|
|
||||||
|
|
||||||
ctx_enc = True
|
|
||||||
|
|
||||||
for c in incl_ctx:
|
|
||||||
ctx_enc = simplify(And(ctx_enc, self.v_ctx[level][c]))
|
|
||||||
for c in excl_ctx:
|
|
||||||
ctx_enc = simplify(And(ctx_enc, Not(self.v_ctx[level][c])))
|
|
||||||
|
|
||||||
cur_trans = simplify(And(src_enc, ctx_enc, dst_enc))
|
|
||||||
|
|
||||||
enc_trans = simplify(Or(enc_trans, cur_trans))
|
|
||||||
|
|
||||||
return enc_trans
|
|
||||||
|
|
||||||
def enc_state(self, level, state):
|
|
||||||
"""Encodes the state at the given level"""
|
|
||||||
|
|
||||||
enc = True
|
|
||||||
|
|
||||||
state_ids = self.rs.get_state_ids(state)
|
|
||||||
|
|
||||||
for entity in state_ids:
|
|
||||||
enc = And(enc, self.v[level][entity])
|
|
||||||
|
|
||||||
not_in_state = set(range(len(self.rs.background_set)))
|
|
||||||
not_in_state = not_in_state.difference(set(state_ids))
|
|
||||||
|
|
||||||
for entity in not_in_state:
|
|
||||||
enc = And(enc, Not(self.v[level][entity]))
|
|
||||||
|
|
||||||
return enc
|
|
||||||
|
|
||||||
def enc_non_exclusive_state(self, level, state):
|
|
||||||
"""Encodes the state at the given level"""
|
|
||||||
|
|
||||||
enc = True
|
|
||||||
|
|
||||||
state_ids = self.rs.get_state_ids(state)
|
|
||||||
|
|
||||||
for entity in state_ids:
|
|
||||||
enc = And(enc, self.v[level][entity])
|
|
||||||
|
|
||||||
return enc
|
|
||||||
|
|
||||||
def enc_state_with_blocking(self, level, prop):
|
|
||||||
"""Encodes the state at the given level with blocking certain concentrations"""
|
|
||||||
|
|
||||||
required,blocked = prop
|
|
||||||
|
|
||||||
enc = True
|
|
||||||
|
|
||||||
required_ids = self.rs.get_state_ids(required)
|
|
||||||
blocked_ids = self.rs.get_state_ids(blocked)
|
|
||||||
|
|
||||||
for e in required_ids:
|
|
||||||
enc = And(enc, self.v[level][e])
|
|
||||||
for e in blocked_ids:
|
|
||||||
enc = And(enc, Not(self.v[level][e]))
|
|
||||||
|
|
||||||
return simplify(enc)
|
|
||||||
|
|
||||||
|
|
||||||
def decode_witness(self, max_level, print_model=False):
|
|
||||||
|
|
||||||
m = self.solver.model()
|
|
||||||
|
|
||||||
if print_model:
|
|
||||||
print(m)
|
|
||||||
|
|
||||||
for level in range(max_level+1):
|
|
||||||
|
|
||||||
print("\n[Level=" + repr(level) + "]")
|
|
||||||
|
|
||||||
print(" State: {", end=""),
|
|
||||||
for var_id in range(len(self.v[level])):
|
|
||||||
if repr(m[self.v[level][var_id]]) == "True":
|
|
||||||
print(" " + self.rs.get_entity_name(var_id), end="")
|
|
||||||
print(" }")
|
|
||||||
|
|
||||||
if level != max_level:
|
|
||||||
print(" Context set: ", end="")
|
|
||||||
print("{", end="")
|
|
||||||
for var_id in range(len(self.v[level])):
|
|
||||||
if repr(m[self.v_ctx[level][var_id]]) == "True":
|
|
||||||
print(" " + self.rs.get_entity_name(var_id), end="")
|
|
||||||
print(" }")
|
|
||||||
|
|
||||||
def check_reachability(self, state, print_witness=True, print_time=True, print_mem=True):
|
|
||||||
"""Main testing function"""
|
|
||||||
|
|
||||||
if not type(state) is tuple:
|
|
||||||
state = (state,[])
|
|
||||||
|
|
||||||
if print_time:
|
|
||||||
# start = time()
|
|
||||||
start = resource.getrusage(resource.RUSAGE_SELF).ru_utime
|
|
||||||
|
|
||||||
self.prepare_all_variables()
|
|
||||||
self.solver.add(self.enc_init_state(0))
|
|
||||||
current_level = 0
|
|
||||||
|
|
||||||
while True:
|
|
||||||
print("-----[ Working at level=" + str(current_level) + " ]-----")
|
|
||||||
stdout.flush()
|
|
||||||
|
|
||||||
self.prepare_all_variables()
|
|
||||||
|
|
||||||
# reachability test:
|
|
||||||
print("[i] Adding the reachability test...")
|
|
||||||
self.solver.push()
|
|
||||||
self.solver.add(self.enc_state_with_blocking(current_level,state))
|
|
||||||
|
|
||||||
result = self.solver.check()
|
|
||||||
if result == sat:
|
|
||||||
print("\n[+] SAT at level=" + str(current_level))
|
|
||||||
if print_witness:
|
|
||||||
self.decode_witness(current_level)
|
|
||||||
break
|
|
||||||
else:
|
|
||||||
self.solver.pop()
|
|
||||||
|
|
||||||
print("[i] Unrolling the transition relation")
|
|
||||||
self.solver.add(self.enc_transition_relation(current_level))
|
|
||||||
|
|
||||||
print("-----[ level=" + str(current_level) + " done ]")
|
|
||||||
current_level += 1
|
|
||||||
|
|
||||||
if print_time:
|
|
||||||
# stop = time()
|
|
||||||
stop = resource.getrusage(resource.RUSAGE_SELF).ru_utime
|
|
||||||
self.verification_time = stop-start
|
|
||||||
print()
|
|
||||||
print("[i] Time: " + repr(self.verification_time))
|
|
||||||
|
|
||||||
if print_mem:
|
|
||||||
print("[i] Memory: " + repr(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss/(1024*1024)) + " MB")
|
|
||||||
|
|
||||||
def get_verification_time(self):
|
|
||||||
return self.verification_time
|
|
||||||
480
smtcheckerrsc.py
480
smtcheckerrsc.py
@@ -1,480 +0,0 @@
|
|||||||
"""
|
|
||||||
SMT-based Model Checking Module for RS with Concentrations and Context Automaton
|
|
||||||
"""
|
|
||||||
|
|
||||||
from z3 import *
|
|
||||||
from time import time
|
|
||||||
from sys import stdout
|
|
||||||
from itertools import chain
|
|
||||||
import resource
|
|
||||||
from colour import *
|
|
||||||
|
|
||||||
# def simplify(x):
|
|
||||||
# return x
|
|
||||||
|
|
||||||
class SmtCheckerRSC(object):
|
|
||||||
|
|
||||||
def __init__(self, rsca):
|
|
||||||
|
|
||||||
rsca.sanity_check()
|
|
||||||
|
|
||||||
if not rsca.is_with_concentrations():
|
|
||||||
raise RuntimeError("RS and CA with concentrations expected")
|
|
||||||
|
|
||||||
self.rs = rsca.rs
|
|
||||||
self.ca = rsca.ca
|
|
||||||
|
|
||||||
self.v = []
|
|
||||||
self.v_ctx = []
|
|
||||||
self.ca_state = []
|
|
||||||
self.next_level_to_encode = 0
|
|
||||||
|
|
||||||
self.solver = Solver()
|
|
||||||
|
|
||||||
self.verification_time = None
|
|
||||||
|
|
||||||
def prepare_all_variables(self):
|
|
||||||
"""Encodes all the variables"""
|
|
||||||
|
|
||||||
self.prepare_state_variables()
|
|
||||||
self.prepare_context_variables()
|
|
||||||
self.next_level_to_encode += 1
|
|
||||||
|
|
||||||
def prepare_context_variables(self):
|
|
||||||
"""Encodes all the context variables"""
|
|
||||||
|
|
||||||
level = self.next_level_to_encode
|
|
||||||
|
|
||||||
variables = []
|
|
||||||
for entity in self.rs.background_set:
|
|
||||||
variables.append(Int("C"+str(level)+"_"+entity))
|
|
||||||
|
|
||||||
self.v_ctx.append(variables)
|
|
||||||
|
|
||||||
def prepare_state_variables(self):
|
|
||||||
"""Encodes all the state variables"""
|
|
||||||
|
|
||||||
level = self.next_level_to_encode
|
|
||||||
|
|
||||||
variables = []
|
|
||||||
for entity in self.rs.background_set:
|
|
||||||
variables.append(Int("L"+str(level)+"_"+entity))
|
|
||||||
self.v.append(variables)
|
|
||||||
|
|
||||||
self.ca_state.append(Int("CA"+str(level)+"_state"))
|
|
||||||
|
|
||||||
def enc_concentration_levels_assertion(self, level):
|
|
||||||
"""Encodes assertions that (some) variables need to be >0
|
|
||||||
|
|
||||||
We do not need to actually control all the variables,
|
|
||||||
only those that can possibly go below 0.
|
|
||||||
"""
|
|
||||||
|
|
||||||
enc_nz = True
|
|
||||||
|
|
||||||
for e_i in range(len(self.rs.background_set)):
|
|
||||||
v = self.v[level][e_i]
|
|
||||||
v_ctx = self.v_ctx[level][e_i]
|
|
||||||
e_max = self.rs.get_max_concentration_level(e_i)
|
|
||||||
enc_nz = simplify(And(enc_nz, v >= 0, v_ctx >= 0, v <= e_max, v_ctx <= e_max))
|
|
||||||
|
|
||||||
return enc_nz
|
|
||||||
|
|
||||||
def enc_init_state(self, level):
|
|
||||||
"""Encodes the initial state at the given level"""
|
|
||||||
|
|
||||||
rs_init_state_enc = True
|
|
||||||
|
|
||||||
for v in self.v[level]:
|
|
||||||
rs_init_state_enc = simplify(And(rs_init_state_enc, v == 0)) # the initial concentration levels are zeroed
|
|
||||||
|
|
||||||
ca_init_state_enc = self.ca_state[level] == self.ca.get_init_state_id()
|
|
||||||
|
|
||||||
init_state_enc = simplify(And(rs_init_state_enc, ca_init_state_enc))
|
|
||||||
|
|
||||||
return init_state_enc
|
|
||||||
|
|
||||||
def enc_produced_concentration(self, level, prod_entity):
|
|
||||||
"""Encodes the produced concentrations for the given level and entity"""
|
|
||||||
|
|
||||||
rcts_for_prod_entity = []
|
|
||||||
if prod_entity in self.rs.get_reactions_by_product():
|
|
||||||
rcts_for_prod_entity = self.rs.get_reactions_by_product()[prod_entity]
|
|
||||||
|
|
||||||
meta_reactions = []
|
|
||||||
if prod_entity in self.rs.meta_reactions:
|
|
||||||
meta_reactions = self.rs.meta_reactions[prod_entity]
|
|
||||||
|
|
||||||
permanency_inhibition = None
|
|
||||||
if prod_entity in self.rs.permanent_entities:
|
|
||||||
permanency_inhibition = self.rs.permanent_entities[prod_entity]
|
|
||||||
|
|
||||||
if rcts_for_prod_entity == [] and meta_reactions == []:
|
|
||||||
return simplify(self.v[level+1][prod_entity] == 0) # this should never happen
|
|
||||||
|
|
||||||
enc_enabledness = False
|
|
||||||
|
|
||||||
# ----------- ordinary reactions --------------------------------------------
|
|
||||||
|
|
||||||
enc_rct_prod = False
|
|
||||||
|
|
||||||
enc_ordinary_reactions_enabledness = False
|
|
||||||
|
|
||||||
for reactants,inhibitors,products in rcts_for_prod_entity:
|
|
||||||
|
|
||||||
enc_reactants = True
|
|
||||||
for reactant,concentration in reactants:
|
|
||||||
enc_reactants = simplify(And(enc_reactants,
|
|
||||||
Or(self.v[level][reactant] >= concentration, self.v_ctx[level][reactant] >= concentration)))
|
|
||||||
|
|
||||||
enc_inhibitors = True
|
|
||||||
for inhibitor,concentration in inhibitors:
|
|
||||||
enc_inhibitors = simplify(And(enc_inhibitors,
|
|
||||||
And(self.v[level][inhibitor] < concentration, self.v_ctx[level][inhibitor] < concentration)))
|
|
||||||
|
|
||||||
enc_rct_enabled = And(enc_reactants, enc_inhibitors)
|
|
||||||
enc_products = self.v[level+1][products[0][0]] == products[0][1]
|
|
||||||
enc_rct_prod = simplify(If(enc_rct_enabled, enc_products, enc_rct_prod))
|
|
||||||
enc_enabledness = simplify(Or(enc_enabledness, enc_rct_enabled))
|
|
||||||
|
|
||||||
enc_ordinary_reactions_enabledness = simplify(Or(enc_ordinary_reactions_enabledness,enc_rct_enabled))
|
|
||||||
|
|
||||||
# for reactants,inhibitors,products in rcts_for_prod_entity:
|
|
||||||
# enc_reactants = True
|
|
||||||
# enc_inhibitors = True
|
|
||||||
# # enc_products -- below
|
|
||||||
#
|
|
||||||
# for reactant,concentration in reactants:
|
|
||||||
# enc_reactants = simplify(And(enc_reactants,
|
|
||||||
# Or(self.v[level][reactant] >= concentration, self.v_ctx[level][reactant] >= concentration)))
|
|
||||||
# for inhibitor,concentration in inhibitors:
|
|
||||||
# enc_inhibitors = simplify(And(enc_inhibitors,
|
|
||||||
# And(self.v[level][inhibitor] < concentration, self.v_ctx[level][inhibitor] < concentration)))
|
|
||||||
#
|
|
||||||
# enc_products = self.v[level+1][products[0][0]] == products[0][1]
|
|
||||||
#
|
|
||||||
# enc_enabledness = simplify(Or(enc_enabledness, And(enc_reactants, enc_inhibitors)))
|
|
||||||
# enc_rct_prod = simplify(Or(enc_rct_prod, And(enc_reactants, enc_inhibitors, enc_products)))
|
|
||||||
|
|
||||||
# -------- meta reactions ---------------------------------------------------
|
|
||||||
|
|
||||||
for r_type,command_entity,reactants,inhibitors in meta_reactions:
|
|
||||||
|
|
||||||
# command entity is e.g. 'inc' for incrementation operation
|
|
||||||
# (inc,W) gives us the value W by which the given entity's value should be incremented
|
|
||||||
|
|
||||||
enc_reactants = True
|
|
||||||
enc_inhibitors = True
|
|
||||||
|
|
||||||
for reactant,concentration in reactants:
|
|
||||||
enc_reactants = simplify(And(enc_reactants,
|
|
||||||
Or(self.v[level][reactant] >= concentration, self.v_ctx[level][reactant] >= concentration)))
|
|
||||||
|
|
||||||
# command entity needs to be present (with concentration level > 0) in order to perform the operation
|
|
||||||
enc_reactants = simplify(And(enc_reactants,
|
|
||||||
Or(self.v[level][command_entity] > 0, self.v_ctx[level][command_entity] > 0)))
|
|
||||||
|
|
||||||
for inhibitor,concentration in inhibitors:
|
|
||||||
enc_inhibitors = simplify(And(enc_inhibitors,
|
|
||||||
And(self.v[level][inhibitor] < concentration, self.v_ctx[level][inhibitor] < concentration)))
|
|
||||||
|
|
||||||
if r_type == "inc":
|
|
||||||
value_after_inc = If(self.v[level][prod_entity]>self.v_ctx[level][prod_entity],self.v[level][prod_entity],self.v_ctx[level][prod_entity]) + \
|
|
||||||
If(self.v[level][command_entity]>self.v_ctx[level][command_entity],self.v[level][command_entity],self.v_ctx[level][command_entity])
|
|
||||||
enc_products = self.v[level+1][prod_entity] == value_after_inc
|
|
||||||
|
|
||||||
elif r_type == "dec":
|
|
||||||
value_after_dec = simplify(If(self.v[level][prod_entity]>self.v_ctx[level][prod_entity],self.v[level][prod_entity],self.v_ctx[level][prod_entity]) - \
|
|
||||||
If(self.v[level][command_entity]>self.v_ctx[level][command_entity],self.v[level][command_entity],self.v_ctx[level][command_entity]))
|
|
||||||
enc_products = self.v[level+1][prod_entity] == If(value_after_dec < 0, 0, value_after_dec)
|
|
||||||
|
|
||||||
else:
|
|
||||||
raise RuntimeError("Unknown meta-reaction type: " + repr(r_type))
|
|
||||||
|
|
||||||
enc_meta_reaction_enabledness = And(enc_reactants, enc_inhibitors, Not(enc_ordinary_reactions_enabledness))
|
|
||||||
enc_enabledness = simplify(Or(enc_enabledness, enc_meta_reaction_enabledness))
|
|
||||||
enc_rct_prod = simplify(Or(enc_rct_prod, And(enc_meta_reaction_enabledness, enc_products)))
|
|
||||||
|
|
||||||
# -----------------------------------------------------------------------------
|
|
||||||
|
|
||||||
if not permanency_inhibition == None:
|
|
||||||
|
|
||||||
enc_reactants = Or(self.v[level][prod_entity] >= concentration, self.v_ctx[level][prod_entity] >= concentration)
|
|
||||||
|
|
||||||
enc_inhibitors = True
|
|
||||||
for inhibitor,concentration in permanency_inhibition:
|
|
||||||
enc_inhibitors = simplify(And(enc_inhibitors,
|
|
||||||
And(self.v[level][inhibitor] < concentration, self.v_ctx[level][inhibitor] < concentration)))
|
|
||||||
enc_products = simplify(self.v[level+1][prod_entity] == \
|
|
||||||
If(self.v[level][prod_entity] > self.v_ctx[level][prod_entity],self.v[level][prod_entity],self.v_ctx[level][prod_entity]))
|
|
||||||
|
|
||||||
enc_permanency_enabledness = And(enc_reactants, enc_inhibitors, Not(enc_ordinary_reactions_enabledness))
|
|
||||||
enc_enabledness = simplify(Or(enc_enabledness, enc_permanency_enabledness))
|
|
||||||
enc_permanency = And(enc_permanency_enabledness, enc_products)
|
|
||||||
enc_rct_prod = simplify(Or(enc_rct_prod, enc_permanency))
|
|
||||||
|
|
||||||
# -----------------------------------------------------------------------------
|
|
||||||
|
|
||||||
enc_when_to_produce_zero_conc = simplify(And(Not(enc_enabledness), self.v[level+1][prod_entity] == 0))
|
|
||||||
|
|
||||||
enc_rct_prod = Or(enc_rct_prod, enc_when_to_produce_zero_conc)
|
|
||||||
return enc_rct_prod
|
|
||||||
|
|
||||||
# def enc_entity_production(self, level, prod_entity):
|
|
||||||
# """Encodes the production of a given entity from a given level at level+1"""
|
|
||||||
#
|
|
||||||
# enc_enab_cond = self.enc_enabledness(level, prod_entity)
|
|
||||||
#
|
|
||||||
# enc_ent_prod = Or(And(enc_enab_cond, self.v[level+1][prod_entity]),
|
|
||||||
# And(Not(enc_enab_cond), Not(self.v[level+1][prod_entity])))
|
|
||||||
#
|
|
||||||
# return simplify(enc_ent_prod)
|
|
||||||
|
|
||||||
|
|
||||||
def enc_transition_relation(self, level):
|
|
||||||
return simplify(And(self.enc_rs_trans(level), self.enc_automaton_trans(level)))
|
|
||||||
|
|
||||||
def enc_rs_trans(self, level):
|
|
||||||
"""Encodes the transition relation"""
|
|
||||||
|
|
||||||
unused_entities = set(range(len(self.rs.background_set)))
|
|
||||||
|
|
||||||
enc_trans = True
|
|
||||||
|
|
||||||
reactions = self.rs.get_reactions_by_product()
|
|
||||||
meta_reactions = self.rs.meta_reactions
|
|
||||||
|
|
||||||
for prod_entity in chain(reactions, meta_reactions):
|
|
||||||
unused_entities.discard(prod_entity)
|
|
||||||
enc_trans = simplify(And(enc_trans, self.enc_produced_concentration(level, prod_entity)))
|
|
||||||
|
|
||||||
for prod_entity in unused_entities:
|
|
||||||
enc_trans = simplify(And(enc_trans, self.v[level+1][prod_entity] == 0))
|
|
||||||
|
|
||||||
return enc_trans
|
|
||||||
|
|
||||||
def enc_automaton_trans(self, level):
|
|
||||||
"""Encodes the transition relation for the context automaton"""
|
|
||||||
|
|
||||||
enc_trans = False
|
|
||||||
|
|
||||||
for src,ctx,dst in self.ca.transitions:
|
|
||||||
src_enc = self.ca_state[level] == src
|
|
||||||
dst_enc = self.ca_state[level+1] == dst
|
|
||||||
|
|
||||||
all_ent = set(range(len(self.rs.background_set)))
|
|
||||||
|
|
||||||
incl_ctx = set([e for e,c in ctx])
|
|
||||||
excl_ctx = all_ent - incl_ctx
|
|
||||||
|
|
||||||
ctx_enc = True
|
|
||||||
|
|
||||||
for e,c in ctx:
|
|
||||||
ctx_enc = simplify(And(ctx_enc, self.v_ctx[level][e] == c))
|
|
||||||
|
|
||||||
for e in excl_ctx:
|
|
||||||
ctx_enc = simplify(And(ctx_enc, self.v_ctx[level][e] == 0))
|
|
||||||
|
|
||||||
cur_trans = simplify(And(src_enc, ctx_enc, dst_enc))
|
|
||||||
enc_trans = simplify(Or(enc_trans, cur_trans))
|
|
||||||
|
|
||||||
return enc_trans
|
|
||||||
|
|
||||||
def enc_exact_state(self, level, state):
|
|
||||||
"""Encodes the state at the given level with the exact concentration values"""
|
|
||||||
|
|
||||||
raise RuntimeError("Should not be used with RSC")
|
|
||||||
|
|
||||||
# enc = True
|
|
||||||
# used_entities_ids = self.rs.get_state_ids(state)
|
|
||||||
#
|
|
||||||
# for ent,conc in state:
|
|
||||||
# e_id = self.rs.get_entity_id(ent)
|
|
||||||
# enc = And(enc, self.v[level][e_id] == conc)
|
|
||||||
#
|
|
||||||
# not_in_state = set(range(len(self.rs.background_set)))
|
|
||||||
# not_in_state = not_in_state.difference(set(used_entities_ids))
|
|
||||||
#
|
|
||||||
# for entity in not_in_state:
|
|
||||||
# enc = And(enc, self.v[level][entity] == 0)
|
|
||||||
#
|
|
||||||
# return simplify(enc)
|
|
||||||
|
|
||||||
def enc_min_state(self, level, state):
|
|
||||||
"""Encodes the state at the given level with the minimal required concentration levels"""
|
|
||||||
|
|
||||||
enc = True
|
|
||||||
for ent,conc in state:
|
|
||||||
e_id = self.rs.get_entity_id(ent)
|
|
||||||
enc = And(enc, self.v[level][e_id] >= conc)
|
|
||||||
|
|
||||||
# state_ids = self.rs.get_state_ids(state)
|
|
||||||
#
|
|
||||||
# for entity in state_ids:
|
|
||||||
# enc = And(enc, self.v[level][entity])
|
|
||||||
|
|
||||||
return simplify(enc)
|
|
||||||
|
|
||||||
def enc_state_with_blocking(self, level, prop):
|
|
||||||
"""Encodes the state at the given level with blocking certain concentrations"""
|
|
||||||
|
|
||||||
required,blocked = prop
|
|
||||||
|
|
||||||
enc = True
|
|
||||||
for ent,conc in required:
|
|
||||||
e_id = self.rs.get_entity_id(ent)
|
|
||||||
enc = And(enc, self.v[level][e_id] >= conc)
|
|
||||||
|
|
||||||
for ent,conc in blocked:
|
|
||||||
e_id = self.rs.get_entity_id(ent)
|
|
||||||
enc = And(enc, self.v[level][e_id] < conc)
|
|
||||||
|
|
||||||
return simplify(enc)
|
|
||||||
|
|
||||||
def decode_witness(self, max_level, print_model=False):
|
|
||||||
|
|
||||||
m = self.solver.model()
|
|
||||||
|
|
||||||
if print_model:
|
|
||||||
print(m)
|
|
||||||
|
|
||||||
for level in range(max_level+1):
|
|
||||||
|
|
||||||
print("\n{: >70}".format("[ level=" + repr(level) + " ]"))
|
|
||||||
|
|
||||||
print(" State: {", end=""),
|
|
||||||
for var_id in range(len(self.v[level])):
|
|
||||||
var_rep = repr(m[self.v[level][var_id]])
|
|
||||||
if not var_rep.isdigit():
|
|
||||||
raise RuntimeError("unexpected: representation is not a positive integer")
|
|
||||||
if int(var_rep) > 0:
|
|
||||||
print(" " + self.rs.get_entity_name(var_id) + "=" + var_rep, end="")
|
|
||||||
# print(" " + repr(m[self.v[level][var_id]]), end="")
|
|
||||||
print(" }")
|
|
||||||
|
|
||||||
if level != max_level:
|
|
||||||
print(" Context set: ", end="")
|
|
||||||
print("{", end="")
|
|
||||||
for var_id in range(len(self.v[level])):
|
|
||||||
var_rep = repr(m[self.v_ctx[level][var_id]])
|
|
||||||
if not var_rep.isdigit():
|
|
||||||
raise RuntimeError("unexpected: representation is not a positive integer")
|
|
||||||
if int(var_rep) > 0:
|
|
||||||
print(" " + self.rs.get_entity_name(var_id) + "=" + var_rep, end="")
|
|
||||||
print(" }")
|
|
||||||
|
|
||||||
def check_reachability(self, state, print_witness=True,
|
|
||||||
print_time=True, print_mem=True, max_level=100):
|
|
||||||
"""Main testing function"""
|
|
||||||
|
|
||||||
if print_time:
|
|
||||||
# start = time()
|
|
||||||
start = resource.getrusage(resource.RUSAGE_SELF).ru_utime
|
|
||||||
|
|
||||||
self.prepare_all_variables()
|
|
||||||
self.solver.add(self.enc_init_state(0))
|
|
||||||
current_level = 0
|
|
||||||
|
|
||||||
self.prepare_all_variables()
|
|
||||||
|
|
||||||
self.solver.add(self.enc_concentration_levels_assertion(0))
|
|
||||||
|
|
||||||
while True:
|
|
||||||
self.prepare_all_variables()
|
|
||||||
self.solver.add(self.enc_concentration_levels_assertion(current_level+1))
|
|
||||||
|
|
||||||
print("\n{:-^70}".format("[ Working at level=" + str(current_level) + " ]"))
|
|
||||||
stdout.flush()
|
|
||||||
|
|
||||||
# reachability test:
|
|
||||||
print("[" + colour_str(C_BOLD, "i") + "] Adding the reachability test...")
|
|
||||||
self.solver.push()
|
|
||||||
|
|
||||||
self.solver.add(self.enc_state_with_blocking(current_level,state))
|
|
||||||
|
|
||||||
result = self.solver.check()
|
|
||||||
if result == sat:
|
|
||||||
print("[" + colour_str(C_BOLD, "+") + "] " + colour_str(C_GREEN, "SAT at level=" + str(current_level)))
|
|
||||||
if print_witness:
|
|
||||||
print("\n{:=^70}".format("[ WITNESS ]"))
|
|
||||||
self.decode_witness(current_level)
|
|
||||||
break
|
|
||||||
else:
|
|
||||||
self.solver.pop()
|
|
||||||
|
|
||||||
print("[" + colour_str(C_BOLD, "i") + "] Unrolling the transition relation")
|
|
||||||
self.solver.add(self.enc_transition_relation(current_level))
|
|
||||||
|
|
||||||
print("{:->70}".format("[ level=" + str(current_level) + " done ]"))
|
|
||||||
current_level += 1
|
|
||||||
|
|
||||||
if current_level > max_level:
|
|
||||||
print("Stopping at level=" + str(max_level))
|
|
||||||
break
|
|
||||||
|
|
||||||
if print_time:
|
|
||||||
# stop = time()
|
|
||||||
stop = resource.getrusage(resource.RUSAGE_SELF).ru_utime
|
|
||||||
self.verification_time = stop-start
|
|
||||||
print()
|
|
||||||
print("\n[i] {: >60}".format(" Time: " + repr(self.verification_time) + " s"))
|
|
||||||
|
|
||||||
if print_mem:
|
|
||||||
print("[i] {: >60}".format(" Memory: " + repr(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss/(1024*1024)) + " MB"))
|
|
||||||
|
|
||||||
def get_verification_time(self):
|
|
||||||
return self.verification_time
|
|
||||||
|
|
||||||
def show_encoding(self, state, print_witness=True,
|
|
||||||
print_time=False, print_mem=False, max_level=100):
|
|
||||||
"""Encoding debug function"""
|
|
||||||
|
|
||||||
self.prepare_all_variables()
|
|
||||||
init_s = self.enc_init_state(0)
|
|
||||||
print(init_s)
|
|
||||||
self.solver.add(init_s)
|
|
||||||
current_level = 0
|
|
||||||
|
|
||||||
self.prepare_all_variables()
|
|
||||||
|
|
||||||
while True:
|
|
||||||
self.prepare_all_variables()
|
|
||||||
|
|
||||||
print("-----[ Working at level=" + str(current_level) + " ]-----")
|
|
||||||
stdout.flush()
|
|
||||||
|
|
||||||
# reachability test:
|
|
||||||
print("[i] Adding the reachability test...")
|
|
||||||
self.solver.push()
|
|
||||||
|
|
||||||
s = self.enc_min_state(current_level,state)
|
|
||||||
print("Test: ", s)
|
|
||||||
|
|
||||||
self.solver.add(s)
|
|
||||||
|
|
||||||
result = self.solver.check()
|
|
||||||
if result == sat:
|
|
||||||
print("\n[+] " + colour_str(C_RED, "SAT at level=" + str(current_level)))
|
|
||||||
if print_witness:
|
|
||||||
self.decode_witness(current_level)
|
|
||||||
break
|
|
||||||
else:
|
|
||||||
self.solver.pop()
|
|
||||||
|
|
||||||
print("[i] Unrolling the transition relation")
|
|
||||||
t = self.enc_transition_relation(current_level)
|
|
||||||
print(t)
|
|
||||||
self.solver.add(t)
|
|
||||||
|
|
||||||
print("-----[ level=" + str(current_level) + " done ]")
|
|
||||||
current_level += 1
|
|
||||||
|
|
||||||
if current_level > max_level:
|
|
||||||
print("Stopping at level=" + str(max_level))
|
|
||||||
break
|
|
||||||
else:
|
|
||||||
x=input("Next level? ")
|
|
||||||
x=x.lower()
|
|
||||||
if not (x == "y" or x == "yes"):
|
|
||||||
break
|
|
||||||
|
|
||||||
Reference in New Issue
Block a user