diff --git a/swmmanywhere/graph_utilities.py b/swmmanywhere/graph_utilities.py index 455cb64f..899a0655 100644 --- a/swmmanywhere/graph_utilities.py +++ b/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): diff --git a/tests/test_geospatial_utilities.py b/tests/test_geospatial_utilities.py index 94830fe7..89c806ab 100644 --- a/tests/test_geospatial_utilities.py +++ b/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]) diff --git a/tests/test_graph_utilities.py b/tests/test_graph_utilities.py index 13a60806..ccc00a0f 100644 --- a/tests/test_graph_utilities.py +++ b/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,9 +78,9 @@ 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) @@ -87,7 +88,7 @@ def test_split_long_edges(): 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,9 +163,9 @@ 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() @@ -172,9 +173,9 @@ def test_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: diff --git a/tests/test_metric_utilities.py b/tests/test_metric_utilities.py index 1a8a8fa3..e1bf37f9 100644 --- a/tests/test_metric_utilities.py +++ b/tests/test_metric_utilities.py @@ -17,8 +17,8 @@ def assert_close(a: float, b: float, rtol: float = 1e-3) -> None: """Assert that two floats are close.""" assert np.isclose(a, b, rtol=rtol).all() - -def get_subs(): +@pytest.fixture +def subs(): """Get a GeoDataFrame of subcatchments.""" subs = [shapely.Polygon([(700262, 5709928), (700262, 5709883), @@ -54,7 +54,8 @@ def get_subs(): crs = 'EPSG:32630') return subs -def get_results(): +@pytest.fixture +def results(): """Get a DataFrame of results.""" results = pd.DataFrame([{'id' : 4253560, 'variable' : 'flow', @@ -159,10 +160,9 @@ def test_kstest_edge_betweenness(): val = mu.metrics.kstest_edge_betweenness(synthetic_G = G_, real_G = G) assert_close(val, 0.38995) -def test_best_outlet_match(): +def test_best_outlet_match(subs): """Test the best_outlet_match and ks_betweenness.""" G = load_graph(Path(__file__).parent / 'test_data' / 'graph_topo_derived.json') - subs = get_subs() sg, outlet = mu.best_outlet_match(synthetic_G = G, real_subs = subs) @@ -210,11 +210,10 @@ def test_relerror_different_length(): yhat = np.array([1])) assert_close(val, (1 - 3.5)/3.5) -def test_outlet_nse_flow(): +def test_outlet_nse_flow(subs): """Test the outlet_nse_flow metric.""" # Load data G = load_graph(Path(__file__).parent / 'test_data' / 'graph_topo_derived.json') - subs = get_subs() # Mock results results = pd.DataFrame([{'id' : 4253560, @@ -287,11 +286,10 @@ def test_outlet_nse_flow(): metric_evaluation = None) assert val == -15.0 -def test_outlet_nse_flooding(): +def test_outlet_nse_flooding(subs): """Test the outlet_nse_flow metric.""" # Load data G = load_graph(Path(__file__).parent / 'test_data' / 'graph_topo_derived.json') - subs = get_subs() # Mock results results = pd.DataFrame([{'id' : 4253560, @@ -381,11 +379,10 @@ def test_outlet_nse_flooding(): metric_evaluation = None) assert val == 0.0 -def test_design_params(): +def test_design_params(subs): """Test the design param related metrics.""" G = load_graph(Path(__file__).parent / 'test_data' / 'graph_topo_derived.json') nx.set_edge_attributes(G, 0.15, 'diameter') - subs = get_subs() # Mock results (only needed for dominant outlet) results = pd.DataFrame([{'id' : 4253560, @@ -472,14 +469,11 @@ def test_netcomp_iterate(): assert metric in netcomp_results assert np.isclose(val, netcomp_results[metric]) -def test_subcatchment_nse_flooding(): +def test_subcatchment_nse_flooding(subs, results): """Test the outlet_nse_flow metric.""" # Load data G = load_graph(Path(__file__).parent / 'test_data' / 'graph_topo_derived.json') - subs = get_subs() - # Mock results - results = get_results() # Calculate NSE (perfect results) val = mu.metrics.subcatchment_nse_flooding(synthetic_G = G, @@ -539,9 +533,8 @@ def test_restirctions(): with pytest.raises(ValueError): mu.metric_factory('outlet_nse_nmanholes') -def test_nodes_to_subs(): +def test_nodes_to_subs(subs): """Test the nodes_to_subs function.""" - subs = get_subs() G = load_graph(Path(__file__).parent / 'test_data' / 'graph_topo_derived.json') nodes = mu.nodes_to_subs(G, subs) nodes = set([(r.id, r.sub_id) for _, r in nodes.iterrows()]) @@ -553,11 +546,9 @@ def test_nodes_to_subs(): (6277683849, 6277683849)} assert nodes == nodes_ -def test_align_by_shape_and_median_coef(): +def test_align_by_shape_and_median_coef(subs, results): """Test the align_by_shape and median_coef_by_group function.""" - subs = get_subs() G = load_graph(Path(__file__).parent / 'test_data' / 'graph_topo_derived.json') - results = get_results() # Align the grid aligned = mu.align_by_shape('flooding', results,