ENH: add get_nearest_node

.github/workflows/test_user_guide.yaml

@@ -27,7 +27,7 @@ jobs:
       - name: setup micromamba
         uses: mamba-org/setup-micromamba@v1
         with:
-          environment-file: ci/envs/312-latest.yaml
+          environment-file: ci/envs/312-dev.yaml
 
       - name: Install momepy
         run: pip install .

momepy/functional/_elements.py

@@ -21,6 +21,7 @@
     "enclosed_tessellation",
     "verify_tessellation",
     "get_nearest_street",
+    "get_nearest_node",
     "generate_blocks",
     "buffered_limit",
 ]
@@ -368,6 +369,57 @@ def get_nearest_street(
     return ids
 
 
+def get_nearest_node(
+    buildings: GeoSeries | GeoDataFrame,
+    nodes: GeoDataFrame,
+    edges: GeoDataFrame,
+    nearest_edge: Series,
+) -> Series:
+    """Identify the nearest node for each building.
+
+    Snap each building to the closest street network node on the closest network edge.
+    This assumes that the nearest street network edge has already been identified using
+    :func:`get_nearest_street`.
+
+    The ``edges`` and ``nodes`` GeoDataFrames are expected to be an outcome of
+    :func:`momepy.nx_to_gdf` or match its structure with ``["node_start", "node_end"]``
+    columns and their meaning.
+
+
+    Parameters
+    ----------
+    buildings : GeoSeries | GeoDataFrame
+        GeoSeries or GeoDataFrame of buildings
+    nodes : GeoDataFrame
+        A GeoDataFrame containing street nodes.
+    edges : GeoDataFrame
+        A GeoDataFrame containing street edges with ``["node_start", "node_end"]``
+        columns marking start and end nodes of each edge. These are the default
+        outcome of :func:`momepy.nx_to_gdf`.
+    nearest_edge : Series
+        A Series aligned with ``buildings`` containing the information on the nearest
+        street edge. Matches the outcome of :func:`get_nearest_street`.
+
+    Returns
+    -------
+    Series
+    """
+    # treat possibly missing edge index
+    a = np.empty(len(buildings))
+    na_mask = np.isnan(nearest_edge)
+    a[na_mask] = np.nan
+
+    streets = edges.loc[nearest_edge[~na_mask]]
+    starts = nodes.loc[streets["node_start"]].distance(buildings[~na_mask], align=False)
+    ends = nodes.loc[streets["node_end"]].distance(buildings[~na_mask], align=False)
+    mask = starts.values > ends.values
+    r = starts.index.to_numpy()
+    r[mask] = ends.index[mask]
+
+    a[~na_mask] = r
+    return pd.Series(a, index=buildings.index)
+
+
 def generate_blocks(
     tessellation: GeoDataFrame, edges: GeoDataFrame, buildings: GeoDataFrame
 ) -> tuple[Series, Series, Series]:

momepy/functional/tests/test_elements.py

@@ -223,6 +223,53 @@ def test_get_nearest_street(self):
         nearest = mm.get_nearest_street(self.df_buildings, streets, 10)
         assert (nearest == None).sum() == 137  # noqa: E711
 
+    def test_get_nearest_node(self):
+        nodes, edges = mm.nx_to_gdf(mm.gdf_to_nx(self.df_streets))
+        edge_index = mm.get_nearest_street(self.df_buildings, edges)
+
+        node_index = mm.get_nearest_node(self.df_buildings, nodes, edges, edge_index)
+
+        assert len(node_index) == len(self.df_buildings)
+        assert_index_equal(node_index.index, self.df_buildings.index)
+        expected = np.array(
+            [
+                0.0,
+                1.0,
+                2.0,
+                3.0,
+                4.0,
+                6.0,
+                9.0,
+                11.0,
+                14.0,
+                15.0,
+                16.0,
+                20.0,
+                22.0,
+                25.0,
+            ]
+        )
+        expected_counts = np.array([9, 31, 12, 10, 11, 2, 23, 8, 2, 8, 3, 6, 12, 7])
+        unique, counts = np.unique(node_index, return_counts=True)
+        np.testing.assert_array_equal(unique, expected)
+        np.testing.assert_array_equal(counts, expected_counts)
+
+    def test_get_nearest_node_missing(self):
+        nodes, edges = mm.nx_to_gdf(mm.gdf_to_nx(self.df_streets))
+        edge_index = mm.get_nearest_street(self.df_buildings, edges, max_distance=20)
+
+        node_index = mm.get_nearest_node(self.df_buildings, nodes, edges, edge_index)
+
+        assert len(node_index) == len(self.df_buildings)
+        assert_index_equal(node_index.index, self.df_buildings.index)
+        expected = np.array(
+            [1.0, 2.0, 3.0, 4.0, 9.0, 11.0, 14.0, 15.0, 16.0, 20.0, 22.0, 25.0, np.nan]
+        )
+        expected_counts = np.array([14, 8, 10, 4, 14, 8, 2, 7, 2, 5, 9, 4, 57])
+        unique, counts = np.unique(node_index, return_counts=True)
+        np.testing.assert_array_equal(unique, expected)
+        np.testing.assert_array_equal(counts, expected_counts)
+
     def test_buffered_limit(self):
         limit = mm.buffered_limit(self.df_buildings, 50)
         assert limit.geom_type == "Polygon"