Merge branch 'topological_metrics' of https://github.com/ImperialColl…

…egeLondon/SWMManywhere into topological_metrics

dev-requirements.txt

@@ -55,6 +55,7 @@ colorama==0.4.6
     # via
     #   build
     #   click
+    #   loguru
     #   pytest
     #   tqdm
 contourpy==1.2.0
@@ -117,6 +118,8 @@ lazy-loader==0.3
     # via scikit-image
 llvmlite==0.41.1
     # via numba
+loguru==0.7.2
+    # via swmmanywhere (pyproject.toml)
 matplotlib==3.8.2
     # via
     #   salib
@@ -311,6 +314,8 @@ virtualenv==20.24.5
     # via pre-commit
 wheel==0.41.3
     # via pip-tools
+win32-setctime==1.1.0
+    # via loguru
 xarray==2023.12.0
     # via
     #   rioxarray

pyproject.toml

@@ -31,6 +31,8 @@ dependencies = [
   "geopandas",
   "geopy",
   "joblib",
+  "GitPython",
+  "loguru",
   "matplotlib",
   "netcdf4",
   "netcomp@ git+https://github.com/barneydobson/NetComp.git",

requirements.txt

@@ -48,6 +48,7 @@ cligj==0.7.2
 colorama==0.4.6
     # via
     #   click
+    #   loguru
     #   tqdm
 contourpy==1.2.0
     # via matplotlib
@@ -93,6 +94,8 @@ lazy-loader==0.3
     # via scikit-image
 llvmlite==0.41.1
     # via numba
+loguru==0.7.2
+    # via swmmanywhere (pyproject.toml)
 matplotlib==3.8.2
     # via
     #   salib
@@ -238,6 +241,8 @@ tzdata==2024.1
     # via pandas
 urllib3==2.1.0
     # via requests
+win32-setctime==1.1.0
+    # via loguru
 xarray==2023.12.0
     # via
     #   rioxarray

swmmanywhere/graph_utilities.py

@@ -22,6 +22,7 @@
 
 from swmmanywhere import geospatial_utilities as go
 from swmmanywhere import parameters
+from swmmanywhere.logging import logger
 
 
 def load_graph(fid: Path) -> nx.Graph:
@@ -508,18 +509,55 @@ def __call__(self, G: nx.Graph,
         # Set the 'surface_slope' attribute for all edges
         nx.set_edge_attributes(G, slope_dict, 'surface_slope')
         return G
+
+@register_graphfcn
+class set_chahinian_slope(BaseGraphFunction,
+                          required_edge_attributes = ['surface_slope'],
+                          adds_edge_attributes = ['chahinian_slope']):
+    """set_chahinian_slope class."""
+    def __call__(self, G: nx.Graph, **kwargs) -> nx.Graph:
+        """set_chahinian_slope class.
+            
+        This function sets the Chahinian slope for each edge. The Chahinian slope is
+        calculated from the surface slope and weighted according to the slope
+        (based on: https://doi.org/10.1016/j.compenvurbsys.2019.101370)
+        
+        Args:
+            G (nx.Graph): A graph
+            **kwargs: Additional keyword arguments are ignored.
+            
+        Returns:
+            G (nx.Graph): A graph
+        """
+        G = G.copy()
 
+        # Values where the weight of the angle can be matched to the values 
+        # in weights
+        angle_points = [-1, 0.3, 0.7, 10] 
+        weights = [1, 0, 0, 1]
+
+        # Calculate weights
+        slope = nx.get_edge_attributes(G, "surface_slope")
+        weights = np.interp(np.asarray(list(slope.values())) * 100, 
+                            angle_points,
+                            weights, 
+                            left=1, 
+                            right=1)
+        nx.set_edge_attributes(G, dict(zip(slope, weights)), "chahinian_slope")
+
+        return G
+
 @register_graphfcn
-class set_chahinan_angle(BaseGraphFunction,
+class set_chahinian_angle(BaseGraphFunction,
                          required_node_attributes = ['x','y'],
-                         adds_edge_attributes = ['chahinan_angle']):
-    """set_chahinan_angle class."""
+                         adds_edge_attributes = ['chahinian_angle']):
+    """set_chahinian_angle class."""
 
     def __call__(self, G: nx.Graph, 
                        **kwargs) -> nx.Graph:
-        """Set the Chahinan angle for each edge.
+        """Set the Chahinian angle for each edge.
 
-        This function sets the Chahinan angle for each edge. The Chahinan angle is
+        This function sets the Chahinian angle for each edge. The Chahinian angle is
         calculated from the geometry of the edge and weighted according to the 
         angle (based on: https://doi.org/10.1016/j.compenvurbsys.2019.101370)
 
@@ -543,14 +581,14 @@ def __call__(self, G: nx.Graph,
                 p2 = (G.nodes[v]['x'], G.nodes[v]['y'])
                 p3 = (G.nodes[node]['x'], G.nodes[node]['y'])
                 angle = go.calculate_angle(p1,p2,p3)
-                chahinan_weight = np.interp(angle,
+                chahinian_weight = np.interp(angle,
                                             [0, 90, 135, 180, 225, 270, 360],
                                             [1, 0.2, 0.7, 0, 0.7, 0.2, 1]
                                             )
-                min_weight = min(chahinan_weight, min_weight)
+                min_weight = min(chahinian_weight, min_weight)
             if min_weight == float('inf'):
                 min_weight = 0
-            d['chahinan_angle'] = min_weight
+            d['chahinian_angle'] = min_weight
         return G
 
 @register_graphfcn
@@ -933,7 +971,7 @@ def process_successors(G: nx.Graph,
         edge_diams[(node,ds_node,0)] = diam
         chamber_floor[ds_node] = surface_elevations[ds_node] - depth
         if ix > 0:
-            print('''a node has multiple successors, 
+            logger.warning('''a node has multiple successors, 
                 not sure how that can happen if using shortest path
                 to derive topology''')
 

swmmanywhere/logging.py

@@ -0,0 +1,55 @@
+# -*- coding: utf-8 -*-
+"""Created on 2024-03-04.
+
+@author: Barney
+"""
+import os
+import sys
+
+import loguru
+
+
+def dynamic_filter(record):
+    """A dynamic filter."""
+    return os.getenv("SWMMANYWHERE_VERBOSE", "false").lower() == "true"
+
+def get_logger() -> loguru.logger:
+    """Get a logger."""
+    logger = loguru.logger
+    logger.configure(
+        handlers=[
+            {
+                "sink": sys.stdout,
+                "filter" : dynamic_filter,
+                "colorize": True,
+                "format": " | ".join(
+                    [
+                        "<cyan>{time:YYYY/MM/DD HH:mm:ss}</>",
+                        "{message}",
+                    ]
+                ),
+            }
+        ]
+    )
+    return logger
+
+# Get the logger
+logger = get_logger()
+
+# Add a test_logger method to the logger
+logger.test_logger = lambda : logger.info("This is a test message.")
+
+# Store the original add method
+original_add = logger.add
+
+# Define a new function that wraps the original add method
+def new_add(sink, **kwargs):
+    """A new add method to wrap existing one but with the filter."""
+    # Include the dynamic filter in the kwargs if not already specified
+    if 'filter' not in kwargs:
+        kwargs['filter'] = dynamic_filter
+    # Call the original add method with the updated kwargs
+    return original_add(sink, **kwargs)
+
+# Replace the logger's add method with new_add
+logger.add = new_add

swmmanywhere/parameters.py

@@ -61,25 +61,25 @@ class OutletDerivation(BaseModel):
 
 class TopologyDerivation(BaseModel):
     """Parameters for topology derivation."""
-    weights: list = Field(default = ['surface_slope',
-                                      'chahinan_angle',
+    weights: list = Field(default = ['chahinian_slope',
+                                      'chahinian_angle',
                                       'length',
                                       'contributing_area'],
                         min_items = 1,
                         unit = "-",
                         description = "Weights for topo derivation")
 
-    surface_slope_scaling: float = Field(default = 1,
+    chahinian_slope_scaling: float = Field(default = 1,
         le = 1,
         ge = 0,
         unit = "-",
         description = "Constant to apply to surface slope in topo derivation")
 
-    chahinan_angle_scaling: float = Field(default = 1,
+    chahinian_angle_scaling: float = Field(default = 1,
         le = 1,
         ge = 0,
         unit = "-",
-        description = "Constant to apply to chahinan angle in topo derivation")
+        description = "Constant to apply to chahinian angle in topo derivation")
 
     length_scaling: float = Field(default = 1,
         le = 1,
@@ -93,17 +93,17 @@ class TopologyDerivation(BaseModel):
         unit = "-",
         description = "Constant to apply to contributing area in topo derivation")
 
-    surface_slope_exponent: float = Field(default = 1,
+    chahinian_slope_exponent: float = Field(default = 1,
         le = 2,
         ge = -2,
         unit = "-",
         description = "Exponent to apply to surface slope in topo derivation")
 
-    chahinan_angle_exponent: float = Field(default = 1,
+    chahinian_angle_exponent: float = Field(default = 1,
         le = 2,
         ge = -2,
         unit = "-",
-        description = "Exponent to apply to chahinan angle in topo derivation")
+        description = "Exponent to apply to chahinian angle in topo derivation")
 
     length_exponent: float = Field(default = 1,
         le = 2,
@@ -132,8 +132,8 @@ def check_weights(cls, values):
 # TODO move this to tests and run it if we're happy with this way of doing things
 class NewTopo(TopologyDerivation):
      """Demo for changing weights that should break the validator."""
-     weights: list = Field(default = ['surface_slope',
-                                      'chahinan_angle',
+     weights: list = Field(default = ['chahinian_slope',
+                                      'chahinian_angle',
                                       'length',
                                       'contributing_area',
                                 'test'],
@@ -149,38 +149,38 @@ class HydraulicDesign(BaseModel):
                             description = """Diameters to consider in 
                             pipe by pipe method""")
     max_fr: float = Field(default = 0.8,
-		upper_limit = 1,
-		lower_limit = 0,
+		le = 1,
+		ge = 0,
 		unit = "-",
 		description = "Maximum filling ratio in pipe by pipe method")
     min_shear: float = Field(default = 2,
-		upper_limit = 3,
-		lower_limit = 0,
+		le = 3,
+		ge = 0,
 		unit = "Pa",
 		description = "Minimum wall shear stress in pipe by pipe method")
     min_v: float = Field(default = 0.75,
-		upper_limit = 2,
-		lower_limit = 0,
+		le = 2,
+		ge = 0,
 		unit = "m/s",
 		description = "Minimum velocity in pipe by pipe method")
     max_v: float = Field(default = 5,
-		upper_limit = 10,
-		lower_limit = 3,
+		le = 10,
+		ge = 3,
 		unit = "m/s",
 		description = "Maximum velocity in pipe by pipe method")
     min_depth: float = Field(default = 0.5,
-		upper_limit = 1,
-		lower_limit = 0,
+		le = 1,
+		ge = 0,
 		unit = "m",
 		description = "Minimum excavation depth in pipe by pipe method")
     max_depth: float = Field(default = 5,
-		upper_limit = 10,
-		lower_limit = 2,
+		le = 10,
+		ge = 2,
 		unit = "m",
 		description = "Maximum excavation depth in pipe by pipe method")
     precipitation: float = Field(default = 0.006,
-		upper_limit = 0.010,
-		lower_limit = 0.001,
+		le = 0.010,
+		ge = 0.001,
 		description = "Depth of design storm in pipe by pipe method",
 		unit = "m")
 

swmmanywhere/prepare_data.py

@@ -17,7 +17,8 @@
 import yaml
 from geopy.geocoders import Nominatim
 
-# Some minor comment (to remove)
+from swmmanywhere.logging import logger
+
 
 def get_country(x: float, 
                 y: float) -> dict[int, str]:
@@ -83,9 +84,9 @@ def download_buildings(file_address: Path,
         # Save data to the specified file address
         with file_address.open("wb") as file:
             file.write(response.content)
-        print(f"Data downloaded and saved to {file_address}")
+        logger.info(f"Data downloaded and saved to {file_address}")
     else:
-        print(f"Error downloading data. Status code: {response.status_code}")
+        logger.error(f"Error downloading data. Status code: {response.status_code}")
     return response.status_code
 
 def download_street(bbox: tuple[float, float, float, float]) -> nx.MultiDiGraph:
@@ -174,10 +175,10 @@ def download_elevation(fid: Path,
         with fid.open('wb') as rast_file:
             shutil.copyfileobj(r.raw, rast_file)
 
-        print('Elevation data downloaded successfully.')
+        logger.info('Elevation data downloaded successfully.')
 
     except requests.exceptions.RequestException as e:
-        print(f'Error downloading elevation data: {e}')
+        logger.error(f'Error downloading elevation data: {e}')
 
     return r.status_code