Skip to content

event_graph

Event graph representation of a temporal graph and related operations.

EventGraph

Bases: pathpyG.core.graph.Graph

A directed acyclic graph whose nodes are time-stamped events.

Each event is a temporal edge of an underlying temporal graph. The node_sequence node attribute of shape (num_events, 2) holds the source and target first-order node of each event, and node_time holds its timestamp.

Source code in src/pathpyG/core/event_graph.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
class EventGraph(Graph):
    """A directed acyclic graph whose nodes are time-stamped events.

    Each event is a temporal edge of an underlying temporal graph. The `node_sequence`
    node attribute of shape `(num_events, 2)` holds the source and target first-order
    node of each event, and `node_time` holds its timestamp.
    """

    # The event graph manages the first-order edge of each event itself.
    _internal_node_attrs: frozenset[str] = frozenset({"node_sequence"})

    # Attributes that are constructed explicitly when lifting a temporal graph and that
    # must therefore not be overwritten by propagated attributes.
    _RESERVED_ATTRS = frozenset({"edge_index", "num_nodes", "node_sequence", "node_time"})

    def __init__(
        self,
        data: Data,
        delta: int,
        first_order_mapping: IndexMap | None = None,
        n_first_order: int | None = None,
        mapping: IndexMap | None = None,
    ) -> None:
        """Create an EventGraph from a `Data` object carrying per-event `node_time`."""
        if "node_time" not in data:
            raise ValueError("EventGraph requires a per-event `node_time` node attribute.")

        super().__init__(data, mapping=mapping)

        self.delta = delta
        self.first_order_mapping = first_order_mapping if first_order_mapping is not None else IndexMap()
        if n_first_order is not None:
            self._n_first_order = int(n_first_order)
        else:
            self._n_first_order = int(self.data.node_sequence.max().item()) + 1

        ei = self.data.edge_index
        self.data.edge_delta = self.data.node_time[ei[1]] - self.data.node_time[ei[0]]

        self._temporal_graph: TemporalGraph | None = None

    @staticmethod
    def build_edge_index(temporal_graph: TemporalGraph, delta: float | int = 1):
        """Build the event-graph edge index by lifting a temporal graph to second order.

        Each temporal edge of `temporal_graph` becomes an event (node); two events are
        connected when the second can continue the first within the time window `delta`.

        Args:
            temporal_graph: Temporal graph to lift.
            delta: Maximum time difference between events to consider them connected.

        Returns:
            ho_index: Edge index of the second-order temporal event graph.
        """
        # first-order edge index
        edge_index, timestamps = temporal_graph.data.edge_index, temporal_graph.data.time

        delta = torch.tensor(delta, device=edge_index.device)  # type: ignore[assignment]
        indices = torch.arange(0, edge_index.size(1), device=edge_index.device)

        unique_t = torch.unique(timestamps, sorted=True)
        second_order = []

        # lift order: find possible continuations for edges in each time stamp
        for t in tqdm(unique_t):
            # find indices of all source edges that occur at unique timestamp t
            src_time_mask = timestamps == t
            src_edge_idx = indices[src_time_mask]

            # find indices of all edges that can possibly continue edges occurring at time t for the given delta
            dst_time_mask = (timestamps > t) & (timestamps <= t + delta)
            dst_edge_idx = indices[dst_time_mask]

            if dst_edge_idx.size(0) > 0 and src_edge_idx.size(0) > 0:
                # compute second-order edges between src and dst idx
                # for all edges where dst in src_edges (edge_index[1, x[:, 0]]) matches src in dst_edges (edge_index[0, x[:, 1]])
                x = torch.cartesian_prod(src_edge_idx, dst_edge_idx)
                ho_edge_index = x[edge_index[1, x[:, 0]] == edge_index[0, x[:, 1]]]
                second_order.append(ho_edge_index)

        ho_index = torch.cat(second_order, dim=0).t().contiguous()
        return ho_index

    @staticmethod
    def lift_attrs(temporal_graph: TemporalGraph) -> dict[str, Any]:
        """Map the attributes of a temporal graph to attributes of the lifted event graph.

        Since every temporal edge becomes an event, edge attributes of the temporal graph
        become node attributes of the event graph and are renamed from `edge_x` to `node_x`
        accordingly. Node attributes refer to first-order nodes, which have no counterpart
        among the events, so they are dropped. Graph attributes are kept unchanged.

        Attributes that are constructed explicitly while lifting (`edge_index`, `num_nodes`,
        `node_sequence` and the timestamps in `time`) are excluded.

        Args:
            temporal_graph: Temporal graph whose attributes shall be lifted.

        Returns:
            dict: mapping from attribute names in the event graph to their values.
        """
        attrs: dict[str, Any] = {}
        for key in temporal_graph.data.keys():
            if key in EventGraph._RESERVED_ATTRS or key == "time":
                continue
            if key.startswith("node_"):
                # attributes of first-order nodes have no counterpart in the event graph
                continue
            if key.startswith("edge_"):
                event_key = "node_" + key[len("edge_") :]
                if event_key in EventGraph._RESERVED_ATTRS:
                    logger.error("Edge attribute %s cannot be lifted to reserved attribute %s", key, event_key)
                    raise ValueError(f"edge attribute '{key}' would be lifted to reserved attribute '{event_key}'")
                attrs[event_key] = _copy_attr(temporal_graph.data[key])
            else:
                # graph-level attributes are kept as they are
                attrs[key] = _copy_attr(temporal_graph.data[key])
        return attrs

    @classmethod
    def from_temporal_graph(cls, temporal_graph: TemporalGraph, delta: int = 1) -> "EventGraph":
        """Build an EventGraph from a temporal graph by lifting its edges into events.

        Attributes of the temporal graph are propagated to the event graph as described in
        [`lift_attrs`][pathpyG.core.event_graph.EventGraph.lift_attrs].
        """
        ho_index = cls.build_edge_index(temporal_graph, delta)
        m = temporal_graph.data.time.size(0)  # number of events (== number of first-order edges)
        node_sequence = temporal_graph.data.edge_index.as_tensor().t().contiguous()  # [m, 2]
        node_time = temporal_graph.data.time.clone()  # [m]

        # Build an event mapping with IDs of the form "a->b@t" for each edge node
        event_ids = [
            f"{temporal_graph.mapping.to_id(u)}->{temporal_graph.mapping.to_id(v)}@{t}"
            for (u, v), t in zip(node_sequence.tolist(), node_time.tolist())
        ]
        mapping = IndexMap(event_ids)

        data = Data(
            edge_index=ho_index,
            num_nodes=m,
            node_sequence=node_sequence,
            node_time=node_time,
        )
        for key, value in cls.lift_attrs(temporal_graph).items():
            data[key] = value

        event_graph = cls(data, delta=delta, first_order_mapping=temporal_graph.mapping, n_first_order=temporal_graph.n, mapping=mapping)

        # Attach a clone of the temporal graph since we already have it
        event_graph._temporal_graph = TemporalGraph(temporal_graph.data.clone(), mapping=temporal_graph.mapping)

        return event_graph

    def _summary(self) -> str:
        """Return a one-line summary of the event graph."""
        return "Event Graph (delta={0}) with {1} first-order nodes, {2} events and {3} edges in [{4}, {5}]".format(
            self.delta,
            self.n_first_order,
            self.num_events,
            self.m,
            self.start_time,
            self.end_time,
        )

    def __len__(self):
        """Return the number of events in the graph."""
        return self.n

    def to(self, device: torch.device) -> "EventGraph":
        """Move the event graph and its underlying temporal graph to the given device."""
        super().to(device)
        self.data.node_sequence = self.data.node_sequence.to(device)
        if self._temporal_graph is not None:
            self._temporal_graph.to(device)
        return self

    def to_temporal_graph(self) -> TemporalGraph:
        """Return the underlying temporal graph, reconstructing it if necessary."""
        if self._temporal_graph is None:
            edge_index = self.data.node_sequence.t().contiguous()  # [2, num_events]
            self._temporal_graph = TemporalGraph(
                Data(
                    edge_index=edge_index,
                    time=self.data.node_time.clone(),
                    num_nodes=self.n_first_order,
                ),
                mapping=self.first_order_mapping,
            )
        return self._temporal_graph

    @property
    def n_first_order(self) -> int:
        """Number of distinct first-order nodes underlying the events."""
        return self._n_first_order

    @property
    def num_events(self) -> int:
        """Number of events (nodes) in the event graph."""
        return self.n

    @property
    def start_time(self) -> Union[int, float]:
        """Return the timestamp of the first event in the event graph."""
        return self.data.node_time.min().item()

    @property
    def end_time(self) -> Union[int, float]:
        """Return the timestamp of the last event in the event graph."""
        return self.data.node_time.max().item()

    def event_time(self, i: int) -> int:
        """Return the timestamp of the i-th event."""
        return self.data.node_time[i].item()

    def edge_delta_map(self) -> dict[tuple[int, int], int]:
        """Return a mapping from each transition edge (src, dst) to its time delta."""
        return {
            tuple(c): d
            for c, d in zip(
                self.data.edge_index.as_tensor().t().tolist(),
                self.data.edge_delta.tolist(),
            )
        }

    def shortest_paths(self) -> Tuple[np.ndarray, np.ndarray]:
        """Return first-order shortest-path distances and predecessors respecting delta."""
        from pathpyG.algorithms.temporal import temporal_shortest_paths

        return temporal_shortest_paths(temporal_graph=None, delta=self.delta, event_graph=self)

    def reduce_delta(self, decrement: int = 1) -> "EventGraph":
        """Return a new EventGraph with a reduced time window `delta - decrement`.

        The events are unchanged, so node and graph attributes are carried over as they are,
        while edge attributes are restricted to the edges that remain for the smaller `delta`.
        """
        new_delta = self.delta - decrement
        if new_delta < 0:
            raise ValueError(
                f"decrement={decrement} exceeds current delta={self.delta}"
            )

        ei = self.data.edge_index
        edge_delta = self.data.node_time[ei[1]] - self.data.node_time[ei[0]]
        mask = edge_delta <= new_delta
        new_edge_index = ei[:, mask].contiguous()

        data = Data(
            edge_index=new_edge_index,
            num_nodes=self.n,
            node_sequence=self.data.node_sequence.clone(),
            node_time=self.data.node_time.clone(),
        )
        for key in self.data.keys():
            if key in self._RESERVED_ATTRS:
                continue
            if key.startswith("edge_"):
                data[key] = self.data[key][mask]
            else:
                data[key] = _copy_attr(self.data[key])

        return EventGraph(
            data,
            delta=new_delta,
            first_order_mapping=self.first_order_mapping,
            n_first_order=self.n_first_order,
            mapping=self.mapping,
        )

end_time property

Return the timestamp of the last event in the event graph.

n_first_order property

Number of distinct first-order nodes underlying the events.

num_events property

Number of events (nodes) in the event graph.

start_time property

Return the timestamp of the first event in the event graph.

__init__

Create an EventGraph from a Data object carrying per-event node_time.

Source code in src/pathpyG/core/event_graph.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def __init__(
    self,
    data: Data,
    delta: int,
    first_order_mapping: IndexMap | None = None,
    n_first_order: int | None = None,
    mapping: IndexMap | None = None,
) -> None:
    """Create an EventGraph from a `Data` object carrying per-event `node_time`."""
    if "node_time" not in data:
        raise ValueError("EventGraph requires a per-event `node_time` node attribute.")

    super().__init__(data, mapping=mapping)

    self.delta = delta
    self.first_order_mapping = first_order_mapping if first_order_mapping is not None else IndexMap()
    if n_first_order is not None:
        self._n_first_order = int(n_first_order)
    else:
        self._n_first_order = int(self.data.node_sequence.max().item()) + 1

    ei = self.data.edge_index
    self.data.edge_delta = self.data.node_time[ei[1]] - self.data.node_time[ei[0]]

    self._temporal_graph: TemporalGraph | None = None

__len__

Return the number of events in the graph.

Source code in src/pathpyG/core/event_graph.py 
194
195
196
def __len__(self):
    """Return the number of events in the graph."""
    return self.n

build_edge_index staticmethod

Build the event-graph edge index by lifting a temporal graph to second order.

Each temporal edge of temporal_graph becomes an event (node); two events are connected when the second can continue the first within the time window delta.

Parameters:

Name Type Description Default
temporal_graph pathpyG.core.temporal_graph.TemporalGraph

Temporal graph to lift.

 required
delta float | int

Maximum time difference between events to consider them connected.

 1

Returns:

Name Type Description
ho_index

Edge index of the second-order temporal event graph.
Source code in src/pathpyG/core/event_graph.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@staticmethod
def build_edge_index(temporal_graph: TemporalGraph, delta: float | int = 1):
    """Build the event-graph edge index by lifting a temporal graph to second order.

    Each temporal edge of `temporal_graph` becomes an event (node); two events are
    connected when the second can continue the first within the time window `delta`.

    Args:
        temporal_graph: Temporal graph to lift.
        delta: Maximum time difference between events to consider them connected.

    Returns:
        ho_index: Edge index of the second-order temporal event graph.
    """
    # first-order edge index
    edge_index, timestamps = temporal_graph.data.edge_index, temporal_graph.data.time

    delta = torch.tensor(delta, device=edge_index.device)  # type: ignore[assignment]
    indices = torch.arange(0, edge_index.size(1), device=edge_index.device)

    unique_t = torch.unique(timestamps, sorted=True)
    second_order = []

    # lift order: find possible continuations for edges in each time stamp
    for t in tqdm(unique_t):
        # find indices of all source edges that occur at unique timestamp t
        src_time_mask = timestamps == t
        src_edge_idx = indices[src_time_mask]

        # find indices of all edges that can possibly continue edges occurring at time t for the given delta
        dst_time_mask = (timestamps > t) & (timestamps <= t + delta)
        dst_edge_idx = indices[dst_time_mask]

        if dst_edge_idx.size(0) > 0 and src_edge_idx.size(0) > 0:
            # compute second-order edges between src and dst idx
            # for all edges where dst in src_edges (edge_index[1, x[:, 0]]) matches src in dst_edges (edge_index[0, x[:, 1]])
            x = torch.cartesian_prod(src_edge_idx, dst_edge_idx)
            ho_edge_index = x[edge_index[1, x[:, 0]] == edge_index[0, x[:, 1]]]
            second_order.append(ho_edge_index)

    ho_index = torch.cat(second_order, dim=0).t().contiguous()
    return ho_index

edge_delta_map

Return a mapping from each transition edge (src, dst) to its time delta.

Source code in src/pathpyG/core/event_graph.py 
244
245
246
247
248
249
250
251
252
def edge_delta_map(self) -> dict[tuple[int, int], int]:
    """Return a mapping from each transition edge (src, dst) to its time delta."""
    return {
        tuple(c): d
        for c, d in zip(
            self.data.edge_index.as_tensor().t().tolist(),
            self.data.edge_delta.tolist(),
        )
    }

event_time

Return the timestamp of the i-th event.

Source code in src/pathpyG/core/event_graph.py 
240
241
242
def event_time(self, i: int) -> int:
    """Return the timestamp of the i-th event."""
    return self.data.node_time[i].item()

from_temporal_graph classmethod

Build an EventGraph from a temporal graph by lifting its edges into events.

Attributes of the temporal graph are propagated to the event graph as described in lift_attrs.

Source code in src/pathpyG/core/event_graph.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
@classmethod
def from_temporal_graph(cls, temporal_graph: TemporalGraph, delta: int = 1) -> "EventGraph":
    """Build an EventGraph from a temporal graph by lifting its edges into events.

    Attributes of the temporal graph are propagated to the event graph as described in
    [`lift_attrs`][pathpyG.core.event_graph.EventGraph.lift_attrs].
    """
    ho_index = cls.build_edge_index(temporal_graph, delta)
    m = temporal_graph.data.time.size(0)  # number of events (== number of first-order edges)
    node_sequence = temporal_graph.data.edge_index.as_tensor().t().contiguous()  # [m, 2]
    node_time = temporal_graph.data.time.clone()  # [m]

    # Build an event mapping with IDs of the form "a->b@t" for each edge node
    event_ids = [
        f"{temporal_graph.mapping.to_id(u)}->{temporal_graph.mapping.to_id(v)}@{t}"
        for (u, v), t in zip(node_sequence.tolist(), node_time.tolist())
    ]
    mapping = IndexMap(event_ids)

    data = Data(
        edge_index=ho_index,
        num_nodes=m,
        node_sequence=node_sequence,
        node_time=node_time,
    )
    for key, value in cls.lift_attrs(temporal_graph).items():
        data[key] = value

    event_graph = cls(data, delta=delta, first_order_mapping=temporal_graph.mapping, n_first_order=temporal_graph.n, mapping=mapping)

    # Attach a clone of the temporal graph since we already have it
    event_graph._temporal_graph = TemporalGraph(temporal_graph.data.clone(), mapping=temporal_graph.mapping)

    return event_graph

lift_attrs staticmethod

Map the attributes of a temporal graph to attributes of the lifted event graph.

Since every temporal edge becomes an event, edge attributes of the temporal graph become node attributes of the event graph and are renamed from edge_x to node_x accordingly. Node attributes refer to first-order nodes, which have no counterpart among the events, so they are dropped. Graph attributes are kept unchanged.

Attributes that are constructed explicitly while lifting (edge_index, num_nodes, node_sequence and the timestamps in time) are excluded.

Parameters:

Name Type Description Default
temporal_graph pathpyG.core.temporal_graph.TemporalGraph

Temporal graph whose attributes shall be lifted.

 required

Returns:

Name Type Description
dict dict[str, typing.Any]

mapping from attribute names in the event graph to their values.
Source code in src/pathpyG/core/event_graph.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
@staticmethod
def lift_attrs(temporal_graph: TemporalGraph) -> dict[str, Any]:
    """Map the attributes of a temporal graph to attributes of the lifted event graph.

    Since every temporal edge becomes an event, edge attributes of the temporal graph
    become node attributes of the event graph and are renamed from `edge_x` to `node_x`
    accordingly. Node attributes refer to first-order nodes, which have no counterpart
    among the events, so they are dropped. Graph attributes are kept unchanged.

    Attributes that are constructed explicitly while lifting (`edge_index`, `num_nodes`,
    `node_sequence` and the timestamps in `time`) are excluded.

    Args:
        temporal_graph: Temporal graph whose attributes shall be lifted.

    Returns:
        dict: mapping from attribute names in the event graph to their values.
    """
    attrs: dict[str, Any] = {}
    for key in temporal_graph.data.keys():
        if key in EventGraph._RESERVED_ATTRS or key == "time":
            continue
        if key.startswith("node_"):
            # attributes of first-order nodes have no counterpart in the event graph
            continue
        if key.startswith("edge_"):
            event_key = "node_" + key[len("edge_") :]
            if event_key in EventGraph._RESERVED_ATTRS:
                logger.error("Edge attribute %s cannot be lifted to reserved attribute %s", key, event_key)
                raise ValueError(f"edge attribute '{key}' would be lifted to reserved attribute '{event_key}'")
            attrs[event_key] = _copy_attr(temporal_graph.data[key])
        else:
            # graph-level attributes are kept as they are
            attrs[key] = _copy_attr(temporal_graph.data[key])
    return attrs

reduce_delta

Return a new EventGraph with a reduced time window delta - decrement.

The events are unchanged, so node and graph attributes are carried over as they are, while edge attributes are restricted to the edges that remain for the smaller delta.

Source code in src/pathpyG/core/event_graph.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
def reduce_delta(self, decrement: int = 1) -> "EventGraph":
    """Return a new EventGraph with a reduced time window `delta - decrement`.

    The events are unchanged, so node and graph attributes are carried over as they are,
    while edge attributes are restricted to the edges that remain for the smaller `delta`.
    """
    new_delta = self.delta - decrement
    if new_delta < 0:
        raise ValueError(
            f"decrement={decrement} exceeds current delta={self.delta}"
        )

    ei = self.data.edge_index
    edge_delta = self.data.node_time[ei[1]] - self.data.node_time[ei[0]]
    mask = edge_delta <= new_delta
    new_edge_index = ei[:, mask].contiguous()

    data = Data(
        edge_index=new_edge_index,
        num_nodes=self.n,
        node_sequence=self.data.node_sequence.clone(),
        node_time=self.data.node_time.clone(),
    )
    for key in self.data.keys():
        if key in self._RESERVED_ATTRS:
            continue
        if key.startswith("edge_"):
            data[key] = self.data[key][mask]
        else:
            data[key] = _copy_attr(self.data[key])

    return EventGraph(
        data,
        delta=new_delta,
        first_order_mapping=self.first_order_mapping,
        n_first_order=self.n_first_order,
        mapping=self.mapping,
    )

shortest_paths

Return first-order shortest-path distances and predecessors respecting delta.

Source code in src/pathpyG/core/event_graph.py 
254
255
256
257
258
def shortest_paths(self) -> Tuple[np.ndarray, np.ndarray]:
    """Return first-order shortest-path distances and predecessors respecting delta."""
    from pathpyG.algorithms.temporal import temporal_shortest_paths

    return temporal_shortest_paths(temporal_graph=None, delta=self.delta, event_graph=self)

to

Move the event graph and its underlying temporal graph to the given device.

Source code in src/pathpyG/core/event_graph.py 
198
199
200
201
202
203
204
def to(self, device: torch.device) -> "EventGraph":
    """Move the event graph and its underlying temporal graph to the given device."""
    super().to(device)
    self.data.node_sequence = self.data.node_sequence.to(device)
    if self._temporal_graph is not None:
        self._temporal_graph.to(device)
    return self

to_temporal_graph

Return the underlying temporal graph, reconstructing it if necessary.

Source code in src/pathpyG/core/event_graph.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
def to_temporal_graph(self) -> TemporalGraph:
    """Return the underlying temporal graph, reconstructing it if necessary."""
    if self._temporal_graph is None:
        edge_index = self.data.node_sequence.t().contiguous()  # [2, num_events]
        self._temporal_graph = TemporalGraph(
            Data(
                edge_index=edge_index,
                time=self.data.node_time.clone(),
                num_nodes=self.n_first_order,
            ),
            mapping=self.first_order_mapping,
        )
    return self._temporal_graph