V0.py 6.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221
  1. from uuid import UUID
  2. from state.base import State
  3. from typing import Any, List
  4. class Bottom:
  5. """
  6. Implements services for LTM bottom (not yet explicitly modelled).
  7. Implemented using the (Modelverse) state graph data structure
  8. """
  9. def __init__(self, state: State):
  10. self.state = state
  11. def create_node(self, value=None) -> UUID:
  12. """
  13. Creates a node, which optionally contains a value.
  14. Args:
  15. value: value to be stored in the node
  16. Returns:
  17. UUID of the node.
  18. """
  19. if value == None:
  20. return self.state.create_node()
  21. else:
  22. return self.state.create_nodevalue(value)
  23. def create_edge(self, source: UUID, target: UUID, label=None):
  24. """
  25. Creates an edge, which optionally is labelled.
  26. Args:
  27. source: source element of the edge
  28. target: target element of the edge
  29. label: value to label the edge with
  30. Returns:
  31. UUID of the edge.
  32. """
  33. if label == None:
  34. return self.state.create_edge(source, target)
  35. else:
  36. return self.state.create_dict(source, label, target)
  37. def read_value(self, node: UUID) -> Any:
  38. """
  39. Reads value stored in a node.
  40. Args:
  41. node: UUID of the node to read value from
  42. Returns:
  43. Value contained in the node. If no value is found, returns None
  44. """
  45. return self.state.read_value(node)
  46. def read_edge_source(self, edge: UUID) -> UUID:
  47. """
  48. Reads source element of an edge.
  49. Args:
  50. edge: UUID of the edge to read source element of
  51. Returns:
  52. UUID of source element of the edge
  53. """
  54. result = self.state.read_edge(edge)
  55. return result[0] if result != None else result
  56. def read_edge_target(self, edge: UUID) -> UUID:
  57. """
  58. Reads target element of an edge.
  59. Args:
  60. edge: UUID of the edge to read target element of
  61. Returns:
  62. UUID of target element of the edge
  63. """
  64. result = self.state.read_edge(edge)
  65. return result[1] if result != None else result
  66. def is_edge(self, elem: UUID) -> bool:
  67. return self.state.is_edge(elem)
  68. def read_incoming_edges(self, target: UUID, label=None) -> List[UUID]:
  69. """
  70. Reads incoming edges of an element. Optionally, filter them based on their label
  71. Args:
  72. target: UUID of the element to read incoming edges for
  73. label: value to filter edge labels by
  74. Returns:
  75. List of UUIDs of incoming edges
  76. """
  77. def read_label(_edge: UUID):
  78. try:
  79. label_edge, = self.state.read_outgoing(_edge)
  80. _, tgt = self.state.read_edge(label_edge)
  81. _label = self.state.read_value(tgt)
  82. return _label
  83. except (TypeError, ValueError):
  84. return None
  85. edges = self.state.read_incoming(target)
  86. if edges == None:
  87. return []
  88. if label != None:
  89. edges = [e for e in edges if read_label(e) == label]
  90. return edges
  91. def read_outgoing_edges(self, source: UUID, label=None) -> List[UUID]:
  92. """
  93. Reads outgoing edges of an element. Optionally, filter them based on their label
  94. Args:
  95. source: UUID of the element to read outgoing edges for
  96. label: value to filter edge labels by
  97. Returns:
  98. List of UUIDs of outgoing edges
  99. """
  100. ### PERFORMANCE OPTIMIZATION ###
  101. if label != None:
  102. fast_result = self.state.read_dict_edge_all(source, label)
  103. # if set(alt) != set(edges):
  104. # raise Exception("WRONG", alt, edges)
  105. return fast_result
  106. ### PERFORMANCE OPTIMIZATION ###
  107. def read_label(_edge: UUID):
  108. try:
  109. label_edge, = self.state.read_outgoing(_edge)
  110. _, tgt = self.state.read_edge(label_edge)
  111. _label = self.state.read_value(tgt)
  112. return _label
  113. except (TypeError, ValueError):
  114. return None
  115. edges = self.state.read_outgoing(source)
  116. if edges == None:
  117. return []
  118. if label != None:
  119. edges = [e for e in edges if read_label(e) == label]
  120. return edges
  121. def read_incoming_elements(self, target: UUID, label=None) -> List[UUID]:
  122. """
  123. Reads elements connected to given element via incoming edges.
  124. Optionally, filter them based on the edge label.
  125. Args:
  126. target: UUID of the element to read incoming elements for
  127. label: value to filter edge labels by
  128. Returns:
  129. List of UUIDs of elements connected via incoming edges
  130. """
  131. edges = self.read_incoming_edges(target, label)
  132. if edges == None or len(edges) == 0:
  133. return []
  134. else:
  135. return [self.read_edge_source(e) for e in edges]
  136. def read_outgoing_elements(self, source: UUID, label=None) -> List[UUID]:
  137. """
  138. Reads elements connected to given element via outgoing edges.
  139. Optionally, filter them based on the edge label.
  140. Args:
  141. source: UUID of the element to read outgoing elements for
  142. label: value to filter edge labels by
  143. Returns:
  144. List of UUIDs of elements connected via outgoing edges
  145. """
  146. edges = self.read_outgoing_edges(source, label)
  147. if edges == None or len(edges) == 0:
  148. return []
  149. else:
  150. return [self.read_edge_target(e) for e in edges]
  151. def read_keys(self, element: UUID) -> List[str]:
  152. """
  153. Retrieve list of outgoing edge labels
  154. Args:
  155. element: UUID of the element to read outgoing edge labels for
  156. Returns:
  157. List of outgoing edge labels
  158. """
  159. key_nodes = self.state.read_dict_keys(element)
  160. unique_keys = {self.state.read_value(node) for node in key_nodes}
  161. return list(unique_keys)
  162. def delete_element(self, element: UUID):
  163. """
  164. Delete an element
  165. Args:
  166. element: UUID of the element to be deleted
  167. Returns:
  168. Nothing
  169. """
  170. src, tgt = self.state.read_edge(element)
  171. if src == None and tgt == None:
  172. # node
  173. self.state.delete_node(element)
  174. else:
  175. # edge
  176. self.state.delete_edge(element)