diff --git a/swmmanywhere/graph_utilities.py b/swmmanywhere/graph_utilities.py index 899a0655..d03780af 100644 --- a/swmmanywhere/graph_utilities.py +++ b/swmmanywhere/graph_utilities.py @@ -180,7 +180,13 @@ def iterate_graphfcns(G: nx.Graph, raise ValueError(f"Graphfcns are not registered:\n{', '.join(not_exists)}") for function in graphfcn_list: G = graphfcns[function](G, addresses = addresses, **params) - logger.info(f"graphfcn: {function} completed.") + if len(_filter_streets(G).edges) == 0: + logger.warning(f"""graphfcn: {function} removed all edges, + returning graph.""") + return G + else: + logger.info(f"graphfcn: {function} completed.") + if verbose(): save_graph(G, addresses.model / f"{function}_graph.json") go.graph_to_geojson(graphfcns.fix_geometries(G), diff --git a/swmmanywhere/swmmanywhere.py b/swmmanywhere/swmmanywhere.py index 26333cc6..d688717b 100644 --- a/swmmanywhere/swmmanywhere.py +++ b/swmmanywhere/swmmanywhere.py @@ -100,6 +100,12 @@ def swmmanywhere(config: dict) -> tuple[Path, dict | None]: G.graph['crs'] ) save_graph(G, addresses.graph) + + # Check any edges + if len(G.edges) == 0: + logger.warning("No edges in graph, returning graph file.") + return addresses.graph, None + # Write to .inp synthetic_write(addresses) diff --git a/tests/test_graph_utilities.py b/tests/test_graph_utilities.py index ccc00a0f..d4e58c10 100644 --- a/tests/test_graph_utilities.py +++ b/tests/test_graph_utilities.py @@ -480,8 +480,7 @@ def test_iterate_graphfcns(): project_name = None, bbox_number = None, model_number = None) - # Needed if VERBOSE is on.. maybe I should turn it off at the top of - # each test, not sure + addresses.model = temp_path G = iterate_graphfcns(G, ['assign_id', @@ -493,6 +492,44 @@ def test_iterate_graphfcns(): 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 _remove_edges(G: nx.Graph,**kw): + """Remove all edges from the graph. + + Args: + G (nx.Graph): The graph to remove edges from. + kw: Additional keyword arguments. (which are ignored) + + Returns: + nx.Graph: The graph with no edges. + """ + G.remove_edges_from(list(G.edges)) + return G + +def test_iterate_graphfcns_noedges(): + """Test the iterate_graphfcns function for a graph with no edges.""" + G = load_graph(Path(__file__).parent / 'test_data' / 'graph_topo_derived.json') + + with tempfile.TemporaryDirectory() as temp_dir: + temp_path = Path(temp_dir) + addresses = parameters.FilePaths(base_dir = None, + project_name = None, + bbox_number = None, + model_number = None) + os.environ['SWMMANYWHERE_VERBOSE'] = 'true' + addresses.model = temp_path + original_function = gu['remove_non_pipe_allowable_links'] + gu['remove_non_pipe_allowable_links'] = _remove_edges + G = iterate_graphfcns(G, + ['assign_id', + 'remove_non_pipe_allowable_links'], + {}, + addresses) + gu['remove_non_pipe_allowable_links'] = original_function + os.environ['SWMMANYWHERE_VERBOSE'] = 'false' + assert (addresses.model / 'assign_id_graph.json').exists() + assert not (addresses.model /\ + 'remove_non_pipe_allowable_links_graph.json').exists() + def test_fix_geometries(): """Test the fix_geometries function.""" # Create a graph with edge geometry not matching node coordinates