123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221 |
- from uuid import UUID
- from state.base import State
- from typing import Any, List
- class Bottom:
- """
- Implements services for LTM bottom (not yet explicitly modelled).
- Implemented using the (Modelverse) state graph data structure
- """
- def __init__(self, state: State):
- self.state = state
- def create_node(self, value=None) -> UUID:
- """
- Creates a node, which optionally contains a value.
- Args:
- value: value to be stored in the node
- Returns:
- UUID of the node.
- """
- if value == None:
- return self.state.create_node()
- else:
- return self.state.create_nodevalue(value)
- def create_edge(self, source: UUID, target: UUID, label=None):
- """
- Creates an edge, which optionally is labelled.
- Args:
- source: source element of the edge
- target: target element of the edge
- label: value to label the edge with
- Returns:
- UUID of the edge.
- """
- if label == None:
- return self.state.create_edge(source, target)
- else:
- return self.state.create_dict(source, label, target)
- def read_value(self, node: UUID) -> Any:
- """
- Reads value stored in a node.
- Args:
- node: UUID of the node to read value from
- Returns:
- Value contained in the node. If no value is found, returns None
- """
- return self.state.read_value(node)
- def read_edge_source(self, edge: UUID) -> UUID:
- """
- Reads source element of an edge.
- Args:
- edge: UUID of the edge to read source element of
- Returns:
- UUID of source element of the edge
- """
- result = self.state.read_edge(edge)
- return result[0] if result != None else result
- def read_edge_target(self, edge: UUID) -> UUID:
- """
- Reads target element of an edge.
- Args:
- edge: UUID of the edge to read target element of
- Returns:
- UUID of target element of the edge
- """
- result = self.state.read_edge(edge)
- return result[1] if result != None else result
- def is_edge(self, elem: UUID) -> bool:
- return self.state.is_edge(elem)
- def read_incoming_edges(self, target: UUID, label=None) -> List[UUID]:
- """
- Reads incoming edges of an element. Optionally, filter them based on their label
- Args:
- target: UUID of the element to read incoming edges for
- label: value to filter edge labels by
- Returns:
- List of UUIDs of incoming edges
- """
- def read_label(_edge: UUID):
- try:
- label_edge, = self.state.read_outgoing(_edge)
- _, tgt = self.state.read_edge(label_edge)
- _label = self.state.read_value(tgt)
- return _label
- except (TypeError, ValueError):
- return None
- edges = self.state.read_incoming(target)
- if edges == None:
- return []
- if label != None:
- edges = [e for e in edges if read_label(e) == label]
- return edges
- def read_outgoing_edges(self, source: UUID, label=None) -> List[UUID]:
- """
- Reads outgoing edges of an element. Optionally, filter them based on their label
- Args:
- source: UUID of the element to read outgoing edges for
- label: value to filter edge labels by
- Returns:
- List of UUIDs of outgoing edges
- """
- ### PERFORMANCE OPTIMIZATION ###
- if label != None:
- fast_result = self.state.read_dict_edge_all(source, label)
- # if set(alt) != set(edges):
- # raise Exception("WRONG", alt, edges)
- return fast_result
- ### PERFORMANCE OPTIMIZATION ###
- def read_label(_edge: UUID):
- try:
- label_edge, = self.state.read_outgoing(_edge)
- _, tgt = self.state.read_edge(label_edge)
- _label = self.state.read_value(tgt)
- return _label
- except (TypeError, ValueError):
- return None
- edges = self.state.read_outgoing(source)
- if edges == None:
- return []
- if label != None:
- edges = [e for e in edges if read_label(e) == label]
- return edges
- def read_incoming_elements(self, target: UUID, label=None) -> List[UUID]:
- """
- Reads elements connected to given element via incoming edges.
- Optionally, filter them based on the edge label.
- Args:
- target: UUID of the element to read incoming elements for
- label: value to filter edge labels by
- Returns:
- List of UUIDs of elements connected via incoming edges
- """
- edges = self.read_incoming_edges(target, label)
- if edges == None or len(edges) == 0:
- return []
- else:
- return [self.read_edge_source(e) for e in edges]
- def read_outgoing_elements(self, source: UUID, label=None) -> List[UUID]:
- """
- Reads elements connected to given element via outgoing edges.
- Optionally, filter them based on the edge label.
- Args:
- source: UUID of the element to read outgoing elements for
- label: value to filter edge labels by
- Returns:
- List of UUIDs of elements connected via outgoing edges
- """
- edges = self.read_outgoing_edges(source, label)
- if edges == None or len(edges) == 0:
- return []
- else:
- return [self.read_edge_target(e) for e in edges]
- def read_keys(self, element: UUID) -> List[str]:
- """
- Retrieve list of outgoing edge labels
- Args:
- element: UUID of the element to read outgoing edge labels for
- Returns:
- List of outgoing edge labels
- """
- key_nodes = self.state.read_dict_keys(element)
- unique_keys = {self.state.read_value(node) for node in key_nodes}
- return list(unique_keys)
- def delete_element(self, element: UUID):
- """
- Delete an element
- Args:
- element: UUID of the element to be deleted
- Returns:
- Nothing
- """
- src, tgt = self.state.read_edge(element)
- if src == None and tgt == None:
- # node
- self.state.delete_node(element)
- else:
- # edge
- self.state.delete_edge(element)
|