dedicated minimal tests for continuity.py (#50)

* dedicated tests for continuity.py

* full coverage continuity testing

* fix typo in continuity() docstring

* add CES description in get_stroke_info() docstring

sgeop/continuity.py

@@ -4,19 +4,27 @@
 
 def continuity(
     roads: geopandas.GeoDataFrame, angle_threshold: float = 120
-) -> geopandas.GeoDataFrame:
+) -> tuple[geopandas.GeoDataFrame, momepy.COINS]:
     """Assign COINS-based information to roads.
 
     Parameters
     ----------
     roads :  geopandas.GeoDataFrame
         Road network.
+    angle_threshold : float = 120
+        See the ``angle_threshold`` keyword argument in ``momepy.COINS()``.
 
     Returns
     -------
-    geopandas.GeoDataFrame
-        The input ``roads`` with additional columns where the original
-        index may be reset (see ``dedup`` keyword argument).
+    roads : geopandas.GeoDataFrame
+        The input ``roads`` with additional columns describing COINS information.
+    coins : momepy.COINS
+        **This is not used in production.**
+
+    Notes
+    -----
+    The returned ``coins`` object is not used in production, but is
+    very helpful in testing & debugging. See gh:sgeop#49.
     """
     roads = roads.copy()
 
@@ -36,24 +44,35 @@ def continuity(
     return roads, coins
 
 
-def get_stroke_info(artifacts, roads):
-    """Generate information about strokes within the artifacts
+def get_stroke_info(
+    artifacts: geopandas.GeoSeries | geopandas.GeoDataFrame,
+    roads: geopandas.GeoSeries | geopandas.GeoDataFrame,
+) -> tuple[list[int]]:
+    """Generate information about strokes within ``artifacts`` and the
+    resulting lists can be assigned as columns to ``artifacts``. Classifies
+    the strokes within the CES typology.
 
-    Resulting lists can be assigned as columns to ``artifacts``.
+        * 'continuing' strokes - continues before and after artifact.
+        * 'ending' strokes - continues only at one end.
+        * 'single' strokes - does not continue.
 
     Parameters
     ----------
-    artifacts : GeoDataFrame | GeoSeries
-        Polygons representing the artifacts
-    roads : GeoDataFrame | GeoSeries
-        LineStrings representing the road network
+    artifacts : GeoSeries | GeoDataFrame
+        Polygons representing the artifacts.
+    roads : GeoSeries | GeoDataFrame
+        LineStrings representing the road network.
 
     Returns
     -------
-    stroke_count : list
-    C_count : list
-    E_count : list
-    S_count : list
+    strokes : list[int]
+        All strokes counts.
+    c_ : list[int]
+        Counts for 'continuing' strokes - continues before and after artifact.
+    e_ : list[int]
+        Counts for 'ending' strokes - continues only at one end.
+    s_ : list[int]
+        Counts for 'single' strokes - does not continue.
     """
     strokes = []
     c_ = []
@@ -63,31 +82,25 @@ def get_stroke_info(artifacts, roads):
         singles = 0
         ends = 0
         edges = roads.iloc[roads.sindex.query(geom, predicate="covers")]
-        if (  # roundabout special case
-            edges.coins_group.nunique() == 1
-            and edges.shape[0] == edges.coins_count.iloc[0]
-        ):
+        ecg = edges.coins_group
+        if ecg.nunique() == 1 and edges.shape[0] == edges.coins_count.iloc[0]:
+            # roundabout special case
             singles = 1
             mains = 0
         else:
             all_ends = edges[edges.coins_end]
-            mains = edges[
-                ~edges.coins_group.isin(all_ends.coins_group)
-            ].coins_group.nunique()
-
+            ae_cg = all_ends.coins_group
+            mains = edges[~ecg.isin(ae_cg)].coins_group.nunique()
             visited = []
-            for coins_count, group in zip(
-                all_ends.coins_count, all_ends.coins_group, strict=True
-            ):
-                if (group not in visited) and (
-                    coins_count == (edges.coins_group == group).sum()
-                ):
-                    singles += 1
-                    visited.append(group)
-                elif group not in visited:
-                    ends += 1
-                    # do not add to visited as they may be disjoint within the artifact
-        strokes.append(edges.coins_group.nunique())
+            for coins_count, group in zip(all_ends.coins_count, ae_cg, strict=True):
+                if group not in visited:
+                    if coins_count == (ecg == group).sum():
+                        singles += 1
+                        visited.append(group)
+                    else:
+                        # do not add to visited -- may be disjoint within the artifact
+                        ends += 1
+        strokes.append(ecg.nunique())
         c_.append(mains)
         e_.append(ends)
         s_.append(singles)

sgeop/tests/test_continuity.py

@@ -0,0 +1,110 @@
+import io
+
+import geopandas.testing
+import momepy
+import pandas
+import pytest
+import shapely
+
+import sgeop
+
+
+@pytest.fixture
+def roads() -> geopandas.GeoDataFrame:
+    """Toy set of 'roads' for testing only."""
+    inita = 2
+    final = 8
+    grid = list(range(inita, final))
+    vert_points = list(zip(grid[:-1], grid[1:], strict=True))
+    hori_points = [(j, i) for i, j in vert_points]
+    vert_lines = [
+        shapely.LineString(i)
+        for i in list(zip(hori_points[:-1], vert_points[1:], strict=True))
+    ]
+    hori_lines = [
+        shapely.LineString(i)
+        for i in list(zip(vert_points[:-1], hori_points[1:], strict=True))
+    ]
+    return geopandas.GeoDataFrame(
+        geometry=(
+            vert_lines
+            + hori_lines
+            + [
+                shapely.LineString(((4, 5), (3, 6))),
+                shapely.LineString(((3, 6), (4, 4))),
+                shapely.LineString(((6, 3), (5, 4))),
+                shapely.LineString(((3, 6), (3, 4))),
+                shapely.LineString(((5, 5), (6, 6))),
+                shapely.LineString(((6, 7), (7, 7))),
+                shapely.LineString(((7, 6), (7, 7))),
+            ]
+        )
+    )
+
+
+def test_continuity(roads):
+    observed_continuity, observed_coins = sgeop.continuity.continuity(roads)
+
+    assert isinstance(observed_continuity, geopandas.GeoDataFrame)
+    known_continuity = (
+        geopandas.GeoDataFrame(
+            pandas.read_csv(
+                io.StringIO(
+                    "geometry	coins_group	coins_end	coins_len	coins_count\n"
+                    "LINESTRING (3 2, 3 4)	0	True	4.0	2\n"
+                    "LINESTRING (4 3, 4 5)	1	True	4.0	2\n"
+                    "LINESTRING (5 4, 5 6)	2	True	11.414213562373096	7\n"
+                    "LINESTRING (6 5, 6 7)	2	False	11.414213562373096	7\n"
+                    "LINESTRING (2 3, 4 3)	1	True	4.0	2\n"
+                    "LINESTRING (3 4, 5 4)	3	True	3.414213562373095	2\n"
+                    "LINESTRING (4 5, 6 5)	2	False	11.414213562373096	7\n"
+                    "LINESTRING (5 6, 7 6)	2	False	11.414213562373096	7\n"
+                    "LINESTRING (4 5, 3 6)	2	True	11.414213562373096	7\n"
+                    "LINESTRING (3 6, 4 4)	4	True	2.23606797749979	1\n"
+                    "LINESTRING (6 3, 5 4)	3	True	3.414213562373095	2\n"
+                    "LINESTRING (3 6, 3 4)	0	True	4.0	2\n"
+                    "LINESTRING (5 5, 6 6)	5	True	1.4142135623730951	1\n"
+                    "LINESTRING (6 7, 7 7)	2	False	11.414213562373096	7\n"
+                    "LINESTRING (7 6, 7 7)	2	False	11.414213562373096	7\n"
+                ),
+                sep="\t",
+            )
+        )
+        .pipe(lambda df: df.assign(**{"geometry": shapely.from_wkt(df["geometry"])}))
+        .set_geometry("geometry")
+    )
+    geopandas.testing.assert_geodataframe_equal(observed_continuity, known_continuity)
+
+    assert isinstance(observed_coins, momepy.COINS)
+    assert observed_coins.already_merged
+    assert observed_coins.merging_list == [
+        [0, 11],
+        [1, 4],
+        [2, 3, 6, 7, 8, 13, 14],
+        [5, 10],
+        [9],
+        [12],
+    ]
+    assert len(observed_coins.angle_pairs) == 36
+
+
+def test_get_stroke_info(roads):
+    known_strokes = [0, 0, 2, 1, 1, 1, 2]
+    known_c_ = [0, 0, 0, 0, 0, 1, 0]
+    known_e_ = [0, 0, 1, 0, 0, 0, 1]
+    known_s_ = [0, 0, 1, 1, 1, 0, 1]
+
+    observed = sgeop.continuity.get_stroke_info(
+        sgeop.artifacts.get_artifacts(roads, threshold=1)[0],
+        sgeop.continuity.continuity(roads.copy())[0],
+    )
+
+    observed_strokes = observed[0]
+    observed_c_ = observed[1]
+    observed_e_ = observed[2]
+    observed_s_ = observed[3]
+
+    assert observed_strokes == known_strokes
+    assert observed_c_ == known_c_
+    assert observed_e_ == known_e_
+    assert observed_s_ == known_s_