algorithms

Algorithms for temporal path calculation and graph metrics.

The functions and submodules in this module allow to compute time-respecting or causal paths in temporal graphs and to calculate (temporal) and higher-order graph metrics like centralities.

Example

# Import pathpyG and configure your torch device if you want to use GPU .
import pathpyG as pp
pp.config['torch']['device'] = 'cuda'

# Generate a toy example for a temporal graph.
g = pp.TemporalGraph.from_edge_list([
    ('b', 'c', 2),
    ('a', 'b', 1),
    ('c', 'd', 3),
    ('d', 'a', 4),
    ('b', 'd', 2),
    ('d', 'a', 6),
    ('a', 'b', 7)
])

# Extract DAG capturing causal interaction sequences in temporal graph.
e_i = pp.algorithms.lift_order_temporal(g, delta=1)
dag = pp.Graph.from_edge_index(e_i)
print(dag)

# Calculate shortest time-respecting pathas
dist, pred = pp.algorithms.temporal.temporal_shortest_paths(g, delta=1)

Graph

A graph object storing nodes, edges, and attributes.

An object than be be used to store directed or undirected graphs with node and edge attributes. Data on nodes and edges are stored in an underlying instance of torch_geometric.Data.

Source code in src/pathpyG/core/graph.py

class Graph:
    """
    A graph object storing nodes, edges, and attributes.

    An object than be be used to store directed or undirected graphs with node
    and edge attributes. Data on nodes and edges are stored in an underlying instance of
    [`torch_geometric.Data`](https://pytorch-geometric.readthedocs.io/en/latest/generated/torch_geometric.data.Data.html#torch_geometric.data.Data).
    """

    def __init__(self, data: Data, mapping: Optional[IndexMap] = None):
        """Generate graph instance from a pyG `Data` object.

        Generate a Graph instance from a `torch_geometric.Data` object that contains an EdgeIndex as well as
        optional node-, edge- or graph-level attributes. An optional mapping can be used to transparently map
        node indices to string identifiers.

        Args:
            data: A pyG Data object containing an EdgeIndex and additional attributes
            mapping: `IndexMap` object that maps node indices to string identifiers

        Example:
            ```py
            import pathpyG as pp
            from torch_geometric.data import Data
            from torch_geometric import EdgeIndex

            data = Data(edge_index=EdgeIndex([[1,1,2],[0,2,1]], sparse_size=(3,3)))
            g = pp.Graph(data)

            g = pp.Graph(data, mapping=pp.IndexMap(['a', 'b', 'c']))
            ```
        """
        if mapping is None:
            self.mapping = IndexMap()
        else:
            self.mapping = mapping

        # set num_nodes property
        if "num_nodes" not in data:
            data.num_nodes = data.edge_index.max().item() + 1

        # turn edge index tensor into EdgeIndex object
        if not isinstance(data.edge_index, EdgeIndex):
            data.edge_index = EdgeIndex(data=data.edge_index, sparse_size=(data.num_nodes, data.num_nodes))

        if (
            data.edge_index.get_sparse_size(dim=0) != data.num_nodes
            or data.edge_index.get_sparse_size(dim=1) != data.num_nodes
        ):
            raise Exception("sparse size of EdgeIndex should match number of nodes!")

        # sort EdgeIndex and validate
        data.edge_index = data.edge_index.sort_by("row").values
        data.edge_index.validate()

        self.data = data

        # create mapping between edge tuples and edge indices
        self.edge_to_index = {
            (e[0].item(), e[1].item()): i for i, e in enumerate([e for e in self.data.edge_index.t()])
        }

        ((self.row_ptr, self.col), _) = self.data.edge_index.get_csr()
        ((self.col_ptr, self.row), _) = self.data.edge_index.get_csc()

        # create node_sequence mapping for higher-order graphs
        if "node_sequence" not in self.data:
            self.data.node_sequence = torch.arange(data.num_nodes).reshape(-1, 1)

    @staticmethod
    def from_edge_index(edge_index: torch.Tensor, mapping: Optional[IndexMap] = None, num_nodes: int = None) -> Graph:
        """Construct a graph from a torch Tensor containing an edge index. An optional mapping can
        be used to transparently map node indices to string identifiers.

        Args:
            edge_index:  torch.Tensor or torch_geometric.EdgeIndex object containing an edge_index
            mapping: `IndexMap` object that maps node indices to string identifiers
            num_nodes: optional number of nodes (default: None). If None, the number of nodes will be
                inferred based on the maximum node index in the edge index, i.e. there will be no isolated nodes.

        Examples:
            You can create a graph from an edge index tensor as follows:

            >>> import torch
            >>> import pathpyG as pp
            >>> g = pp.Graph.from_edge_index(torch.LongTensor([[1, 1, 2], [0, 2, 1]]))
            >>> print(g)
            Directed graph with 3 nodes and 3 edges ...

            You can also include a mapping of node IDs:

            >>> g = pp.Graph.from_edge_index(torch.LongTensor([[1, 1, 2], [0, 2, 1]]),
            >>>                              mapping=pp.IndexMap(['a', 'b', 'c']))
            >>> print(g.mapping)
            a -> 0
            b -> 1
            c -> 2
        """

        if not num_nodes:
            d = Data(edge_index=edge_index)
        else:
            d = Data(edge_index=edge_index, num_nodes=num_nodes)
        return Graph(d, mapping=mapping)

    @staticmethod
    def from_edge_list(
        edge_list: Iterable[Tuple[str, str]],
        is_undirected: bool = False,
        mapping: Optional[IndexMap] = None,
        num_nodes: Optional[int] = None,
    ) -> Graph:
        """Generate a Graph based on an edge list.

        Edges can be given as string or integer tuples. If strings are used and no mapping is given,
        a mapping of node IDs to indices will be automatically created based on a lexicographic ordering of
        node IDs.

        Args:
            edge_list: Iterable of edges represented as tuples
            is_undirected: Whether the edge list contains all bidorectional edges
            mapping: optional mapping of string IDs to node indices
            num_nodes: optional number of nodes (useful in case not all nodes have incident edges)

        Examples:
            >>> import pathpyG as pp
            >>> l = [('a', 'b'), ('a', 'c'), ('b', 'c')]
            >>> g = pp.Graph.from_edge_list(l)
            >>> print(list(g.edges))
            [('a', 'b'), ('a', 'c'), ('b', 'c')]
        """

        # handle empty graph
        if len(edge_list) == 0:
            return Graph(Data(edge_index=torch.tensor([[], []], dtype=torch.int32), num_nodes=0), mapping=IndexMap())

        if mapping is None:
            edge_array = np.array(edge_list)
            node_ids = np.unique(edge_array)
            if np.issubdtype(node_ids.dtype, str) and np.char.isnumeric(node_ids).all():
                node_ids = np.sort(node_ids.astype(int)).astype(str)
            mapping = IndexMap(node_ids)

        if num_nodes is None:
            num_nodes = mapping.num_ids()

        edge_index = EdgeIndex(
            mapping.to_idxs(edge_list).T.contiguous(),
            sparse_size=(num_nodes, num_nodes),
            is_undirected=is_undirected,
        )
        return Graph(Data(edge_index=edge_index, num_nodes=num_nodes), mapping=mapping)

    def to_undirected(self) -> Graph:
        """
        Returns an undirected version of a directed graph.

        This method transforms the current graph instance into an undirected graph by
        adding all directed edges in opposite direction. It applies [`ToUndirected`](https://pytorch-geometric.readthedocs.io/en/latest/generated/torch_geometric.transforms.ToUndirected.html#torch_geometric.transforms.ToUndirected)
        transform to the underlying [`torch_geometric.Data`](https://pytorch-geometric.readthedocs.io/en/latest/generated/torch_geometric.data.Data.html#torch_geometric.data.Data) object, which automatically
        duplicates edge attributes for newly created directed edges.

        Examples:
            >>> import pathpyG as pp
            >>> g = pp.Graph.from_edge_list([('a', 'b'), ('b', 'c'), ('c', 'a')])
            >>> g_u = g.to_undirected()
            >>> print(g_u)
            Undirected graph with 3 nodes and 6 (directed) edges
        """
        tf = ToUndirected()
        d = tf(self.data)
        # unfortunately, the application of a transform creates a new edge_index of type tensor
        # so we have to recreate the EdgeIndex tensor and sort it again

        e = EdgeIndex(data=d.edge_index, sparse_size=(self.data.num_nodes, self.data.num_nodes), is_undirected=True)
        d.edge_index = e
        d.num_nodes = self.data.num_nodes
        return Graph(d, self.mapping)

    def to_weighted_graph(self) -> Graph:
        """Coalesces multi-edges to single-edges with an additional weight attribute

        If the graph contains multiple edges between the same nodes, this method will coalesce
        them into a single edge with an additional weight attribute called `edge_weight` that
        contains the number of coalesced edges. The method returns a new graph instance with
        the coalesced edges.

        Returns:
            Graph: Graph with coalesced edges
        """
        i, w = torch_geometric.utils.coalesce(
            self.data.edge_index.as_tensor(), torch.ones(self.m, device=self.data.edge_index.device)
        )
        return Graph(Data(edge_index=i, edge_weight=w, num_nodes=self.data.num_nodes), mapping=self.mapping)

    def node_attrs(self) -> List[str]:
        """
        Return a list of node attributes.

        This method returns a list containing the names of all node-level attributes,
        ignoring the special `node_sequence` attribute.

        Returns:
            list: list of node attributes
        """
        attrs = []
        for k in self.data.keys():
            if k != "node_sequence" and k.startswith("node_"):
                attrs.append(k)
        return attrs

    def edge_attrs(self) -> List[str]:
        """
        Return a list of edge attributes.

        This method returns a list containing the names of all edge-level attributes,
        ignoring the special `edge_index` attribute.

        Returns:
            list: list of edge attributes
        """
        attrs = []
        for k in self.data.keys():
            if k != "edge_index" and k.startswith("edge_"):
                attrs.append(k)
        return attrs

    @property
    def nodes(self) -> list:
        """
        Return indices or IDs of all nodes in the graph.

        This method returns a list object that contains all nodes.
        If an IndexMap is used, nodes are returned as string IDs.
        If no IndexMap is used, nodes are returned as integer indices.

        Returns:
            list: list of all nodes using IDs or indices (if no mapping is used)
        """
        node_list = self.mapping.to_ids(np.arange(self.n)).tolist()
        if self.order > 1:
            return list(map(tuple, node_list))
        return node_list

    @property
    def edges(self) -> list:
        """Return all edges in the graph.

        This method returns a list object that contains all edges, where each
        edge is a tuple of two elements. If an IndexMap is used to map node
        indices to string IDs, edges are returned as tuples of string IDs.
        If no mapping is used, edges are returned as tuples of integer indices.

        Returns:
            list: list object yielding all edges using IDs or indices (if no mapping is used)
        """
        edge_list = self.mapping.to_ids(self.data.edge_index.t()).tolist()
        if self.order > 1:
            return [tuple(map(tuple, x)) for x in edge_list]
        return list(map(tuple, edge_list))

    def get_successors(self, row_idx: int) -> torch.Tensor:
        """Return a tensor containing the indices of all successor nodes for a given node identified by an index.

        Args:
            row_idx:   Index of node for which predecessors shall be returned.

        Returns:
            tensor: tensor containing indices of all successor nodes of the node indexed by `row_idx`
        """

        if row_idx + 1 < self.row_ptr.size(0):
            row_start = self.row_ptr[row_idx]
            row_end = self.row_ptr[row_idx + 1]
            return self.col[row_start:row_end]
        else:
            return torch.tensor([], device=self.data.edge_index.device)

    def get_predecessors(self, col_idx: int) -> torch.Tensor:
        """Return a tensor containing the indices of all predecessor nodes for a given node identified by an index.

        Args:
            col_idx:   Index of node for which predecessors shall be returned.

        Returns:
            tensor: tensor containing indices of all predecessor nodes of the node indexed by `col_idx`
        """
        if col_idx + 1 < self.col_ptr.size(0):
            col_start = self.col_ptr[col_idx]
            col_end = self.col_ptr[col_idx + 1]
            return self.row[col_start:col_end]
        else:
            return torch.tensor([], device=self.data.edge_index.device)

    def successors(self, node: Union[int, str] | tuple) -> list:
        """Return all successors of a given node.

        This method returns a generator object that yields all successors of a
        given node. If an IndexMap is used, successors are returned
        as string IDs. If no mapping is used, successors are returned as indices.

        Args:
            node:   Index or string ID of node for which successors shall be returned.

        Returns:
            list: list with all successors of the node identified
                by `node` using ID or index (if no mapping is used)
        """

        node_list = self.mapping.to_ids(self.get_successors(self.mapping.to_idx(node))).tolist()  # type: ignore

        if self.order > 1:
            return list(map(tuple, node_list))
        return node_list

    def predecessors(self, node: Union[str, int] | tuple) -> list:
        """Return the predecessors of a given node.

        This method returns a generator object that yields all predecessors of a
        given node. If a `node_id` mapping is used, predecessors will be returned
        as string IDs. If no mapping is used, predecessors are returned as indices.

        Args:
            node:   Index or string ID of node for which predecessors shall be returned.

        Returns:
            list: list with all predecessors of the node identified
                by `node` using ID or index (if no mapping is used)
        """
        node_list = self.mapping.to_ids(self.get_predecessors(self.mapping.to_idx(node))).tolist()  # type: ignore

        if self.order > 1:
            return list(map(tuple, node_list))
        return node_list

    def is_edge(self, v: Union[str, int], w: Union[str, int]) -> bool:
        """Return whether edge $(v,w)$ exists in the graph.

        If an index to ID mapping is used, nodes are assumed to be string IDs. If no
        mapping is used, nodes are assumed to be integer indices.

        Args:
            v: source node of edge as integer index or string ID
            w: target node of edge as integer index or string ID

        Returns:
            bool: True if edge exists, False otherwise
        """
        row = self.mapping.to_idx(v)
        row_start = self.row_ptr[row]
        row_end = self.row_ptr[row + 1]

        return self.mapping.to_idx(w) in self.col[row_start:row_end]

    def sparse_adj_matrix(self, edge_attr: Any = None) -> Any:
        """Return sparse adjacency matrix representation of (weighted) graph.

        Args:
            edge_attr: the edge attribute that shall be used as edge weight

        Returns:
            scipy.sparse.coo_matrix: sparse adjacency matrix representation of graph
        """
        if edge_attr is None:
            return torch_geometric.utils.to_scipy_sparse_matrix(self.data.edge_index.as_tensor(), num_nodes=self.n)
        else:
            return torch_geometric.utils.to_scipy_sparse_matrix(
                self.data.edge_index.as_tensor(), edge_attr=self.data[edge_attr], num_nodes=self.n
            )

    @property
    def in_degrees(self) -> Dict[str, float]:
        """Return in-degrees of nodes in directed network.

        Returns:
            dict: dictionary containing in-degrees of nodes
        """
        return self.degrees(mode="in")

    @property
    def out_degrees(self) -> Dict[str, float]:
        """Return out-degrees of nodes in directed network.

        Returns:
            dict: dictionary containing out-degrees of nodes
        """
        return self.degrees(mode="out")

    def degrees(self, mode: str = "in") -> Dict[str, float]:
        """
        Return degrees of nodes.

        Args:
            mode: `in` or `out` to calculate the in- or out-degree for
                directed networks.

        Returns:
            dict: dictionary containing degrees of nodes
        """
        if mode == "in":
            d = torch_geometric.utils.degree(self.data.edge_index[1], num_nodes=self.n, dtype=torch.int)
        else:
            d = torch_geometric.utils.degree(self.data.edge_index[0], num_nodes=self.n, dtype=torch.int)
        return {self.mapping.to_id(i): d[i].item() for i in range(self.n)}

    def weighted_outdegrees(self) -> torch.Tensor:
        """
        Compute the weighted outdegrees of each node in the graph.

        Args:
            graph (Graph): pathpy graph object.

        Returns:
            tensor: Weighted outdegrees of nodes.
        """
        weighted_outdegree = scatter(
            self.data.edge_weight, self.data.edge_index[0], dim=0, dim_size=self.data.num_nodes, reduce="sum"
        )
        return weighted_outdegree

    def transition_probabilities(self) -> torch.Tensor:
        """
        Compute transition probabilities based on weighted outdegrees.

        Returns:
            tensor: Transition probabilities.
        """
        weighted_outdegree = self.weighted_outdegrees()
        source_ids = self.data.edge_index[0]
        return self.data.edge_weight / weighted_outdegree[source_ids]

    def laplacian(self, normalization: Any = None, edge_attr: Any = None) -> Any:
        """Return Laplacian matrix for a given graph.

        This wrapper method will use [`torch_geometric.utils.laplacian`](https://pytorch-geometric.readthedocs.io/en/latest/modules/utils.html#torch_geometric.utils.laplacian)
        to return a Laplcian matrix representation of a given graph.

        Args:
            normalization: normalization parameter passed to pyG `get_laplacian`
                function
            edge_attr: optinal name of numerical edge attribute that shall
                be passed to pyG `get_laplacian` function as edge weight

        Returns:
            scipy.sparse.coo_matrix: Laplacian matrix representation of graph
        """
        if edge_attr is None:
            index, weight = torch_geometric.utils.get_laplacian(
                self.data.edge_index.as_tensor(), normalization=normalization
            )
            return torch_geometric.utils.to_scipy_sparse_matrix(index, weight)
        else:
            index, weight = torch_geometric.utils.get_laplacian(
                self.data.edge_index.as_tensor(),
                normalization=normalization,
                edge_weight=self.data[edge_attr],
            )
            return torch_geometric.utils.to_scipy_sparse_matrix(index, weight)

    def __getitem__(self, key: Union[tuple, str]) -> Any:
        """Return node, edge, or graph attribute.

        Args:
            key: name of attribute to be returned
        """
        if not isinstance(key, tuple):
            if key in self.data.keys():
                return self.data[key]
            else:
                raise KeyError(key + " is not a graph attribute")
        elif key[0] in self.node_attrs():
            return self.data[key[0]][self.mapping.to_idx(key[1])]
        elif key[0] in self.edge_attrs():
            return self.data[key[0]][self.edge_to_index[self.mapping.to_idx(key[1]), self.mapping.to_idx(key[2])]]
        else:
            raise KeyError(key[0] + " is not a node or edge attribute")

    def __setitem__(self, key: str, val: torch.Tensor) -> None:
        """Store node, edge, or graph attribute.

        Args:
            key: name of attribute to be stored
            val: value of attribute
        """
        if not isinstance(key, tuple):
            if key.startswith("node_"):
                if val.size(0) != self.n:
                    raise ValueError("Attribute must have same length as number of nodes")
                self.data[key] = val
            elif key.startswith("edge_"):
                if val.size(0) != self.m:
                    raise ValueError("Attribute must have same length as number of edges")
                self.data[key] = val
            else:
                self.data[key] = val
        elif key[0].startswith("node_"):  # type: ignore
            if key[0] not in self.data.keys():
                raise KeyError(
                    "Attribute does not yet exist. Setting the value of a specific node attribute"
                    + "requires that the attribute already exists."
                )
            self.data[key[0]][self.mapping.to_idx(key[1])] = val
        elif key[0].startswith("edge_"):  # type: ignore
            if key[0] not in self.data.keys():
                raise KeyError(
                    "Attribute does not yet exist. Setting the value of a specific node attribute"
                    + "requires that the attribute already exists."
                )
            self.data[key[0]][self.edge_to_index[self.mapping.to_idx(key[1]), self.mapping.to_idx(key[2])]] = val
        else:
            raise KeyError("node and edge specific attributes should be prefixed with 'node_' or 'edge_'")

    @property
    def n(self) -> int:
        """
        Return number of nodes.

        Returns:
            int: number of nodes in the graph
        """
        return self.data.num_nodes  # type: ignore

    @property
    def m(self) -> int:
        """
        Return number of edges.

        Returns the number of edges in the graph. For an undirected graph, the number of directed edges is returned.

        Returns:
            int: number of edges in the graph
        """
        return self.data.num_edges  # type: ignore

    @property
    def order(self) -> int:
        """
        Return order of graph.

        Returns:
            int: order of the (De Bruijn) graph
        """
        return self.data.node_sequence.size(1)  # type: ignore

    def is_directed(self) -> bool:
        """Return whether graph is directed.

        Returns:
            bool: True if graph is directed, False otherwise
        """
        return not self.data.edge_index.is_undirected

    def is_undirected(self) -> bool:
        """Return whether graph is undirected.

        Returns:
            bool: True if graph is undirected, False otherwise
        """
        return self.data.edge_index.is_undirected

    def has_self_loops(self) -> bool:
        """Return whether graph contains self-loops.

        Returns:
            bool: True if graph contains self-loops, False otherwise
        """
        return self.data.has_self_loops()

    def __add__(self, other: Graph) -> Graph:
        """Combine Graph object with other Graph object.

        The semantics of this operation depends on the optional IndexMap
        of both graphs. If no IndexMap is included, the two underlying data objects
        are concatenated, thus merging edges from both graphs while leaving node indices
        unchanged. If both graphs include IndexMaps that assign node IDs to indices,
        indiced will be adjusted, creating a new mapping for the union of node Ids in both graphs.

        Node IDs of graphs to be combined can be disjoint, partly overlapping or non-overlapping.

        Examples:
            Adding two graphs without node IDs:

            >>> g1 = pp.Graph.from_edge_index(torch.Tensor([[0,1,1],[1,2,3]]))
            >>> g1 = pp.Graph.from_edge_index(torch.Tensor([[0,2,3],[3,2,1]]))
            >>> print(g1 + g2)
            Graph with 3 nodes and 6 edges

            Adding two graphs with identical node IDs:

            >>> g1 = pp.Graph.from_edge_list([('a', 'b'), ('b', 'c')])
            >>> g2 = pp.Graph.from_edge_list([('a', 'c'), ('c', 'b')])
            >>> print(g1 + g2)
            Graph with 3 nodes and 4 edges

            Adding two graphs with non-overlapping node IDs:

            >>> g1 = pp.Graph.from_edge_list([('a', 'b'), ('b', 'c')])
            >>> g2 = pp.Graph.from_edge_list([('c', 'd'), ('d', 'e')])
            >>> print(g1 + g2)
            Graph with 6 nodes and 4 edges

            Adding two graphs with partly overlapping node IDs:

            >>> g1 = pp.Graph.from_edge_list([('a', 'b'), ('b', 'c')])
            >>> g2 = pp.Graph.from_edge_list([('b', 'd'), ('d', 'e')])
            >>> print(g1 + g2)
            Graph with 5 nodes and 4 edges
        """

        if self.order > 1:
            raise NotImplementedError("Add operator can only be applied to order 1 graphs")

        d1 = self.data.clone()
        m1 = self.mapping

        d2 = other.data.clone()
        m2 = other.mapping

        # compute overlap and additional nodes in g2 over g1
        overlap = set(m2.node_ids).intersection(m1.node_ids)
        additional_nodes = set(m2.node_ids).difference(m1.node_ids)

        d2_idx_translation = {}
        node_ids = [""] * (self.n + len(additional_nodes))
        # keep mappings of nodes in g1
        for v in m1.node_ids:
            node_ids[m1.to_idx(v)] = v
        for v in m2.node_ids:
            d2_idx_translation[m2.to_idx(v)] = m2.to_idx(v)
        # for overlapping node IDs we must correct node indices in m2
        for v in overlap:
            d2_idx_translation[m2.to_idx(v)] = m1.to_idx(v)
        # add mapping for nodes in g2 that are not in g1 and correct indices in g2
        for v in additional_nodes:
            new_idx = m2.to_idx(v) + self.n - len(overlap)
            node_ids[new_idx] = v
            d2_idx_translation[m2.to_idx(v)] = new_idx
        # apply index translation to d2
        # fast dictionary based mapping using torch
        palette, key = zip(*d2_idx_translation.items())
        key = torch.tensor(key)
        palette = torch.tensor(palette)

        index = torch.bucketize(d2.edge_index.ravel(), palette)
        d2.edge_index = key[index].reshape(d2.edge_index.shape)
        d = d1.concat(d2)
        mapping = IndexMap(node_ids)
        d.num_nodes = self.n + len(additional_nodes)
        d.edge_index = EdgeIndex(d.edge_index, sparse_size=(d.num_nodes, d.num_nodes))
        return Graph(d, mapping=mapping)

    def __str__(self) -> str:
        """Return a string representation of the graph."""

        attr = self.data.to_dict()
        attr_types = {}
        for k in attr:
            t = type(attr[k])
            if t == torch.Tensor:
                attr_types[k] = str(t) + " -> " + str(attr[k].size())
            else:
                attr_types[k] = str(t)

        from pprint import pformat

        if self.is_undirected():
            s = "Undirected graph with {0} nodes and {1} (directed) edges\n".format(self.n, self.m)
        else:
            s = "Directed graph with {0} nodes and {1} edges\n".format(self.n, self.m)

        attribute_info = {"Node Attributes": {}, "Edge Attributes": {}, "Graph Attributes": {}}
        for a in self.node_attrs():
            attribute_info["Node Attributes"][a] = attr_types[a]
        for a in self.edge_attrs():
            attribute_info["Edge Attributes"][a] = attr_types[a]
        for a in self.data.keys():
            if not self.data.is_node_attr(a) and not self.data.is_edge_attr(a):
                attribute_info["Graph Attributes"][a] = attr_types[a]
        s += pformat(attribute_info, indent=4, width=160)
        return s

edges property

Return all edges in the graph.

This method returns a list object that contains all edges, where each edge is a tuple of two elements. If an IndexMap is used to map node indices to string IDs, edges are returned as tuples of string IDs. If no mapping is used, edges are returned as tuples of integer indices.

Returns:

Name	Type	Description
list	list	list object yielding all edges using IDs or indices (if no mapping is used)

in_degrees property

Return in-degrees of nodes in directed network.

Returns:

Name	Type	Description
dict	typing.Dict[str, float]	dictionary containing in-degrees of nodes

m property

Return number of edges.

Returns the number of edges in the graph. For an undirected graph, the number of directed edges is returned.

Returns:

Name	Type	Description
int	int	number of edges in the graph

n property

Return number of nodes.

Returns:

Name	Type	Description
int	int	number of nodes in the graph

nodes property

Return indices or IDs of all nodes in the graph.

This method returns a list object that contains all nodes. If an IndexMap is used, nodes are returned as string IDs. If no IndexMap is used, nodes are returned as integer indices.

Returns:

Name	Type	Description
list	list	list of all nodes using IDs or indices (if no mapping is used)

order property

Return order of graph.

Returns:

Name	Type	Description
int	int	order of the (De Bruijn) graph

out_degrees property

Return out-degrees of nodes in directed network.

Returns:

Name	Type	Description
dict	typing.Dict[str, float]	dictionary containing out-degrees of nodes

__add__

Combine Graph object with other Graph object.

The semantics of this operation depends on the optional IndexMap of both graphs. If no IndexMap is included, the two underlying data objects are concatenated, thus merging edges from both graphs while leaving node indices unchanged. If both graphs include IndexMaps that assign node IDs to indices, indiced will be adjusted, creating a new mapping for the union of node Ids in both graphs.

Node IDs of graphs to be combined can be disjoint, partly overlapping or non-overlapping.

Examples:

Adding two graphs without node IDs:

>>> g1 = pp.Graph.from_edge_index(torch.Tensor([[0,1,1],[1,2,3]]))
>>> g1 = pp.Graph.from_edge_index(torch.Tensor([[0,2,3],[3,2,1]]))
>>> print(g1 + g2)
Graph with 3 nodes and 6 edges

Adding two graphs with identical node IDs:

>>> g1 = pp.Graph.from_edge_list([('a', 'b'), ('b', 'c')])
>>> g2 = pp.Graph.from_edge_list([('a', 'c'), ('c', 'b')])
>>> print(g1 + g2)
Graph with 3 nodes and 4 edges

Adding two graphs with non-overlapping node IDs:

>>> g1 = pp.Graph.from_edge_list([('a', 'b'), ('b', 'c')])
>>> g2 = pp.Graph.from_edge_list([('c', 'd'), ('d', 'e')])
>>> print(g1 + g2)
Graph with 6 nodes and 4 edges

Adding two graphs with partly overlapping node IDs:

>>> g1 = pp.Graph.from_edge_list([('a', 'b'), ('b', 'c')])
>>> g2 = pp.Graph.from_edge_list([('b', 'd'), ('d', 'e')])
>>> print(g1 + g2)
Graph with 5 nodes and 4 edges

Source code in src/pathpyG/core/graph.py

def __add__(self, other: Graph) -> Graph:
    """Combine Graph object with other Graph object.

    The semantics of this operation depends on the optional IndexMap
    of both graphs. If no IndexMap is included, the two underlying data objects
    are concatenated, thus merging edges from both graphs while leaving node indices
    unchanged. If both graphs include IndexMaps that assign node IDs to indices,
    indiced will be adjusted, creating a new mapping for the union of node Ids in both graphs.

    Node IDs of graphs to be combined can be disjoint, partly overlapping or non-overlapping.

    Examples:
        Adding two graphs without node IDs:

        >>> g1 = pp.Graph.from_edge_index(torch.Tensor([[0,1,1],[1,2,3]]))
        >>> g1 = pp.Graph.from_edge_index(torch.Tensor([[0,2,3],[3,2,1]]))
        >>> print(g1 + g2)
        Graph with 3 nodes and 6 edges

        Adding two graphs with identical node IDs:

        >>> g1 = pp.Graph.from_edge_list([('a', 'b'), ('b', 'c')])
        >>> g2 = pp.Graph.from_edge_list([('a', 'c'), ('c', 'b')])
        >>> print(g1 + g2)
        Graph with 3 nodes and 4 edges

        Adding two graphs with non-overlapping node IDs:

        >>> g1 = pp.Graph.from_edge_list([('a', 'b'), ('b', 'c')])
        >>> g2 = pp.Graph.from_edge_list([('c', 'd'), ('d', 'e')])
        >>> print(g1 + g2)
        Graph with 6 nodes and 4 edges

        Adding two graphs with partly overlapping node IDs:

        >>> g1 = pp.Graph.from_edge_list([('a', 'b'), ('b', 'c')])
        >>> g2 = pp.Graph.from_edge_list([('b', 'd'), ('d', 'e')])
        >>> print(g1 + g2)
        Graph with 5 nodes and 4 edges
    """

    if self.order > 1:
        raise NotImplementedError("Add operator can only be applied to order 1 graphs")

    d1 = self.data.clone()
    m1 = self.mapping

    d2 = other.data.clone()
    m2 = other.mapping

    # compute overlap and additional nodes in g2 over g1
    overlap = set(m2.node_ids).intersection(m1.node_ids)
    additional_nodes = set(m2.node_ids).difference(m1.node_ids)

    d2_idx_translation = {}
    node_ids = [""] * (self.n + len(additional_nodes))
    # keep mappings of nodes in g1
    for v in m1.node_ids:
        node_ids[m1.to_idx(v)] = v
    for v in m2.node_ids:
        d2_idx_translation[m2.to_idx(v)] = m2.to_idx(v)
    # for overlapping node IDs we must correct node indices in m2
    for v in overlap:
        d2_idx_translation[m2.to_idx(v)] = m1.to_idx(v)
    # add mapping for nodes in g2 that are not in g1 and correct indices in g2
    for v in additional_nodes:
        new_idx = m2.to_idx(v) + self.n - len(overlap)
        node_ids[new_idx] = v
        d2_idx_translation[m2.to_idx(v)] = new_idx
    # apply index translation to d2
    # fast dictionary based mapping using torch
    palette, key = zip(*d2_idx_translation.items())
    key = torch.tensor(key)
    palette = torch.tensor(palette)

    index = torch.bucketize(d2.edge_index.ravel(), palette)
    d2.edge_index = key[index].reshape(d2.edge_index.shape)
    d = d1.concat(d2)
    mapping = IndexMap(node_ids)
    d.num_nodes = self.n + len(additional_nodes)
    d.edge_index = EdgeIndex(d.edge_index, sparse_size=(d.num_nodes, d.num_nodes))
    return Graph(d, mapping=mapping)

__getitem__

Return node, edge, or graph attribute.

Parameters:

Name	Type	Description	Default
key	typing.Union[tuple, str]	name of attribute to be returned	required

Source code in src/pathpyG/core/graph.py

def __getitem__(self, key: Union[tuple, str]) -> Any:
    """Return node, edge, or graph attribute.

    Args:
        key: name of attribute to be returned
    """
    if not isinstance(key, tuple):
        if key in self.data.keys():
            return self.data[key]
        else:
            raise KeyError(key + " is not a graph attribute")
    elif key[0] in self.node_attrs():
        return self.data[key[0]][self.mapping.to_idx(key[1])]
    elif key[0] in self.edge_attrs():
        return self.data[key[0]][self.edge_to_index[self.mapping.to_idx(key[1]), self.mapping.to_idx(key[2])]]
    else:
        raise KeyError(key[0] + " is not a node or edge attribute")

__init__

Generate graph instance from a pyG Data object.

Generate a Graph instance from a torch_geometric.Data object that contains an EdgeIndex as well as optional node-, edge- or graph-level attributes. An optional mapping can be used to transparently map node indices to string identifiers.

Parameters:

Name	Type	Description	Default
data	torch_geometric.data.Data	A pyG Data object containing an EdgeIndex and additional attributes	required
mapping	typing.Optional[pathpyG.core.index_map.IndexMap]	IndexMap object that maps node indices to string identifiers	None

Example

import pathpyG as pp
from torch_geometric.data import Data
from torch_geometric import EdgeIndex

data = Data(edge_index=EdgeIndex([[1,1,2],[0,2,1]], sparse_size=(3,3)))
g = pp.Graph(data)

g = pp.Graph(data, mapping=pp.IndexMap(['a', 'b', 'c']))

Source code in src/pathpyG/core/graph.py

def __init__(self, data: Data, mapping: Optional[IndexMap] = None):
    """Generate graph instance from a pyG `Data` object.

    Generate a Graph instance from a `torch_geometric.Data` object that contains an EdgeIndex as well as
    optional node-, edge- or graph-level attributes. An optional mapping can be used to transparently map
    node indices to string identifiers.

    Args:
        data: A pyG Data object containing an EdgeIndex and additional attributes
        mapping: `IndexMap` object that maps node indices to string identifiers

    Example:
        ```py
        import pathpyG as pp
        from torch_geometric.data import Data
        from torch_geometric import EdgeIndex

        data = Data(edge_index=EdgeIndex([[1,1,2],[0,2,1]], sparse_size=(3,3)))
        g = pp.Graph(data)

        g = pp.Graph(data, mapping=pp.IndexMap(['a', 'b', 'c']))
        ```
    """
    if mapping is None:
        self.mapping = IndexMap()
    else:
        self.mapping = mapping

    # set num_nodes property
    if "num_nodes" not in data:
        data.num_nodes = data.edge_index.max().item() + 1

    # turn edge index tensor into EdgeIndex object
    if not isinstance(data.edge_index, EdgeIndex):
        data.edge_index = EdgeIndex(data=data.edge_index, sparse_size=(data.num_nodes, data.num_nodes))

    if (
        data.edge_index.get_sparse_size(dim=0) != data.num_nodes
        or data.edge_index.get_sparse_size(dim=1) != data.num_nodes
    ):
        raise Exception("sparse size of EdgeIndex should match number of nodes!")

    # sort EdgeIndex and validate
    data.edge_index = data.edge_index.sort_by("row").values
    data.edge_index.validate()

    self.data = data

    # create mapping between edge tuples and edge indices
    self.edge_to_index = {
        (e[0].item(), e[1].item()): i for i, e in enumerate([e for e in self.data.edge_index.t()])
    }

    ((self.row_ptr, self.col), _) = self.data.edge_index.get_csr()
    ((self.col_ptr, self.row), _) = self.data.edge_index.get_csc()

    # create node_sequence mapping for higher-order graphs
    if "node_sequence" not in self.data:
        self.data.node_sequence = torch.arange(data.num_nodes).reshape(-1, 1)

__setitem__

Store node, edge, or graph attribute.

Parameters:

Name	Type	Description	Default
key	str	name of attribute to be stored	required
val	torch.Tensor	value of attribute	required

Source code in src/pathpyG/core/graph.py

def __setitem__(self, key: str, val: torch.Tensor) -> None:
    """Store node, edge, or graph attribute.

    Args:
        key: name of attribute to be stored
        val: value of attribute
    """
    if not isinstance(key, tuple):
        if key.startswith("node_"):
            if val.size(0) != self.n:
                raise ValueError("Attribute must have same length as number of nodes")
            self.data[key] = val
        elif key.startswith("edge_"):
            if val.size(0) != self.m:
                raise ValueError("Attribute must have same length as number of edges")
            self.data[key] = val
        else:
            self.data[key] = val
    elif key[0].startswith("node_"):  # type: ignore
        if key[0] not in self.data.keys():
            raise KeyError(
                "Attribute does not yet exist. Setting the value of a specific node attribute"
                + "requires that the attribute already exists."
            )
        self.data[key[0]][self.mapping.to_idx(key[1])] = val
    elif key[0].startswith("edge_"):  # type: ignore
        if key[0] not in self.data.keys():
            raise KeyError(
                "Attribute does not yet exist. Setting the value of a specific node attribute"
                + "requires that the attribute already exists."
            )
        self.data[key[0]][self.edge_to_index[self.mapping.to_idx(key[1]), self.mapping.to_idx(key[2])]] = val
    else:
        raise KeyError("node and edge specific attributes should be prefixed with 'node_' or 'edge_'")

__str__

Return a string representation of the graph.

Source code in src/pathpyG/core/graph.py

def __str__(self) -> str:
    """Return a string representation of the graph."""

    attr = self.data.to_dict()
    attr_types = {}
    for k in attr:
        t = type(attr[k])
        if t == torch.Tensor:
            attr_types[k] = str(t) + " -> " + str(attr[k].size())
        else:
            attr_types[k] = str(t)

    from pprint import pformat

    if self.is_undirected():
        s = "Undirected graph with {0} nodes and {1} (directed) edges\n".format(self.n, self.m)
    else:
        s = "Directed graph with {0} nodes and {1} edges\n".format(self.n, self.m)

    attribute_info = {"Node Attributes": {}, "Edge Attributes": {}, "Graph Attributes": {}}
    for a in self.node_attrs():
        attribute_info["Node Attributes"][a] = attr_types[a]
    for a in self.edge_attrs():
        attribute_info["Edge Attributes"][a] = attr_types[a]
    for a in self.data.keys():
        if not self.data.is_node_attr(a) and not self.data.is_edge_attr(a):
            attribute_info["Graph Attributes"][a] = attr_types[a]
    s += pformat(attribute_info, indent=4, width=160)
    return s

degrees

Return degrees of nodes.

Parameters:

Name	Type	Description	Default
mode	str	in or out to calculate the in- or out-degree for directed networks.	'in'

Returns:

Name	Type	Description
dict	typing.Dict[str, float]	dictionary containing degrees of nodes

Source code in src/pathpyG/core/graph.py

def degrees(self, mode: str = "in") -> Dict[str, float]:
    """
    Return degrees of nodes.

    Args:
        mode: `in` or `out` to calculate the in- or out-degree for
            directed networks.

    Returns:
        dict: dictionary containing degrees of nodes
    """
    if mode == "in":
        d = torch_geometric.utils.degree(self.data.edge_index[1], num_nodes=self.n, dtype=torch.int)
    else:
        d = torch_geometric.utils.degree(self.data.edge_index[0], num_nodes=self.n, dtype=torch.int)
    return {self.mapping.to_id(i): d[i].item() for i in range(self.n)}

edge_attrs

Return a list of edge attributes.

This method returns a list containing the names of all edge-level attributes, ignoring the special edge_index attribute.

Returns:

Name	Type	Description
list	typing.List[str]	list of edge attributes

Source code in src/pathpyG/core/graph.py

def edge_attrs(self) -> List[str]:
    """
    Return a list of edge attributes.

    This method returns a list containing the names of all edge-level attributes,
    ignoring the special `edge_index` attribute.

    Returns:
        list: list of edge attributes
    """
    attrs = []
    for k in self.data.keys():
        if k != "edge_index" and k.startswith("edge_"):
            attrs.append(k)
    return attrs

from_edge_index staticmethod

Construct a graph from a torch Tensor containing an edge index. An optional mapping can be used to transparently map node indices to string identifiers.

Parameters:

Name	Type	Description	Default
edge_index	torch.Tensor	torch.Tensor or torch_geometric.EdgeIndex object containing an edge_index	required
mapping	typing.Optional[pathpyG.core.index_map.IndexMap]	IndexMap object that maps node indices to string identifiers	None
num_nodes	int	optional number of nodes (default: None). If None, the number of nodes will be inferred based on the maximum node index in the edge index, i.e. there will be no isolated nodes.	None

Examples:

You can create a graph from an edge index tensor as follows:

>>> import torch
>>> import pathpyG as pp
>>> g = pp.Graph.from_edge_index(torch.LongTensor([[1, 1, 2], [0, 2, 1]]))
>>> print(g)
Directed graph with 3 nodes and 3 edges ...

You can also include a mapping of node IDs:

>>> g = pp.Graph.from_edge_index(torch.LongTensor([[1, 1, 2], [0, 2, 1]]),
>>>                              mapping=pp.IndexMap(['a', 'b', 'c']))
>>> print(g.mapping)
a -> 0
b -> 1
c -> 2

Source code in src/pathpyG/core/graph.py

@staticmethod
def from_edge_index(edge_index: torch.Tensor, mapping: Optional[IndexMap] = None, num_nodes: int = None) -> Graph:
    """Construct a graph from a torch Tensor containing an edge index. An optional mapping can
    be used to transparently map node indices to string identifiers.

    Args:
        edge_index:  torch.Tensor or torch_geometric.EdgeIndex object containing an edge_index
        mapping: `IndexMap` object that maps node indices to string identifiers
        num_nodes: optional number of nodes (default: None). If None, the number of nodes will be
            inferred based on the maximum node index in the edge index, i.e. there will be no isolated nodes.

    Examples:
        You can create a graph from an edge index tensor as follows:

        >>> import torch
        >>> import pathpyG as pp
        >>> g = pp.Graph.from_edge_index(torch.LongTensor([[1, 1, 2], [0, 2, 1]]))
        >>> print(g)
        Directed graph with 3 nodes and 3 edges ...

        You can also include a mapping of node IDs:

        >>> g = pp.Graph.from_edge_index(torch.LongTensor([[1, 1, 2], [0, 2, 1]]),
        >>>                              mapping=pp.IndexMap(['a', 'b', 'c']))
        >>> print(g.mapping)
        a -> 0
        b -> 1
        c -> 2
    """

    if not num_nodes:
        d = Data(edge_index=edge_index)
    else:
        d = Data(edge_index=edge_index, num_nodes=num_nodes)
    return Graph(d, mapping=mapping)

from_edge_list staticmethod

Generate a Graph based on an edge list.

Edges can be given as string or integer tuples. If strings are used and no mapping is given, a mapping of node IDs to indices will be automatically created based on a lexicographic ordering of node IDs.

Parameters:

Name	Type	Description	Default
edge_list	typing.Iterable[typing.Tuple[str, str]]	Iterable of edges represented as tuples	required
is_undirected	bool	Whether the edge list contains all bidorectional edges	False
mapping	typing.Optional[pathpyG.core.index_map.IndexMap]	optional mapping of string IDs to node indices	None
num_nodes	typing.Optional[int]	optional number of nodes (useful in case not all nodes have incident edges)	None

Examples:

>>> import pathpyG as pp
>>> l = [('a', 'b'), ('a', 'c'), ('b', 'c')]
>>> g = pp.Graph.from_edge_list(l)
>>> print(list(g.edges))
[('a', 'b'), ('a', 'c'), ('b', 'c')]

Source code in src/pathpyG/core/graph.py

@staticmethod
def from_edge_list(
    edge_list: Iterable[Tuple[str, str]],
    is_undirected: bool = False,
    mapping: Optional[IndexMap] = None,
    num_nodes: Optional[int] = None,
) -> Graph:
    """Generate a Graph based on an edge list.

    Edges can be given as string or integer tuples. If strings are used and no mapping is given,
    a mapping of node IDs to indices will be automatically created based on a lexicographic ordering of
    node IDs.

    Args:
        edge_list: Iterable of edges represented as tuples
        is_undirected: Whether the edge list contains all bidorectional edges
        mapping: optional mapping of string IDs to node indices
        num_nodes: optional number of nodes (useful in case not all nodes have incident edges)

    Examples:
        >>> import pathpyG as pp
        >>> l = [('a', 'b'), ('a', 'c'), ('b', 'c')]
        >>> g = pp.Graph.from_edge_list(l)
        >>> print(list(g.edges))
        [('a', 'b'), ('a', 'c'), ('b', 'c')]
    """

    # handle empty graph
    if len(edge_list) == 0:
        return Graph(Data(edge_index=torch.tensor([[], []], dtype=torch.int32), num_nodes=0), mapping=IndexMap())

    if mapping is None:
        edge_array = np.array(edge_list)
        node_ids = np.unique(edge_array)
        if np.issubdtype(node_ids.dtype, str) and np.char.isnumeric(node_ids).all():
            node_ids = np.sort(node_ids.astype(int)).astype(str)
        mapping = IndexMap(node_ids)

    if num_nodes is None:
        num_nodes = mapping.num_ids()

    edge_index = EdgeIndex(
        mapping.to_idxs(edge_list).T.contiguous(),
        sparse_size=(num_nodes, num_nodes),
        is_undirected=is_undirected,
    )
    return Graph(Data(edge_index=edge_index, num_nodes=num_nodes), mapping=mapping)

get_predecessors

Return a tensor containing the indices of all predecessor nodes for a given node identified by an index.

Parameters:

Name	Type	Description	Default
col_idx	int	Index of node for which predecessors shall be returned.	required

Returns:

Name	Type	Description
tensor	torch.Tensor	tensor containing indices of all predecessor nodes of the node indexed by col_idx

Source code in src/pathpyG/core/graph.py

def get_predecessors(self, col_idx: int) -> torch.Tensor:
    """Return a tensor containing the indices of all predecessor nodes for a given node identified by an index.

    Args:
        col_idx:   Index of node for which predecessors shall be returned.

    Returns:
        tensor: tensor containing indices of all predecessor nodes of the node indexed by `col_idx`
    """
    if col_idx + 1 < self.col_ptr.size(0):
        col_start = self.col_ptr[col_idx]
        col_end = self.col_ptr[col_idx + 1]
        return self.row[col_start:col_end]
    else:
        return torch.tensor([], device=self.data.edge_index.device)

get_successors

Return a tensor containing the indices of all successor nodes for a given node identified by an index.

Parameters:

Name	Type	Description	Default
row_idx	int	Index of node for which predecessors shall be returned.	required

Returns:

Name	Type	Description
tensor	torch.Tensor	tensor containing indices of all successor nodes of the node indexed by row_idx

Source code in src/pathpyG/core/graph.py

def get_successors(self, row_idx: int) -> torch.Tensor:
    """Return a tensor containing the indices of all successor nodes for a given node identified by an index.

    Args:
        row_idx:   Index of node for which predecessors shall be returned.

    Returns:
        tensor: tensor containing indices of all successor nodes of the node indexed by `row_idx`
    """

    if row_idx + 1 < self.row_ptr.size(0):
        row_start = self.row_ptr[row_idx]
        row_end = self.row_ptr[row_idx + 1]
        return self.col[row_start:row_end]
    else:
        return torch.tensor([], device=self.data.edge_index.device)

has_self_loops

Return whether graph contains self-loops.

Returns:

Name	Type	Description
bool	bool	True if graph contains self-loops, False otherwise

Source code in src/pathpyG/core/graph.py

def has_self_loops(self) -> bool:
    """Return whether graph contains self-loops.

    Returns:
        bool: True if graph contains self-loops, False otherwise
    """
    return self.data.has_self_loops()

is_directed

Return whether graph is directed.

Returns:

Name	Type	Description
bool	bool	True if graph is directed, False otherwise

Source code in src/pathpyG/core/graph.py

def is_directed(self) -> bool:
    """Return whether graph is directed.

    Returns:
        bool: True if graph is directed, False otherwise
    """
    return not self.data.edge_index.is_undirected

is_edge

Return whether edge \((v,w)\) exists in the graph.

If an index to ID mapping is used, nodes are assumed to be string IDs. If no mapping is used, nodes are assumed to be integer indices.

Parameters:

Name	Type	Description	Default
v	typing.Union[str, int]	source node of edge as integer index or string ID	required
w	typing.Union[str, int]	target node of edge as integer index or string ID	required

Returns:

Name	Type	Description
bool	bool	True if edge exists, False otherwise

Source code in src/pathpyG/core/graph.py

def is_edge(self, v: Union[str, int], w: Union[str, int]) -> bool:
    """Return whether edge $(v,w)$ exists in the graph.

    If an index to ID mapping is used, nodes are assumed to be string IDs. If no
    mapping is used, nodes are assumed to be integer indices.

    Args:
        v: source node of edge as integer index or string ID
        w: target node of edge as integer index or string ID

    Returns:
        bool: True if edge exists, False otherwise
    """
    row = self.mapping.to_idx(v)
    row_start = self.row_ptr[row]
    row_end = self.row_ptr[row + 1]

    return self.mapping.to_idx(w) in self.col[row_start:row_end]

is_undirected

Return whether graph is undirected.

Returns:

Name	Type	Description
bool	bool	True if graph is undirected, False otherwise

Source code in src/pathpyG/core/graph.py

def is_undirected(self) -> bool:
    """Return whether graph is undirected.

    Returns:
        bool: True if graph is undirected, False otherwise
    """
    return self.data.edge_index.is_undirected

laplacian

Return Laplacian matrix for a given graph.

This wrapper method will use torch_geometric.utils.laplacian to return a Laplcian matrix representation of a given graph.

Parameters:

Name	Type	Description	Default
normalization	typing.Any	normalization parameter passed to pyG get_laplacian function	None
edge_attr	typing.Any	optinal name of numerical edge attribute that shall be passed to pyG get_laplacian function as edge weight	None

Returns:

Type	Description
typing.Any	scipy.sparse.coo_matrix: Laplacian matrix representation of graph

Source code in src/pathpyG/core/graph.py

def laplacian(self, normalization: Any = None, edge_attr: Any = None) -> Any:
    """Return Laplacian matrix for a given graph.

    This wrapper method will use [`torch_geometric.utils.laplacian`](https://pytorch-geometric.readthedocs.io/en/latest/modules/utils.html#torch_geometric.utils.laplacian)
    to return a Laplcian matrix representation of a given graph.

    Args:
        normalization: normalization parameter passed to pyG `get_laplacian`
            function
        edge_attr: optinal name of numerical edge attribute that shall
            be passed to pyG `get_laplacian` function as edge weight

    Returns:
        scipy.sparse.coo_matrix: Laplacian matrix representation of graph
    """
    if edge_attr is None:
        index, weight = torch_geometric.utils.get_laplacian(
            self.data.edge_index.as_tensor(), normalization=normalization
        )
        return torch_geometric.utils.to_scipy_sparse_matrix(index, weight)
    else:
        index, weight = torch_geometric.utils.get_laplacian(
            self.data.edge_index.as_tensor(),
            normalization=normalization,
            edge_weight=self.data[edge_attr],
        )
        return torch_geometric.utils.to_scipy_sparse_matrix(index, weight)

node_attrs

Return a list of node attributes.

This method returns a list containing the names of all node-level attributes, ignoring the special node_sequence attribute.

Returns:

Name	Type	Description
list	typing.List[str]	list of node attributes

Source code in src/pathpyG/core/graph.py

def node_attrs(self) -> List[str]:
    """
    Return a list of node attributes.

    This method returns a list containing the names of all node-level attributes,
    ignoring the special `node_sequence` attribute.

    Returns:
        list: list of node attributes
    """
    attrs = []
    for k in self.data.keys():
        if k != "node_sequence" and k.startswith("node_"):
            attrs.append(k)
    return attrs

predecessors

Return the predecessors of a given node.

This method returns a generator object that yields all predecessors of a given node. If a node_id mapping is used, predecessors will be returned as string IDs. If no mapping is used, predecessors are returned as indices.

Parameters:

Name	Type	Description	Default
node	typing.Union[str, int] | tuple	Index or string ID of node for which predecessors shall be returned.	required

Returns:

Name	Type	Description
list	list	list with all predecessors of the node identified by node using ID or index (if no mapping is used)

Source code in src/pathpyG/core/graph.py

def predecessors(self, node: Union[str, int] | tuple) -> list:
    """Return the predecessors of a given node.

    This method returns a generator object that yields all predecessors of a
    given node. If a `node_id` mapping is used, predecessors will be returned
    as string IDs. If no mapping is used, predecessors are returned as indices.

    Args:
        node:   Index or string ID of node for which predecessors shall be returned.

    Returns:
        list: list with all predecessors of the node identified
            by `node` using ID or index (if no mapping is used)
    """
    node_list = self.mapping.to_ids(self.get_predecessors(self.mapping.to_idx(node))).tolist()  # type: ignore

    if self.order > 1:
        return list(map(tuple, node_list))
    return node_list

sparse_adj_matrix

Return sparse adjacency matrix representation of (weighted) graph.

Parameters:

Name	Type	Description	Default
edge_attr	typing.Any	the edge attribute that shall be used as edge weight	None

Returns:

Type	Description
typing.Any	scipy.sparse.coo_matrix: sparse adjacency matrix representation of graph

Source code in src/pathpyG/core/graph.py

def sparse_adj_matrix(self, edge_attr: Any = None) -> Any:
    """Return sparse adjacency matrix representation of (weighted) graph.

    Args:
        edge_attr: the edge attribute that shall be used as edge weight

    Returns:
        scipy.sparse.coo_matrix: sparse adjacency matrix representation of graph
    """
    if edge_attr is None:
        return torch_geometric.utils.to_scipy_sparse_matrix(self.data.edge_index.as_tensor(), num_nodes=self.n)
    else:
        return torch_geometric.utils.to_scipy_sparse_matrix(
            self.data.edge_index.as_tensor(), edge_attr=self.data[edge_attr], num_nodes=self.n
        )

successors

Return all successors of a given node.

This method returns a generator object that yields all successors of a given node. If an IndexMap is used, successors are returned as string IDs. If no mapping is used, successors are returned as indices.

Parameters:

Name	Type	Description	Default
node	typing.Union[int, str] | tuple	Index or string ID of node for which successors shall be returned.	required

Returns:

Name	Type	Description
list	list	list with all successors of the node identified by node using ID or index (if no mapping is used)

Source code in src/pathpyG/core/graph.py

def successors(self, node: Union[int, str] | tuple) -> list:
    """Return all successors of a given node.

    This method returns a generator object that yields all successors of a
    given node. If an IndexMap is used, successors are returned
    as string IDs. If no mapping is used, successors are returned as indices.

    Args:
        node:   Index or string ID of node for which successors shall be returned.

    Returns:
        list: list with all successors of the node identified
            by `node` using ID or index (if no mapping is used)
    """

    node_list = self.mapping.to_ids(self.get_successors(self.mapping.to_idx(node))).tolist()  # type: ignore

    if self.order > 1:
        return list(map(tuple, node_list))
    return node_list

to_undirected

Returns an undirected version of a directed graph.

This method transforms the current graph instance into an undirected graph by adding all directed edges in opposite direction. It applies ToUndirected transform to the underlying torch_geometric.Data object, which automatically duplicates edge attributes for newly created directed edges.

Examples:

>>> import pathpyG as pp
>>> g = pp.Graph.from_edge_list([('a', 'b'), ('b', 'c'), ('c', 'a')])
>>> g_u = g.to_undirected()
>>> print(g_u)
Undirected graph with 3 nodes and 6 (directed) edges

Source code in src/pathpyG/core/graph.py

def to_undirected(self) -> Graph:
    """
    Returns an undirected version of a directed graph.

    This method transforms the current graph instance into an undirected graph by
    adding all directed edges in opposite direction. It applies [`ToUndirected`](https://pytorch-geometric.readthedocs.io/en/latest/generated/torch_geometric.transforms.ToUndirected.html#torch_geometric.transforms.ToUndirected)
    transform to the underlying [`torch_geometric.Data`](https://pytorch-geometric.readthedocs.io/en/latest/generated/torch_geometric.data.Data.html#torch_geometric.data.Data) object, which automatically
    duplicates edge attributes for newly created directed edges.

    Examples:
        >>> import pathpyG as pp
        >>> g = pp.Graph.from_edge_list([('a', 'b'), ('b', 'c'), ('c', 'a')])
        >>> g_u = g.to_undirected()
        >>> print(g_u)
        Undirected graph with 3 nodes and 6 (directed) edges
    """
    tf = ToUndirected()
    d = tf(self.data)
    # unfortunately, the application of a transform creates a new edge_index of type tensor
    # so we have to recreate the EdgeIndex tensor and sort it again

    e = EdgeIndex(data=d.edge_index, sparse_size=(self.data.num_nodes, self.data.num_nodes), is_undirected=True)
    d.edge_index = e
    d.num_nodes = self.data.num_nodes
    return Graph(d, self.mapping)

to_weighted_graph

Coalesces multi-edges to single-edges with an additional weight attribute

If the graph contains multiple edges between the same nodes, this method will coalesce them into a single edge with an additional weight attribute called edge_weight that contains the number of coalesced edges. The method returns a new graph instance with the coalesced edges.

Returns:

Name	Type	Description
Graph	pathpyG.core.graph.Graph	Graph with coalesced edges

Source code in src/pathpyG/core/graph.py

def to_weighted_graph(self) -> Graph:
    """Coalesces multi-edges to single-edges with an additional weight attribute

    If the graph contains multiple edges between the same nodes, this method will coalesce
    them into a single edge with an additional weight attribute called `edge_weight` that
    contains the number of coalesced edges. The method returns a new graph instance with
    the coalesced edges.

    Returns:
        Graph: Graph with coalesced edges
    """
    i, w = torch_geometric.utils.coalesce(
        self.data.edge_index.as_tensor(), torch.ones(self.m, device=self.data.edge_index.device)
    )
    return Graph(Data(edge_index=i, edge_weight=w, num_nodes=self.data.num_nodes), mapping=self.mapping)

transition_probabilities

Compute transition probabilities based on weighted outdegrees.

Returns:

Name	Type	Description
tensor	torch.Tensor	Transition probabilities.

Source code in src/pathpyG/core/graph.py

def transition_probabilities(self) -> torch.Tensor:
    """
    Compute transition probabilities based on weighted outdegrees.

    Returns:
        tensor: Transition probabilities.
    """
    weighted_outdegree = self.weighted_outdegrees()
    source_ids = self.data.edge_index[0]
    return self.data.edge_weight / weighted_outdegree[source_ids]

weighted_outdegrees

Compute the weighted outdegrees of each node in the graph.

Parameters:

Name	Type	Description	Default
graph	pathpyG.core.graph.Graph	pathpy graph object.	required

Returns:

Name	Type	Description
tensor	torch.Tensor	Weighted outdegrees of nodes.

Source code in src/pathpyG/core/graph.py

def weighted_outdegrees(self) -> torch.Tensor:
    """
    Compute the weighted outdegrees of each node in the graph.

    Args:
        graph (Graph): pathpy graph object.

    Returns:
        tensor: Weighted outdegrees of nodes.
    """
    weighted_outdegree = scatter(
        self.data.edge_weight, self.data.edge_index[0], dim=0, dim_size=self.data.num_nodes, reduce="sum"
    )
    return weighted_outdegree

TemporalGraph

Bases: pathpyG.Graph

Source code in src/pathpyG/core/temporal_graph.py

class TemporalGraph(Graph):
    def __init__(self, data: Data, mapping: IndexMap | None = None) -> None:
        """Creates an instance of a temporal graph from a `TemporalData` object.

        Args:
            data: xxx
            mapping: xxx

        Example:
            ```py
            from pytorch_geometric.data import TemporalData
            import pathpyG as pp

            d = Data(edge_index=[[0,0,1], [1,2,2]], time=[0,1,2])
            t = pp.TemporalGraph(d, mapping)
            print(t)
            ```
        """
        if not isinstance(data.edge_index, EdgeIndex):
            data.edge_index = data.edge_index = EdgeIndex(
                data=data.edge_index.contiguous(), sparse_size=(data.num_nodes, data.num_nodes)
            )

        # reorder temporal data
        # TODO: Fix in PyG
        if data.num_nodes != data.num_edges:
            self.data = data.sort_by_time()
        else:
            sorted_idx = torch.argsort(data.time)
            data.time = data.time[sorted_idx]
            for edge_attr in data.edge_attrs():
                if edge_attr == "edge_index":
                    data.edge_index = data.edge_index[:, sorted_idx]
                else:
                    data[edge_attr] = data[edge_attr][sorted_idx]
            self.data = data

        if mapping is not None:
            self.mapping = mapping
        else:
            self.mapping = IndexMap()

        # create mapping between edge index and edge tuples
        self.edge_to_index = {
            (e[0].item(), e[1].item()): i for i, e in enumerate([e for e in self.data.edge_index.t()])
        }

        self.start_time = self.data.time[0].item()
        self.end_time = self.data.time[-1].item()

    @staticmethod
    def from_edge_list(edge_list, num_nodes: Optional[int] = None) -> TemporalGraph:
        edge_array = np.array(edge_list)
        ts = edge_array[:, 2].astype(np.number)

        index_map = IndexMap(np.unique(edge_array[:, :2]))
        edge_index = index_map.to_idxs(edge_array[:, :2].T)

        if not num_nodes:
            num_nodes = index_map.num_ids()

        return TemporalGraph(
            data=Data(
                edge_index=edge_index,
                time=torch.Tensor(ts),
                num_nodes=num_nodes,
            ),
            mapping=index_map,
        )

    @property
    def temporal_edges(self) -> Generator[Tuple[int, int, int], None, None]:
        """Iterator that yields each edge as a tuple of source and destination node as well as the corresponding timestamp."""
        return [(*self.mapping.to_ids(e), t.item()) for e, t in zip(self.data.edge_index.t(), self.data.time)]

    @property
    def order(self) -> int:
        """Return order 1, since all temporal graphs must be order one."""
        return 1

    def shuffle_time(self) -> None:
        """Randomly shuffle the temporal order of edges by randomly permuting timestamps."""
        self.data.time = self.data.time[torch.randperm(len(self.data.time))]

    def to_static_graph(self, weighted: bool = False, time_window: Optional[Tuple[int, int]] = None) -> Graph:
        """Return weighted time-aggregated instance of [`Graph`][pathpyG.Graph] graph.

        Args:
            weighted: whether or not to return a weighted time-aggregated graph
            time_window: A tuple with start and end time of the aggregation window

        Returns:
            Graph: A static graph object
        """
        if time_window is not None:
            idx = (self.data.time >= time_window[0]).logical_and(self.data.time < time_window[1]).nonzero().ravel()
            edge_index = self.data.edge_index[:, idx]
        else:
            edge_index = self.data.edge_index

        n = edge_index.max().item() + 1

        if weighted:
            i, w = torch_geometric.utils.coalesce(
                edge_index.as_tensor(), torch.ones(edge_index.size(1), device=self.data.edge_index.device)
            )
            return Graph(Data(edge_index=EdgeIndex(data=i, sparse_size=(n, n)), edge_weight=w), self.mapping)
        else:
            return Graph.from_edge_index(EdgeIndex(data=edge_index, sparse_size=(n, n)), self.mapping)

    def to_undirected(self) -> TemporalGraph:
        """Return an undirected version of a directed graph.

        This method transforms the current graph instance into an undirected graph by
        adding all directed edges in opposite direction. It applies [`ToUndirected`](https://pytorch-geometric.readthedocs.io/en/latest/generated/torch_geometric.transforms.ToUndirected.html#torch_geometric.transforms.ToUndirected)
        transform to the underlying [`torch_geometric.Data`](https://pytorch-geometric.readthedocs.io/en/latest/generated/torch_geometric.data.Data.html#torch_geometric.data.Data) object, which automatically
        duplicates edge attributes for newly created directed edges.

        Example:
            ```py
            import pathpyG as pp
            g = pp.TemporalGraph.from_edge_list([('a', 'b', 1), ('b', 'c', 2), ('c', 'a', 3)])
            g_u = g.to_undirected()
            print(g_u)
            ```
        """
        rev_edge_index = self.data.edge_index.flip([0])
        edge_index = torch.cat([self.data.edge_index, rev_edge_index], dim=1)
        times = torch.cat([self.data.time, self.data.time])
        return TemporalGraph(data=Data(edge_index=edge_index, time=times), mapping=self.mapping)

    def get_batch(self, start_idx: int, end_idx: int) -> TemporalGraph:
        """Return an instance of the TemporalGraph that captures all time-stamped
        edges in a given batch defined by start and (non-inclusive) end, where start
        and end refer to the index of the first and last event in the time-ordered list of events."""

        return TemporalGraph(
            data=Data(edge_index=self.data.edge_index[:, start_idx:end_idx], time=self.data.time[start_idx:end_idx]),
            mapping=self.mapping,
        )

    def get_window(self, start_time: int, end_time: int) -> TemporalGraph:
        """Return an instance of the TemporalGraph that captures all time-stamped
        edges in a given time window defined by start and (non-inclusive) end, where start
        and end refer to the time stamps"""

        return TemporalGraph(data=self.data.snapshot(start_time, end_time), mapping=self.mapping)

    def __str__(self) -> str:
        """
        Return a string representation of the graph
        """
        s = "Temporal Graph with {0} nodes, {1} unique edges and {2} events in [{3}, {4}]\n".format(
            self.data.num_nodes,
            self.data.edge_index.unique(dim=1).size(dim=1),
            self.data.edge_index.size(1),
            self.start_time,
            self.end_time,
        )

        attr = self.data.to_dict()
        attr_types = {}
        for k in attr:
            t = type(attr[k])
            if t == torch.Tensor:
                attr_types[k] = str(t) + " -> " + str(attr[k].size())
            else:
                attr_types[k] = str(t)

        from pprint import pformat

        attribute_info = {"Node Attributes": {}, "Edge Attributes": {}, "Graph Attributes": {}}
        for a in self.node_attrs():
            attribute_info["Node Attributes"][a] = attr_types[a]
        for a in self.edge_attrs():
            attribute_info["Edge Attributes"][a] = attr_types[a]
        for a in self.data.keys():
            if not self.data.is_node_attr(a) and not self.data.is_edge_attr(a):
                attribute_info["Graph Attributes"][a] = attr_types[a]
        s += pformat(attribute_info, indent=4, width=160)
        return s

order property

Return order 1, since all temporal graphs must be order one.

temporal_edges property

Iterator that yields each edge as a tuple of source and destination node as well as the corresponding timestamp.

__init__

Creates an instance of a temporal graph from a TemporalData object.

Parameters:

Name	Type	Description	Default
data	torch_geometric.data.Data	xxx	required
mapping	pathpyG.core.index_map.IndexMap | None	xxx	None

Example

from pytorch_geometric.data import TemporalData
import pathpyG as pp

d = Data(edge_index=[[0,0,1], [1,2,2]], time=[0,1,2])
t = pp.TemporalGraph(d, mapping)
print(t)

Source code in src/pathpyG/core/temporal_graph.py

def __init__(self, data: Data, mapping: IndexMap | None = None) -> None:
    """Creates an instance of a temporal graph from a `TemporalData` object.

    Args:
        data: xxx
        mapping: xxx

    Example:
        ```py
        from pytorch_geometric.data import TemporalData
        import pathpyG as pp

        d = Data(edge_index=[[0,0,1], [1,2,2]], time=[0,1,2])
        t = pp.TemporalGraph(d, mapping)
        print(t)
        ```
    """
    if not isinstance(data.edge_index, EdgeIndex):
        data.edge_index = data.edge_index = EdgeIndex(
            data=data.edge_index.contiguous(), sparse_size=(data.num_nodes, data.num_nodes)
        )

    # reorder temporal data
    # TODO: Fix in PyG
    if data.num_nodes != data.num_edges:
        self.data = data.sort_by_time()
    else:
        sorted_idx = torch.argsort(data.time)
        data.time = data.time[sorted_idx]
        for edge_attr in data.edge_attrs():
            if edge_attr == "edge_index":
                data.edge_index = data.edge_index[:, sorted_idx]
            else:
                data[edge_attr] = data[edge_attr][sorted_idx]
        self.data = data

    if mapping is not None:
        self.mapping = mapping
    else:
        self.mapping = IndexMap()

    # create mapping between edge index and edge tuples
    self.edge_to_index = {
        (e[0].item(), e[1].item()): i for i, e in enumerate([e for e in self.data.edge_index.t()])
    }

    self.start_time = self.data.time[0].item()
    self.end_time = self.data.time[-1].item()

__str__

Return a string representation of the graph

Source code in src/pathpyG/core/temporal_graph.py

def __str__(self) -> str:
    """
    Return a string representation of the graph
    """
    s = "Temporal Graph with {0} nodes, {1} unique edges and {2} events in [{3}, {4}]\n".format(
        self.data.num_nodes,
        self.data.edge_index.unique(dim=1).size(dim=1),
        self.data.edge_index.size(1),
        self.start_time,
        self.end_time,
    )

    attr = self.data.to_dict()
    attr_types = {}
    for k in attr:
        t = type(attr[k])
        if t == torch.Tensor:
            attr_types[k] = str(t) + " -> " + str(attr[k].size())
        else:
            attr_types[k] = str(t)

    from pprint import pformat

    attribute_info = {"Node Attributes": {}, "Edge Attributes": {}, "Graph Attributes": {}}
    for a in self.node_attrs():
        attribute_info["Node Attributes"][a] = attr_types[a]
    for a in self.edge_attrs():
        attribute_info["Edge Attributes"][a] = attr_types[a]
    for a in self.data.keys():
        if not self.data.is_node_attr(a) and not self.data.is_edge_attr(a):
            attribute_info["Graph Attributes"][a] = attr_types[a]
    s += pformat(attribute_info, indent=4, width=160)
    return s

get_batch

Return an instance of the TemporalGraph that captures all time-stamped edges in a given batch defined by start and (non-inclusive) end, where start and end refer to the index of the first and last event in the time-ordered list of events.

Source code in src/pathpyG/core/temporal_graph.py

def get_batch(self, start_idx: int, end_idx: int) -> TemporalGraph:
    """Return an instance of the TemporalGraph that captures all time-stamped
    edges in a given batch defined by start and (non-inclusive) end, where start
    and end refer to the index of the first and last event in the time-ordered list of events."""

    return TemporalGraph(
        data=Data(edge_index=self.data.edge_index[:, start_idx:end_idx], time=self.data.time[start_idx:end_idx]),
        mapping=self.mapping,
    )

get_window

Return an instance of the TemporalGraph that captures all time-stamped edges in a given time window defined by start and (non-inclusive) end, where start and end refer to the time stamps

Source code in src/pathpyG/core/temporal_graph.py

def get_window(self, start_time: int, end_time: int) -> TemporalGraph:
    """Return an instance of the TemporalGraph that captures all time-stamped
    edges in a given time window defined by start and (non-inclusive) end, where start
    and end refer to the time stamps"""

    return TemporalGraph(data=self.data.snapshot(start_time, end_time), mapping=self.mapping)

shuffle_time

Randomly shuffle the temporal order of edges by randomly permuting timestamps.

Source code in src/pathpyG/core/temporal_graph.py

def shuffle_time(self) -> None:
    """Randomly shuffle the temporal order of edges by randomly permuting timestamps."""
    self.data.time = self.data.time[torch.randperm(len(self.data.time))]

to_static_graph

Return weighted time-aggregated instance of Graph graph.

Parameters:

Name	Type	Description	Default
weighted	bool	whether or not to return a weighted time-aggregated graph	False
time_window	typing.Optional[typing.Tuple[int, int]]	A tuple with start and end time of the aggregation window	None

Returns:

Name	Type	Description
Graph	pathpyG.Graph	A static graph object

Source code in src/pathpyG/core/temporal_graph.py

def to_static_graph(self, weighted: bool = False, time_window: Optional[Tuple[int, int]] = None) -> Graph:
    """Return weighted time-aggregated instance of [`Graph`][pathpyG.Graph] graph.

    Args:
        weighted: whether or not to return a weighted time-aggregated graph
        time_window: A tuple with start and end time of the aggregation window

    Returns:
        Graph: A static graph object
    """
    if time_window is not None:
        idx = (self.data.time >= time_window[0]).logical_and(self.data.time < time_window[1]).nonzero().ravel()
        edge_index = self.data.edge_index[:, idx]
    else:
        edge_index = self.data.edge_index

    n = edge_index.max().item() + 1

    if weighted:
        i, w = torch_geometric.utils.coalesce(
            edge_index.as_tensor(), torch.ones(edge_index.size(1), device=self.data.edge_index.device)
        )
        return Graph(Data(edge_index=EdgeIndex(data=i, sparse_size=(n, n)), edge_weight=w), self.mapping)
    else:
        return Graph.from_edge_index(EdgeIndex(data=edge_index, sparse_size=(n, n)), self.mapping)

to_undirected

Return an undirected version of a directed graph.

This method transforms the current graph instance into an undirected graph by adding all directed edges in opposite direction. It applies ToUndirected transform to the underlying torch_geometric.Data object, which automatically duplicates edge attributes for newly created directed edges.

Example

import pathpyG as pp
g = pp.TemporalGraph.from_edge_list([('a', 'b', 1), ('b', 'c', 2), ('c', 'a', 3)])
g_u = g.to_undirected()
print(g_u)

Source code in src/pathpyG/core/temporal_graph.py

def to_undirected(self) -> TemporalGraph:
    """Return an undirected version of a directed graph.

    This method transforms the current graph instance into an undirected graph by
    adding all directed edges in opposite direction. It applies [`ToUndirected`](https://pytorch-geometric.readthedocs.io/en/latest/generated/torch_geometric.transforms.ToUndirected.html#torch_geometric.transforms.ToUndirected)
    transform to the underlying [`torch_geometric.Data`](https://pytorch-geometric.readthedocs.io/en/latest/generated/torch_geometric.data.Data.html#torch_geometric.data.Data) object, which automatically
    duplicates edge attributes for newly created directed edges.

    Example:
        ```py
        import pathpyG as pp
        g = pp.TemporalGraph.from_edge_list([('a', 'b', 1), ('b', 'c', 2), ('c', 'a', 3)])
        g_u = g.to_undirected()
        print(g_u)
        ```
    """
    rev_edge_index = self.data.edge_index.flip([0])
    edge_index = torch.cat([self.data.edge_index, rev_edge_index], dim=1)
    times = torch.cat([self.data.time, self.data.time])
    return TemporalGraph(data=Data(edge_index=edge_index, time=times), mapping=self.mapping)

temporal_shortest_paths

Compute shortest time-respecting paths in a temporal graph.

Parameters:

Name	Type	Description	Default
g	pathpyG.core.temporal_graph.TemporalGraph	Temporal graph to compute shortest paths on.	required
delta	int	Maximum time difference between events in a path.	required

Returns:

Type	Description
numpy.ndarray	Tuple of two numpy arrays:
numpy.ndarray	dist: Shortest time-respecting path distances between all first-order nodes.
typing.Tuple[numpy.ndarray, numpy.ndarray]	pred: Predecessor matrix for shortest time-respecting paths between all first-order nodes.

Source code in src/pathpyG/algorithms/temporal.py

def temporal_shortest_paths(g: TemporalGraph, delta: int) -> Tuple[np.ndarray, np.ndarray]:
    """Compute shortest time-respecting paths in a temporal graph.

    Args:
        g: Temporal graph to compute shortest paths on.
        delta: Maximum time difference between events in a path.

    Returns:
        Tuple of two numpy arrays:
        - dist: Shortest time-respecting path distances between all first-order nodes.
        - pred: Predecessor matrix for shortest time-respecting paths between all first-order nodes.
    """
    # generate temporal event DAG
    edge_index = lift_order_temporal(g, delta)

    # Add indices of first-order nodes as src and dst of paths in augmented
    # temporal event DAG
    src_edges_src = g.data.edge_index[0] + g.m
    src_edges_dst = torch.arange(0, g.data.edge_index.size(1), device=g.data.edge_index.device)

    dst_edges_src = torch.arange(0, g.data.edge_index.size(1), device=g.data.edge_index.device)
    dst_edges_dst = g.data.edge_index[1] + g.m + g.n

    # add edges from source to edges and from edges to destinations
    src_edges = torch.stack([src_edges_src, src_edges_dst])
    dst_edges = torch.stack([dst_edges_src, dst_edges_dst])
    edge_index = torch.cat([edge_index, src_edges, dst_edges], dim=1)

    # create sparse scipy matrix
    event_graph = Graph.from_edge_index(edge_index, num_nodes=g.m + 2 * g.n)
    m = event_graph.sparse_adj_matrix()

    # print(f"Created temporal event DAG with {event_graph.n} nodes and {event_graph.m} edges")

    # run disjktra for all source nodes
    dist, pred = dijkstra(
        m, directed=True, indices=np.arange(g.m, g.m + g.n), return_predecessors=True, unweighted=True
    )

    # limit to first-order destinations and correct distances
    dist_fo = dist[:, g.m + g.n :] - 1
    np.fill_diagonal(dist_fo, 0)

    # limit to first-order destinations and correct predecessors
    pred_fo = pred[:, g.n + g.m :]
    pred_fo[pred_fo == -9999] = -1
    idx_map = np.concatenate([to_numpy(g.data.edge_index[0].cpu()), [-1]])
    pred_fo = idx_map[pred_fo]
    np.fill_diagonal(pred_fo, np.arange(g.n))

    return dist_fo, pred_fo

to_numpy

Convert a tensor or tensor subclasses like torch_geometric.Edge_Index to numpy.

Parameters:

Name	Type	Description	Default
tensor	torch.Tensor	Tensor or tensor subclass.	required

Returns:

Type	Description
numpy.ndarray	Numpy array.

Source code in src/pathpyG/utils/convert.py

def to_numpy(tensor: torch.Tensor) -> np.ndarray:
    """
    Convert a tensor or tensor subclasses like `torch_geometric.Edge_Index` to numpy.

    Args:
        tensor: Tensor or tensor subclass.

    Returns:
        Numpy array.
    """
    if isinstance(tensor, (EdgeIndex, Index)):
        return tensor.as_tensor().numpy()
    return tensor.numpy()