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

working on RSNA

Browse Source
This commit is contained in:
Artur Meski
2016-12-29 20:21:56 +01:00
parent ba8cb798e4
commit ebee3aa03e
6 changed files with 132 additions and 263 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
Makefile
View File

@@ -1,6 +1,7 @@
clean:
rm -rf *.pyc __pycache__
rm -rf rs/__pycache__ smt/rs/__pycache__ __pycache__
find . -name '*.pyc' -exec rm -f {} \;
cleanall: clean
rm -rf *.log

4
rs/extended_context_automaton.py
View File

@@ -9,6 +9,10 @@ class ExtendedContextAutomaton(ContextAutomaton):
self._actions = []
super(ExtendedContextAutomaton, self).__init__(reaction_system)
@property
def number_of_actions(self):
return len(self._actions)
def add_transition(self, src, actions, ctx_reaction, dst):
"""Adds a transition

10
rs/network_of_context_automata.py
View File

@@ -3,12 +3,16 @@ from sys import exit
class NetworkOfContextAutomata(object):
def __init__(self, context_automata):
self.cas = list(context_automata)
self.automata = list(context_automata)
@property
def number_of_automata(self):
return len(self.automata)
def show(self):
for ca in self.cas:
for ca in self.automata:
print()
ca.show()
def add(self, aut):
self.cas.append(aut)
self.automata.append(aut)

1
rs_testing.py
View File

@@ -39,6 +39,7 @@ def test_extended_automaton():
checker = SmtCheckerRSNA(rna)
checker.check_reachability([])
def process():

56
smt/smt_checker_rs.py
View File

@@ -43,8 +43,8 @@ class SmtCheckerRS(object):
self.v_ctx.append(variables)
def prepare_state_variables(self):
"""Encodes all the state variables"""
def prepare_rs_state_variables(self):
"""Encodes all the state variables of the reaction system"""
level = self.next_level_to_encode
@@ -53,19 +53,36 @@ class SmtCheckerRS(object):
variables.append(Bool("L"+str(level)+"_"+entity))
self.v.append(variables)
def prepare_context_controller_variables(self):
"""Encodes all the variables required for controlling context sequences"""
level = self.next_level_to_encode
self.ca_state.append(Int("CA"+str(level)+"_state"))
def prepare_state_variables(self):
"""Encodes all the state variables"""
self.prepare_rs_state_variables()
self.prepare_context_controller_variables()
def enc_rs_init_state(self, level):
"""Encodes the initial state for the reaction system"""
rs_init_state_enc = True
for v in self.v[level]:
rs_init_state_enc = simplify(And(rs_init_state_enc, Not(v))) # the initial state is empty
return rs_init_state_enc
def enc_context_controller_init_state(self, level):
"""Encodes the initial state for controlling context sequences"""
return self.ca_state[level] == self.ca.get_init_state_id()
def enc_init_state(self, level):
"""Encodes the initial state at the given level"""
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))
init_state_enc = simplify(And(self.enc_rs_init_state(level), self.enc_context_controller_init_state(level)))
return init_state_enc
@@ -103,6 +120,8 @@ class SmtCheckerRS(object):
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)))
def enc_rs_trans(self, level):
@@ -122,12 +141,10 @@ class SmtCheckerRS(object):
return enc_trans
def enc_automaton_trans(self, level):
"""Encodes the transition relation for the context automaton"""
def enc_automaton_single_trans(self, level, transition):
enc_trans = False
src,ctx,dst = transition
for src,ctx,dst in self.ca.transitions:
src_enc = self.ca_state[level] == src
dst_enc = self.ca_state[level+1] == dst
@@ -142,9 +159,16 @@ class SmtCheckerRS(object):
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_single_trans = simplify(And(src_enc, ctx_enc, dst_enc))
enc_trans = simplify(Or(enc_trans, cur_trans))
return enc_single_trans
def enc_automaton_trans(self, level):
"""Encodes the transition relation for the context automaton"""
enc_trans = False
for transition in self.ca.transitions:
enc_trans = simplify(Or(enc_trans, self.enc_automaton_single_trans(level, transition)))
return enc_trans

255
smt/smt_checker_rs_na.py
View File

@@ -7,128 +7,63 @@ from time import time
from sys import stdout
import resource
class SmtCheckerRSNA(object):
from smt.smt_checker_rs import SmtCheckerRS
class SmtCheckerRSNA(SmtCheckerRS):
"""SMT-based Model Checking for Reaction Systems with Network of Automata"""
def __init__(self, reaction_system_with_netaut):
self.rs = reaction_system_with_netaut.rs
self.ca = reaction_system_with_netaut.cas
self.canet = reaction_system_with_netaut.cas
self.number_of_automata = self.canet.number_of_automata
self.v = []
self.v_ctx = []
self.ca_state = []
self.v_canet_states = []
self.v_canet_actions = []
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):
def prepare_context_controller_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)
ca_states = []
for ca_id in range(self.number_of_automata):
ca_states.append(Int("CA"+str(level)+"_a"+str(ca_id)+"_state"))
self.v_canet_states.append(ca_states)
# TODO: tutaj musimy mieć zestaw zmiennych dla każdego automatu
# We do not encode actions when there are no transitions needed.
if level > 0:
ca_actions = []
for ca in self.canet.automata:
for act_id in range(ca.number_of_actions):
ca_actions.append(Bool("CA"+str(level-1)+"_a"+str(self.canet.automata.index(ca))+"_act"+str(act_id)))
self.v_canet_actions.append(ca_actions)
self.ca_state.append(Int("CA"+str(level)+"_state"))
def enc_context_controller_init_state(self, level):
"""Encodes the initial state for the network of automata"""
def enc_init_state(self, level):
"""Encodes the initial state at the given level"""
canet_init_state_enc = True
for ca_idx in range(self.number_of_automata):
ca_init_state_enc = self.v_canet_states[level][ca_idx] == self.canet.automata[ca_idx].get_init_state_id()
canet_init_state_enc = simplify(And(canet_init_state_enc, ca_init_state_enc))
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)
return canet_init_state_enc
def enc_transition_relation(self, level):
return simplify(And(self.enc_rs_trans(level), self.enc_automaton_trans(level)))
def enc_rs_trans(self, level):
"""Encodes the transition relation"""
def enc_automaton_single_trans(self, level, transition):
unused_entities = list(range(len(self.rs.background_set)))
src,actions,ctx_reaction,dst = transition
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
@@ -143,134 +78,34 @@ class SmtCheckerRSNA(object):
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_single_trans = simplify(And(src_enc, ctx_enc, dst_enc))
enc_trans = simplify(Or(enc_trans, cur_trans))
return enc_single_trans
return enc_trans
def enc_single_automaton_trans(self, level, automaton):
def enc_state(self, level, state):
"""Encodes the state at the given level"""
enc_aut_trans = False
for transition in automaton.transitions:
enc_aut_trans = simplify(Or(enc_aut_trans, self.enc_automaton_single_trans(level, transition)))
enc = True
return enc_aut_trans
state_ids = self.rs.get_state_ids(state)
def enc_automaton_trans(self, level):
"""Encodes the transition relation for the context automaton"""
for entity in state_ids:
enc = And(enc, self.v[level][entity])
enc_canet = True
for aut in self.canet:
enc_canet = simplify(And(enc_canet, self.enc_single_automaton_trans(level, aut)))
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(" }")
return enc_canet
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()
self.prepare_all_variables()
self.prepare_all_variables()
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