Handle cases where a graphfunction has removed all potential pipe carrying edges

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),

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)
 

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