feat: 90 implement logic to sort/filter the reinforcement types (#202)

koswat/cost_report/summary/koswat_summary_builder.py

@@ -98,12 +98,13 @@ def _get_final_reinforcement_per_location(
         locations_profile_report_list: list[MultiLocationProfileCostReport],
         available_locations: list[PointSurroundings],
     ) -> dict[PointSurroundings, ReinforcementProfile]:
-        _matrix = KoswatSummaryLocationMatrixBuilder(
+        _matrix, _reinforcements = KoswatSummaryLocationMatrixBuilder(
             available_locations=available_locations,
             locations_profile_report_list=locations_profile_report_list,
         ).build()
         _strategy_input = StrategyInput(
             strategy_locations=_matrix,
+            strategy_reinforcements=_reinforcements,
             reinforcement_min_buffer=self.run_scenario_settings.surroundings.obstacle_surroundings_wrapper.reinforcement_min_buffer,
             reinforcement_min_length=self.run_scenario_settings.surroundings.obstacle_surroundings_wrapper.reinforcement_min_separation,
         )

koswat/cost_report/summary/koswat_summary_location_matrix_builder.py

@@ -8,6 +8,7 @@
 )
 from koswat.dike.surroundings.point.point_surroundings import PointSurroundings
 from koswat.strategies.strategy_input import StrategyLocationInput
+from koswat.strategies.strategy_reinforcement_input import StrategyReinforcementInput
 from koswat.strategies.strategy_reinforcement_type_costs import (
     StrategyReinforcementTypeCosts,
 )
@@ -39,7 +40,6 @@ def create_strategy_location_reinforcement_costs():
                     locations_profile.profile_cost_report.reinforced_profile
                 ),
                 base_costs=locations_profile.profile_cost_report.total_cost,
-                ground_level_surface=locations_profile.profile_cost_report.reinforced_profile.new_ground_level_surface,
             )
 
         _dict_matrix = defaultdict(create_strategy_location_reinforcement_costs)
@@ -61,7 +61,16 @@ def _get_list_summary_matrix_for_locations_with_reinforcements(
 
     def build(
         self,
-    ) -> list[StrategyLocationInput]:
+    ) -> tuple[list[StrategyLocationInput], list[StrategyReinforcementInput]]:
+        """
+        Build the locations-reinforcements matrix.
+
+        Returns:
+            tuple[list[StrategyLocationInput], list[StrategyReinforcementInput]]:
+                Tuple containing:
+                - list[StrategyLocationInput]: The locations-reinforcements matrix.
+                - list[StrategyReinforcementInput]: The list of applied reinforcements.
+        """
         # 1. First we get all the possible reinforcements per point.
 
         logging.info("Initalizing locations-reinforcements matrix.")
@@ -71,15 +80,48 @@ def build(
 
         # 2. Then we initialize the matrix with all available locations,
         # but no reinforcements.
-
-        _strategy_locations = dict((_ps, []) for _ps in self.available_locations)
+        _strategy_locations_dict = dict((_ps, []) for _ps in self.available_locations)
+        # Initialize the list of reinforcement types used in the strategy.
+        _strategy_reinforcements_list = []
 
         # 3. Last, we merge the reinforcements dictionary into the matrix.
-        for _location in _strategy_locations.keys():
+        # Next to this we build a list of reinforcements used in the strategy.
+        def get_reinforcement(
+            reinforcement_cost: StrategyReinforcementTypeCosts,
+        ) -> StrategyReinforcementInput:
+            _reinforcement = next(
+                (
+                    _pcr.profile_cost_report.reinforced_profile
+                    for _pcr in self.locations_profile_report_list
+                    if type(_pcr.profile_cost_report.reinforced_profile)
+                    == reinforcement_cost.reinforcement_type
+                ),
+                None,
+            )
+            if not _reinforcement:
+                raise ValueError(
+                    f"Reinforcement type {reinforcement_cost.reinforcement_type} not found in profile reports."
+                )
+            return StrategyReinforcementInput(
+                reinforcement_type=reinforcement_cost.reinforcement_type,
+                base_costs=reinforcement_cost.base_costs,
+                ground_level_surface=_reinforcement.new_ground_level_surface,
+            )
+
+        for _loc_key, _strat_locs in _strategy_locations_dict.items():
             for _reinforce_matrix_dict in _reinforce_matrix_dict_list:
-                if _location in _reinforce_matrix_dict:
-                    _strategy_locations[_location].append(
-                        _reinforce_matrix_dict[_location]
+                # Add the reinforcement to the matrix if location exists.
+                if _loc_key in _reinforce_matrix_dict:
+                    _strat_locs.append(_reinforce_matrix_dict[_loc_key])
+                else:
+                    continue
+                # Add to the reinforcement to the list if not already present.
+                if (
+                    _reinforce_matrix_dict[_loc_key]
+                    not in _strategy_reinforcements_list
+                ):
+                    _strategy_reinforcements_list.append(
+                        get_reinforcement(_reinforce_matrix_dict[_loc_key])
                     )
 
         # 4. Sort matrix by traject order for normalized usage in Koswat.
@@ -96,12 +138,13 @@ def to_strategy_location(
                 to_strategy_location,
                 dict(
                     sorted(
-                        _strategy_locations.items(), key=lambda x: x[0].traject_order
+                        _strategy_locations_dict.items(),
+                        key=lambda x: x[0].traject_order,
                     )
                 ).items(),
             )
         )
 
         logging.info("Finalized locations-reinforcements matrix.")
 
-        return _strategy_locations
+        return _strategy_locations, _strategy_reinforcements_list

koswat/strategies/README.md

@@ -4,12 +4,13 @@ This modules contains the logic to choose which measure will be applied for a gi
 
 - `StrategyInput`, wraps all required properties to apply a strategy: 
     - strategy_locations (`list[StrategyLocationInput]`), contains all the available reinforcements that can be applied at each location and their related costs.
+    - strategy_reinforcements (`list[StrategyReinforcementInput]`), all the reinforcements that can be used at this location and their related costs and width.
     - structure_min_buffer (`float`), how many extra meters a structure requires to support its reinforcement.
     - structure_min_length (`float`), how many minimal meters are required for a structure to "exist". This rule can, at times, have certain exceptions.
 
 - `StrategyLocationInput`, gathers all the input required for a strategy to determine which reinforcement can be applied based on its location (`point_surrounding`) and `available_reinforcements`.
     - point_surrounding (`PointSurroundings`), a point (meter) in the dike traject.
-    - strategy_reinforcement_type_costs (`list[StrategyReinforcementTypeCosts]`), all the reinforcements that can be used at this location and their related costs and width.
+    - strategy_reinforcement (`list[StrategyReinforcementInput]`), all the reinforcements that can be used at this location and their related cost and width.
     - cheapest_reinforcement (`StrategyReinforcementTypeCosts`), returns which "available reinforcment" has the lower total costs at this location.
     - available_measures (`Type[ReinforcementProfileProtocol]`), returns only the reinforcement type from the `strategy_reinforcement_type_costs` collection.
 
@@ -18,6 +19,10 @@ This modules contains the logic to choose which measure will be applied for a gi
     - base_costs (`float`), the costs only related to the reinforcement's required space (thus excluding infrastructure costs).
     - infrastructure_costs (`float`), the costs associated **only** to infrastructures.
     - total_costs (`float`), the addition of `base_costs` and `infrastructure_costs`.
+
+- `StrategyReinforcementInput`, contains the reinforcement types that are relevant to the strategy, as they were selected for one or more locations included in the strategy.
+    - reinforcement_type (`Type[ReinforcementProfileProtocol]`), the mapped reinforcement type.
+    - base_costs (`float`), the costs only related to the reinforcement's required space (thus excluding infrastructure costs).
     - ground_level_surface (`float`), profile's width from outside (waterside) crest point.
 
 - `StrategyLocationReinforcement`, represents a mapped location to a selected measure.
@@ -31,6 +36,6 @@ This modules contains the logic to choose which measure will be applied for a gi
 
 The following strategies are currently available, please refer to the official documentation for a more in-detail explanation of each of them:
 
-- [__Default__] Order based (`OrderBased`). A strategy is chosen based on a pre-defined measure priority order.
+- [__Default__] Order based (`OrderBased`). A strategy is chosen based on a dynamically determined order of reinforcements. This order is determined from least to most restrictive, where reinforcements are omitted when they are less restrictive and more expensive than other reinforcement(s). Cofferdam is forced as the last reinforcement of this order.
 - Infra-priority based (`InfraPriorityStrategy`). Clusters are created based on the cheapest total cost (including infrastructure reworks). This strategy is applied __after__  _Order based_, the clusters are then modified into a reinforcement that requires less space (thus more expensive) but induce less infrastructure costs, therefore becoming cheaper.
 

koswat/strategies/order_strategy/order_strategy.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from itertools import pairwise
+
 from koswat.dike_reinforcements.reinforcement_profile import (
     CofferdamReinforcementProfile,
     PipingWallReinforcementProfile,
@@ -21,6 +23,7 @@
     StrategyLocationReinforcement,
 )
 from koswat.strategies.strategy_protocol import StrategyProtocol
+from koswat.strategies.strategy_reinforcement_input import StrategyReinforcementInput
 
 
 class OrderStrategy(StrategyProtocol):
@@ -31,7 +34,7 @@ def get_default_order_for_reinforcements() -> list[
         """
         Give the default order for reinforcements types,
         assuming they are sorted from cheapest to most expensive
-        and least restrictive to most restrictive.
+        and least to most restrictive.
 
         Returns:
             list[type[ReinforcementProfileProtocol]]: list of reinforcement types
@@ -46,18 +49,66 @@ def get_default_order_for_reinforcements() -> list[
 
     def get_strategy_order_for_reinforcements(
         self,
+        strategy_reinforcements: list[StrategyReinforcementInput],
     ) -> list[type[ReinforcementProfileProtocol]]:
         """
-        Give the ordered reinforcement types for this strategy,
-        from cheapest to most expensive,
-        possibly omitting reinforcement types that are more expensive and more restrictive than others.
+        Give the ordered reinforcement types for this strategy, from cheapest to most expensive,
+        possibly removing reinforcement types that are more expensive and more restrictive than others.
         Cofferdam should always be the last reinforcement type.
 
+        Input:
+            strategy_reinforcements (list[StrategyReinforcementInput]): list of reinforcement types with costs and surface
+
         Returns:
             list[type[ReinforcementProfileProtocol]]: list of reinforcement types
         """
-        # TODO Implement this method
-        return self.get_default_order_for_reinforcements()
+        if not strategy_reinforcements:
+            return []
+
+        def split_reinforcements() -> tuple[
+            list[StrategyReinforcementInput], list[StrategyReinforcementInput]
+        ]:
+            _last, _other = [], []
+            for obj in strategy_reinforcements:
+                if not obj:
+                    continue
+                if obj.reinforcement_type == CofferdamReinforcementProfile:
+                    _last.append(obj)
+                else:
+                    _other.append(obj)
+
+            return (_other, _last)
+
+        # Split in a list to be sorted (least to most restrictive) and a list to be put last (Cofferdam for now)
+        _unsorted, _last = split_reinforcements()
+        _sorted = sorted(
+            _unsorted,
+            key=lambda x: (x.ground_level_surface, x.base_costs),
+            reverse=True,
+        )
+
+        def check_reinforcement(
+            pair: tuple[StrategyReinforcementInput, StrategyReinforcementInput],
+        ) -> StrategyReinforcementInput | None:
+            # Only keep the less restrictive reinforcement if it is cheaper
+            if (
+                pair[0].ground_level_surface > pair[1].ground_level_surface
+                and pair[0].base_costs < pair[1].base_costs
+            ):
+                return pair[0]
+            return None
+
+        # Check if the current (more expensive) reinforcement is more restrictive than the previous
+        # (the last needs to be appended as it is always kept)
+        _sorted_pairs = pairwise(_sorted + _last)
+        _kept = (
+            list(
+                filter(lambda x: x is not None, map(check_reinforcement, _sorted_pairs))
+            )
+            + _last
+        )
+
+        return [x.reinforcement_type for x in _kept]
 
     @staticmethod
     def get_strategy_reinforcements(
@@ -84,7 +135,9 @@ def get_strategy_reinforcements(
     def apply_strategy(
         self, strategy_input: StrategyInput
     ) -> list[StrategyLocationReinforcement]:
-        _reinforcement_order = self.get_strategy_order_for_reinforcements()
+        _reinforcement_order = self.get_strategy_order_for_reinforcements(
+            strategy_input.strategy_reinforcements
+        )
         _strategy_reinforcements = self.get_strategy_reinforcements(
             strategy_input.strategy_locations, _reinforcement_order
         )

koswat/strategies/strategy_input.py

@@ -1,10 +1,12 @@
 from dataclasses import dataclass
 
 from koswat.strategies.strategy_location_input import StrategyLocationInput
+from koswat.strategies.strategy_reinforcement_input import StrategyReinforcementInput
 
 
 @dataclass
 class StrategyInput:
     strategy_locations: list[StrategyLocationInput]
+    strategy_reinforcements: list[StrategyReinforcementInput]
     reinforcement_min_buffer: float
     reinforcement_min_length: float

koswat/strategies/strategy_reinforcement_input.py

@@ -0,0 +1,12 @@
+from dataclasses import dataclass
+
+from koswat.dike_reinforcements.reinforcement_profile.reinforcement_profile_protocol import (
+    ReinforcementProfileProtocol,
+)
+
+
+@dataclass
+class StrategyReinforcementInput:
+    reinforcement_type: type[ReinforcementProfileProtocol]
+    base_costs: float = 0.0
+    ground_level_surface: float = 0.0

koswat/strategies/strategy_reinforcement_type_costs.py

@@ -1,5 +1,4 @@
 from dataclasses import dataclass
-from typing import Type
 
 from koswat.dike_reinforcements.reinforcement_profile.reinforcement_profile_protocol import (
     ReinforcementProfileProtocol,
@@ -8,10 +7,9 @@
 
 @dataclass
 class StrategyReinforcementTypeCosts:
-    reinforcement_type: Type[ReinforcementProfileProtocol]
+    reinforcement_type: type[ReinforcementProfileProtocol]
     base_costs: float = 0.0
     infrastructure_costs: float = 0.0
-    ground_level_surface: float = 0.0
 
     @property
     def total_costs(self) -> float:

tests/cost_report/summary/test_koswat_summary_location_matrix_builder.py

@@ -32,7 +32,7 @@ def test_given_no_profile_report_list_then_returns_expected_matrix(self):
         _builder.locations_profile_report_list = []
 
         # 2. Run test.
-        _strategy_locations = _builder.build()
+        _strategy_locations, _strategy_reinforcements = _builder.build()
 
         # 3. Verify final expectations.
         assert isinstance(_strategy_locations, list)
@@ -41,6 +41,9 @@ def test_given_no_profile_report_list_then_returns_expected_matrix(self):
             assert _sl.point_surrounding in _builder.available_locations
             assert _sl.strategy_reinforcement_type_costs == []
 
+        assert isinstance(_strategy_reinforcements, list)
+        assert len(_strategy_reinforcements) == 0
+
     def test_given_profile_report_list_then_returns_expected_matrix(self):
         class MyMockedReinforcementProfile(ReinforcementProfileProtocol):
             output_name: str = "MockedReinforcementProfile"
@@ -65,7 +68,7 @@ class MyMockedReinforcementProfile(ReinforcementProfileProtocol):
         _builder.locations_profile_report_list = [_profile_report]
 
         # 2. Run test.
-        _strategy_locations = _builder.build()
+        _strategy_locations, _strategy_reinforcements = _builder.build()
 
         # 3. Verify final expectations.
         assert isinstance(_strategy_locations, list)
@@ -81,3 +84,10 @@ class MyMockedReinforcementProfile(ReinforcementProfileProtocol):
         for _sl in _strategy_locations[1:]:
             assert _sl.point_surrounding in _builder.available_locations
             assert _sl.strategy_reinforcement_type_costs == []
+
+        assert isinstance(_strategy_reinforcements, list)
+        assert len(_strategy_reinforcements) == 1
+        assert (
+            _strategy_reinforcements[0].reinforcement_type
+            == MyMockedReinforcementProfile
+        )

tests/strategies/conftest.py

@@ -3,13 +3,9 @@
 import pytest
 
 from koswat.dike.surroundings.point.point_surroundings import PointSurroundings
-from koswat.dike_reinforcements.reinforcement_profile.outside_slope.cofferdam_reinforcement_profile import (
+from koswat.dike_reinforcements.reinforcement_profile import (
     CofferdamReinforcementProfile,
-)
-from koswat.dike_reinforcements.reinforcement_profile.standard.soil_reinforcement_profile import (
     SoilReinforcementProfile,
-)
-from koswat.dike_reinforcements.reinforcement_profile.standard.stability_wall_reinforcement_profile import (
     StabilityWallReinforcementProfile,
 )
 from koswat.strategies.order_strategy.order_strategy import OrderStrategy
@@ -18,6 +14,7 @@
 from koswat.strategies.strategy_location_reinforcement import (
     StrategyLocationReinforcement,
 )
+from koswat.strategies.strategy_reinforcement_input import StrategyReinforcementInput
 from koswat.strategies.strategy_reinforcement_type_costs import (
     StrategyReinforcementTypeCosts,
 )
@@ -67,9 +64,18 @@ def _get_example_strategy_input() -> Iterator[StrategyInput]:
         )
         for _idx, _rt in enumerate(_initial_state_per_location)
     ]
+    _strategy_reinforcements = [
+        StrategyReinforcementInput(
+            reinforcement_type=_rtc.reinforcement_type,
+            base_costs=_rtc.base_costs,
+            ground_level_surface=10 * (len(_reinforcement_type_default_order) - _idx),
+        )
+        for _idx, _rtc in enumerate(_levels_data)
+    ]
 
     yield StrategyInput(
         strategy_locations=_strategy_locations,
+        strategy_reinforcements=_strategy_reinforcements,
         reinforcement_min_buffer=1,
         reinforcement_min_length=5,
     )