handles levels and tunnels in cleaning

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "cityseer"
-version = '4.16.13'
+version = '4.16.14'
 description = "Computational tools for network-based pedestrian-scale urban analysis"
 readme = "README.md"
 requires-python = ">=3.10, <3.14"

pysrc/cityseer/tools/graphs.py

@@ -233,9 +233,7 @@ def nx_remove_filler_nodes(nx_multigraph: MultiGraph) -> MultiGraph:
 
 
 def nx_remove_dangling_nodes(
-    nx_multigraph: MultiGraph,
-    despine: int = 15,
-    remove_disconnected: int = 100,
+    nx_multigraph: MultiGraph, despine: int = 15, remove_disconnected: int = 100, remove_deadend_tunnels: bool = True
 ) -> MultiGraph:
     """
     Remove disconnected components and optionally removes short dead-end street stubs.
@@ -250,6 +248,8 @@ def nx_remove_dangling_nodes(
     remove_disconnected: int
         Remove disconnected components with fewer nodes than specified by this parameter. Defaults to 100. Set to 0 to
         keep all disconnected components.
+    remove_deadend_tunnels: bool
+        Remove dead-end tunnels. Default of True.
 
     Returns
     -------
@@ -289,36 +289,43 @@ def nx_remove_dangling_nodes(
             if nx.degree(g_multi_copy, nd_key) == 1:
                 # only a single neighbour, so index-in directly and update at key = 0
                 nb_nd_key: NodeKey = list(nx.neighbors(g_multi_copy, nd_key))[0]
-                if g_multi_copy[nd_key][nb_nd_key][0]["geom"].length <= despine:
+                edge_data = g_multi_copy[nd_key][nb_nd_key][0]
+                if (
+                    remove_deadend_tunnels is True and "is_tunnel" in edge_data and edge_data["is_tunnel"] is True
+                ) or edge_data["geom"].length <= despine:
                     remove_nodes.append(nd_key)
         g_multi_copy.remove_nodes_from(remove_nodes)
 
-    g_multi_copy = nx_remove_filler_nodes(g_multi_copy)
-
     return g_multi_copy
 
 
 def _extract_tags_to_set(
     tags_list: list[str] | None = None,
-) -> set[str]:
+) -> set[str | int]:
     """Converts a `list` of `str` tags to a `set` of small caps `str`."""
     tags = set()
     if tags_list is not None:
         if not isinstance(tags_list, list | set | tuple):
             raise ValueError(f"Tags should be provided as a `list` of `str` instead of {type(tags_list)}.")
-        tags_list = [t.strip().lower() for t in tags_list if t not in ["", " ", None]]
+        cleaned_tags_list = []
+        for t in tags_list:
+            if isinstance(t, str):
+                if t not in ["", " ", None]:
+                    cleaned_tags_list.append(t.strip().lower())
+            else:
+                cleaned_tags_list.append(t)
         tags.update(tags_list)
     return tags
 
 
-def _tags_from_edge_key(edge_data: EdgeData, edge_key: str) -> set[str]:
+def _tags_from_edge_key(edge_data: EdgeData, edge_key: str) -> set[str | int]:
     """Fetches tags from a given edge key and returns as `set` of `str`."""
     if edge_key in edge_data:
         return _extract_tags_to_set(edge_data[edge_key])
     return set()
 
 
-def _gather_nb_tags(nx_multigraph: MultiGraph, nd_key: NodeKey, edge_key: str) -> set[str]:
+def _gather_nb_tags(nx_multigraph: MultiGraph, nd_key: NodeKey, edge_key: str) -> set[str | int]:
     """Fetches tags from edges neighbouring a node and returns as a `set` of `str`."""
     nb_tags = set()
     for nb_nd_key in nx_multigraph.neighbors(nd_key):
@@ -327,14 +334,14 @@ def _gather_nb_tags(nx_multigraph: MultiGraph, nd_key: NodeKey, edge_key: str) -
     return nb_tags
 
 
-def _gather_name_tags(edge_data: EdgeData) -> set[str]:
+def _gather_name_tags(edge_data: EdgeData) -> set[str | int]:
     """Fetches `names` and `routes` tags from the provided edge and returns as a `set` of `str`."""
     names_tags = _tags_from_edge_key(edge_data, "names")
     routes_tags = _tags_from_edge_key(edge_data, "routes")
     return names_tags.union(routes_tags)
 
 
-def _gather_nb_name_tags(nx_multigraph: MultiGraph, nd_key: NodeKey) -> set[str]:
+def _gather_nb_name_tags(nx_multigraph: MultiGraph, nd_key: NodeKey) -> set[str | int]:
     """Fetches `names` and `routes` tags from edges neighbouring a node and returns as a `set` of `str`."""
     names_tags = _gather_nb_tags(nx_multigraph, nd_key, "names")
     routes_tags = _gather_nb_tags(nx_multigraph, nd_key, "routes")
@@ -515,7 +522,10 @@ def nx_snap_endpoints(nx_multigraph: MultiGraph) -> MultiGraph:
 
 
 def nx_iron_edges(
-    nx_multigraph: MultiGraph, simplify_by_angle: int = 100, min_self_loop_length: int = 100
+    nx_multigraph: MultiGraph,
+    simplify_by_angle: int = 100,
+    min_self_loop_length: int = 100,
+    max_foot_tunnel_length: int = 50,
 ) -> MultiGraph:
     """
     Simplifies edges.
@@ -529,6 +539,8 @@ def nx_iron_edges(
         The maximum angle to permit for a given edge. Angles greater than this will be reduced.
     min_self_loop_length: int
         Maximum self loop length to permit for a given edge.
+    max_foot_tunnel_length: int
+        Maximum tunnel length to permit for non motorised edges.
 
     Returns
     -------
@@ -549,6 +561,26 @@ def nx_iron_edges(
         if start_nd_key == end_nd_key and edge_geom.length < min_self_loop_length:
             remove_edges.append((start_nd_key, end_nd_key, edge_idx))
             continue
+        # drop long foot tunnels
+        if (
+            "is_tunnel" in edge_data
+            and edge_data["is_tunnel"] is True
+            and edge_data["geom"].length > max_foot_tunnel_length
+        ):
+            hwy_tags = _tags_from_edge_key(edge_data, "highways")
+            if not hwy_tags.intersection(
+                [
+                    "trunk",
+                    "primary",
+                    "secondary",
+                    "tertiary",
+                    "residential",
+                    "service",
+                ]
+            ):
+                remove_edges.append((start_nd_key, end_nd_key, edge_idx))
+                continue
+        # simplify
         line_coords = simplify_line_by_angle(edge_geom.coords, simplify_by_angle)
         g_multi_copy[start_nd_key][end_nd_key][edge_idx]["geom"] = geometry.LineString(line_coords)
     g_multi_copy.remove_edges_from(remove_edges)
@@ -885,8 +917,9 @@ def recursive_squash(
         y: float,
         node_group: set[NodeKey],
         processed_nodes: set[NodeKey],
-        _hwy_tags: set[str],
-        _name_tags: set[str],
+        _hwy_tags: set,
+        _name_tags: set,
+        _levels_tags: set,
         recursive: bool = False,
     ) -> set[NodeKey]:
         # keep track of which nodes have been processed as part of recursion
@@ -908,6 +941,11 @@ def recursive_squash(
                     continue
                 if neighbour_policy == "direct" and j_nd_key not in neighbours:
                     continue
+            # levels
+            if _levels_tags:
+                _nb_level_tags = _gather_nb_tags(nx_multigraph, j_nd_key, "levels")
+                if not _levels_tags.intersection(_nb_level_tags):
+                    continue
             # hwy tags
             if osm_hwy_target_tags:
                 _nb_hwy_tags = _gather_nb_tags(nx_multigraph, j_nd_key, "highways")
@@ -931,6 +969,7 @@ def recursive_squash(
                     processed_nodes,
                     _hwy_tags,
                     _name_tags,
+                    _levels_tags,
                     recursive=crawl,
                 )
         return node_group
@@ -948,6 +987,8 @@ def recursive_squash(
         nb_hwy_tags = _gather_nb_tags(nx_multigraph, nd_key, "highways")
         if osm_hwy_target_tags and not hwy_tags.intersection(nb_hwy_tags):
             continue
+        # get levels info for matching against potential nodes
+        nb_levels_tags = _gather_nb_tags(nx_multigraph, nd_key, "levels")
         # get name tags for matching against potential matches
         nb_name_tags = _gather_nb_name_tags(nx_multigraph, nd_key)
         # recurse
@@ -959,6 +1000,7 @@ def recursive_squash(
             set(),  # processed nodes tracked through recursion
             hwy_tags,
             nb_name_tags,
+            nb_levels_tags,
             crawl,
         )  # whether to recursively probe neighbours per distance
         # update removed nodes
@@ -1011,6 +1053,7 @@ def nx_snap_gapped_endings(
             nb_hwy_tags = _gather_nb_tags(nx_multigraph, nd_key, "highways")
             if not hwy_tags.intersection(nb_hwy_tags):
                 continue
+        nb_levels_tags = _gather_nb_tags(nx_multigraph, nd_key, "levels")
         # get name tags for matching against potential gapped edges
         nb_name_tags = _gather_nb_name_tags(nx_multigraph, nd_key)
         # get all other nodes within the buffer distance
@@ -1044,6 +1087,11 @@ def nx_snap_gapped_endings(
                 edge_hwy_tags = _gather_nb_tags(nx_multigraph, j_nd_key, "highways")
                 if not hwy_tags.intersection(edge_hwy_tags):
                     continue
+            # levels
+            if nb_levels_tags:
+                edge_level_tags = _gather_nb_tags(nx_multigraph, j_nd_key, "levels")
+                if not nb_levels_tags.intersection(edge_level_tags):
+                    continue
             # names tags
             if osm_matched_tags_only is True:
                 edge_name_tags = _gather_nb_name_tags(nx_multigraph, j_nd_key)
@@ -1078,6 +1126,7 @@ def nx_snap_gapped_endings(
                     names=[],
                     routes=[],
                     highways=[],
+                    levels=[],
                     geom=new_geom,
                 )
 
@@ -1208,6 +1257,11 @@ def recurse_child_keys(
                 continue
         # get name tags for matching against potential gapped edges
         nb_name_tags = _gather_nb_name_tags(nx_multigraph, nd_key)
+        # get levels info for matching against potential gapped edges
+        nb_levels_tags = _gather_nb_tags(nx_multigraph, nd_key, "levels")
+        # only split from ground level nodes
+        if nb_levels_tags and 0 not in nb_levels_tags:
+            continue
         # neighbours for filtering out
         neighbours = list(nx.neighbors(nx_multigraph, nd_key))
         # get all other edges within the buffer distance
@@ -1266,6 +1320,15 @@ def recurse_child_keys(
         # iter gapped edges
         for start_nd_key, end_nd_key, edge_idx, edge_data in distinct_edges:
             edge_geom = edge_data["geom"]
+            # don't split on tunnels
+            if "is_tunnel" in edge_data and edge_data["is_tunnel"] is True:
+                continue
+            # level tags
+            if nb_levels_tags:
+                # only split on ground levels
+                edge_levels_tags = _tags_from_edge_key(edge_data, "levels")
+                if edge_levels_tags and 0 not in edge_levels_tags:
+                    continue
             # hwy tags
             if osm_hwy_target_tags:
                 edge_hwy_tags = _tags_from_edge_key(edge_data, "highways")
@@ -1431,6 +1494,7 @@ def recurse_child_keys(
                         names=[],
                         routes=[],
                         highways=[],
+                        levels=[],
                         geom=new_geom,
                     )
     # squashing nodes can result in edge duplicates

pysrc/cityseer/tools/io.py

@@ -340,9 +340,9 @@ def _auto_clean_network(
             squash_nodes=True,
             centroid_by_itx=True,
             osm_hwy_target_tags=tags,
+            osm_matched_tags_only=True,
             prioritise_by_hwy_tag=True,
             simplify_line_angles=simplify_line_angles,
-            contains_buffer_dist=50,
         )
     for dist, tags, simplify_line_angles in (
         (28, ["trunk"], 95),
@@ -356,9 +356,9 @@ def _auto_clean_network(
             crawl=False,
             centroid_by_itx=True,
             osm_hwy_target_tags=tags,
+            osm_matched_tags_only=True,
             prioritise_by_hwy_tag=True,
             simplify_line_angles=simplify_line_angles,
-            contains_buffer_dist=50,
         )
         G = graphs.nx_remove_filler_nodes(G)
     # snap gapped endings - don't clean danglers before this
@@ -373,7 +373,7 @@ def _auto_clean_network(
         squash_nodes=False,
     )
     # remove danglers
-    G = graphs.nx_remove_dangling_nodes(G, despine=50)
+    G = graphs.nx_remove_dangling_nodes(G, despine=40)
     # do smaller scale cleaning
     simplify_angles = 95
     for dist in final_clean_distances:
@@ -401,7 +401,6 @@ def _auto_clean_network(
             ],
             prioritise_by_hwy_tag=True,
             simplify_line_angles=simplify_angles,
-            contains_buffer_dist=50,
         )
         G = graphs.nx_consolidate_nodes(
             G,
@@ -427,10 +426,11 @@ def _auto_clean_network(
             ],
             prioritise_by_hwy_tag=True,
             simplify_line_angles=simplify_angles,
-            contains_buffer_dist=50,
         )
     G = graphs.nx_remove_filler_nodes(G)
-    G = graphs.nx_iron_edges(G)
+    G = graphs.nx_merge_parallel_edges(G, merge_edges_by_midline=True, contains_buffer_dist=50)
+    G = graphs.nx_remove_dangling_nodes(G, despine=25, remove_deadend_tunnels=True)
+    G = graphs.nx_iron_edges(G, min_self_loop_length=100, max_foot_tunnel_length=50)
 
     return G
 
@@ -654,6 +654,22 @@ def get_merged_nd_keys(_idx: int) -> tuple[str, str]:
                 name = [tags["name"].lower()] if "name" in tags else []
                 ref = [tags["ref"].lower()] if "ref" in tags else []
                 highway = [tags["highway"].lower()] if "highway" in tags else []
+                levels = [0]
+                if "level" in tags:
+                    try:
+                        if ":" in tags["level"]:
+                            levels = tags["level"].split(":")
+                        elif ";" in tags["level"]:
+                            levels = tags["level"].split(";")
+                        else:
+                            levels = [tags["level"]]
+                        levels = [int(round(float(level))) for level in levels]
+                    except Exception:
+                        logger.warning(f'Unable to parse level info: {tags["level"]}')
+                is_tunnel = False
+                if "tunnel" in tags:
+                    # tends to be "yes" or "building_passage"
+                    is_tunnel = True
                 for idx in range(count - 1):
                     start_nd_key, end_nd_key = get_merged_nd_keys(idx)
                     nx_multigraph.add_edge(
@@ -662,6 +678,8 @@ def get_merged_nd_keys(_idx: int) -> tuple[str, str]:
                         names=name,
                         routes=ref,
                         highways=highway,
+                        levels=levels,
+                        is_tunnel=is_tunnel,
                     )
             else:
                 for idx in range(count - 1):
@@ -1371,8 +1389,8 @@ def _node_key(node_coords):
         for k in ["geom", "geometry"]:
             if k in props:
                 del props[k]
-        # names, routes, highways
-        for k in ["names", "routes", "highways"]:
+        # names, routes, highways, levels
+        for k in ["names", "routes", "highways", "levels"]:
             if k not in props:
                 props[k] = []  # type: ignore
             else: