from uuid import UUID from state.base import State from services.bottom.V0 import Bottom from services.primitives.integer_type import Integer from services.primitives.string_type import String import re class PN: """ Implements services for the petri nets LTM. Implementation is done in terms of services provided by LTM-bottom. Implementation is very similar to that in scd.py, which has more extensive comments """ def __init__(self, model: UUID, state: State): ltm_pn_id = state.read_dict(state.read_root(), "PN") self.ltm_pn = UUID(state.read_value(ltm_pn_id)) self.model = model self.bottom = Bottom(state) def create_place(self, name: str, tokens: int): """ Creates a place element. Args: name: name of the place Returns: Nothing. """ # instantiate Place class place_node = self.bottom.create_node() # create place node self.bottom.create_edge(self.model, place_node, name) # attach to model morph_node, = self.bottom.read_outgoing_elements(self.ltm_pn, "P") # retrieve type self.bottom.create_edge(place_node, morph_node, "Morphism") # create morphism link # instantiate name attribute name_model = self.bottom.create_node() String(name_model, self.bottom.state).create(name) name_node = self.bottom.create_node(str(name_model)) self.bottom.create_edge(self.model, name_node, f"{name}.n") name_link = self.bottom.create_edge(place_node, name_node) self.bottom.create_edge(self.model, name_link, f"{name}.n_link") ltm_pn_node, = self.bottom.read_outgoing_elements(self.ltm_pn, "String") ltm_pn_link, = self.bottom.read_outgoing_elements(self.ltm_pn, "P_n") self.bottom.create_edge(name_node, ltm_pn_node, "Morphism") self.bottom.create_edge(name_link, ltm_pn_link, "Morphism") # instantiate tokens attribute tokens_model = self.bottom.create_node() Integer(tokens_model, self.bottom.state).create(tokens) tokens_node = self.bottom.create_node(str(tokens_model)) self.bottom.create_edge(self.model, tokens_node, f"{name}.t") tokens_link = self.bottom.create_edge(place_node, tokens_node) self.bottom.create_edge(self.model, tokens_link, f"{name}.t_link") ltm_pn_node, = self.bottom.read_outgoing_elements(self.ltm_pn, "Integer") ltm_pn_link, = self.bottom.read_outgoing_elements(self.ltm_pn, "P_t") self.bottom.create_edge(tokens_node, ltm_pn_node, "Morphism") self.bottom.create_edge(tokens_link, ltm_pn_link, "Morphism") def create_transition(self, name: str): """ Creates a transition element. Args: name: name of the transition Returns: Nothing. """ # instantiate Transition class transition_node = self.bottom.create_node() # create transition node self.bottom.create_edge(self.model, transition_node, name) # attach to model morph_node, = self.bottom.read_outgoing_elements(self.ltm_pn, "T") # retrieve type self.bottom.create_edge(transition_node, morph_node, "Morphism") # create morphism link # instantiate name attribute name_model = self.bottom.create_node() String(name_model, self.bottom.state).create(name) name_node = self.bottom.create_node(str(name_model)) self.bottom.create_edge(self.model, name_node, f"{name}.name") name_link = self.bottom.create_edge(transition_node, name_node) self.bottom.create_edge(self.model, name_link, f"{name}.name_link") ltm_pn_node, = self.bottom.read_outgoing_elements(self.ltm_pn, "String") ltm_pn_link, = self.bottom.read_outgoing_elements(self.ltm_pn, "T_name") self.bottom.create_edge(name_node, ltm_pn_node, "Morphism") self.bottom.create_edge(name_link, ltm_pn_link, "Morphism") def create_p2t(self, place: str, transition: str, weight: int): """ Creates a place to transition link. Args: place: source of the link transition: target of the link weight: weight of the link Returns: Nothing. """ # create p2t link + morphism links edge = self.bottom.create_edge( *self.bottom.read_outgoing_elements(self.model, place), *self.bottom.read_outgoing_elements(self.model, transition), ) self.bottom.create_edge(self.model, edge, f"{place}_to_{transition}") # attach to model morph_node, = self.bottom.read_outgoing_elements(self.ltm_pn, "P2T") # retrieve type self.bottom.create_edge(edge, morph_node, "Morphism") # create morphism link # weight attribute weight_model = self.bottom.create_node() Integer(weight_model, self.bottom.state).create(weight) weight_node = self.bottom.create_node(str(weight_model)) self.bottom.create_edge(self.model, weight_node, f"{place}_to_{transition}.weight") weight_link = self.bottom.create_edge(edge, weight_node) self.bottom.create_edge(self.model, weight_link, f"{place}_to_{transition}.weight_link") scd_node, = self.bottom.read_outgoing_elements(self.ltm_pn, "Integer") scd_link, = self.bottom.read_outgoing_elements(self.ltm_pn, "P2T_weight") self.bottom.create_edge(weight_node, scd_node, "Morphism") self.bottom.create_edge(weight_link, scd_link, "Morphism") def create_t2p(self, transition: str, place: str, weight: int): """ Creates a transition to place link. Args: transition: source of the link place: target of the link weight: weight of the link Returns: Nothing. """ # create t2p link + morphism links edge = self.bottom.create_edge( *self.bottom.read_outgoing_elements(self.model, transition), *self.bottom.read_outgoing_elements(self.model, place), ) self.bottom.create_edge(self.model, edge, f"{transition}_to_{place}") # attach to model morph_node, = self.bottom.read_outgoing_elements(self.ltm_pn, "T2P") # retrieve type self.bottom.create_edge(edge, morph_node, "Morphism") # create morphism link # weight attribute weight_model = self.bottom.create_node() Integer(weight_model, self.bottom.state).create(weight) weight_node = self.bottom.create_node(str(weight_model)) self.bottom.create_edge(self.model, weight_node, f"{transition}_to_{place}.weight") weight_link = self.bottom.create_edge(edge, weight_node) self.bottom.create_edge(self.model, weight_link, f"{transition}_to_{place}.weight_link") scd_node, = self.bottom.read_outgoing_elements(self.ltm_pn, "Integer") scd_link, = self.bottom.read_outgoing_elements(self.ltm_pn, "T2P_weight") self.bottom.create_edge(weight_node, scd_node, "Morphism") self.bottom.create_edge(weight_link, scd_link, "Morphism") def list_elements(self): """ Lists elements in the model. Returns: A list of elements in alphabetical order. """ pn_names = {} for key in self.bottom.read_keys(self.ltm_pn): element, = self.bottom.read_outgoing_elements(self.ltm_pn, key) pn_names[element] = key unsorted = [] for key in self.bottom.read_keys(self.model): element, = self.bottom.read_outgoing_elements(self.model, key) element_types = self.bottom.read_outgoing_elements(element, "Morphism") type_model_elements = self.bottom.read_outgoing_elements(self.ltm_pn) element_type_node, = [e for e in element_types if e in type_model_elements] unsorted.append((key, pn_names[element_type_node])) for elem in sorted(unsorted, key=lambda e: e[0]): print("{} : {}".format(*elem)) def delete_element(self, name: str): """ Deletes an element from the model. Args: name: name of the element to delete Returns: Nothing. """ keys = self.bottom.read_keys(self.model) r = re.compile(r"{}\..*".format(name)) to_delete = list(filter(r.match, keys)) for key in to_delete: # TODO: find way to solve memory leak, primitive models are not deleted this way node, = self.bottom.read_outgoing_elements(self.model, label=key) self.bottom.delete_element(node) def to_bottom(self): # already implemented in terms of LTM bottom pass def from_bottom(self): # already implemented in terms of LTM bottom pass