Moved files to 'reactics-smt'

This commit is contained in:
Artur Meski
2019-03-03 22:26:00 +00:00
parent c3a19d81dc
commit 5eb8e3b63d
33 changed files with 0 additions and 0 deletions

View File

@@ -0,0 +1,14 @@
from rs.reaction_system import ReactionSystem
from rs.context_automaton import ContextAutomaton
from rs.reaction_system_with_concentrations import ReactionSystemWithConcentrations
from rs.reaction_system_with_concentrations_param import ReactionSystemWithConcentrationsParam, ParameterObj, is_param
from rs.context_automaton_with_concentrations import ContextAutomatonWithConcentrations
from rs.extended_context_automaton import ExtendedContextAutomaton
from rs.network_of_context_automata import NetworkOfContextAutomata
from rs.reaction_system_with_automaton import ReactionSystemWithAutomaton
from rs.reaction_system_with_autnet import ReactionSystemWithNetworkOfAutomata
# EOF

View File

@@ -0,0 +1,171 @@
from sys import exit
from colour import *
class ContextAutomaton(object):
def __init__(self, reaction_system):
self._states = []
self._transitions = []
self._init_state = None
self._reaction_system = reaction_system
self._name = ""
self._prod_entities = set()
@property
def states(self):
return self._states
@property
def transitions(self):
return self._transitions
@property
def prod_entities(self):
return self._prod_entities
@property
def reaction_system(self):
return self._reaction_system
@property
def name(self):
return self._name
@name.setter
def name(self, automaton_name):
self._name = automaton_name
def add_state(self, name):
if name not in self._states:
self._states.append(name)
else:
print("\'%s\' already added. skipping..." % (name,))
def add_states(self, states_set):
for st in states_set:
self.add_state(st)
def add_init_state(self, name):
self.add_state(name)
self._init_state = self._states.index(name)
def get_init_state_name(self):
if self._init_state == None:
return None
return self._states[self._init_state]
def is_state(self, name):
if name in self._states:
return True
else:
return False
def get_state_id(self, name):
try:
return self._states.index(name)
except ValueError:
print_error("Undefined context automaton state: " + repr(name))
exit(1)
def get_state_name(self, state_id):
return self._states[state_id]
def get_init_state_id(self):
return self._init_state
def print_states(self):
for state in self._states:
print(state)
def is_valid_rs_set(self, elements):
if set(elements).issubset(self._reaction_system.background_set):
return True
else:
return False
def is_valid_context(self, context):
return self.is_valid_rs_set(context)
def get_set_of_ids(self, elements):
"""Converts a set/list/tuple of entities into a set of their ids"""
new_set = set()
for e in set(elements):
new_set.add(self._reaction_system.get_entity_id(e))
return new_set
def add_transition(self, src, context_set, dst):
if not type(context_set) is set and not type(context_set) is list:
print("Contexts set must be of type set or list")
if not self.is_valid_context(context_set):
raise RuntimeError(
"one of the entities in the context set is unknown (undefined)!")
if not self.is_state(src):
raise RuntimeError(
"\"" + src + "\" is an unknown (undefined) state")
if not self.is_state(dst):
raise RuntimeError(
"\"" + dst + "\" is an unknown (undefined) state")
new_context_set = set()
for e in set(context_set):
new_context_set.add(self._reaction_system.get_entity_id(e))
self._prod_entities |= new_context_set
self._transitions.append(
(self.get_state_id(src),
new_context_set, self.get_state_id(dst)))
def rsset2str(self, elements):
"""Converts the set of entities ids into the string with their names"""
if len(elements) == 0:
return "0"
s = "{"
for c in elements:
s += " " + self._reaction_system.get_entity_name(c)
s += " }"
return s
def context2str(self, ctx):
return self.rsset2str(ctx)
def show_transitions(self):
print(C_MARK_INFO + " Context automaton transitions:")
for transition in self._transitions:
str_transition = str(transition[0]) + " --( "
str_transition += self.context2str(transition[1])
str_transition += " )--> " + str(transition[2])
print(" - " + str_transition)
def show_states(self):
init_state_name = self.get_init_state_name()
print(C_MARK_INFO + " Context automaton states:")
for state in self._states:
print(" - " + state, end="")
if state == init_state_name:
print(" [init]")
else:
print()
def show_header(self):
if self.name:
name_string = ": " + colour_str(C_BOLD, self.name)
print(C_MARK_INFO + " Context automaton" + name_string)
def show_prod_entities(self):
print(C_MARK_INFO + " Context automaton possible products:")
for entity in self._prod_entities:
print(" - " + self._reaction_system.get_entity_name(entity))
def show(self):
self.show_header()
self.show_states()
self.show_transitions()
self.show_prod_entities()
# EOF

View File

@@ -0,0 +1,83 @@
from sys import exit
from colour import *
from rs.context_automaton import ContextAutomaton
class ContextAutomatonWithConcentrations(ContextAutomaton):
def __init__(self, reaction_system):
super(ContextAutomatonWithConcentrations,
self).__init__(reaction_system)
def is_valid_context(self, context):
if set(
[e for e, lvl in context]).issubset(
self._reaction_system.background_set):
return True
else:
return False
def context2str(self, ctx):
if len(ctx) == 0:
return "0"
s = "{"
for ent, lvl in ctx:
s += " " + str((self._reaction_system.get_entity_name(ent), lvl))
s += " }"
return s
def add_transition(self, src, context_set, dst):
if not type(context_set) is set and not type(context_set) is list:
print("Contexts set must be of type set or list")
if not self.is_valid_context(context_set):
raise RuntimeError(
"one of the entities in the context set is unknown (undefined)!")
if not self.is_state(src):
raise RuntimeError(
"\"" + src + "\" is an unknown (undefined) state")
if not self.is_state(dst):
raise RuntimeError(
"\"" + dst + "\" is an unknown (undefined) state")
new_context_set = set()
for ent, lvl in set(context_set):
new_context_set.add(
(self._reaction_system.get_entity_id(ent), lvl))
self._transitions.append(
(self.get_state_id(src),
new_context_set, self.get_state_id(dst)))
def get_automaton_with_flat_contexts(self, ordinary_reaction_system):
ca = ContextAutomaton(ordinary_reaction_system)
ca._states = self._states
ca._init_state = self._init_state
for src, ctx, dst in self._transitions:
new_ctx = set()
for ent, conc in ctx:
for i in range(1, conc+1):
n = self._reaction_system.get_entity_name(
ent) + "#" + str(i)
ca._reaction_system.ensure_bg_set_entity(n)
new_ctx.add(n)
ca.add_transition(
ca.get_state_name(src),
new_ctx, ca.get_state_name(dst))
return ca
def show(self):
self.show_header()
self.show_states()
self.show_transitions()
# EOF

View File

@@ -0,0 +1,176 @@
from sys import exit
from colour import *
from rs.context_automaton import ContextAutomaton
class ExtendedContextAutomaton(ContextAutomaton):
"""Extended Context Automaton
Supports transitions with actions.
Each transitions is additionally guarded with
reactants and inhibitors.
The provided context entities are the products
of the reactions labelling the transition taken.
"""
def __init__(self, reaction_system):
super(ExtendedContextAutomaton, self).__init__(reaction_system)
self._actions = []
self._transitions_for_products = dict()
self._actions_for_products = dict()
@property
def number_of_actions(self):
return len(self._actions)
@property
def actions(self):
return self._actions
def has_action(self, action):
"""Checks if the automaton supports a given action"""
return action in self._actions
def get_transitions_producing_entity(self, entity):
"""Returns the transitions that produce a given entity"""
if entity in self._transitions_for_products:
return self._transitions_for_products[entity]
else:
return []
def get_actions_producing_entity(self, entity):
"""Returns the actions that produce a given entity"""
if entity in self._actions_for_products:
return self._actions_for_products[entity]
else:
return set()
def can_produce_entity(self, entity):
"""Check if the automaton can produce an entity"""
if entity in self._actions_for_products:
return True
else:
return False
def add_transition(self, src, actions, ctx_reaction, dst):
"""Adds a transition
src: is the source state name
dst: is the destination state name
actions: is the set of actions with which the transitions is synchronised
ctx_reaction: is the context reaction associated with the transition
"""
ctx_reactants, ctx_inhibitors, ctx_products = ctx_reaction
if not type(ctx_products) is set and not type(ctx_products) is list:
print("Contexts set (context products) must be of type set or list")
if not self.is_valid_rs_set(ctx_reactants):
raise RuntimeError(
"one of the entities in the reactants set is unknown (undefined)!")
if not self.is_valid_rs_set(ctx_inhibitors):
raise RuntimeError(
"one of the entities in the inhibitors set is unknown (undefined)!")
if not self.is_valid_rs_set(ctx_products):
raise RuntimeError(
"one of the entities in the context set is unknown (undefined)!")
if not self.is_state(src):
raise RuntimeError(
"\"" + src + "\" is an unknown (undefined) state")
if not self.is_state(dst):
raise RuntimeError(
"\"" + dst + "\" is an unknown (undefined) state")
src_id = self.get_state_id(src)
dst_id = self.get_state_id(dst)
act_ids = self.get_set_of_action_ids(actions)
r_ids = self.get_set_of_ids(ctx_reactants)
i_ids = self.get_set_of_ids(ctx_inhibitors)
p_ids = self.get_set_of_ids(ctx_products)
new_transition = (src_id, act_ids, (r_ids, i_ids, p_ids), dst_id)
for product_id in p_ids:
self._transitions_for_products.setdefault(product_id, [])
self._transitions_for_products[product_id].append(new_transition)
self._actions_for_products.setdefault(product_id, set())
self._actions_for_products[product_id] |= set(actions)
self._prod_entities |= p_ids
self._transitions.append(new_transition)
def show_transitions(self):
"""Prints the set of registered transitions"""
print(C_MARK_INFO + " Context automaton transitions:")
for src_id, act_id, reaction, dst_id in self._transitions:
str_transition = self.get_state_name(src_id) + " --( "
str_transition += "<" + self.get_actions_str(act_id) + "> | "
str_transition += "( " + self.rsset2str(reaction[0]) + "," + self.rsset2str(
reaction[1]) + "," + self.rsset2str(reaction[2]) + " )"
str_transition += " )--> " + self.get_state_name(dst_id)
print(" - " + str_transition)
def add_action(self, action_name):
"""Registers an action"""
if action_name not in self._actions:
self._actions.append(action_name)
else:
print("\'%s\' already added. skipping..." % (action_name,))
def get_action_id(self, action_name):
"""For an action name returns its id"""
try:
return self._actions.index(action_name)
except ValueError:
print_error("Undefined context automaton action: " +
repr(action_name))
exit(1)
def get_action_name(self, action_id):
return self._actions[action_id]
def get_set_of_action_ids(self, actions):
"""Converts a set of actions into the set of their ids"""
act_ids = set()
for act in actions:
act_ids.add(self.get_action_id(act))
return act_ids
def get_actions_str(self, actions):
"""Returns the string for the set of action ids given by actions"""
s = ""
for act in actions:
s += self.get_action_name(act) + ", "
s = s[:-2]
return s
def show_actions(self):
"""Prints all the actions"""
print(C_MARK_INFO + " Context automaton actions:")
for act in self._actions:
print(" - " + act + " (id=" + str(self.get_action_id(act)) + ")")
def show(self):
super(ExtendedContextAutomaton, self).show()
self.show_actions()
# EOF

View File

@@ -0,0 +1,111 @@
from sys import exit
from colour import *
class NetworkOfContextAutomata(object):
def __init__(self, reaction_system, context_automata):
self.automata = []
self._reaction_system = reaction_system
self._actions = set()
self._prod_entities = set()
self._automata_for_actions = dict()
self._actions_for_products = dict()
if len(context_automata) < 1:
print("Context automata network must contain at least one automaton!")
exit(1)
for automaton in context_automata:
self.add(automaton)
self.sanity_check()
@property
def number_of_automata(self):
return len(self.automata)
@property
def reaction_system(self):
return self._reaction_system
@property
def prod_entities(self):
"""Returns the set of entities that can potentially be produced by the automata in the network"""
return self._prod_entities
@property
def automata_ids(self):
return set(range(len(self.automata)))
def sanity_check(self):
"""Performs a sanity check of the network of automata"""
for automaton in self.automata:
if automaton.reaction_system != self._reaction_system:
print_error(
"Mismatching reaction system used in \"" +
str(automaton.name) + "\"!!!")
exit(1)
def show_prod_entities(self):
print(
C_MARK_INFO +
" Possible context-products for the network of automata:")
for entity in self.prod_entities:
print(" - " + self._reaction_system.get_entity_name(entity))
def show_actions(self):
print(C_MARK_INFO + " Actions of the network:")
for action in self._actions:
print(" - " + action)
def register_action(self, action, aut):
"""Associates an action with an automaton"""
self._automata_for_actions.setdefault(action, set())
aut_index = self.automata.index(aut)
self._automata_for_actions[action].add(aut_index)
def get_actions_producing_entity(self, entity):
"""Returns the set of actions producing an entity"""
if entity in self._actions_for_products:
return self._actions_for_products[entity]
else:
return set()
def get_automata_with_action(self, action):
"""Returns the set of automata that support an action"""
if action in self._automata_for_actions:
return self._automata_for_actions[action]
else:
return set()
def add(self, aut):
"""Adds an automaton to the network"""
self.automata.append(aut)
self._prod_entities |= aut.prod_entities
self._actions |= set(aut.actions)
for action in aut.actions:
self.register_action(action, aut)
for entity in aut.prod_entities:
self._actions_for_products.setdefault(entity, set())
self._actions_for_products[entity] |= aut.get_actions_producing_entity(
entity)
def show(self):
print()
print(C_MARK_INFO + " NETWORK OF CONTEXT AUTOMATA")
for ca in self.automata:
print()
ca.show()
print()
self.show_prod_entities()
self.show_actions()
# EOF

View File

@@ -0,0 +1,203 @@
from sys import exit
from colour import *
class ReactionSystem(object):
def __init__(self):
self.reactions = []
self.background_set = []
# self.reactions_by_agents = [] # each element is 'reactions_by_prod'
self.reactions_by_prod = None
# legacy:
self.init_contexts = []
self.context_entities = []
@property
def background_set_size(self):
return len(self.background_set)
@property
def set_of_bgset_ids(self):
return set(range(self.background_set_size))
@property
def ordered_list_of_bgset_ids(self):
return list(range(self.background_set_size))
def assume_not_in_bgset(self, name):
if self.is_in_background_set(name):
raise RuntimeError(
"The entity " + name + " is already on the list")
def add_bg_set_entity(self, name):
self.assume_not_in_bgset(name)
self.background_set.append(name)
def ensure_bg_set_entity(self, name):
if not self.is_in_background_set(name):
self.background_set.append(name)
def add_bg_set_entities(self, elements):
for e in elements:
self.add_bg_set_entity(e)
def is_in_background_set(self, entity):
"""Checks if the given name is valid wrt the background set="""
if entity in self.background_set:
return True
else:
return False
def get_entity_id(self, name):
try:
return self.background_set.index(name)
except ValueError:
print("Undefined background set entity: " + repr(name))
exit(1)
def get_state_ids(self, state):
ids = []
for entity in state:
ids.append(self.get_entity_id(entity))
return ids
def get_entity_name(self, entity_id):
"""Returns the string corresponding to the entity"""
return self.background_set[entity_id]
def add_reaction(self, R, I, P):
"""Adds a reaction"""
if R == [] or P == []:
raise RuntimeError("No reactants or products defined")
reactants = []
for entity in R:
reactants.append(self.get_entity_id(entity))
inhibitors = []
for entity in I:
inhibitors.append(self.get_entity_id(entity))
products = []
for entity in P:
products.append(self.get_entity_id(entity))
self.reactions.append((reactants, inhibitors, products))
def add_initial_context_set(self, context_set):
if context_set == []:
print("Empty context set is not allowed")
raise
integers = []
for entity in context_set:
if not entity in self.background_set:
print("The entity", entity, "is not in the background set")
raise
else:
integers.append(self.get_entity_id(entity))
self.init_contexts.append(integers)
def set_context_entities(self, entities):
for entity in entities:
entity_id = self.get_entity_id(entity)
self.context_entities.append(entity_id)
def entities_names_set_to_str(self, entities):
s = ""
for entity in entities:
s += entity + ", "
s = s[:-2]
return s
def entities_ids_set_to_str(self, entities):
s = ""
for entity in entities:
s += self.get_entity_name(entity) + ", "
s = s[:-2]
return s
def state_to_str(self, state):
return self.entities_ids_set_to_str(state)
def show_reactions(self, soft=False):
print(C_MARK_INFO + " Reactions:")
if soft and len(self.reactions) > 50:
print(" -> there are more than 50 reactions (" +
str(len(self.reactions)) + ")")
else:
print(
" "*4 + "{0: ^35}{1: ^25}{2: ^15}".format("reactants", " inhibitors", " products"))
for reaction in self.reactions:
# print("\t( R={" + self.state_to_str(reaction[0]) + "}, I={" + self.state_to_str(reaction[1]) + "}, P={" + self.state_to_str(reaction[2]) + "} )")
print(" " + "- {0: ^35}{1: ^25}{2: ^15}".format("{ " + self.state_to_str(reaction[0]) + " }",
" { " + self.state_to_str(reaction[1]) + " }",
" { " + self.state_to_str(reaction[2]) + " }"))
def show_background_set(self):
print(
C_MARK_INFO + " Background set: {" + self.entities_names_set_to_str(self.background_set) + "}")
def show_initial_contexts(self):
if len(self.init_contexts) > 0:
print(C_MARK_INFO + " Initial context sets:")
for ctx in self.init_contexts:
print(" - {" + self.entities_ids_set_to_str(ctx) + "}")
def show_context_entities(self):
if len(self.context_entities) > 0:
print(
C_MARK_INFO + " Context entities: " + self.entities_ids_set_to_str(self.context_entities))
def show(self, soft=False):
self.show_background_set()
self.show_initial_contexts()
self.show_reactions(soft)
self.show_context_entities()
def get_reactions_by_product(self):
"""Sorts reactions by their products and returns a dictionary of products"""
if self.reactions_by_prod != None:
return self.reactions_by_prod
producible_entities = set()
for reaction in self.reactions:
producible_entities = producible_entities.union(set(reaction[2]))
reactions_by_prod = {}
for prod_entity in producible_entities:
reactions_by_prod[prod_entity] = []
for reaction in self.reactions:
if prod_entity in reaction[2]:
reactions_by_prod[prod_entity].append(
[reaction[0], reaction[1]])
# save in cache
self.reactions_by_prod = reactions_by_prod
return reactions_by_prod
def sanity_check(self):
"""Performs a sanity check on the defined reaction system"""
if self.reactions == []:
print("No reactions defined")
exit(1)
if self.background_set == []:
print("Empty background set")
exit(1)
# EOF

View File

@@ -0,0 +1,12 @@
class ReactionSystemWithNetworkOfAutomata(object):
def __init__(self, reaction_system, context_automata):
self.rs = reaction_system
self.cas = context_automata
def show(self, soft=False):
self.rs.show(soft)
self.cas.show()
# EOF

View File

@@ -0,0 +1,45 @@
from rs.reaction_system_with_concentrations import ReactionSystemWithConcentrations
from rs.reaction_system_with_concentrations_param import ReactionSystemWithConcentrationsParam
from rs.context_automaton_with_concentrations import ContextAutomatonWithConcentrations
class ReactionSystemWithAutomaton(object):
def __init__(self, reaction_system, context_automaton):
self.rs = reaction_system
self.ca = context_automaton
def show(self, soft=False):
self.rs.show(soft)
self.ca.show()
def is_concentr_and_param_compatible(self):
"""
Checks if the underlying RS/CA are compatible
with parameters and concentrations
"""
if not isinstance(self.rs, ReactionSystemWithConcentrationsParam):
return False
if not isinstance(self.ca, ContextAutomatonWithConcentrations):
return False
return True
def is_with_concentrations(self):
if not isinstance(self.rs, ReactionSystemWithConcentrations):
return False
if not isinstance(self.ca, ContextAutomatonWithConcentrations):
return False
return True
def sanity_check(self):
pass
def get_ordinary_reaction_system_with_automaton(self):
if not self.is_with_concentrations():
raise RuntimeError("Not RS/CA with concentrations")
ors = self.rs.get_reaction_system()
oca = self.ca.get_automaton_with_flat_contexts(ors)
return ReactionSystemWithAutomaton(ors, oca)

View File

@@ -0,0 +1,433 @@
from sys import exit
from colour import *
from rs.reaction_system import ReactionSystem
class ReactionSystemWithConcentrations(ReactionSystem):
def __init__(self):
self.reactions = []
self.meta_reactions = dict()
self.permanent_entities = dict()
self.background_set = []
self.context_entities = [] # legacy. to be removed
self.reactions_by_prod = None
self.max_concentration = 0
self.max_conc_per_ent = dict()
def add_bg_set_entity(self, e):
name = ""
def_max_conc = -1
if type(e) is tuple and len(e) == 2:
name, def_max_conc = e
elif type(e) is str:
name = e
print("\nWARNING: no maximal concentration level specified for:", e, "\n")
else:
raise RuntimeError(
"Bad entity type when adding background set element")
self.assume_not_in_bgset(name)
self.background_set.append(name)
if def_max_conc != -1:
ent_id = self.get_entity_id(name)
self.max_conc_per_ent.setdefault(ent_id, 0)
if self.max_conc_per_ent[ent_id] < def_max_conc:
self.max_conc_per_ent[ent_id] = def_max_conc
if self.max_concentration < def_max_conc:
self.max_concentration = def_max_conc
def get_max_concentration_level(self, e):
if e in self.max_conc_per_ent:
return self.max_conc_per_ent[e]
else:
return self.max_concentration
def is_valid_entity_with_concentration(self, e):
"""Sanity check for entities with concentration"""
if type(e) is tuple:
if len(e) == 2 and type(e[1]) is int:
return True
if type(e) is list:
if len(e) == 2 and type(e[1]) is int:
return True
print("FATAL. Invalid entity+concentration: {:s}".format(e))
exit(1)
return False
def get_state_ids(self, state):
"""Returns entities of the given state without levels"""
return [e for e, c in state]
def has_non_zero_concentration(self, elem):
if elem[1] < 1:
raise RuntimeError(
"Unexpected concentration level in state: " + str(elem))
def process_rip(self, R, I, P, ignore_empty_R=False):
"""Chcecks concentration levels and converts entities names into their ids"""
if R == [] and not ignore_empty_R:
raise RuntimeError("No reactants defined")
reactants = []
for r in R:
self.is_valid_entity_with_concentration(r)
self.has_non_zero_concentration(r)
entity, level = r
reactants.append((self.get_entity_id(entity), level))
if self.max_concentration < level:
self.max_concentration = level
inhibitors = []
for i in I:
self.is_valid_entity_with_concentration(i)
self.has_non_zero_concentration(i)
entity, level = i
inhibitors.append((self.get_entity_id(entity), level))
if self.max_concentration < level:
self.max_concentration = level
products = []
for p in P:
self.is_valid_entity_with_concentration(p)
self.has_non_zero_concentration(p)
entity, level = p
products.append((self.get_entity_id(entity), level))
return reactants, inhibitors, products
def add_reaction(self, R, I, P):
"""Adds a reaction"""
if P == []:
raise RuntimeError("No products defined")
reaction = self.process_rip(R, I, P)
self.reactions.append(reaction)
def add_reaction_without_reactants(self, R, I, P):
"""Adds a reaction"""
if P == []:
raise RuntimeError("No products defined")
reaction = self.process_rip(R, I, P, ignore_empty_R=True)
self.reactions.append(reaction)
def add_reaction_inc(self, incr_entity, incrementer, R, I):
"""Adds a macro/meta reaction for increasing the value of incr_entity"""
reactants, inhibitors, products = self.process_rip(
R, I, [], ignore_empty_R=True)
incr_entity_id = self.get_entity_id(incr_entity)
self.meta_reactions.setdefault(incr_entity_id, [])
self.meta_reactions[incr_entity_id].append(
("inc", self.get_entity_id(incrementer), reactants, inhibitors))
def add_reaction_dec(self, decr_entity, decrementer, R, I):
"""Adds a macro/meta reaction for decreasing the value of incr_entity"""
reactants, inhibitors, products = self.process_rip(
R, I, [], ignore_empty_R=True)
decr_entity_id = self.get_entity_id(decr_entity)
self.meta_reactions.setdefault(decr_entity_id, [])
self.meta_reactions[decr_entity_id].append(
("dec", self.get_entity_id(decrementer), reactants, inhibitors))
def add_permanency(self, ent, I):
"""Sets entity to be permanent unless it is inhibited"""
ent_id = self.get_entity_id(ent)
if ent_id in self.permanent_entities:
raise RuntimeError(
"Permanency for {0} already defined.".format(ent))
inhibitors = self.process_rip([], I, [], ignore_empty_R=True)[1]
self.permanent_entities[ent_id] = inhibitors
def set_context_entities(self, entities):
raise NotImplementedError
def entities_names_set_to_str(self, entities):
s = ""
for entity in entities:
s += entity + ", "
s = s[:-2]
return s
def entities_ids_set_to_str(self, entities):
s = ""
for entity in entities:
s += self.get_entity_name(entity) + ", "
s = s[:-2]
return s
def state_to_str(self, state):
s = ""
for ent, level in state:
s += self.get_entity_name(ent) + "=" + str(level) + ", "
s = s[:-2]
return s
def show_background_set(self):
print(
C_MARK_INFO + " Background set: {" + self.entities_names_set_to_str(self.background_set) + "}")
def show_meta_reactions(self):
print(C_MARK_INFO + " Meta reactions:")
for param_ent, reactions in self.meta_reactions.items():
for r_type, command, reactants, inhibitors in reactions:
if r_type == "inc" or r_type == "dec":
print(" - [ Type=" + repr(r_type) + " Operand=( " + self.get_entity_name(param_ent) + " ) Command=( " + self.get_entity_name(
command) + " ) ] -- ( R={" + self.state_to_str(reactants) + "}, I={" + self.state_to_str(inhibitors) + "} )")
else:
raise RuntimeError(
"Unknown meta-reaction type: " + repr(r_type))
def show_max_concentrations(self):
print(
C_MARK_INFO +
" Maximal allowed concentration levels (for optimized translation to RS):")
for e, max_conc in self.max_conc_per_ent.items():
print(" - {0:^20} = {1:<6}".format(self.get_entity_name(e), max_conc))
def show_permanent_entities(self):
print(C_MARK_INFO + " Permanent entities:")
for e, inhibitors in self.permanent_entities.items():
print(" - {0:^20}{1:<6}".format(self.get_entity_name(e) + ": ",
"I={" + self.state_to_str(inhibitors) + "}"))
def show(self, soft=False):
self.show_background_set()
self.show_reactions(soft)
self.show_permanent_entities()
self.show_meta_reactions()
self.show_max_concentrations()
def get_reactions_by_product(self):
"""Sorts reactions by their products and returns a dictionary of products"""
if self.reactions_by_prod != None:
return self.reactions_by_prod
producible_entities = set()
for reaction in self.reactions:
product_entities = [e for e, c in reaction[2]]
producible_entities = producible_entities.union(
set(product_entities))
reactions_by_prod = {}
for p_e in producible_entities:
reactions_by_prod[p_e] = []
rcts_for_p_e = reactions_by_prod[p_e]
for r in self.reactions:
product_entities = [e for e, c in r[2]]
if p_e in product_entities:
reactants = r[0]
inhibitors = r[1]
products = [(e, c) for e, c in r[2] if e == p_e]
prod_conc = products[0][1]
insert_place = None
# we need to order the reactions w.r.t. the concentration levels produced (increasing order)
for i in range(0, len(rcts_for_p_e)):
checked_conc = rcts_for_p_e[i][2][0][1]
if prod_conc <= checked_conc:
insert_place = i
break
if insert_place == None: # empty or the is only one element which is smaller than the element being added
# we append (to the end)
rcts_for_p_e.append((reactants, inhibitors, products))
else:
rcts_for_p_e.insert(
insert_place, (reactants, inhibitors, products))
# save in cache
self.reactions_by_prod = reactions_by_prod
return reactions_by_prod
def get_reaction_system(self):
rs = ReactionSystem()
for reactants, inhibitors, products in self.reactions:
new_reactants = []
new_inhibitors = []
new_products = []
for ent, conc in reactants:
n = self.get_entity_name(ent) + "#" + str(conc)
rs.ensure_bg_set_entity(n)
new_reactants.append(n)
for ent, conc in inhibitors:
n = self.get_entity_name(ent) + "#" + str(conc)
rs.ensure_bg_set_entity(n)
new_inhibitors.append(n)
for ent, conc in products:
for i in range(1, conc+1):
n = self.get_entity_name(ent) + "#" + str(i)
rs.ensure_bg_set_entity(n)
new_products.append(n)
rs.add_reaction(new_reactants, new_inhibitors, new_products)
for param_ent, reactions in self.meta_reactions.items():
for r_type, command, reactants, inhibitors in reactions:
param_ent_name = self.get_entity_name(param_ent)
new_reactants = []
new_inhibitors = []
for ent, conc in reactants:
n = self.get_entity_name(ent) + "#" + str(conc)
rs.ensure_bg_set_entity(n)
new_reactants.append(n)
for ent, conc in inhibitors:
n = self.get_entity_name(ent) + "#" + str(conc)
rs.ensure_bg_set_entity(n)
new_inhibitors.append(n)
max_cmd_c = self.max_concentration
if command in self.max_conc_per_ent:
max_cmd_c = self.max_conc_per_ent[command]
else:
print(
"WARNING:\n\tThere is no maximal concentration level defined for "
+ self.get_entity_name(command))
print("\tThis is a very bad idea -- expect degraded performance\n")
for l in range(1, max_cmd_c+1):
cmd_ent = self.get_entity_name(command) + "#" + str(l)
rs.ensure_bg_set_entity(cmd_ent)
if r_type == "inc":
# pre_conc -- predecessor concentration
# succ_conc -- successor concentration concentration
for i in range(1, self.max_concentration):
pre_conc = param_ent_name + "#" + str(i)
rs.ensure_bg_set_entity(pre_conc)
new_products = []
succ_value = i+l
for j in range(1, succ_value+1):
if j > self.max_concentration:
break
new_p = param_ent_name + "#" + str(j)
rs.ensure_bg_set_entity(new_p)
new_products.append(new_p)
if new_products != []:
rs.add_reaction(
set(new_reactants + [pre_conc, cmd_ent]),
set(new_inhibitors),
set(new_products))
elif r_type == "dec":
for i in range(1, self.max_concentration+1):
pre_conc = param_ent_name + "#" + str(i)
rs.ensure_bg_set_entity(pre_conc)
new_products = []
succ_value = i-l
for j in range(1, succ_value+1):
if j > self.max_concentration:
break
new_p = param_ent_name + "#" + str(j)
rs.ensure_bg_set_entity(new_p)
new_products.append(new_p)
if new_products != []:
rs.add_reaction(
set(new_reactants + [pre_conc, cmd_ent]),
set(new_inhibitors),
set(new_products))
else:
raise RuntimeError(
"Unknown meta-reaction type: " + repr(r_type))
for ent, inhibitors in self.permanent_entities.items():
max_c = self.max_concentration
if ent in self.max_conc_per_ent:
max_c = self.max_conc_per_ent[ent]
else:
print(
"WARNING:\n\tThere is no maximal concentration level defined for "
+ self.get_entity_name(ent))
print("\tThis is a very bad idea -- expect degraded performance\n")
def e_value(i):
return self.get_entity_name(ent) + "#" + str(i)
for value in range(1, max_c+1):
new_reactants = []
new_inhibitors = []
new_products = []
new_reactants = [e_value(value)]
for e_inh, conc in inhibitors:
n = self.get_entity_name(e_inh) + "#" + str(conc)
rs.ensure_bg_set_entity(n)
new_inhibitors.append(n)
for i in range(1, value+1):
new_products.append(e_value(i))
rs.add_reaction(new_reactants, new_inhibitors, new_products)
return rs
class ReactionSystemWithAutomaton(object):
def __init__(self, reaction_system, context_automaton):
self.rs = reaction_system
self.ca = context_automaton
def show(self, soft=False):
self.rs.show(soft)
self.ca.show()
def is_with_concentrations(self):
if not isinstance(self.rs, ReactionSystemWithConcentrations):
return False
if not isinstance(self.ca, ContextAutomatonWithConcentrations):
return False
return True
def sanity_check(self):
pass
def get_ordinary_reaction_system_with_automaton(self):
if not self.is_with_concentrations():
raise RuntimeError("Not RS/CA with concentrations")
ors = self.rs.get_reaction_system()
oca = self.ca.get_automaton_with_flat_contexts(ors)
return ReactionSystemWithAutomaton(ors, oca)
# EOF

View File

@@ -0,0 +1,439 @@
from sys import exit
from colour import *
from rs.reaction_system import ReactionSystem
class ParameterObj(object):
def __init__(self, name):
self.name = name
def __repr__(self):
return "@{0}".format(self.name)
def is_param(some_object):
if isinstance(some_object, ParameterObj):
return True
else:
return False
class ReactionSystemWithConcentrationsParam(ReactionSystem):
def __init__(self):
self.reactions = []
self.parameters = dict()
self.meta_reactions = dict()
self.permanent_entities = dict()
self.background_set = []
self.context_entities = [] # legacy. to be removed
self.reactions_by_prod = None
self.max_concentration = 1
self.max_conc_per_ent = dict()
def add_bg_set_entity(self, e):
name = ""
def_max_conc = -1
if type(e) is tuple and len(e) == 2:
name, def_max_conc = e
elif type(e) is str:
name = e
print("\nWARNING: no maximal concentration level specified for:", e, "\n")
else:
raise RuntimeError(
"Bad entity type when adding background set element")
self.assume_not_in_bgset(name)
self.background_set.append(name)
if def_max_conc != -1:
ent_id = self.get_entity_id(name)
self.max_conc_per_ent.setdefault(ent_id, 0)
if self.max_conc_per_ent[ent_id] < def_max_conc:
self.max_conc_per_ent[ent_id] = def_max_conc
if self.max_concentration < def_max_conc:
self.max_concentration = def_max_conc
def get_param(self, name):
if self.has_param(name):
return self.parameters[name]
else:
param = ParameterObj(name)
self.add_param(param)
return param
def has_param(self, name):
return name in self.parameters
def add_param(self, param):
if param in self.parameters:
raise RuntimeError("Parameter {:s} already exists".format(param))
param_key = param.name
self.parameters[param_key] = param
self.parameters[param_key].idx = len(self.parameters)
def get_max_concentration_level(self, e):
if e in self.max_conc_per_ent:
return self.max_conc_per_ent[e]
else:
return self.max_concentration
def is_valid_entity_with_concentration(self, e):
"""Sanity check for entities with concentration"""
if type(e) is tuple:
if len(e) == 2 and type(e[1]) is int:
return True
if type(e) is list:
if len(e) == 2 and type(e[1]) is int:
return True
print("FATAL. Invalid entity+concentration: {:s}".format(e))
exit(1)
return False
def get_state_ids(self, state):
"""Returns entities of the given state without levels"""
return [self.get_entity_id(e) for e in state]
def has_non_zero_concentration(self, elem):
if elem[1] < 1:
raise RuntimeError(
"Unexpected concentration level in state: " + str(elem))
def process_rip(self, R, I, P, ignore_empty_R=False):
"""Chcecks concentration levels and converts entities names into their ids"""
if R == [] and not ignore_empty_R:
raise RuntimeError("No reactants defined")
#
# REACTANTS
#
reactants = []
if isinstance(R, ParameterObj):
reactants = R
else:
for r in R:
self.is_valid_entity_with_concentration(r)
self.has_non_zero_concentration(r)
entity, level = r
reactants.append((self.get_entity_id(entity), level))
if self.max_concentration < level:
self.max_concentration = level
#
# INHIBITORS
#
inhibitors = []
if isinstance(I, ParameterObj):
inhibitors = I
else:
for i in I:
self.is_valid_entity_with_concentration(i)
self.has_non_zero_concentration(i)
entity, level = i
inhibitors.append((self.get_entity_id(entity), level))
if self.max_concentration < level:
self.max_concentration = level
#
# PRODUCTS
#
products = []
if isinstance(P, ParameterObj):
products = P
else:
for p in P:
self.is_valid_entity_with_concentration(p)
self.has_non_zero_concentration(p)
entity, level = p
products.append((self.get_entity_id(entity), level))
return reactants, inhibitors, products
def is_parametric_reaction(self, reaction):
result = any([isinstance(r_set, ParameterObj) for r_set in reaction])
return result
def add_reaction(self, R, I, P):
"""Adds a reaction
R, I, and P are sets of entities (not their IDs)
"""
if P == []:
raise RuntimeError("No products defined")
reaction = self.process_rip(R, I, P)
self.reactions.append(reaction)
def add_reaction_without_reactants(self, R, I, P):
"""Adds a reaction"""
if P == []:
raise RuntimeError("No products defined")
reaction = self.process_rip(R, I, P, ignore_empty_R=True)
self.reactions.append(reaction)
def add_reaction_inc(self, incr_entity, incrementer, R, I):
"""Adds a macro/meta reaction for increasing the value of incr_entity"""
reactants, inhibitors, products = self.process_rip(
R, I, [], ignore_empty_R=True)
incr_entity_id = self.get_entity_id(incr_entity)
self.meta_reactions.setdefault(incr_entity_id, [])
self.meta_reactions[incr_entity_id].append(
("inc", self.get_entity_id(incrementer), reactants, inhibitors))
def add_reaction_dec(self, decr_entity, decrementer, R, I):
"""Adds a macro/meta reaction for decreasing the value of incr_entity"""
reactants, inhibitors, products = self.process_rip(
R, I, [], ignore_empty_R=True)
decr_entity_id = self.get_entity_id(decr_entity)
self.meta_reactions.setdefault(decr_entity_id, [])
self.meta_reactions[decr_entity_id].append(
("dec", self.get_entity_id(decrementer), reactants, inhibitors))
def add_permanency(self, ent, I):
"""Sets entity to be permanent unless it is inhibited"""
ent_id = self.get_entity_id(ent)
if ent_id in self.permanent_entities:
raise RuntimeError(
"Permanency for {0} already defined.".format(ent))
inhibitors = self.process_rip([], I, [], ignore_empty_R=True)[1]
self.permanent_entities[ent_id] = inhibitors
def set_context_entities(self, entities):
raise NotImplementedError
def entities_names_set_to_str(self, entities):
s = ""
for entity in entities:
s += entity + ", "
s = s[:-2]
return s
def entities_ids_set_to_str(self, entities):
s = ""
for entity in entities:
s += self.get_entity_name(entity) + ", "
s = s[:-2]
return s
def state_to_str(self, state):
"""
If state is a parameter, we return
the string representation of the whole state
which should be the name of the parameter
"""
if isinstance(state, ParameterObj):
return str(state)
else:
s = ""
for ent, level in state:
s += self.get_entity_name(ent) + "=" + str(level) + ", "
s = s[:-2]
return s
def show_background_set(self):
print(
C_MARK_INFO + " Background set: {" + self.entities_names_set_to_str(self.background_set) + "}")
def show_meta_reactions(self):
print(C_MARK_INFO + " Meta reactions:")
for param_ent, reactions in self.meta_reactions.items():
for r_type, command, reactants, inhibitors in reactions:
if r_type == "inc" or r_type == "dec":
print(" - [ Type=" + repr(r_type) + " Operand=( " + self.get_entity_name(param_ent) + " ) Command=( " + self.get_entity_name(
command) + " ) ] -- ( R={" + self.state_to_str(reactants) + "}, I={" + self.state_to_str(inhibitors) + "} )")
else:
raise RuntimeError(
"Unknown meta-reaction type: " + repr(r_type))
def show_max_concentrations(self):
print(C_MARK_INFO + " Maximal allowed concentration levels:")
for e, max_conc in self.max_conc_per_ent.items():
print(" - {0:^20} = {1:<6}".format(self.get_entity_name(e), max_conc))
def show_permanent_entities(self):
print(C_MARK_INFO + " Permanent entities:")
for e, inhibitors in self.permanent_entities.items():
print(" - {0:^20}{1:<6}".format(self.get_entity_name(e) + ": ",
"I={" + self.state_to_str(inhibitors) + "}"))
def show(self, soft=False):
self.show_background_set()
self.show_reactions(soft)
# self.show_param_reactions(soft)
self.show_permanent_entities()
self.show_meta_reactions()
self.show_max_concentrations()
def get_producible_entities(self):
"""
Returns the set of entities that appear as products of
reactions.
"""
producible_entities = set()
for reaction in self.reactions:
product_entities = [e for e, c in reaction[2] if c > 0]
producible_entities = producible_entities.union(
set(product_entities))
return producible_entities
def get_reaction_system(self):
"""
Translates RSC into RS
"""
rs = ReactionSystem()
for reactants, inhibitors, products in self.reactions:
new_reactants = []
new_inhibitors = []
new_products = []
for ent, conc in reactants:
n = self.get_entity_name(ent) + "#" + str(conc)
rs.ensure_bg_set_entity(n)
new_reactants.append(n)
for ent, conc in inhibitors:
n = self.get_entity_name(ent) + "#" + str(conc)
rs.ensure_bg_set_entity(n)
new_inhibitors.append(n)
for ent, conc in products:
for i in range(1, conc + 1):
n = self.get_entity_name(ent) + "#" + str(i)
rs.ensure_bg_set_entity(n)
new_products.append(n)
rs.add_reaction(new_reactants, new_inhibitors, new_products)
for param_ent, reactions in self.meta_reactions.items():
for r_type, command, reactants, inhibitors in reactions:
param_ent_name = self.get_entity_name(param_ent)
new_reactants = []
new_inhibitors = []
for ent, conc in reactants:
n = self.get_entity_name(ent) + "#" + str(conc)
rs.ensure_bg_set_entity(n)
new_reactants.append(n)
for ent, conc in inhibitors:
n = self.get_entity_name(ent) + "#" + str(conc)
rs.ensure_bg_set_entity(n)
new_inhibitors.append(n)
max_cmd_c = self.max_concentration
if command in self.max_conc_per_ent:
max_cmd_c = self.max_conc_per_ent[command]
else:
print(
"WARNING:\n\tThere is no maximal concentration level defined for "
+ self.get_entity_name(command))
print("\tThis is a very bad idea -- expect degraded performance\n")
for l in range(1, max_cmd_c + 1):
cmd_ent = self.get_entity_name(command) + "#" + str(l)
rs.ensure_bg_set_entity(cmd_ent)
if r_type == "inc":
# pre_conc -- predecessor concentration
# succ_conc -- successor concentration concentration
for i in range(1, self.max_concentration):
pre_conc = param_ent_name + "#" + str(i)
rs.ensure_bg_set_entity(pre_conc)
new_products = []
succ_value = i + l
for j in range(1, succ_value + 1):
if j > self.max_concentration:
break
new_p = param_ent_name + "#" + str(j)
rs.ensure_bg_set_entity(new_p)
new_products.append(new_p)
if new_products != []:
rs.add_reaction(
set(new_reactants + [pre_conc, cmd_ent]),
set(new_inhibitors),
set(new_products))
elif r_type == "dec":
for i in range(1, self.max_concentration + 1):
pre_conc = param_ent_name + "#" + str(i)
rs.ensure_bg_set_entity(pre_conc)
new_products = []
succ_value = i - l
for j in range(1, succ_value + 1):
if j > self.max_concentration:
break
new_p = param_ent_name + "#" + str(j)
rs.ensure_bg_set_entity(new_p)
new_products.append(new_p)
if new_products != []:
rs.add_reaction(
set(new_reactants + [pre_conc, cmd_ent]),
set(new_inhibitors),
set(new_products))
else:
raise RuntimeError(
"Unknown meta-reaction type: " + repr(r_type))
for ent, inhibitors in self.permanent_entities.items():
max_c = self.max_concentration
if ent in self.max_conc_per_ent:
max_c = self.max_conc_per_ent[ent]
else:
print(
"WARNING:\n\tThere is no maximal concentration level defined for "
+ self.get_entity_name(ent))
print("\tThis is a very bad idea -- expect degraded performance\n")
def e_value(i):
return self.get_entity_name(ent) + "#" + str(i)
for value in range(1, max_c + 1):
new_reactants = []
new_inhibitors = []
new_products = []
new_reactants = [e_value(value)]
for e_inh, conc in inhibitors:
n = self.get_entity_name(e_inh) + "#" + str(conc)
rs.ensure_bg_set_entity(n)
new_inhibitors.append(n)
for i in range(1, value + 1):
new_products.append(e_value(i))
rs.add_reaction(new_reactants, new_inhibitors, new_products)
return rs
# EOF