Add complexipy (#326)

* Add dep

* Update lockfile

* Update workflow

* Fix incorrect directory

* Refactor

* Split out warnings

* Refactor

* Refactor

* Refactor

* Refactor

* Add directive

* Refactor

* Refactor

* Minor refactor

* Add ignore directive

Author: ppinchuk

.github/workflows/ci-python.yml

@@ -16,20 +16,30 @@ on:
 
 jobs:
   lint:
-    name: Lint Python Code Base with Ruff
+    name: Lint Python Code Base
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
-      - uses: astral-sh/ruff-action@278981a28ce3188b1e39527901f38254bf3aac89 # v3.6.1
+      - name: Lint Python Code (Ruff)
+        uses: astral-sh/ruff-action@278981a28ce3188b1e39527901f38254bf3aac89 # v3.6.1
         with:
           version: "latest"
           args: "check"
           src: "./revrt"
-      - uses: astral-sh/ruff-action@278981a28ce3188b1e39527901f38254bf3aac89 # v3.6.1
+      - name: Check Python Code Format (Ruff)
+        uses: astral-sh/ruff-action@278981a28ce3188b1e39527901f38254bf3aac89 # v3.6.1
         with:
           version: "latest"
           args: "format --check"
           src: "./revrt"
+      - name: Check Python Code Complexity (Complexipy)
+        uses: rohaquinlop/complexipy-action@e2b05bcc06d899a24e2b6bb8b1354ac42800a95e # v7.0.1
+        with:
+          paths: "./revrt"
+          max_complexity_allowed: 10
+          failed: false  # true
+          sort: desc
+          ignore_complexity: false  # Set to true to ignore complexity checks
 
   locked-tests:
     needs: lint

pixi.lock

@@ -53,6 +53,7 @@ dependencies = [
 
 [project.optional-dependencies]
 dev = [
+  "complexipy>=7.0.1,<8",
   "contextily>=1.7.0,<2",
   "datashader>=0.19.0,<0.20",
   "geoviews>=1.15.1,<2",
@@ -294,6 +295,9 @@ rust-src = ">=1.96.1,<1.97"
 maturin = ">=1.13.1,<2"
 rattler-build = ">=0.69.1,<0.70"
 
+[tool.pixi.feature.dev.pypi-dependencies]
+complexipy = ">=7.0.1,<8"
+
 [tool.pixi.feature.test.dependencies]
 hypothesis = ">=6.152.1,<7"
 pytest = ">=9.0.0,<9.1"
@@ -382,3 +386,12 @@ omit = [
 [tool.pytest.ini_options]
 addopts = "--disable-warnings"
 testpaths = ["tests/python/unit", "tests/python/integration"]
+
+
+[tool.complexipy]
+paths = ["revrt"]
+max-complexity-allowed = 10
+exclude = ["tests/**"]
+failed = true
+sort = "desc"
+check-script = true

pyproject.toml

@@ -319,6 +319,40 @@ def _compute_multipliers(
         default_multipliers=None,
     ):
         """Create costs multiplier raster"""
+        multipliers, regions_mask = self._process_iso_multipliers(
+            iso_multipliers,
+            iso_lookup,
+            iso_layer,
+            land_use_layer,
+            slope_layer,
+            land_use_classes,
+        )
+        if default_multipliers is not None:
+            multipliers = self._process_default_region(
+                multipliers,
+                regions_mask,
+                default_multipliers,
+                land_use_layer,
+                slope_layer,
+                land_use_classes,
+            )
+
+        # Set water multiplier last so we don't get super high
+        # multipliers at water body boundaries next to steep slopes
+        return da.where(
+            land_use_layer == WATER_NLCD_CODE, WATER_MULTIPLIER, multipliers
+        )
+
+    def _process_iso_multipliers(
+        self,
+        iso_multipliers,
+        iso_lookup,
+        iso_layer,
+        land_use_layer,
+        slope_layer,
+        land_use_classes,
+    ):
+        """Create ISO multipliers"""
         multipliers = da.ones(
             self.shape, dtype=self._dtype, chunks=self.chunks
         )
@@ -357,43 +391,42 @@ def _compute_multipliers(
                 multipliers = da.where(
                     mask, multipliers * slope_multipliers, multipliers
                 )
+        return multipliers, regions_mask
 
-        # Calculate multipliers for regions not defined in `config`
+    def _process_default_region(
+        self,
+        multipliers,
+        regions_mask,
+        default_multipliers,
+        land_use_layer,
+        slope_layer,
+        land_use_classes,
+    ):
+        """Calculate multipliers for regions not defined in `config`"""
         logger.debug("Processing default region")
-        if default_multipliers is not None:
-            default_mask = ~regions_mask
+        default_mask = ~regions_mask
 
-            if "land_use" in default_multipliers:
-                region_land_use = da.where(
-                    default_mask, land_use_layer, da.nan
-                )
-                lum_dict = default_multipliers["land_use"]
-                lum = compute_land_use_multipliers(
-                    region_land_use,
-                    lum_dict,
-                    land_use_classes,
-                    chunks=self.chunks,
-                )
-                multipliers = da.where(
-                    default_mask, multipliers * lum, multipliers
-                )
-
-            if "slope" in default_multipliers:
-                region_slope = da.where(default_mask, slope_layer, da.nan)
-                slope_multipliers = compute_slope_multipliers(
-                    region_slope,
-                    chunks=self.chunks,
-                    config=default_multipliers["slope"],
-                )
-                multipliers = da.where(
-                    default_mask, multipliers * slope_multipliers, multipliers
-                )
+        if "land_use" in default_multipliers:
+            region_land_use = da.where(default_mask, land_use_layer, da.nan)
+            lum_dict = default_multipliers["land_use"]
+            lum = compute_land_use_multipliers(
+                region_land_use, lum_dict, land_use_classes, chunks=self.chunks
+            )
+            multipliers = da.where(
+                default_mask, multipliers * lum, multipliers
+            )
 
-        # Set water multiplier last so we don't get super high
-        # multipliers at water body boundaries next to steep slopes
-        return da.where(
-            land_use_layer == WATER_NLCD_CODE, WATER_MULTIPLIER, multipliers
-        )
+        if "slope" in default_multipliers:
+            region_slope = da.where(default_mask, slope_layer, da.nan)
+            slope_multipliers = compute_slope_multipliers(
+                region_slope,
+                chunks=self.chunks,
+                config=default_multipliers["slope"],
+            )
+            multipliers = da.where(
+                default_mask, multipliers * slope_multipliers, multipliers
+            )
+        return multipliers
 
     def _compute_base_line_costs(
         self, capacity, base_line_costs, iso_layer, iso_lookup

revrt/costs/dry_costs_creator.py

@@ -452,31 +452,8 @@ def _process_forced_inclusions(self, data, fi_layers, tiff_chunks="file"):
         fi = da.zeros(self.shape, dtype=self._dtype, chunks=self.chunks)
 
         for fname, config in fi_layers.items():
-            if Path(fname).suffix.lower() not in TIFF_EXTENSIONS:
-                msg = (
-                    f"Forced inclusion file {fname!r} does not end with .tif."
-                    " GeoTIFFs are the only format allowed for forced "
-                    "inclusions."
-                )
-                raise revrtValueError(msg)
-
-            global_value_given = config.global_value is not None
-            map_given = config.map is not None
-            range_given = config.bins is not None
-            rasterize_given = config.rasterize is not None
-            bad_input_given = (
-                global_value_given
-                or map_given
-                or range_given
-                or rasterize_given
-            )
-            if bad_input_given:
-                msg = (
-                    "`global_value`, `map`, `bins`, and `rasterize` are "
-                    "not allowed if `forced_inclusion` is True, but one "
-                    f"was found in config: {fname!r}: {config}"
-                )
-                raise revrtValueError(msg)
+            _validate_fi_file_extension(fname)
+            _validate_fi_config_input(config, fname)
 
             # Past guard clauses, process FI
             if config.extent != ALL:
@@ -550,6 +527,35 @@ def _check_tiff_layer_config(config, fname):
         raise revrtValueError(msg)
 
 
+def _validate_fi_file_extension(fname):
+    """Validate that the forced inclusion file has correct extension"""
+    if Path(fname).suffix.lower() not in TIFF_EXTENSIONS:
+        msg = (
+            f"Forced inclusion file {fname!r} does not end with .tif."
+            " GeoTIFFs are the only format allowed for forced "
+            "inclusions."
+        )
+        raise revrtValueError(msg)
+
+
+def _validate_fi_config_input(config, fname):
+    """Validate that the forced inclusion config is correct"""
+    global_value_given = config.global_value is not None
+    map_given = config.map is not None
+    range_given = config.bins is not None
+    rasterize_given = config.rasterize is not None
+    bad_input_given = (
+        global_value_given or map_given or range_given or rasterize_given
+    )
+    if bad_input_given:
+        msg = (
+            "`global_value`, `map`, `bins`, and `rasterize` are "
+            "not allowed if `forced_inclusion` is True, but one "
+            f"was found in config: {fname!r}: {config}"
+        )
+        raise revrtValueError(msg)
+
+
 def _validate_bin_range(bins):
     """Check for correctness in bin range"""
     for input_bin in bins:
@@ -570,26 +576,38 @@ def _validate_bin_continuity(bins):
     sorted_bins = sorted(bins, key=lambda x: x.min)
     last_max = float("-inf")
     for i, input_bin in enumerate(sorted_bins):
-        if input_bin.min < last_max:
-            last_bin = sorted_bins[i - 1] if i > 0 else "-infinity"
-            msg = (
-                "Overlapping bins detected between "
-                f"{last_bin!r} and {input_bin!r}"
-            )
-            warn(msg, revrtWarning)
+        last_bin = sorted_bins[i - 1] if i > 0 else "-infinity"
 
-        if input_bin.min > last_max:
-            last_bin = sorted_bins[i - 1] if i > 0 else "-infinity"
-            msg = f"Gap detected between {last_bin!r} and {input_bin!r}"
-            warn(msg, revrtWarning)
-
-        if i + 1 == len(sorted_bins) and input_bin.max < float("inf"):
-            msg = f"Gap detected between {input_bin!r} and 'infinity'"
-            warn(msg, revrtWarning)
+        _warn_about_overlapping_bins(input_bin, last_bin, last_max)
+        _warn_about_gap_in_bins(input_bin, last_bin, last_max)
+        _warn_about_unbounded_bins(i, sorted_bins, input_bin)
 
         last_max = input_bin.max
 
 
+def _warn_about_overlapping_bins(input_bin, last_bin, last_max):
+    """Warn about overlapping bins"""
+    if input_bin.min < last_max:
+        msg = (
+            f"Overlapping bins detected between {last_bin!r} and {input_bin!r}"
+        )
+        warn(msg, revrtWarning)
+
+
+def _warn_about_gap_in_bins(input_bin, last_bin, last_max):
+    """Warn about gaps in bin continuity"""
+    if input_bin.min > last_max:
+        msg = f"Gap detected between {last_bin!r} and {input_bin!r}"
+        warn(msg, revrtWarning)
+
+
+def _warn_about_unbounded_bins(i, sorted_bins, input_bin):
+    """Warn if the last bin is not unbounded"""
+    if i + 1 == len(sorted_bins) and input_bin.max < float("inf"):
+        msg = f"Gap detected between {input_bin!r} and 'infinity'"
+        warn(msg, revrtWarning)
+
+
 def _vector_raster_dtype(burn_value, default_dtype):
     """Choose a compact dtype for vector rasterization"""
     try:

revrt/costs/layer_creator.py

@@ -235,6 +235,22 @@ def _empty_cost_layer_data_array(self):
 
     def _build_cost_layer_from_option(self, option, config):
         """Build a single routing option's cost layer"""
+        cost_layers = self._build_cost_layers_single_option(config, option)
+        scaled_layers = self._scale_single_option_cost_layers(
+            config, *cost_layers
+        )
+        option_cost, option_li_cost, option_untracked_cost = scaled_layers
+
+        self.costs[option] = option_cost
+        self.li_costs[option] = option_li_cost
+        self._build_final_routing_layer_from_option(
+            option,
+            config,
+            option_cost + option_li_cost + option_untracked_cost,
+        )
+
+    def _build_cost_layers_single_option(self, config, option):
+        """Build cost layers for a single routing option"""
         option_cost = self._empty_cost_layer_data_array()
         option_li_cost = self._empty_cost_layer_data_array()
         option_untracked_cost = self._empty_cost_layer_data_array()
@@ -243,14 +259,10 @@ def _build_cost_layer_from_option(self, option, config):
             cost = self._extract_and_scale_layer(layer_info)
             cost.values = da.where(cost > 0, cost, 0)
             is_li = layer_info.get("is_invariant", False)
-            if layer_info.get("include_in_final_cost", True):
-                if is_li:
-                    option_li_cost += cost
-                else:
-                    option_cost += cost
-            else:
-                option_untracked_cost += cost
-
+            base_cost_layer = _select_base_cost_layer(
+                layer_info, option_li_cost, option_cost, option_untracked_cost
+            )
+            base_cost_layer += cost
             if layer_info.get("include_in_report", True):
                 report_key = (layer_info["layer_name"], is_li)
                 reported_costs[report_key] = (
@@ -267,6 +279,12 @@ def _build_cost_layer_from_option(self, option, config):
                 )
             )
 
+        return option_cost, option_li_cost, option_untracked_cost
+
+    def _scale_single_option_cost_layers(
+        self, config, option_cost, option_li_cost, option_untracked_cost
+    ):
+        """Scale the cost layers for a single routing option"""
         mult = config.get("cost_multiplier_scalar", 1) or 1
         option_cost *= mult
         option_li_cost *= mult
@@ -281,13 +299,7 @@ def _build_cost_layer_from_option(self, option, config):
             option_li_cost *= multiplier
             option_untracked_cost *= multiplier
 
-        self.costs[option] = option_cost
-        self.li_costs[option] = option_li_cost
-        self._build_final_routing_layer_from_option(
-            option,
-            config,
-            option_cost + option_li_cost + option_untracked_cost,
-        )
+        return option_cost, option_li_cost, option_untracked_cost
 
     def _build_final_routing_layer_from_option(
         self, option, config, option_layer
@@ -828,3 +840,14 @@ def _driver_zones_for_rust(zones):
             out_zone["mask_operator"] = mask_operator
             out_zone["mask_threshold"] = mask_threshold
         yield out_zone
+
+
+def _select_base_cost_layer(
+    layer_info, option_li_cost, option_cost, option_untracked_cost
+):
+    """Select the appropriate base cost layer based on config"""
+    if layer_info.get("include_in_final_cost", True):
+        if layer_info.get("is_invariant", False):
+            return option_li_cost
+        return option_cost
+    return option_untracked_cost