dedicated minimal tests -- nodes.induce_nodes()

sgeop/nodes.py

@@ -24,7 +24,7 @@ def split(
         Line geometries to be split with ``split_points``.
     crs : str | pyproj.CRS
         Anything accepted by ``pyproj.CRS``.
-    eps : float
+    eps : float = 1e-4
         Tolerance epsilon for point snapping.
 
     Returns
@@ -287,28 +287,39 @@ def fix_topology(
     return remove_false_nodes(roads_w_nodes, **kwargs)
 
 
-def induce_nodes(roads, eps=1e-4):
-    """
-    adding potentially missing nodes
-    on intersections of individual LineString endpoints with the remaining network. The
-    idea behind is that if a line ends on an intersection with another, there should be
-    a node on both of them.
+def induce_nodes(roads: gpd.GeoDataFrame, eps: float = 1e-4) -> gpd.GeoDataFrame:
+    """Adding potentially missing nodes on intersections of individual LineString
+    endpoints with the remaining network. The idea behind is that if a line ends
+    on an intersection with another, there should be a node on both of them.
+
+    Parameters
+    ----------
+    roads : geopandas.GeoDataFrame
+        Input LineString geometries.
+    eps : float = 1e-4
+        Tolerance epsilon for point snapping passed into ``nodes.split()``.
+
+    Returns
+    -------
+    geopandas.GeoDataFrame
+        Updated ``roads`` with (potentially) added nodes.
     """
     nodes_w_degree = momepy.nx_to_gdf(
         momepy.node_degree(momepy.gdf_to_nx(roads)), lines=False
     )
     nodes_ix, roads_ix = roads.sindex.query(
         nodes_w_degree.geometry, predicate="dwithin", distance=1e-4
     )
-    intersects = sparse.coo_array(
-        ([True] * len(nodes_ix), (nodes_ix, roads_ix)),
-        shape=(len(nodes_w_degree), len(roads)),
-        dtype=np.bool_,
-    )
+
+    coo_vals = ([True] * len(nodes_ix), (nodes_ix, roads_ix))
+    coo_shape = (len(nodes_w_degree), len(roads))
+    intersects = sparse.coo_array(coo_vals, shape=coo_shape, dtype=np.bool_)
+
     nodes_w_degree["expected_degree"] = intersects.sum(axis=1)
-    nodes_to_induce = nodes_w_degree[
-        nodes_w_degree.degree != nodes_w_degree.expected_degree
-    ]
+
+    degree_mistmatch = nodes_w_degree.degree != nodes_w_degree.expected_degree
+    nodes_to_induce = nodes_w_degree[degree_mistmatch]
+
     return split(nodes_to_induce.geometry, roads, roads.crs, eps=eps)
 
 

sgeop/tests/test_nodes.py

@@ -436,3 +436,35 @@ def test_momepy_suite(self):
         known = df_streets.drop([4, 7, 17, 22]).reset_index(drop=True)
         observed = sgeop.nodes.remove_false_nodes(known).reset_index(drop=True)
         geopandas.testing.assert_geodataframe_equal(observed, known)
+
+
+class TestInduceNodes:
+    def setup_method(self):
+        p10 = shapely.Point(1, 0)
+        p20 = shapely.Point(2, 0)
+        p30 = shapely.Point(3, 0)
+        p21 = shapely.Point(2, 1)
+        p201 = shapely.Point(2, 0.1)
+
+        self.line1020 = shapely.LineString((p10, p20))
+        self.line2030 = shapely.LineString((p20, p30))
+        self.line1030 = shapely.LineString((p10, p30))
+        self.line2021 = shapely.LineString((p20, p21))
+        self.line20121 = shapely.LineString((p201, p21))
+
+    def test_induced(self):
+        known = geopandas.GeoDataFrame(
+            {
+                "geometry": [self.line2021, self.line1020, self.line2030],
+                "_status": [numpy.nan, "changed", "changed"],
+            }
+        )
+        frame = geopandas.GeoDataFrame(geometry=[self.line1030, self.line2021])
+        observed = sgeop.nodes.induce_nodes(frame)
+        geopandas.testing.assert_geodataframe_equal(observed, known)
+
+    def test_not_induced(self):
+        known = geopandas.GeoDataFrame(geometry=[self.line1030, self.line20121])
+        frame = geopandas.GeoDataFrame(geometry=[self.line1030, self.line20121])
+        observed = sgeop.nodes.induce_nodes(frame)
+        geopandas.testing.assert_geodataframe_equal(observed, known)