What to do when no variance in weight bounds

swmmanywhere/graph_utilities.py

@@ -1007,11 +1007,13 @@ def __call__(self, G: nx.Graph,
         # Calculate bounds to normalise between
         bounds: Dict[Any, List[float]] = defaultdict(lambda: [np.Inf, -np.Inf])
 
-        for (u, v, d), w in product(G.edges(data=True), 
-                                    topology_derivation.weights):
-            bounds[w][0] = min(bounds[w][0], d.get(w, np.Inf))
-            bounds[w][1] = max(bounds[w][1], d.get(w, -np.Inf))
+        for w in topology_derivation.weights:
+            bounds[w][0] = min(nx.get_edge_attributes(G, w).values()) # lower bound
+            bounds[w][1] = max(nx.get_edge_attributes(G, w).values()) # upper bound
 
+        # Avoid division by zero
+        bounds = {w : [b[0], b[1]] for w, b in bounds.items() if b[0] != b[1]}
+
         G = G.copy()
         eps = np.finfo(float).eps
         for u, v, d in G.edges(data=True):

tests/test_geospatial_utilities.py

@@ -8,6 +8,7 @@
 import geopandas as gpd
 import networkx as nx
 import numpy as np
+import pytest
 import rasterio as rst
 from scipy.interpolate import RegularGridInterpolator
 from shapely import geometry as sgeom
@@ -17,7 +18,8 @@
 from swmmanywhere.misc.debug_derive_rc import derive_rc_alt
 
 
-def load_street_network():
+@pytest.fixture
+def street_network():
     """Load a street network."""
     G = ge.load_graph(Path(__file__).parent / 'test_data' / 'street_graph.json')
     return G
@@ -217,19 +219,18 @@ def test_burn_shape_in_raster():
         raster_fid.unlink(missing_ok=True)
         new_raster_fid.unlink(missing_ok=True)
 
-def test_derive_subcatchments():
+def test_derive_subcatchments(street_network):
     """Test the derive_subcatchments function."""
-    G = load_street_network()
     elev_fid = Path(__file__).parent / 'test_data' / 'elevation.tif'
     for method in ['pysheds', 'pyflwdir']:
-        polys = go.derive_subcatchments(G, elev_fid,method=method)
+        polys = go.derive_subcatchments(street_network, elev_fid,method=method)
         assert 'slope' in polys.columns
         assert 'area' in polys.columns
         assert 'geometry' in polys.columns
         assert 'id' in polys.columns
         assert polys.shape[0] > 0
         assert polys.dropna().shape == polys.shape
-        assert polys.crs == G.graph['crs']
+        assert polys.crs == street_network.graph['crs']
 
         # Pyflwdir and pysheds catchment derivation aren't absolutely identical
         assert almost_equal(polys.set_index('id').loc[2623975694, 'area'], 
@@ -239,10 +240,9 @@ def test_derive_subcatchments():
         assert almost_equal(polys.set_index('id').loc[2623975694, 'width'], 
                             21.845, tol = 0.001)
 
-def test_derive_rc():
+def test_derive_rc(street_network):
     """Test the derive_rc function."""
-    G = load_street_network()
-    crs = G.graph['crs']
+    crs = street_network.graph['crs']
     eg_bldg = sgeom.Polygon([(700291,5709928), 
                        (700331,5709927),
                        (700321,5709896), 
@@ -276,7 +276,7 @@ def test_derive_rc():
                             (700329, 5709883),
                             (700351, 5709883)])]
 
-    streetcover = [d['geometry'].buffer(5) for u,v,d in G.edges(data=True)]
+    streetcover = [d['geometry'].buffer(5) for u,v,d in street_network.edges(data=True)]
     streetcover = gpd.GeoDataFrame(geometry = streetcover, crs = crs)
 
     subs = gpd.GeoDataFrame(data = {'id' : [107733,
@@ -410,27 +410,26 @@ def test_remove_intersections():
     assert polys_.set_index('id')[['area']].equals(
         targets.set_index('id')[['area']])
 
-def test_graph_to_geojson():
+def test_graph_to_geojson(street_network):
     """Test the graph_to_geojson function."""
-    G = load_street_network()
-    crs = G.graph['crs']
+    crs = street_network.graph['crs']
     with tempfile.TemporaryDirectory() as temp_dir:
         temp_path = Path(temp_dir)
-        go.graph_to_geojson(G, 
+        go.graph_to_geojson(street_network, 
                             temp_path / 'graph_nodes.geojson', 
                             temp_path / 'graph_edges.geojson', 
                             crs)
         gdf = gpd.read_file(temp_path / 'graph_nodes.geojson')
         assert gdf.crs == crs
-        assert gdf.shape[0] == len(G.nodes)
+        assert gdf.shape[0] == len(street_network.nodes)
 
         gdf = gpd.read_file(temp_path / 'graph_edges.geojson')
-        assert gdf.shape[0] == len(G.edges)
+        assert gdf.shape[0] == len(street_network.edges)
 
-def test_merge_points():
+def test_merge_points(street_network):
     """Test the merge_points function."""
-    G = load_street_network()
-    mapping = go.merge_points([(d['x'], d['y']) for u,d in G.nodes(data=True)],
+    mapping = go.merge_points([(d['x'], d['y']) 
+                               for u,d in street_network.nodes(data=True)],
                               20)
     assert set(mapping.keys()) == set([2,3,5,15,16,18,22])
     assert set([x['maps_to'] for x in mapping.values()]) == set([2,5,15])

tests/test_graph_utilities.py

@@ -25,16 +25,17 @@
 from swmmanywhere.graph_utilities import graphfcns as gu
 
 
-def load_street_network():
+@pytest.fixture
+def street_network():
     """Load a street network."""
     bbox = (-0.11643,51.50309,-0.11169,51.50549)
     G = load_graph(Path(__file__).parent / 'test_data' / 'street_graph.json')
     return G, bbox
 
-def test_save_load():
+def test_save_load(street_network):
     """Test the save_graph and load_graph functions."""
+    G, _ = street_network
     # Load a street network
-    G,_ = load_street_network()
     with tempfile.TemporaryDirectory() as temp_dir:
         # Save the graph
         save_graph(G, Path(temp_dir) / 'test_graph.json')
@@ -43,25 +44,25 @@ def test_save_load():
         # Check if the loaded graph is the same as the original graph
         assert nx.is_isomorphic(G, G_new)
 
-def test_assign_id():
+def test_assign_id(street_network):
     """Test the assign_id function."""
-    G, _ = load_street_network()
+    G, _ = street_network
     G = gu.assign_id(G)
     for u, v, data in G.edges(data=True):
         assert 'id' in data.keys()
         assert isinstance(data['id'], str)
 
-def test_double_directed():
+def test_double_directed(street_network):
     """Test the double_directed function."""
-    G, _ = load_street_network()
+    G, _ = street_network
     G = gu.assign_id(G)
     G = gu.double_directed(G)
     for u, v in G.edges():
         assert (v,u) in G.edges
 
-def test_calculate_streetcover():
+def test_calculate_streetcover(street_network):
     """Test the calculate_streetcover function."""
-    G, _ = load_street_network()
+    G, _ = street_network
     params = parameters.SubcatchmentDerivation()
     addresses = parameters.FilePaths(base_dir = None,
                                         project_name = None,
@@ -77,17 +78,17 @@ def test_calculate_streetcover():
         assert len(gdf) == len(G.edges)
         assert gdf.geometry.area.sum() > 0
 
-def test_split_long_edges():
+def test_split_long_edges(street_network):
     """Test the split_long_edges function."""
-    G, _ = load_street_network()
+    G, _ = street_network
     G = gu.assign_id(G)
     max_length = 40
     params = parameters.SubcatchmentDerivation(max_street_length = max_length)
     G = gu.split_long_edges(G, params)
     for u, v, data in G.edges(data=True):
         assert data['length'] <= (max_length * 2)
 
-def test_derive_subcatchments():
+def test_derive_subcatchments(street_network):
     """Test the derive_subcatchments function."""
     with tempfile.TemporaryDirectory() as temp_dir:
         temp_path = Path(temp_dir)
@@ -101,7 +102,7 @@ def test_derive_subcatchments():
         addresses.streetcover = temp_path / 'building.geojson'
         addresses.subcatchments = temp_path / 'subcatchments.geojson'
         params = parameters.SubcatchmentDerivation()
-        G, bbox = load_street_network()
+        G, _ = street_network
 
         # mock up buildings
         eg_bldg = sgeom.Polygon([(700291.346,5709928.922), 
@@ -122,9 +123,9 @@ def test_derive_subcatchments():
             assert 'contributing_area' in data.keys()
             assert isinstance(data['contributing_area'], float)
 
-def test_set_elevation_and_slope():
+def test_set_elevation_and_slope(street_network):
     """Test the set_elevation, set_surface_slope, chahinian_slope function."""
-    G, _ = load_street_network()
+    G, _ = street_network
     with tempfile.TemporaryDirectory() as temp_dir:
         temp_path = Path(temp_dir)
         addresses = parameters.FilePaths(base_dir = temp_path, 
@@ -162,19 +163,19 @@ def test_set_elevation_and_slope():
 
 
 
-def test_chahinian_angle():
+def test_chahinian_angle(street_network):
     """Test the chahinian_angle function."""
-    G, _ = load_street_network()
+    G, _ = street_network
     G = gu.set_chahinian_angle(G)
     for u, v, data in G.edges(data=True):
         assert 'chahinian_angle' in data.keys()
         assert math.isfinite(data['chahinian_angle'])
 
 
 
-def test_calculate_weights():
+def test_calculate_weights(street_network):
     """Test the calculate_weights function."""
-    G, _ = load_street_network()
+    G, _ = street_network
     params = parameters.TopologyDerivation()
     for weight in params.weights:
         for ix, (u,v,data) in enumerate(G.edges(data=True)):
@@ -184,10 +185,23 @@ def test_calculate_weights():
     for u, v, data in G.edges(data=True):
         assert 'weight' in data.keys()
         assert math.isfinite(data['weight'])
-
-def test_identify_outlets_no_river():
+
+def test_calculate_weights_novar(street_network):
+    """Test the calculate_weights function with no variance."""
+    G, _ = street_network
+    params = parameters.TopologyDerivation()
+    for weight in params.weights:
+        for ix, (u,v,data) in enumerate(G.edges(data=True)):
+            data[weight] = 1.5
+
+    G = gu.calculate_weights(G, params)
+    for u, v, data in G.edges(data=True):
+        assert 'weight' in data.keys()
+        assert math.isfinite(data['weight'])
+
+def test_identify_outlets_no_river(street_network):
     """Test the identify_outlets in the no river case."""
-    G, _ = load_street_network()
+    G, _ = street_network
     G = gu.assign_id(G)
     G = gu.double_directed(G)
     elev_fid = Path(__file__).parent / 'test_data' / 'elevation.tif'
@@ -205,9 +219,9 @@ def test_identify_outlets_no_river():
     outlets = [(u,v,d) for u,v,d in G.edges(data=True) if d['edge_type'] == 'outlet']
     assert len(outlets) == 1
 
-def test_identify_outlets_sg():
+def test_identify_outlets_sg(street_network):
     """Test the identify_outlets with subgraphs."""
-    G, _ = load_street_network()
+    G, _ = street_network
 
     G = gu.assign_id(G)
     G = gu.double_directed(G)
@@ -277,9 +291,9 @@ def test_identify_outlets_sg():
     outlets = [(u,v,d) for u,v,d in G_.edges(data=True) if d['edge_type'] == 'outlet']
     assert len(outlets) == 3
 
-def test_identify_outlets_and_derive_topology():
+def test_identify_outlets_and_derive_topology(street_network):
     """Test the identify_outlets and derive_topology functions."""
-    G, _ = load_street_network()
+    G, _ = street_network
     G = gu.assign_id(G)
     G = gu.double_directed(G)
     for ix, (u,v,d) in enumerate(G.edges(data=True)):
@@ -342,9 +356,9 @@ def test_identify_outlets_and_derive_topology():
     outlets = [(u,v,d) for u,v,d in G_.edges(data=True) if d['edge_type'] == 'outlet']
     assert len(outlets) == 1
 
-def test_identify_outlets_and_derive_topology_withtopo():
+def test_identify_outlets_and_derive_topology_withtopo(street_network):
     """Test the identify_outlets and derive_topology functions."""
-    G, _ = load_street_network()
+    G, _ = street_network
     G = gu.assign_id(G)
     G = gu.double_directed(G)
     for ix, (u,v,d) in enumerate(G.edges(data=True)):
@@ -499,18 +513,18 @@ def almost_equal(a, b, tol=1e-6):
     """Check if two numbers are almost equal."""
     return abs(a-b) < tol
 
-def test_merge_street_nodes():
+def test_merge_street_nodes(street_network):
     """Test the merge_street_nodes function."""
-    G, _ = load_street_network()
+    G, _ = street_network
     subcatchment_derivation = parameters.SubcatchmentDerivation(
         node_merge_distance = 20)
     G_ = gu.merge_street_nodes(G, subcatchment_derivation)
     assert not set([107736,266325461,2623975694,32925453]).intersection(G_.nodes)
     assert almost_equal(G_.nodes[25510321]['x'], 700445.0112082)
 
-def test_clip_to_catchments():
+def test_clip_to_catchments(street_network):
     """Test the clip_to_catchments function."""
-    G, _ = load_street_network()
+    G, _ = street_network
 
     with tempfile.TemporaryDirectory() as temp_dir: