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

Resources reporting. Chain reaction benchmark, etc.

Browse Source
This commit is contained in:
Artur Meski
2016-02-24 22:49:40 +01:00
parent 9c7cbc74f7
commit 8a739b0760
5 changed files with 309 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
rctsys.py
View File

@@ -453,6 +453,14 @@ class ReactionSystemWithConcentrations(ReactionSystem):
self.meta_reactions.setdefault(incr_entity_id,[])
self.meta_reactions[incr_entity_id].append(("inc", reactants, inhibitors))
def add_reaction_dec(self, incr_entity, R, I):
"""Adds a macro/meta reaction for decreasing the value of incr_entity"""
reactants,inhibitors,products = self.process_rip(R,I,[])
decr_entity_id = self.get_entity_id(decr_entity)
self.meta_reactions.setdefault(decr_entity_id,[])
self.meta_reactions[decr_entity_id].append(("dec", reactants, inhibitors))
def set_context_entities(self, entities):
raise NotImplementedError
@@ -482,10 +490,10 @@ class ReactionSystemWithConcentrations(ReactionSystem):
def show_meta_reactions(self):
print("[*] Meta reactions:")
for incr_ent,reactions in self.meta_reactions.items():
for param_ent,reactions in self.meta_reactions.items():
for r_type,reactants,inhibitors in reactions:
if r_type == "inc":
print("\t - [ Type=" + repr(r_type) + " Parameter=( " + self.get_entity_name(incr_ent) + \
if r_type == "inc" or r_type == "dec":
print("\t - [ Type=" + repr(r_type) + " Parameter=( " + self.get_entity_name(param_ent) + \
" ) ] -- ( R={" + self.state_to_str(reactants) + "}, \tI={" + self.state_to_str(inhibitors) + "} )")
else:
raise RuntimeError("Unknown meta-reaction type: " + repr(r_type))
@@ -552,11 +560,10 @@ class ReactionSystemWithConcentrations(ReactionSystem):
rs.add_reaction(new_reactants,new_inhibitors,new_products)
for incr_ent,reactions in self.meta_reactions.items():
for param_ent,reactions in self.meta_reactions.items():
for r_type,reactants,inhibitors in reactions:
if r_type == "inc":
incr_ent_name = self.get_entity_name(incr_ent)
param_ent_name = self.get_entity_name(param_ent)
new_reactants = []
new_inhibitors = []
@@ -571,19 +578,34 @@ class ReactionSystemWithConcentrations(ReactionSystem):
rs.ensure_bg_set_entity(n)
new_inhibitors.append(n)
pre_conc = incr_ent_name + "_" + str(1)
rs.ensure_bg_set_entity(pre_conc)
if r_type == "inc":
# pre_conc -- predecessor concentration
# succ_conc -- successor concentration concentration
for i in range(1,self.max_concentration):
succ_conc = incr_ent_name + "_" + str(i+1)
rs.ensure_bg_set_entity(succ_conc)
pre_conc = param_ent_name + "_" + str(i)
rs.ensure_bg_set_entity(pre_conc)
new_products = []
for j in range(1,i+1):
new_p = incr_ent_name + "_" + str(j)
succ_value = i+1
for j in range(1,succ_value+1):
new_p = param_ent_name + "_" + str(j)
rs.ensure_bg_set_entity(new_p)
new_products.append(new_p)
rs.add_reaction(new_reactants + [pre_conc], new_inhibitors, new_products)
elif r_type == "dec":
for i in range(1,self.max_concentration):
pre_conc = param_ent_name + "_" + str(i+1)
rs.ensure_bg_set_entity(pre_conc)
new_products = []
succ_value = i
for j in range(1,succ_value+1):
new_p = param_ent_name + "_" + str(j)
rs.ensure_bg_set_entity(new_p)
new_products.append(new_p)
new_products.append(succ_conc)
rs.add_reaction(new_reactants + [pre_conc], new_inhibitors, new_products)
pre_conc = succ_conc
else:
raise RuntimeError("Unknown meta-reaction type: " + repr(r_type))

130
rs_examples.py
View File

@@ -1,8 +1,13 @@
#!/usr/bin/env python
from rctsys import ReactionSystem
from rctsys import ReactionSystemWithAutomaton,ContextAutomaton
from rctsys import ReactionSystem,ReactionSystemWithConcentrations,ContextAutomatonWithConcentrations,ReactionSystemWithAutomaton
from distrib_rctsys import DistributedReactionSystem
from smtchecker import SmtChecker
from smtcheckerpgrs import SmtCheckerPGRS
from smtcheckerdistribrs import SmtCheckerDistribRS
from smtcheckerrsc import SmtCheckerRSC
import sys
import resource
def toy_ex1():
@@ -261,3 +266,124 @@ def drs_mutex_property1(k):
state.extend([[] for i in range(k)])
return state
def run_counter_exp():
if len(sys.argv) < 1+1:
print("provide N")
exit(1)
N=int(sys.argv[1])
r = ReactionSystemWithConcentrations()
r.add_bg_set_entity("e")
r.add_bg_set_entity("inc")
r.add_reaction_inc("e",[("e",1),("inc",1)],[("e",N)])
# for i in range(1,N):
# r.add_reaction([("e",i),("inc",1)],[("e",N)],[("e",i+1)])
# r.show()
c = ContextAutomatonWithConcentrations(r)
c.add_init_state("init")
c.add_state("working")
c.add_transition("init", [("e",1),("inc",1)], "working")
c.add_transition("working", [("inc",1)], "working")
# c.show()
rc = ReactionSystemWithAutomaton(r,c)
rc.show()
smt_rsc = SmtCheckerRSC(rc)
smt_rsc.check_reachability([('e',N)],print_time=True,max_level=N)
orc = rc.get_ordinary_reaction_system_with_automaton()
orc.show()
smt_tr_rs = SmtCheckerPGRS(orc)
smt_tr_rs.check_reachability(['e_' + str(N)],print_time=True)
print("Reaction System with Concentrations:", smt_rsc.get_verification_time())
print("Reaction System from translating RSC:", smt_tr_rs.get_verification_time())
def chain_reaction(print_system=False):
if len(sys.argv) < 1+3:
print("provide N M B")
print(" B=1 - RSC")
print(" B=0 - Translated RSC into RS")
exit(1)
chainLen=int(sys.argv[1]) # chain length
maxConc=int(sys.argv[2]) # depth (max concentration)
verify_rsc=bool(int(sys.argv[3]))
if chainLen < 1 or maxConc < 1:
print("be reasonable")
exit(1)
r = ReactionSystemWithConcentrations()
r.add_bg_set_entity("inc")
r.add_bg_set_entity("dec")
for i in range(1,chainLen+1):
r.add_bg_set_entity("e_" + str(i))
for i in range(1,chainLen+1):
ent = "e_" + str(i)
r.add_reaction_inc(ent, [(ent, 1),("inc",1)],[(ent,maxConc)])
if i < chainLen:
r.add_reaction([(ent,maxConc)],[],[("e_"+str(i+1),1)])
r.add_reaction([("e_" + str(chainLen),maxConc)],[("dec",1)],[("e_" + str(chainLen),maxConc)])
c = ContextAutomatonWithConcentrations(r)
c.add_init_state("init")
c.add_state("working")
c.add_transition("init", [("e_1",1),("inc",1)], "working")
c.add_transition("working", [("inc",1)], "working")
rc = ReactionSystemWithAutomaton(r,c)
if print_system:
rc.show()
if verify_rsc:
smt_rsc = SmtCheckerRSC(rc)
smt_rsc.check_reachability([('e_'+str(chainLen),maxConc)],max_level=maxConc*chainLen+10)
# smt_rsc.show_encoding([('e_'+str(1),1)],print_time=True,max_level=maxConc*chainLen+10)
if not verify_rsc:
orc = rc.get_ordinary_reaction_system_with_automaton()
if print_system:
orc.show()
smt_tr_rs = SmtCheckerPGRS(orc)
smt_tr_rs.check_reachability(['e_'+str(chainLen)+"_"+str(maxConc)])
# print("Reaction System with Concentrations:", smt_rsc.get_verification_time())
# print("Reaction System from translating RSC:", smt_tr_rs.get_verification_time())
filename=""
time=0
mem_usage=resource.getrusage(resource.RUSAGE_SELF).ru_maxrss/(1024*1024)
if verify_rsc:
filename_t="bench_rsc_time.log"
filename_m="bench_rsc_mem.log"
time=smt_rsc.get_verification_time()
else:
filename_t="bench_tr_rs_time.log"
filename_m="bench_tr_rs_mem.log"
time=smt_tr_rs.get_verification_time()
f=open(filename_t, 'a')
log_str="(" + str(chainLen) + "," + str(maxConc) + "," + str(time) + ")\n"
f.write(log_str)
f.close()
f=open(filename_m, 'a')
log_str="(" + str(chainLen) + "," + str(maxConc) + "," + str(mem_usage) + ")\n"
f.write(log_str)
f.close()

98
rssmt.py
View File

@@ -26,76 +26,19 @@ rsmc_banner = """
*** Author: Artur Męski <meski@ipipan.waw.pl> / <artur.meski@ncl.ac.uk>
"""
def process():
# rs_examples.run_counter_exp()
rs_examples.chain_reaction()
##################################################################
def main():
if len(sys.argv) < 1+1:
print("provide N")
exit(1)
N=int(sys.argv[1])
print(rsmc_banner)
# PGRS:
# rsca = rs_examples.ca_toy_ex1()
# rsca.show()
# smt = SmtCheckerPGRS(rsca)
# smt.check_reachability(rs_examples.ca_toy_ex1_property1(), print_time=True)
process()
# rsca = rs_examples.ca_bitctr(N)
# rsca.show(True)
# smt = SmtCheckerPGRS(rsca)
# smt.check_reachability(rs_examples.ca_bitctr_property(N), print_time=True)
# Distributed RS:
#drs = rs_examples.drs_mutex(N)
#drs.show()
#smt = SmtCheckerDistribRS(drs,debug_level=0)
#smt.check_reachability(rs_examples.drs_mutex_property1(N),
# exclusive_state=False,
# max_level=10,
# print_time=True,
# print_mem=True)
# drs = rs_examples.drs_toy_ex1()
# drs.show()
#
# smt = SmtCheckerDistribRS(drs,debug_level=3)
# smt.check_reachability(rs_examples.drs_toy_ex1_property1(),max_level=10,print_time=True)
r = ReactionSystemWithConcentrations()
r.add_bg_set_entity("e")
r.add_bg_set_entity("inc")
r.add_reaction_inc("e",[("e",1),("inc",1)],[("e",N)])
# for i in range(1,N):
# r.add_reaction([("e",i),("inc",1)],[("e",N)],[("e",i+1)])
# r.show()
c = ContextAutomatonWithConcentrations(r)
c.add_init_state("init")
c.add_state("working")
c.add_transition("init", [("e",1),("inc",1)], "working")
c.add_transition("working", [("inc",1)], "working")
# c.show()
rc = ReactionSystemWithAutomaton(r,c)
rc.show()
smt_rsc = SmtCheckerRSC(rc)
smt_rsc.check_reachability([('e',N)],print_time=True,max_level=N)
orc = rc.get_ordinary_reaction_system_with_automaton()
orc.show()
smt_tr_rs = SmtCheckerPGRS(orc)
smt_tr_rs.check_reachability(['e_' + str(N)],print_time=True)
print("Reaction System with Concentrations:", smt_rsc.get_verification_time())
print("Reaction System from translating RSC:", smt_tr_rs.get_verification_time())
if __name__ == "__main__":
try:
@@ -108,3 +51,30 @@ if __name__ == "__main__":
print("\nQuitting...")
sys.exit(99)
# PGRS:
# rsca = rs_examples.ca_toy_ex1()
# rsca.show()
# smt = SmtCheckerPGRS(rsca)
# smt.check_reachability(rs_examples.ca_toy_ex1_property1(), print_time=True)
# rsca = rs_examples.ca_bitctr(N)
# rsca.show(True)
# smt = SmtCheckerPGRS(rsca)
# smt.check_reachability(rs_examples.ca_bitctr_property(N), print_time=True)
# Distributed RS:
#drs = rs_examples.drs_mutex(N)
#drs.show()
#smt = SmtCheckerDistribRS(drs,debug_level=0)
#smt.check_reachability(rs_examples.drs_mutex_property1(N),
# exclusive_state=False,
# max_level=10,
# print_time=True,
# print_mem=True)
# drs = rs_examples.drs_toy_ex1()
# drs.show()
#
# smt = SmtCheckerDistribRS(drs,debug_level=3)
# smt.check_reachability(rs_examples.drs_toy_ex1_property1(),max_level=10,print_time=True)

12
smtcheckerpgrs.py
View File

@@ -5,6 +5,7 @@ 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):
@@ -202,11 +203,12 @@ class SmtCheckerPGRS(object):
print(" " + self.rs.get_entity_name(var_id), end="")
print(" }")
def check_reachability(self, state, exact_state=False, print_witness=True, print_time=False):
def check_reachability(self, state, exact_state=False, print_witness=True, print_time=True, print_mem=True):
"""Main testing function"""
if print_time:
start = time()
# start = time()
start = resource.getrusage(resource.RUSAGE_SELF).ru_utime
self.prepare_all_variables()
self.solver.add(self.enc_init_state(0))
@@ -242,10 +244,14 @@ class SmtCheckerPGRS(object):
current_level += 1
if print_time:
stop = 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

82
smtcheckerrsc.py
View File

@@ -6,6 +6,10 @@ from z3 import *
from time import time
from sys import stdout
from itertools import chain
import resource
# def simplify(x):
# return x
class SmtCheckerRSC(object):
@@ -112,7 +116,6 @@ class SmtCheckerRSC(object):
for r_type,reactants,inhibitors in meta_reactions:
if r_type == "inc":
enc_reactants = True
enc_inhibitors = True
@@ -123,13 +126,19 @@ class SmtCheckerRSC(object):
enc_inhibitors = simplify(And(enc_inhibitors,
And(self.v[level][inhibitor] < concentration, self.v_ctx[level][inhibitor] < concentration)))
if r_type == "inc":
enc_products = self.v[level+1][prod_entity] == self.v[level][prod_entity]+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)))
elif r_type == "dec":
enc_products = self.v[level+1][prod_entity] == self.v[level][prod_entity]-1
else:
raise RuntimeError("Unknown meta-reaction type: " + repr(r_type))
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)))
# -----------------------------------------------------------------------------
@@ -155,7 +164,7 @@ class SmtCheckerRSC(object):
def enc_rs_trans(self, level):
"""Encodes the transition relation"""
unused_entities = list(range(len(self.rs.background_set)))
unused_entities = set(range(len(self.rs.background_set)))
enc_trans = True
@@ -163,7 +172,7 @@ class SmtCheckerRSC(object):
meta_reactions = self.rs.meta_reactions
for prod_entity in chain(reactions, meta_reactions):
unused_entities.remove(prod_entity)
unused_entities.discard(prod_entity)
enc_trans = simplify(And(enc_trans, self.enc_produced_concentration(level, prod_entity)))
for prod_entity in unused_entities:
@@ -266,11 +275,12 @@ class SmtCheckerRSC(object):
print(" }")
def check_reachability(self, state, print_witness=True,
print_time=False, print_mem=False, max_level=100):
print_time=True, print_mem=True, max_level=100):
"""Main testing function"""
if print_time:
start = time()
# start = time()
start = resource.getrusage(resource.RUSAGE_SELF).ru_utime
self.prepare_all_variables()
self.solver.add(self.enc_init_state(0))
@@ -310,11 +320,67 @@ class SmtCheckerRSC(object):
break
if print_time:
stop = 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
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(s)
self.solver.add(s)
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")
t = self.enc_transition_relation(current_level)
print(t)
self.solver.add(t)
print("-----[ level=" + str(current_level) + " done ]")
current_level += 1
x=input("Next level? ")
x=x.lower()
if not (x == "y" or x == "yes"):
break
if current_level > max_level:
print("Stopping at level=" + str(max_level))
break