Add voltage/polarity-based transition costs (#328)

* Add new type

* Use new type

* Add `validate_transition_cost_configs`

* Update function

* Assume input is validated

* Validator now resolves transition costs

* Update docstrings

* Add tests

* Add tests

* Fix test

* Update directives

* Add test

* move validation

* Assume input is validated

* Fix tests

* Fix test

* Add warning

* Fix tests

* Minor update

Author: ppinchuk

docs/source/guides/routing_layers.md

@@ -142,6 +142,28 @@ other hand, friction values near ``-1`` can be used to represent incentives
 individual friction values are {math}`\gt -1` (i.e. {math}`(1 + F) > 0` is
 upheld), so the cost values themselves can never flip signs.
 
+```{warning}
+Large finite friction values at an endpoint can make routing unexpectedly
+slow and cause cost chunks to be computed far beyond the area between the
+endpoints. The endpoint remains valid, but every complete route must pay its
+large cost. Dijkstra-based algorithms may therefore explore all cells with a
+lower cumulative cost before they can prove that the selected route is
+optimal.
+
+The endpoint itself does not need to have high friction for this to occur. A
+high-friction region that surrounds an endpoint, or otherwise forms an
+unavoidable bottleneck near it, can have the same effect. This is especially
+noticeable with bidirectional routing because the search starting at the
+endpoint must cross that region before its frontier can advance.
+
+Check the effective routing cost {math}`R` at and around every endpoint when
+using large friction multipliers. Prefer moving an endpoint to a nearby
+representative low-friction cell or reducing the friction where every route
+must enter or leave the endpoint. When ``save_routing_layer`` is enabled, the
+saved chunk footprint represents the union of the search area explored for
+the route batch, not only the final routes or their endpoint bounds.
+```
+
 ### Building friction layers
 Friction layers are built similarly to cost layers. A single friction
 layer could be defined as follows:

revrt/models/routing.py

@@ -32,6 +32,9 @@
 type MultiplierLayerInput = list[str] | None
 """One or more layer names used as spatial multipliers"""
 
+type TransitionCostValue = float | dict[str, float | dict[str, float]]
+"""A scalar transition cost or a voltage/polarity-dependent cost"""
+
 
 class RoutingCostLayer(BaseModel, extra="forbid"):
     """Config for one cost layer in a routing option
@@ -481,11 +484,13 @@ class TransitionCostRule(BaseModel, extra="forbid"):
     transition cost applies.
     """
 
-    cost: float
+    cost: TransitionCostValue
     """The transition cost
 
     This is the transition cost (in $) applied when a route switches
-    between the specified options.
+    between the specified options. It may be a scalar, a mapping from
+    voltage to scalar cost, or a mapping from voltage and polarity to
+    scalar cost.
     """
 
 
@@ -496,8 +501,12 @@ class TransitionCostsConfig(BaseModel, extra="forbid"):
     ignored.**
     """
 
-    default: float = 0
-    """Fallback cost applied when no pairwise rule is configured"""
+    default: TransitionCostValue = 0
+    """Fallback cost applied when no pairwise rule is configured
+
+    This may be a scalar, a mapping from voltage to scalar cost, or a
+    mapping from voltage and polarity to scalar cost.
+    """
 
     pairwise: list[TransitionCostRule] = Field(default_factory=list)
     """Explicit transition costs between routing options"""
@@ -538,8 +547,8 @@ def validate_driver_configs(drivers, routing_options):
     return _flatten_driver_config(validated)
 
 
-def validate_transition_cost_configs(transition_costs, routing_options):
-    """[NOT PUBLIC API] Normalize transition costs"""
+def validate_transition_cost_input(transition_costs, routing_options):
+    """[NOT PUBLIC API] Normalize user-supplied transition costs"""
     if transition_costs is None:
         return None
 
@@ -630,8 +639,9 @@ def _validation_error_to_config_error(error):
     "RoutingOptionsMap",
     "TrackedLayer",
     "TransitionCostRule",
+    "TransitionCostValue",
     "TransitionCostsConfig",
     "validate_driver_configs",
     "validate_routing_options",
-    "validate_transition_cost_configs",
+    "validate_transition_cost_input",
 ]

revrt/routing/base.py

@@ -15,11 +15,6 @@
 
 from revrt import simplify_using_slopes
 from revrt.models.cost_layers import BarrierLayer
-from revrt.models.routing import (
-    validate_driver_configs,
-    validate_routing_options,
-    validate_transition_cost_configs,
-)
 from revrt.routing.utilities import compute_lens
 from revrt.utilities.parsing import (
     parse_comparison_values,
@@ -62,11 +57,8 @@ def __init__(
             :class:`~revrt.models.routing.TrackedLayer` for the
             canonical schema.
         routing_options : dict
-            Mapping of routing-option names to dictionaries containing
-            cost, friction, and barrier definitions. See
-            :class:`~revrt.models.routing.RoutingOptionConfig` for the
-            canonical schema that is serialized for the Rust routing
-            core.
+            Validated mapping of routing-option names to dictionaries
+            containing cost, friction, and barrier definitions.
         drivers : dict, optional
             Optional driver-rule configuration keyed by routing option.
             See :class:`~revrt.models.routing.DriverConfig` and
@@ -91,11 +83,9 @@ def __init__(
             By default, ``"bidirectional_long_range_dijkstra"``.
         """
         self.cost_fpath = cost_fpath
-        self.routing_options = validate_routing_options(routing_options)
-        self.drivers = validate_driver_configs(drivers, self.routing_options)
-        self.transition_costs = validate_transition_cost_configs(
-            transition_costs, self.routing_options
-        )
+        self.routing_options = routing_options
+        self.drivers = drivers
+        self.transition_costs = transition_costs
         self.tracked_layers = tracked_layers or []
         self.invalid_costs_block_routing = invalid_costs_block_routing
         self.algorithm = algorithm

revrt/routing/cli/base.py

@@ -14,7 +14,11 @@
 import geopandas as gpd
 import xarray as xr
 
-from revrt.models.routing import validate_routing_options
+from revrt.models.routing import (
+    validate_driver_configs,
+    validate_routing_options,
+    validate_transition_cost_input,
+)
 from revrt.routing.cli.utilities import routing_layer_mover
 from revrt.routing.base import RoutingScenario
 from revrt.routing.processing import BatchRouteProcessor
@@ -64,11 +68,12 @@ def __init__(
         route_points : pandas.DataFrame
             DataFrame defining the points to be routed. This DataFrame
             should contain route definitions to be transformed and
-            passed down to the Rust routing algorithm. Route values for
-            polarity and voltage must be provided either as one shared
-            `polarity` / `voltage` pair that applies to every routing
-            option, or as a full per-option set of
-            `polarity_<option>` / `voltage_<option>` columns.
+            passed down to the Rust routing algorithm. Provide route
+            values for polarity and voltage when cost multipliers or
+            transition costs depend on them. Values may be supplied as
+            one shared ``polarity`` / ``voltage`` pair or as a full
+            per-option set of ``polarity_<option>`` /
+            ``voltage_<option>`` columns.
         out_fp : path-like
             Path to output file where computed routes will be saved.
             This file will be checked for existing routes to avoid
@@ -82,15 +87,27 @@ def __init__(
             path to JSON/JSON5 file containing this dictionary. See the
             description of
             :func:`revrt.routing.cli.point_to_point.compute_lcp_routes`
-            for more details.
+            for more details. By default, ``None``.
+        drivers : dict, optional
+            Optional driver-rule configuration keyed by routing option.
+            By default, ``None``.
+        transition_costs : dict, optional
+            Optional transition-cost configuration between routing
+            options. A voltage/polarity-dependent cost requires matching
+            values for every affected routing option.
+            By default, ``None``.
         """
         self.cost_fpath = cost_fpath
         self.out_fp = Path(out_fp)
         self.transmission_config = transmission_config
-        self.drivers = drivers
-        self.transition_costs = transition_costs
         self._input_route_points = route_points
         self._routing_options = RoutingOptions(routing_options)
+        self.drivers = validate_driver_configs(
+            drivers, self._routing_options.routing_options
+        )
+        self.transition_costs = validate_transition_cost_input(
+            transition_costs, self._routing_options.routing_options
+        )
 
     @cached_property
     def route_points(self):
@@ -169,10 +186,18 @@ def __iter__(self):
                 pv_by_option=pv_by_option,
                 transmission_config=self.transmission_config,
             )
+            transition_costs = _resolve_transition_costs(
+                self.transition_costs, pv_by_option
+            )
             route_definitions, route_attrs = (
                 self._convert_to_route_definitions(routes)
             )
-            yield route_options, route_definitions, route_attrs
+            yield (
+                route_options,
+                transition_costs,
+                route_definitions,
+                route_attrs,
+            )
 
     @property
     def _paths_to_compute(self):
@@ -330,12 +355,17 @@ def _run_all_lcp_batches(
     """Run LCP routing for all batches of routes and save results"""
     out_fp = Path(out_fp)
     save_paths = out_fp.suffix.lower() == ".gpkg"
-    for route_options, route_definitions, route_attrs in routes_to_compute:
+    for (
+        route_options,
+        transition_costs,
+        route_definitions,
+        route_attrs,
+    ) in routes_to_compute:
         scenario = RoutingScenario(
             cost_fpath=cost_fpath,
             routing_options=route_options,
             drivers=routes_to_compute.drivers,
-            transition_costs=routes_to_compute.transition_costs,
+            transition_costs=transition_costs,
             tracked_layers=tracked_layers,
             invalid_costs_block_routing=invalid_costs_block_routing,
             algorithm=algorithm,
@@ -427,6 +457,110 @@ def _handle_maybe_missing_column(points, option, base_col):
     return points
 
 
+def _resolve_transition_costs(transition_costs, pv_by_option):
+    """Resolve transition-cost mappings from route values"""
+    if transition_costs is None:
+        return None
+
+    resolved = {}
+    if "default" in transition_costs:
+        resolved["default"] = _resolve_transition_cost_value(
+            transition_costs["default"],
+            pv_by_option,
+            list(pv_by_option),
+            context="transition_costs.default",
+        )
+
+    if "pairwise" in transition_costs:
+        resolved["pairwise"] = []
+        for rule in transition_costs["pairwise"]:
+            options = rule["between"]
+            resolved["pairwise"].append(
+                {
+                    "between": options,
+                    "cost": _resolve_transition_cost_value(
+                        rule["cost"],
+                        pv_by_option,
+                        options,
+                        context=(
+                            "transition_costs.pairwise cost between "
+                            f"{options[0]!r} and {options[1]!r}"
+                        ),
+                    ),
+                }
+            )
+
+    return resolved
+
+
+def _resolve_transition_cost_value(value, pv_by_option, options, context):
+    """Resolve one scalar or voltage/polarity transition-cost value"""
+    if not isinstance(value, dict):
+        return value
+
+    option_values = {
+        option: _normalized_transition_values(pv_by_option[option], option)
+        for option in options
+    }
+    unique_values = set(option_values.values())
+    if len(unique_values) != 1:
+        received = ", ".join(
+            f"{option}(polarity={polarity!r}, voltage={voltage!r})"
+            for option, (voltage, polarity) in option_values.items()
+        )
+        msg = (
+            f"{context} requires matching voltage and polarity across "
+            f"routing options. Received: {received}"
+        )
+        raise revrtConfigurationError(msg)
+
+    voltage, polarity = next(iter(unique_values))
+    try:
+        resolved = value[voltage]
+    except KeyError as error:
+        msg = (
+            f"{context} has no cost for voltage {voltage!r}. Available "
+            f"voltages: {list(value)}"
+        )
+        raise revrtKeyError(msg) from error
+
+    if not isinstance(resolved, dict):
+        return resolved
+
+    try:
+        return resolved[polarity]
+    except KeyError as error:
+        msg = (
+            f"{context} has no cost for voltage {voltage!r} and polarity "
+            f"{polarity!r}. Available polarities: {list(resolved)}"
+        )
+        raise revrtKeyError(msg) from error
+
+
+def _normalized_transition_values(values, option):
+    """tuple[str, str]: Normalized values for one routing option"""
+    polarity = values.get(_POLARITY)
+    voltage = values.get(_VOLTAGE)
+    unknowns = {None, "None", "unknown"}
+    if polarity in unknowns or voltage in unknowns:
+        msg = (
+            "voltage/polarity-dependent transition costs require known "
+            f"values for routing option {option!r}"
+        )
+        raise revrtConfigurationError(msg)
+
+    try:
+        voltage = str(int(voltage))
+    except (TypeError, ValueError, OverflowError) as error:
+        msg = (
+            "voltage/polarity-dependent transition costs require an integer "
+            f"voltage for routing option {option!r}, received {voltage!r}"
+        )
+        raise revrtConfigurationError(msg) from error
+
+    return voltage, str(polarity)
+
+
 def update_multipliers(
     layers, polarity, voltage, routing_option, transmission_config
 ):

revrt/routing/cli/point_to_feature.py

@@ -283,7 +283,12 @@ def compute_lcp_routes(  # ruff:ignore[too-many-arguments, too-many-positional-a
         Optional transition-cost configuration between routing
         options. See
         :class:`~revrt.models.routing.TransitionCostsConfig` for
-        details.
+        details. The ``default`` and each pairwise ``cost`` value may
+        be a scalar, a mapping from voltage to scalar cost, or a nested
+        mapping from voltage and polarity to cost. A voltage scalar
+        applies to every polarity at that voltage. Dependent pairwise
+        costs require matching voltage and polarity for both options;
+        a dependent default requires matching values for every option.
     tracked_layers : list, optional
         List of dictionaries defining route-characterization layers.
         These layers do not influence the routing objective and are

revrt/routing/cli/point_to_point.py

@@ -152,7 +152,12 @@ def compute_lcp_routes(  # ruff:ignore[too-many-arguments, too-many-positional-a
         Optional transition-cost configuration between routing
         options. See
         :class:`~revrt.models.routing.TransitionCostsConfig` for
-        details.
+        details. The ``default`` and each pairwise ``cost`` value may
+        be a scalar, a mapping from voltage to scalar cost, or a nested
+        mapping from voltage and polarity to cost. A voltage scalar
+        applies to every polarity at that voltage. Dependent pairwise
+        costs require matching voltage and polarity for both options;
+        a dependent default requires matching values for every option.
     tracked_layers : list, optional
         List of dictionaries defining route-characterization layers.
         These layers do not influence the routing objective and are

revrt/utilities/handlers.py

@@ -992,7 +992,7 @@ def _compute_lat_lon(ny, nx, src_crs, transform, chunk_x=2048, chunk_y=2048):
         _proj_to_lon_lat,
         x_mesh_transformed,
         y_mesh_transformed,
-        src_crs.to_string(),
+        src_crs.to_wkt(),
         dtype="float32",
         new_axis=(0,),  # we add a new leading axis of length 2
         chunks=((2,), *x_mesh_transformed.chunks),  # chunk sizes for [2, y, x]