muMLE/services/bottom/V0.py

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)