Subselect certain types of roads that pipes can be under

dev-requirements.txt

@@ -178,7 +178,7 @@ numpy==1.26.4
     #   swmmanywhere (pyproject.toml)
     #   tifffile
     #   xarray
-osmnx==1.8.1
+osmnx==1.9.1
     # via swmmanywhere (pyproject.toml)
 packaging==23.2
     # via

requirements.txt

@@ -148,7 +148,7 @@ numpy==1.26.4
     #   swmmanywhere (pyproject.toml)
     #   tifffile
     #   xarray
-osmnx==1.8.1
+osmnx==1.9.1
     # via swmmanywhere (pyproject.toml)
 packaging==23.2
     # via

swmmanywhere/graph_utilities.py

@@ -220,6 +220,50 @@ def __call__(self,
         for u, v, key in edges_to_remove:
             G.remove_edge(u, v, key)
         return G
+
+@register_graphfcn
+class remove_non_pipe_allowable_links(BaseGraphFunction):
+    """remove_non_pipe_allowable_links class."""
+    def __call__(self,
+                 G: nx.Graph,
+                 topology_derivation: parameters.TopologyDerivation,
+                 **kwargs) -> nx.Graph:
+        """Remove non-pipe allowable links.
+
+        This function removes links that are not allowable for pipes. The non-
+        allowable links are specified in the `omit_edges` attribute of the 
+        topology_derivation parameter. There two cases handled:
+        1. The `highway` property of the edge. In `osmnx`, `highway` is a category
+            that contains the road type, e.g., motorway, trunk, primary. If the
+            edge contains a value in the `highway` property that is in `omit_edges`, 
+            the edge is removed.
+        2. Any other properties of the edge that are in `omit_edges`. If the 
+            property is not null in the edge data, the edge is removed. e.g.,
+            if `bridge` is in `omit_edges` and the `bridge` entry of the edge 
+            is NULL, then the edge is retained, if it is something like 'yes', 
+            or 'viaduct' then the edge is removed.
+
+        Args:
+            G (nx.Graph): A graph
+            topology_derivation (parameters.TopologyDerivation): A TopologyDerivation
+                parameter object
+            **kwargs: Additional keyword arguments are ignored.
+
+        Returns:
+            G (nx.Graph): A graph
+        """
+        edges_to_remove = set()
+        for u, v, keys, data in G.edges(data=True,keys = True):
+            for omit in topology_derivation.omit_edges:
+                if data.get('highway', None) == omit:
+                    # Check whether the 'highway' property is 'omit'
+                    edges_to_remove.add((u, v, keys))
+                elif data.get(omit, None):
+                    # Check whether the 'omit' property of edge is not None 
+                    edges_to_remove.add((u, v, keys))
+        for edges in edges_to_remove:
+            G.remove_edge(*edges)
+        return G
 
 @register_graphfcn
 class format_osmnx_lanes(BaseGraphFunction,

swmmanywhere/parameters.py

@@ -81,6 +81,14 @@ class TopologyDerivation(BaseModel):
                         unit = "-",
                         description = "Weights for topo derivation")
 
+    omit_edges: list = Field(default = ['motorway', 
+                                        'motorway_link',
+                                        'bridge', 
+                                        'tunnel'],
+                        min_items = 1,
+                        unit = "-",
+                        description = "OSM paths pipes are not allowed under")
+
     chahinian_slope_scaling: float = Field(default = 1,
         le = 1,
         ge = 0,

swmmanywhere/prepare_data.py

@@ -103,8 +103,9 @@ def download_street(bbox: tuple[float, float, float, float]) -> nx.MultiDiGraph:
             ``truncate_by_edge set`` to True.
     """
     west, south, east, north = bbox
+    bbox = (north, south, east, west) # not sure why osmnx uses this order
     graph = ox.graph_from_bbox(
-        north, south, east, west, network_type="drive", truncate_by_edge=True
+        bbox = bbox, network_type="drive", truncate_by_edge=True
     )
     return cast("nx.MultiDiGraph", graph)
 

tests/test_data/demo_config.yml

@@ -15,6 +15,7 @@ starting_graph: null
 graphfcn_list:
   - assign_id
   - format_osmnx_lanes
+  - remove_non_pipe_allowable_links
   - double_directed
   - fix_geometries
   - split_long_edges

tests/test_graph_utilities.py

@@ -243,7 +243,34 @@ def test_pipe_by_pipe():
     for u, d in G.nodes(data=True):
         assert 'chamber_floor_elevation' in d.keys()
         assert math.isfinite(d['chamber_floor_elevation'])
-
+
+def get_edge_types(G):
+    """Get the edge types in the graph."""
+    edge_types = set()
+    for u,v,d in G.edges(data=True):
+        if isinstance(d['highway'], list):
+            edge_types.union(d['highway'])
+        else:
+            edge_types.add(d['highway'])
+    return edge_types
+
+def test_remove_non_pipe_allowable_links():
+    """Test the remove_non_pipe_allowable_links function."""
+    G = load_graph(Path(__file__).parent / 'test_data' / 'street_graph.json')
+    # Ensure some invalid paths
+    topology_params = parameters.TopologyDerivation(omit_edges = ['primary', 'bridge'])
+
+    # Test that an edge has a non-None 'bridge' entry
+    assert len(set([d.get('bridge',None) for u,v,d in G.edges(data=True)])) > 1
+
+    # Test that an edge has a 'primary' entry under highway
+    assert 'primary' in get_edge_types(G)
+
+    G_ = gu.remove_non_pipe_allowable_links(G, topology_params)
+    assert 'primary' not in get_edge_types(G_)
+    assert len(set([d.get('bridge',None) for u,v,d in G_.edges(data=True)])) == 1
+
+
 def test_iterate_graphfcns():
     """Test the iterate_graphfcns function."""
     G = load_graph(Path(__file__).parent / 'test_data' / 'graph_topo_derived.json')