Skip to content

Reference

Graphs

unweaver.graphs.DiGraphGPKG

Bases: DiGraphGPKGView

Mutable directed graph backed by a GeoPackage. An extension of unweaver.graphs.DiGraphGPKGView.

Parameters:

Name Type Description Default
path Optional[str]

An optional path to database file (or :memory:-type string).

 None
network Optional[GeoPackageNetwork]

An optional path to a custom GeoPackageNetwork instance.

 None
**kwargs Any

Keyword arguments compatible with networkx.DiGraph.

 {}
Source code in unweaver/graphs/digraphgpkg/digraphgpkg.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 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
class DiGraphGPKG(DiGraphGPKGView):
    """Mutable directed graph backed by a GeoPackage. An extension of
    unweaver.graphs.DiGraphGPKGView.

    :param path: An optional path to database file (or :memory:-type string).
    :param network: An optional path to a custom GeoPackageNetwork instance.
    :param **kwargs: Keyword arguments compatible with networkx.DiGraph.

    """

    node_dict_factory = Nodes
    adjlist_outer_dict_factory = OuterSuccessors
    # TODO: consider creating a read-only Mapping in the case of immutable
    #       graphs.
    adjlist_inner_dict_factory = dict
    edge_attr_dict_factory = Edge

    def __init__(
        self,
        path: Optional[str] = None,
        network: Optional[GeoPackageNetwork] = None,
        **kwargs: Any
    ):
        # TODO: Consider adding database file existence checker rather than
        #       always checking on initialization?
        if network is None:
            # FIXME: should path be allowed to be None?
            if path is None:
                raise UnderspecifiedGraphError()
            else:
                if not os.path.exists(path):
                    raise UnderspecifiedGraphError(
                        "DB file does not exist. Consider using "
                        "DiGraphGPKG.create_graph"
                    )

                network = GeoPackageNetwork(path)

        super().__init__(path=path, network=network, **kwargs)
        self.mutable = True

    @classmethod
    def create_graph(cls, path: str = None, **kwargs: Any) -> DiGraphGPKG:
        """Create a new DiGraphGPKG (.gpkg) at a given path.

        :param path: The path of the new GeoPackage (.gpkg).
        :param **kwargs: Any other keyword arguments to pass to the new
        DiGraphGPKG instance.
        :returns: A new DiGraphGPKG instance.

        """
        network = GeoPackageNetwork(path)
        return DiGraphGPKG(network=network, **kwargs)

    def add_edges_from(
        self,
        ebunch: Iterable[EdgeTuple],
        _batch_size: int = 1000,
        counter: Optional[ProgressBar] = None,
        **attr: Any
    ) -> None:
        """Equivalent to add_edges_from in networkx but with batched SQL writes.

        :param ebunch: edge bunch, identical to nx ebunch_to_add.
        :param _batch_size: Number of rows to commit to the database at a time.
        :param **attr: Default attributes, identical to nx attr.

        """
        if _batch_size < 2:
            # User has entered invalid number (negative, zero) or 1. Use
            # default behavior.
            super().add_edges_from(self, ebunch, **attr)
            return

        # TODO: length check on each edge
        features = (
            {"_u": edge[0], "_v": edge[1], **edge[2]} for edge in ebunch
        )
        self.network.edges.write_features(
            features, batch_size=_batch_size, counter=counter
        )

    def update_edges(self, ebunch: Iterable[EdgeTuple]) -> None:
        """Update edges as a batch.

        :param ebunch: Any iterable of edge tuples (u, v, d).

        """
        # FIXME: this doesn't actually work. Implement update / upsert
        #        logic for GeoPackage feature tables, then use that.
        self.network.edges.update_edges(ebunch)

add_edges_from(ebunch, _batch_size=1000, counter=None, **attr)

Equivalent to add_edges_from in networkx but with batched SQL writes.

Parameters:

Name Type Description Default
ebunch Iterable[EdgeTuple]

edge bunch, identical to nx ebunch_to_add.

 required
_batch_size int

Number of rows to commit to the database at a time.

 1000
**attr Any

Default attributes, identical to nx attr.

 {}
Source code in unweaver/graphs/digraphgpkg/digraphgpkg.py 
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
def add_edges_from(
    self,
    ebunch: Iterable[EdgeTuple],
    _batch_size: int = 1000,
    counter: Optional[ProgressBar] = None,
    **attr: Any
) -> None:
    """Equivalent to add_edges_from in networkx but with batched SQL writes.

    :param ebunch: edge bunch, identical to nx ebunch_to_add.
    :param _batch_size: Number of rows to commit to the database at a time.
    :param **attr: Default attributes, identical to nx attr.

    """
    if _batch_size < 2:
        # User has entered invalid number (negative, zero) or 1. Use
        # default behavior.
        super().add_edges_from(self, ebunch, **attr)
        return

    # TODO: length check on each edge
    features = (
        {"_u": edge[0], "_v": edge[1], **edge[2]} for edge in ebunch
    )
    self.network.edges.write_features(
        features, batch_size=_batch_size, counter=counter
    )

create_graph(path=None, **kwargs) classmethod

Create a new DiGraphGPKG (.gpkg) at a given path.

Parameters:

Name Type Description Default
path str

The path of the new GeoPackage (.gpkg).

 None
**kwargs Any

Any other keyword arguments to pass to the new DiGraphGPKG instance.

 {}

Returns:

Type Description
DiGraphGPKG

A new DiGraphGPKG instance.
Source code in unweaver/graphs/digraphgpkg/digraphgpkg.py 
60
61
62
63
64
65
66
67
68
69
70
71
@classmethod
def create_graph(cls, path: str = None, **kwargs: Any) -> DiGraphGPKG:
    """Create a new DiGraphGPKG (.gpkg) at a given path.

    :param path: The path of the new GeoPackage (.gpkg).
    :param **kwargs: Any other keyword arguments to pass to the new
    DiGraphGPKG instance.
    :returns: A new DiGraphGPKG instance.

    """
    network = GeoPackageNetwork(path)
    return DiGraphGPKG(network=network, **kwargs)

update_edges(ebunch)

Update edges as a batch.

Parameters:

Name Type Description Default
ebunch Iterable[EdgeTuple]

Any iterable of edge tuples (u, v, d).

 required
Source code in unweaver/graphs/digraphgpkg/digraphgpkg.py 
101
102
103
104
105
106
107
108
109
def update_edges(self, ebunch: Iterable[EdgeTuple]) -> None:
    """Update edges as a batch.

    :param ebunch: Any iterable of edge tuples (u, v, d).

    """
    # FIXME: this doesn't actually work. Implement update / upsert
    #        logic for GeoPackage feature tables, then use that.
    self.network.edges.update_edges(ebunch)

unweaver.graphs.DiGraphGPKGView

Bases: nx.DiGraph

An immutable, networkx-compatible directed graph view over a routable GeoPackage. Either the path to a .gpkg file or an existing instance of unweaver.network_adapters.geopackagenetwork.GeoPackageNetwork must be provided.

Parameters:

Name Type Description Default
incoming_graph_data Optional[nx.DiGraph]

Any class derived from networkx.DiGraph.

 None
path Optional[str]

A path to the GeoPackage file (.gpkg). If no file exists at this path, one will be created.

 None
network Optional[GeoPackageNetwork]

An existing GeoPackageNetwork instance.

 None
**attr Any

Any parameters to be attached as graph attributes.

 {}
Source code in unweaver/graphs/digraphgpkg/digraphgpkg_view.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 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
class DiGraphGPKGView(nx.DiGraph):
    """An immutable, `networkx`-compatible directed graph view over a
    routable GeoPackage. Either the path to a .gpkg file or an existing
    instance of
    unweaver.network_adapters.geopackagenetwork.GeoPackageNetwork must be
    provided.

    :param incoming_graph_data: Any class derived from `networkx.DiGraph`.
    :param path: A path to the GeoPackage file (.gpkg). If no file exists
    at this path, one will be created.
    :param network: An existing GeoPackageNetwork instance.
    :param **attr: Any parameters to be attached as graph attributes.

    """

    node_dict_factory = NodesView
    adjlist_outer_dict_factory = OuterSuccessorsView
    # In networkx, inner adjlist is only ever invoked without parameters in
    # order to assign new nodes or edges with no attr. Therefore, its
    # functionality can be accounted for elsewhere: via __getitem__ and
    # __setitem__ on the outer adjacency list.
    adjlist_inner_dict_factory = dict
    edge_attr_dict_factory = EdgeView

    def __init__(
        self,
        incoming_graph_data: Optional[nx.DiGraph] = None,
        path: Optional[str] = None,
        network: Optional[GeoPackageNetwork] = None,
        **attr: Any,
    ):
        # Path attr overrides sqlite attr
        if path:
            network = GeoPackageNetwork(path)
        elif network is None:
            raise ValueError("Path or network must be set")

        self.network = network

        # The factories of nx dict-likes need to be informed of the connection
        self.adjlist_inner_dict_factory = self.adjlist_inner_dict_factory

        # FIXME: should use a persistent table/container for .graph as well.
        self.graph = {}
        self._node = self.node_dict_factory(self.network)
        self._succ = self._adj = self.adjlist_outer_dict_factory(self.network)
        self._pred = OuterPredecessorsView(self.network)

        if incoming_graph_data is not None:
            nx.convert.to_networkx_graph(
                incoming_graph_data, create_using=self
            )
        self.graph.update(attr)

        # Set custom flag for read-only graph DBs
        self.mutable = False

    def size(self, weight: Optional[str] = None) -> int:
        """The 'size' of the directed graph, with the same behavior as
        `networkx.DiGraph`: if the weight parameter is set to a string, it's
        the sum of values for all edges that have that string as a key in their
        data dictionaries. If the weight parameter is unset, then it's the
        count of edges.

        :param weight: The string to use as an edge dictionary key to
        calculate a weighted sum over all edges.
        :returns: Either the number of edges (weight=None) or the sum of edge
        properties for the weight string.

        """
        if weight is None:
            return len(self.network.edges)
        else:
            return super().size(weight=weight)

    def iter_edges(self) -> Iterable[EdgeTuple]:
        """Roughly equivalent to the .edges interface, but much faster.

        :returns: generator of (u, v, d) similar to .edges, but where d is a
                  dictionary, not an Edge that syncs to database.

        """
        # FIXME: handle case where initializing with ddict data from query.
        # If implemented here (adding **d to the edge factory arguments), it
        # will always attempt to update the database on a per-read basis!
        return (
            (u, v, dict(self.edge_attr_dict_factory(self.network, u, v)))
            for u, v, d in self.network.edges.iter_edges()
        )

    def edges_dwithin(
        self, lon: float, lat: float, distance: float, sort: bool = False
    ) -> Iterable[EdgeTuple]:
        """Retrieve an iterable of edges within N meters of a
        latitude-longitude coordinate pair.

        :param lon: Longitude of the query point.
        :param lat: Latitude of the query point.
        :param distance: Search radius for the edge, can be thought of as
        defining the circle radius with which edges will be tested for
        intersection. Is in meters.
        :param sort: Whether to sort the iterable by distance such that the
        nearest edges are iterated over first.
        :returns: A generator of edge tuples, possibly sorted by distance
        (if the `sort` argument is set to True).

        """
        # TODO: document self.network.edges instead?
        return self.network.edges.dwithin_edges(lon, lat, distance, sort=sort)

    def to_in_memory(self) -> DiGraphGPKGView:
        """Copy the GeoPackage, itself an SQLite database, into an in-memory
        SQLite database. This may speed up queries and is useful if you want to
        create an ephemeral graph.

        :returns: A new instance of this class, backed by an in-memory SQLite
        database.

        """
        # TODO: make into 'copy' method instead, taking path as a parameter?
        db_id = uuid.uuid4()
        path = f"file:unweaver-{db_id}?mode=memory&cache=shared"
        new_network = self.network.copy(path)
        return self.__class__(network=new_network)

edges_dwithin(lon, lat, distance, sort=False)

Retrieve an iterable of edges within N meters of a latitude-longitude coordinate pair.

Parameters:

Name Type Description Default
lon float

Longitude of the query point.

 required
lat float

Latitude of the query point.

 required
distance float

Search radius for the edge, can be thought of as defining the circle radius with which edges will be tested for intersection. Is in meters.

 required
sort bool

Whether to sort the iterable by distance such that the nearest edges are iterated over first.

 False

Returns:

Type Description
Iterable[EdgeTuple]

A generator of edge tuples, possibly sorted by distance (if the sort argument is set to True).
Source code in unweaver/graphs/digraphgpkg/digraphgpkg_view.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def edges_dwithin(
    self, lon: float, lat: float, distance: float, sort: bool = False
) -> Iterable[EdgeTuple]:
    """Retrieve an iterable of edges within N meters of a
    latitude-longitude coordinate pair.

    :param lon: Longitude of the query point.
    :param lat: Latitude of the query point.
    :param distance: Search radius for the edge, can be thought of as
    defining the circle radius with which edges will be tested for
    intersection. Is in meters.
    :param sort: Whether to sort the iterable by distance such that the
    nearest edges are iterated over first.
    :returns: A generator of edge tuples, possibly sorted by distance
    (if the `sort` argument is set to True).

    """
    # TODO: document self.network.edges instead?
    return self.network.edges.dwithin_edges(lon, lat, distance, sort=sort)

iter_edges()

Roughly equivalent to the .edges interface, but much faster.

Returns:

Type Description
Iterable[EdgeTuple]

generator of (u, v, d) similar to .edges, but where d is a dictionary, not an Edge that syncs to database.
Source code in unweaver/graphs/digraphgpkg/digraphgpkg_view.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def iter_edges(self) -> Iterable[EdgeTuple]:
    """Roughly equivalent to the .edges interface, but much faster.

    :returns: generator of (u, v, d) similar to .edges, but where d is a
              dictionary, not an Edge that syncs to database.

    """
    # FIXME: handle case where initializing with ddict data from query.
    # If implemented here (adding **d to the edge factory arguments), it
    # will always attempt to update the database on a per-read basis!
    return (
        (u, v, dict(self.edge_attr_dict_factory(self.network, u, v)))
        for u, v, d in self.network.edges.iter_edges()
    )

size(weight=None)

The 'size' of the directed graph, with the same behavior as networkx.DiGraph: if the weight parameter is set to a string, it's the sum of values for all edges that have that string as a key in their data dictionaries. If the weight parameter is unset, then it's the count of edges.

Parameters:

Name Type Description Default
weight Optional[str]

The string to use as an edge dictionary key to calculate a weighted sum over all edges.

 None

Returns:

Type Description
int

Either the number of edges (weight=None) or the sum of edge properties for the weight string.
Source code in unweaver/graphs/digraphgpkg/digraphgpkg_view.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def size(self, weight: Optional[str] = None) -> int:
    """The 'size' of the directed graph, with the same behavior as
    `networkx.DiGraph`: if the weight parameter is set to a string, it's
    the sum of values for all edges that have that string as a key in their
    data dictionaries. If the weight parameter is unset, then it's the
    count of edges.

    :param weight: The string to use as an edge dictionary key to
    calculate a weighted sum over all edges.
    :returns: Either the number of edges (weight=None) or the sum of edge
    properties for the weight string.

    """
    if weight is None:
        return len(self.network.edges)
    else:
        return super().size(weight=weight)

to_in_memory()

Copy the GeoPackage, itself an SQLite database, into an in-memory SQLite database. This may speed up queries and is useful if you want to create an ephemeral graph.

Returns:

Type Description
DiGraphGPKGView

A new instance of this class, backed by an in-memory SQLite database.
Source code in unweaver/graphs/digraphgpkg/digraphgpkg_view.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def to_in_memory(self) -> DiGraphGPKGView:
    """Copy the GeoPackage, itself an SQLite database, into an in-memory
    SQLite database. This may speed up queries and is useful if you want to
    create an ephemeral graph.

    :returns: A new instance of this class, backed by an in-memory SQLite
    database.

    """
    # TODO: make into 'copy' method instead, taking path as a parameter?
    db_id = uuid.uuid4()
    path = f"file:unweaver-{db_id}?mode=memory&cache=shared"
    new_network = self.network.copy(path)
    return self.__class__(network=new_network)

unweaver.graphs.AugmentedDiGraphGPKGView

Bases: nx.DiGraph

A wrapper over DiGraphGPKGView that allows for overlaying an in-memory DiGraph but with a seamless interface. When querying the graph, such as asking for a particular edge based on a (u, d) pair (G[u][v]), an AugmentedDiGraphGPKGView will first attempt to retrieve this edge from the in-memory DiGraph, then check the DiGraphGPKGView.

This wrapper is particularly useful for adding temporary nodes and edges for the purposes of running a graph analysis algorithm. For example, Unweaver uses AugmentedDiGraphGPKGView when it's necessary to start "part-way along" an edge for a shortest-path query using Dijkstra's algorithm. There is often no on-graph node near the physical locationfrom which someone wants to begin searching for shortest paths, so Unweaver creates two new temporary edges starting from the nearest point on the nearest edge, connecting them to the on-graph nodes for that edge, and creates an AugmentedDiGraphGPKGView using those temporary edges.

Parameters:

Name Type Description Default
G DiGraphGPKGView

A DiGraphGPKGView, usually the main graph data.

 required
G_overlay nx.DiGraph

A dict-of-dict-of-dicts (or networkx.DiGraph) to overlay.

 required
Source code in unweaver/graphs/augmented.py 
 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
class AugmentedDiGraphGPKGView(nx.DiGraph):
    """A wrapper over DiGraphGPKGView that allows for overlaying an in-memory
    DiGraph but with a seamless interface. When querying the graph, such as
    asking for a particular edge based on a (u, d) pair (G[u][v]), an
    AugmentedDiGraphGPKGView will first attempt to retrieve this edge from
    the in-memory DiGraph, then check the DiGraphGPKGView.

    This wrapper is particularly useful for adding temporary nodes and edges
    for the purposes of running a graph analysis algorithm. For example,
    Unweaver uses AugmentedDiGraphGPKGView when it's necessary to start
    "part-way along" an edge for a shortest-path query using Dijkstra's
    algorithm. There is often no on-graph node near the physical locationfrom
    which someone wants to begin searching for shortest paths, so Unweaver
    creates two new temporary edges starting from the nearest point on the
    nearest edge, connecting them to the on-graph nodes for that edge, and
    creates an AugmentedDiGraphGPKGView using those temporary edges.

    :param G: A DiGraphGPKGView, usually the main graph data.
    :param G_overlay: A dict-of-dict-of-dicts (or networkx.DiGraph) to overlay.

    """

    node_dict_factory: Callable = AugmentedNodesView
    adjlist_outer_dict_factory: Callable = AugmentedOuterSuccessorsView
    # In networkx, inner adjlist is only ever invoked without parameters in
    # order to assign new nodes or edges with no attr. Therefore, its
    # functionality can be accounted for elsewhere: via __getitem__ and
    # __setitem__ on the outer adjacency list.
    adjlist_inner_dict_factory = dict
    edge_attr_dict_factory: Callable = dict

    def __init__(self, G: DiGraphGPKGView, G_overlay: nx.DiGraph):
        # The factories of nx dict-likes need to be informed of the connection
        setattr(
            self,
            "node_dict_factory",
            partial(self.node_dict_factory, _G=G, _G_overlay=G_overlay),
        )
        setattr(
            self,
            "adjlist_outer_dict_factory",
            partial(
                self.adjlist_outer_dict_factory, _G=G, _G_overlay=G_overlay
            ),
        )
        # Not 'partial' on this?
        setattr(
            self, "adjlist_inner_dict_factory", self.adjlist_inner_dict_factory
        )
        setattr(
            self,
            "edge_attr_dict_factory",
            partial(self.edge_attr_dict_factory, _G=G, _G_overlay=G_overlay),
        )

        self.graph: Dict[Any, Any] = {}
        setattr(self, "_node", self.node_dict_factory())
        self._succ = self._adj = self.adjlist_outer_dict_factory()
        self._pred = AugmentedOuterPredecessorsView(_G=G, _G_overlay=G_overlay)

        self.network = G.network

    @classmethod
    def prepare_augmented(
        cls: Type[T], G: DiGraphGPKGView, candidate: ProjectedNode
    ) -> T:
        """Create an AugmentedDiGraphGPKGView based on a DiGraphGPKGView and
        a start point candidatee (a ProjectedNode class instance). This will
        embed an overlay node and two edges.

        :param G: The base DiGraphGPKGView.
        :param candidate: The potential start point candidate.

        """
        temp_edges = []
        if candidate.edges_in:
            for e in candidate.edges_in:
                temp_edges.append(e)
        if candidate.edges_out:
            for e in candidate.edges_out:
                temp_edges.append(e)

        G_overlay = nx.DiGraph()
        if temp_edges:
            G_overlay.add_edges_from(temp_edges)
            for u, v, d in temp_edges:
                # TODO: 'add_edges_from' should automatically add geometry info
                # to nodes. This is a workaround for the fact that it doesn't.
                G_overlay.nodes[u][G.network.nodes.geom_column] = Point(
                    d[G.network.edges.geom_column]["coordinates"][0]
                )
                G_overlay.nodes[v][G.network.nodes.geom_column] = Point(
                    d[G.network.edges.geom_column]["coordinates"][-1]
                )
        G_augmented = AugmentedDiGraphGPKGView(G=G, G_overlay=G_overlay)

        return G_augmented

prepare_augmented(G, candidate) classmethod

Create an AugmentedDiGraphGPKGView based on a DiGraphGPKGView and a start point candidatee (a ProjectedNode class instance). This will embed an overlay node and two edges.

Parameters:

Name Type Description Default
G DiGraphGPKGView

The base DiGraphGPKGView.

 required
candidate ProjectedNode

The potential start point candidate.

 required
Source code in unweaver/graphs/augmented.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
182
@classmethod
def prepare_augmented(
    cls: Type[T], G: DiGraphGPKGView, candidate: ProjectedNode
) -> T:
    """Create an AugmentedDiGraphGPKGView based on a DiGraphGPKGView and
    a start point candidatee (a ProjectedNode class instance). This will
    embed an overlay node and two edges.

    :param G: The base DiGraphGPKGView.
    :param candidate: The potential start point candidate.

    """
    temp_edges = []
    if candidate.edges_in:
        for e in candidate.edges_in:
            temp_edges.append(e)
    if candidate.edges_out:
        for e in candidate.edges_out:
            temp_edges.append(e)

    G_overlay = nx.DiGraph()
    if temp_edges:
        G_overlay.add_edges_from(temp_edges)
        for u, v, d in temp_edges:
            # TODO: 'add_edges_from' should automatically add geometry info
            # to nodes. This is a workaround for the fact that it doesn't.
            G_overlay.nodes[u][G.network.nodes.geom_column] = Point(
                d[G.network.edges.geom_column]["coordinates"][0]
            )
            G_overlay.nodes[v][G.network.nodes.geom_column] = Point(
                d[G.network.edges.geom_column]["coordinates"][-1]
            )
    G_augmented = AugmentedDiGraphGPKGView(G=G, G_overlay=G_overlay)

    return G_augmented