API¶

class Vertex:
    """
    Defines a vertex/node with a given label and incoming and outgoing edges.
    """

    def __init__(
        self,
        graph: "Graph",
        vertex_id: int,
        label: str,
        vertex_type: str | None = None,
        metadata: dict[str, Any] | None = None,
        **kwargs,
    ):
        """
        Creates a new vertex inside the graph.

        :param graph: The graph where the vertex is a member
        :param vertex_id: The unique id of the vertex inside the graph
        :param label: The label of the vertex as string
        :param vertex_type: The v_type of the vertex
        :param metadata: Optional metadata for the vertex
        """
        self._graph: Graph = graph
        self._vertex_id = vertex_id
        self.outgoing_edges: list[Edge] = []
        self.incoming_edges: list[Edge] = []
        self._label = label
        self.metadata: dict[str, Any] = {}
        if metadata:
            self.metadata.update(metadata)
        self.metadata.update(**kwargs)
        if vertex_type is None:
            vertex_type = self.metadata.get("v_type", None)
        if vertex_type is None:
            vertex_type = ""
        self.metadata["v_type"] = vertex_type
        _ = self.vertex_type_id
        self._label_id = self.graph.graph_collection.get_set_label_id(self._label, self.metadata.get("v_type", None))
        self.metrics = Vertex._VertexMetrics(self)

    def add_edge(self, other_vertex: "Vertex", directed=True, edge_metadata: dict[str, Any] | None = None) -> "Vertex":
        """
        Add an outgoing edge to this vertex. The other vertex must be present in the same graph.

        :param other_vertex: The other vertex
        :param directed: Defines if the edge is directed or not
        :param edge_metadata: The optional metadata for the edge
        :return: this vertex
        """
        self.graph.add_edge(self, other_vertex, directed=directed, edge_metadata=edge_metadata)
        return self

    @property
    def label(self) -> str:
        """The label property."""
        return self._label

    @label.setter
    def label(self, value):
        self._label = value
        self._label_id = self.graph.graph_collection.get_set_label_id(self._label, self.metadata.get("v_type", None))

    @property
    def vertex_type(self) -> str:
        return self.metadata.get("v_type", "")

    @property
    def vertex_type_id(self) -> int:
        return self.graph.graph_collection.get_set_label_id(self.vertex_type, self.vertex_type)

    @property
    def label_id(self) -> int:
        """The id of the label"""
        return self._label_id

    @property
    def graph(self) -> "Graph":
        """The graph of the vertex"""
        return self._graph

    @property
    def vertex_id(self) -> int:
        """The id of the vertex"""
        return self._vertex_id

    def as_str_rep(self, variant) -> str:
        """
        To string representation:
        The id of the vertex is printed, as well as the label_id as integer and the type id of the vertex
        (No metadata are saved)

        'v id label type'

        e.g.: 'v 1 3 5'
        :return: A string representation of the vertex
        """
        if variant == "subdue":
            return f'v {self.vertex_id + 1} "{self.label}"'
        elif variant == "gspan":
            return f"v {self.vertex_id} {self.label_id}"
        return f"v {self.vertex_id} {self.label_id} {self.vertex_type_id}"

    @property
    def proceeding_vertices(self) -> list["Vertex"]:
        """
        Return all proceeding vertices (from outgoing edges).

        :return: All proceeding vertices (from outgoing edges)
        """
        return [e.to_vertex for e in self.outgoing_edges]

    @property
    def preceding_vertices(self) -> list["Vertex"]:
        """
        Return all preceding vertices (from incoming edges).

        :return: All preceding vertices (from incoming edges)
        """
        return [e.from_vertex for e in self.incoming_edges]

    def get_edge(self, other: "Vertex") -> Optional["Edge"]:
        """
        Get an outgoing edge to the other Vertex, return None if there is no edge

        :param other: The other Vertex
        :return: An edge or None if there is no edge
        """
        m1 = self.graph.edge_mapping.get(self.vertex_id)
        if m1 is None:
            return None
        return m1.get(other.vertex_id)

    def has_edge(self, other: "Vertex") -> bool:
        """
        Check if there is an outgoing edge to the other Vertex

        :param other: The other Vertex
        :return: True if there is an edge, else False
        """
        return self.get_edge(other) is not None

    def __eq__(self, other):
        """
        Two vertices are equal if the vertex_id is equal. The metadata values are not compared.

        :param other: Vertex
        :return: True if both vertices are equal
        """
        if not isinstance(other, Vertex):
            return False
        elif self is other:
            return True
        return self.graph.graph_id == other.graph.graph_id and self.vertex_id == other.vertex_id

    def __str__(self):
        return f"Vertex(#{self.vertex_id}, {self.label} ({self.vertex_type}))"

    def __repr__(self):
        return str(self)

    class _VertexMetrics:
        def __init__(self, vertex: "Vertex"):
            self.vertex = vertex

        @property
        def indegree(self):
            """
            Get the indegree of this vertex = sum of all incoming edges.

            :return: The indegee (int) of this vertex
            """
            return len(self.vertex.incoming_edges)

        @property
        def outdegree(self):
            """
            Get the outdegree of this vertex = sum of all outgoining edges.

            :return: The outdegree (int) of this vertex
            """
            return len(self.vertex.outgoing_edges)

        @property
        def strength(self):
            """
            Get the strength of this vertex = indegree - outdegree

            :return: The strength of this vertex
            """
            return self.indegree - self.outdegree

class Edge:
    """
    An edge represents a directed arc between two vertices.
    """

    def __init__(
        self,
        graph: "Graph",
        from_vertex: Vertex,
        to_vertex: Vertex,
        metadata: dict[str, Any] | None = None,
        **kwargs,
    ):
        """
        Creates a new directed edge between two vertices.

        :param graph: The graph, where the edge is a member
        :param from_vertex: The source vertex
        :param to_vertex: The sink vertex
        :param metadata: Optional metadata about the edge
        """
        self._graph: Graph = graph
        assert self.graph is from_vertex.graph is to_vertex.graph
        self._from_vertex = from_vertex
        self._from_vertex.outgoing_edges.append(self)
        self._to_vertex = to_vertex
        self._to_vertex.incoming_edges.append(self)
        self.metadata: dict[str, Any] = {}
        if metadata:
            self.metadata.update(metadata)
        self.metadata.update(**kwargs)

    @property
    def graph(self) -> "Graph":
        """The graph of the edge"""
        return self._graph

    @property
    def to_vertex(self) -> Vertex:
        """The sink vertex of this edge"""
        return self._to_vertex

    @property
    def from_vertex(self) -> Vertex:
        """The source vertex of this edge"""
        return self._from_vertex

    def as_str_rep(self, variant="gspan") -> str:
        """
        The str representation of the edge where the ids of the vertices are used.
        (No metadata are saved)

        'e source sink type'

        e.g.:

        'e 1 2 1'

        :return: A string representation of the edge
        """
        if variant == "subdue":
            return f"e {self.from_vertex.vertex_id + 1} {self.to_vertex.vertex_id + 1} 1"
        return f"e {self.from_vertex.vertex_id} {self.to_vertex.vertex_id} 1"

    @property
    def as_edge_rep(self):
        """
        Save as edge representation. (No metadata are saved)

        :return: A tuple of the source vertex label and the sink vertex label, and the type
        """
        return (self.from_vertex.label, self.to_vertex.label), 1

    def __eq__(self, other):
        """
        The other edge is only equal, if the source vertex and the sink vertex are equal.
        The metadata values are not compared.

        :param other: the other edge
        :return: True if both edges are equal
        """
        if not isinstance(other, Edge):
            return False
        elif self is other:
            return True
        return self.from_vertex == other.from_vertex and self.to_vertex == other.to_vertex

    def __str__(self):
        return f"{self.from_vertex} --> {self.to_vertex}"

    def __repr__(self):
        return str(self)

    def __hash__(self):
        return hash((self.from_vertex.label, self.to_vertex.label))

@dataclass
class GraphSetCluster:
    cluster_id: int
    representative: Graph | None
    graphs: set[Graph] = field(default_factory=lambda: set())
    attributes: dict[str, Any] = field(default_factory=dict)

    def zip_images(self, path: str, format: str = "svg"):
        """
        Exports the graphs of a GraphCollection cluster to a zip file (images of the graphs).

        :param path: The base path where the zip file will be saved.
        :param format: The format of the images, Default: svg
        """
        zip_images_of_graphs(list(self.graphs), path, format)

class Graph:
    """
    A graph represents a set of vertices and a set of edges which connects the vertices.
    """

    def __init__(
        self,
        graph_collection: "GraphCollection",
        graph_id: int,
        cluster_id: int | None = None,
        metadata: dict[str, Any] | None = None,
        **kwargs,
    ):
        """
        Creates a new graph. Each graph is part of a graph collection
        (which contains information about the label mapping)

        :param graph_collection: The Graph Collection
        :param graph_id: The id of the graph
        :param cluster_id: The id of the cluster the graph belongs to
        :param metadata: Optional metadata for the graph
        """
        self.graph_collection = graph_collection
        self.graph_id = graph_id
        self.cluster_id = cluster_id
        self.vertices: dict[int, Vertex] = {}
        self.label_vertices_mapping: dict[int, list[Vertex]] = defaultdict(list)
        self.edges: list[Edge] = []
        self.edge_mapping: dict[int, dict[int, Edge]] = defaultdict(dict)
        self.metadata: dict[str, Any] = {}
        if metadata:
            self.metadata.update(metadata)
        self.metadata.update(**kwargs)

    def add_edge(
        self, from_vertex: Vertex, to_vertex: Vertex, directed=True, edge_metadata: dict[str, Any] | None = None
    ):
        """
        Add a new edge in this graph. Both vertices must be part of this graph.

        :param from_vertex: The source vertex
        :param to_vertex: The sink vertex
        :param directed: if False, the edge is created twice. In the second edge the source and sink vertex are swapped.
        :param edge_metadata: Optional metadata for the edge
        """
        # both, from_vertex and to_vertex should be created first!
        if from_vertex.graph is not self:
            raise ValueError(
                f"From Vertex does not belongs to this graph! "
                f"Was graph {from_vertex.graph.graph_id}; this graph: {self.graph_id}"
            )
        if to_vertex.graph is not self:
            raise ValueError(
                f"To Vertex does not belongs to this graph! "
                f"Was graph {from_vertex.graph.graph_id}; this graph: {self.graph_id}"
            )
        new_edge = Edge(self, from_vertex, to_vertex, metadata=edge_metadata)
        self.edges.append(new_edge)
        self.edge_mapping[from_vertex.vertex_id][to_vertex.vertex_id] = new_edge
        if not directed:
            new_reverse_edge = Edge(self, to_vertex, from_vertex, metadata=edge_metadata)
            self.edges.append(new_reverse_edge)
            self.edge_mapping[to_vertex.vertex_id][from_vertex.vertex_id] = new_reverse_edge

    def get_vertex(
        self,
        label: str,
        *,
        force_create=False,
        vertex_id: int | None = None,
        vertex_type: str | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> Vertex:
        """
        Get the vertex with the given label. If the vertex does not exist, it will be created.

        :param label: The label of the requested vertex
        :param force_create: Force create a new vertex, even if there is a vertex with the same label
        :param vertex_id: If provided, the vertex with this id is returned if exising, else created with this id.
          If the force create is True, and no vertex_id is provided, a new vertex id will be created.
        :param vertex_type: The optional vertex type of the vertex
        :param metadata: The optional metadata added to the vertex if the vertex is created.
            This parameter will be ignored if the vertex already exists.
        :return: The vertex with the given label
        """
        if vertex_id is not None and vertex_id in self.vertices:
            v = self.get_vertex_by_id(vertex_id)
            if v is not None:
                return v
        if vertex_type is None and metadata is not None:
            vertex_type = metadata.get("v_type", None)
        label_id = self.graph_collection.get_set_label_id(label, vertex_type)
        if not force_create and label_id in self.label_vertices_mapping:
            return self.label_vertices_mapping[label_id][-1]
        if vertex_id is None:
            vertex_id = len(self.vertices)
            while vertex_id in self.vertices:
                vertex_id += 1
        new_vertex = Vertex(self, vertex_id, label, vertex_type=vertex_type, metadata=metadata)
        self.vertices[vertex_id] = new_vertex
        self.label_vertices_mapping[label_id].append(new_vertex)
        return new_vertex

    def get_vertices_by_metadata(self, **kwargs) -> dict[int, Vertex]:
        """
        Get all vertices with the given metadata attributes

        :param kwargs: The attribute key, value pair the vertex metadata must satisfy
        :return: The dict of vertices matching the provided metadata (id: Vertex)
        """
        result = {}
        for v_id, vertex in self.vertices.items():
            accept = True
            for key, value in kwargs.items():
                v_value = vertex.metadata.get(key, None)
                if value != v_value:
                    accept = False
                    break
            if accept:
                result[v_id] = vertex
        return result

    def get_edges_by_metadata(self, **kwargs) -> list[Edge]:
        """
        Get all edges with the given metadata attributes

        :param kwargs: The attribute key, value pair the edge metadata must satisfy
        :return: The dict of edges matching the provided metadata (id: Vertex)
        """
        result = []
        for edge in self.edges:
            accept = True
            for key, value in kwargs.items():
                e_value = edge.metadata.get(key, None)
                if value != e_value:
                    accept = False
                    break
            if accept:
                result.append(edge)

        return result

    def get_edges_by_vertices(self, vertices: Iterable[Vertex]) -> tuple[set[Edge], bool]:
        vertices_ids = {v.vertex_id for v in vertices}
        return self.get_edges_by_vertices_ids(vertices_ids)

    def get_edges_by_vertices_ids(self, vertices: Iterable[int]) -> tuple[set[Edge], bool]:
        result = set()
        vertices = set(vertices)
        visited_vertices = set()
        for v_id in vertices:
            for v_e_id, edge in self.edge_mapping[v_id].items():
                if v_e_id in vertices:
                    visited_vertices.add(v_id)
                    visited_vertices.add(v_e_id)
                    result.add(edge)
        return result, len(vertices) == len(visited_vertices)

    def get_vertex_by_id(self, vertex_id: int) -> Vertex | None:
        """
        Get the vertex with the given vertex_id. If the vertex does not exist, the function will return None.

        :param vertex_id: The label of the requested vertex
        :return: The vertex with the given vertex_id or None, if the vertex does not exist.
        """
        return self.vertices.get(vertex_id, None)

    def copy_into(
        self, graph_collection: "GraphCollection", additional_metadata: dict[str, Any] | None = None
    ) -> "Graph":
        """
        Copy this graph into a graph_collection

        :param graph_collection: The other graph collection
        :return: The new graph on the other graph_collection
        """
        metadata = self.metadata.copy()
        if additional_metadata is not None:
            metadata.update(additional_metadata)
        new_graph = graph_collection.new_graph(metadata=metadata)
        for v_id, v in self.vertices.items():
            new_graph.get_vertex(v.label, force_create=True, vertex_id=v_id, metadata=v.metadata.copy())
        for e in self.edges:
            new_graph.add_edge(
                new_graph.get_vertex_by_id(e.from_vertex.vertex_id),
                new_graph.get_vertex_by_id(e.to_vertex.vertex_id),
                edge_metadata=e.metadata.copy(),
            )
        return new_graph

    def as_str_rep(self, variant="gspan") -> str:
        """
        Returns a string of the graph with all vertices and edges.
        (No metadata are saved)

         - t # graph_id
         - v vertex_id vertex_label_id (vertex_type_id)
         - v vertex_id vertex_label_id (vertex_type_id)
         - ...
         - e vertex_source_id vertex_sink_id edge_type
         - e vertex_source_id vertex_sink_id edge_type
         - ...

        :return: The string representation of the graph
        """
        lines = [f"t # {self.graph_id}"]
        if variant == "subdue":
            lines = ["XP"]
        lines.extend(v.as_str_rep(variant) for _, v in self.vertices.items())
        lines.extend(e.as_str_rep(variant) for e in self.edges)
        return "\n".join(lines)

    @property
    def as_edge_rep(self) -> dict:
        """
        Returns a dict of the graph with all edges.
        The keys are tuples of strings (the two connected vertices) The vertices are defined by their label (string).
        The value is the type of the edge. (No metadata are saved)

        e.g.

        {('A','B'):2,('B','C'):1}

        :return: The edge representation of a graph
        """
        return {e[0]: e[1] for e in (x.as_edge_rep for x in self.edges)}

    @property
    def as_edge_tuple(self) -> list[tuple]:
        """
        Returns a dict of the graph with all edges.
        The keys are tuples of strings (the two connected vertices) The vertices are defined by their label (string).
        The value is the type of the edge.
        e.g.

        {('A','B'):2,('B','C'):1}

        :return: The edge representation of a graph
        """
        return [e[0] for e in (x.as_edge_rep for x in self.edges)]

    @property
    def as_networkx_digraph(self) -> nx.DiGraph:
        """
        Converts the graph into a python networkx graph object
        :return: DiGraph object of this graph
        """
        dg = nx.DiGraph()
        for v in self.vertices.values():
            dg.add_node(v.vertex_id, label=v.label, label_id=v.label_id, vertex_type_id=v.vertex_type_id, **v.metadata)
        dg.add_edges_from((e.from_vertex.vertex_id, e.to_vertex.vertex_id, e.metadata) for e in self.edges)
        dg.graph_id = self.graph_id
        return dg

    @property
    def as_graph_map(self) -> GraphMap:
        """
        Converts the graph into a graph map. (No metadata are saved)
        :return: A Graph Map
        """
        vertices = {str(vertex_id): MapVertex(vertex.label, vertex_id) for vertex_id, vertex in self.vertices.items()}
        edges = {
            f"{edge.from_vertex.vertex_id}:{edge.to_vertex.vertex_id}": MapEdge(
                vertices[str(edge.from_vertex.vertex_id)], vertices[str(edge.to_vertex.vertex_id)]
            )
            for edge in self.edges
        }
        return GraphMap(vertices, edges, str(self.graph_id))

    def to_dot_digraph(self, path: str | None = None, rankdir: str = "TB") -> Digraph:
        """
        Create a digraph object of the graph.

        :param path: if filled with a path to a file name (.png/.svg) the digraph is saved to the given path
        :param rankdir: the rank direction of the graph, "TB" or "LR"
        :return: a digraph
        """
        with tempfile.NamedTemporaryFile(suffix=".gv", delete=False) as tmp_file:
            tmp_file_name = tmp_file.name
        viz = Digraph(
            "",
            filename=tmp_file_name,
            engine="dot",
            graph_attr={"bgcolor": "transparent", "rankdir": rankdir},
        )
        viz.format = "svg"
        if path:
            viz.format = path.split(".")[-1]
        # draw nodes
        for v in self.vertices.values():
            bg = v.metadata.get("fillcolor", None)
            if bg is None:
                if v.vertex_type == "activity" or v.vertex_type == "":
                    bg = "#800000"
                else:
                    bg = "#000088"

            fc = v.metadata.get("fontcolor") or get_font_color(background_color=bg)
            shape = v.metadata.get("shape", None)
            if shape is None:
                if v.vertex_type == "activity" or v.vertex_type == "":
                    shape = "box"
                else:
                    shape = "oval"

            rank = v.metadata.get("rank")
            viz.node(
                str(v.vertex_id),
                v.label,
                shape=shape,
                style="filled",
                fontcolor=fc,
                fillcolor=bg,
                rank=rank,
                fontsize=str(12),
            )

        # draw edges
        for edge in self.edges:
            label = edge.metadata.get("label", "")
            dir_value = "none"
            penwidth = edge.metadata.get("penwidth", "0.8")
            style = "dashed"
            if (
                not edge.metadata.get("undirected", True)
                or (edge.from_vertex.vertex_type == "activity" or edge.from_vertex.vertex_type == "")
                and (edge.to_vertex.vertex_type == "activity" or edge.to_vertex.vertex_type == "")
            ):
                dir_value = "forward"
                penwidth = edge.metadata.get("penwidth", "1.2")
                style = ""
            viz.edge(
                str(edge.from_vertex.vertex_id),
                str(edge.to_vertex.vertex_id),
                label=label,
                penwidth=penwidth,
                fontsize=str(12),
                dir=dir_value,
                style=style,
            )
        viz.attr(overlap="false")
        viz.render(tmp_file.name, outfile=path, cleanup=True)
        return viz

    def _repr_svg_(self):
        dot = self.to_dot_digraph()
        data = dot.pipe(format="svg")
        svg_str = data.decode("utf-8")
        return svg_str

    @property
    def as_adjacency_matrix(self) -> pd.DataFrame:
        """
        Create an adjacency matrix of this graph as Pandas DataFrame.

        :return: the adjacency matrix of the graph as Pandas DataFrame
        """
        v_l = [v_id for v_id, v in sorted(self.vertices.items())]
        am = pd.DataFrame(np.zeros(shape=(len(v_l), len(v_l)), dtype=int), columns=v_l, index=v_l)
        for e in self.edges:
            am.at[e.from_vertex.vertex_id, e.to_vertex.vertex_id] += 1
        return am

    def unlink(self) -> "Graph":
        self.graph_collection = None
        return self

    def is_subgraph_of(self, other: "Graph") -> bool:
        """
        Check if the current graph is a subgraph of the other graph.
        A graph is a subgraph, only if all its vertices are in the other graph
        and all its edges are also in the other graph.

        :param other: The other graph (the super graph)
        :return: True if the current graph is a subgraph of the other graph.
        """
        # TODO: DOES NOT WORK!
        return all(v_s in other.vertices for v_s in self.vertices) and all(e_s in other.edges for e_s in self.edges)

    def __str__(self):
        return f"Graph ({self.graph_id}) with {len(self.vertices)} vertices and {len(self.edges)} edges"

    def __repr__(self):
        return str(self)

    def __eq__(self, other):
        """
        A graph is equal to another graph if the other graph hase the same edges and vertices.
        The metadata are not compared.

        :param other: The other graph
        :return: True if the graphs are equal
        """
        if not isinstance(other, Graph):
            return False
        elif self is other:
            return True
        else:
            return len(self.vertices) == len(other.vertices) and set(self.edges) == set(other.edges)

    def __len__(self):
        """
        The len of a graph is equal to len(graph.vertices)

        :return: The len of the graph
        """
        return len(self.vertices)

    def __hash__(self):
        return hash(self.graph_id)