From 98ff7fc15cbd8c19e65c4080896c7a0238f9b669 Mon Sep 17 00:00:00 2001 From: Ardt Klapwijk <59741981+ArdtK@users.noreply.github.com> Date: Mon, 28 Oct 2024 14:53:38 +0100 Subject: [PATCH] feat: 90 implement logic to sort/filter the reinforcement types (#202) --- .../summary/koswat_summary_builder.py | 3 +- .../koswat_summary_location_matrix_builder.py | 63 ++++++++++++++--- koswat/strategies/README.md | 9 ++- .../order_strategy/order_strategy.py | 67 +++++++++++++++++-- koswat/strategies/strategy_input.py | 2 + .../strategy_reinforcement_input.py | 12 ++++ .../strategy_reinforcement_type_costs.py | 4 +- ..._koswat_summary_location_matrix_builder.py | 14 +++- tests/strategies/conftest.py | 16 +++-- .../order_strategy/test_order_strategy.py | 67 ++++++++++++++++++- 10 files changed, 225 insertions(+), 32 deletions(-) create mode 100644 koswat/strategies/strategy_reinforcement_input.py diff --git a/koswat/cost_report/summary/koswat_summary_builder.py b/koswat/cost_report/summary/koswat_summary_builder.py index 805c7ff9..21c1188d 100644 --- a/koswat/cost_report/summary/koswat_summary_builder.py +++ b/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, ) diff --git a/koswat/cost_report/summary/koswat_summary_location_matrix_builder.py b/koswat/cost_report/summary/koswat_summary_location_matrix_builder.py index 3fcd4fd4..9b447dbc 100644 --- a/koswat/cost_report/summary/koswat_summary_location_matrix_builder.py +++ b/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,7 +138,8 @@ 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(), ) @@ -104,4 +147,4 @@ def to_strategy_location( logging.info("Finalized locations-reinforcements matrix.") - return _strategy_locations + return _strategy_locations, _strategy_reinforcements_list diff --git a/koswat/strategies/README.md b/koswat/strategies/README.md index e2ef9bfd..0526bdeb 100644 --- a/koswat/strategies/README.md +++ b/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. diff --git a/koswat/strategies/order_strategy/order_strategy.py b/koswat/strategies/order_strategy/order_strategy.py index 844008a9..eb78fc3b 100644 --- a/koswat/strategies/order_strategy/order_strategy.py +++ b/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 ) diff --git a/koswat/strategies/strategy_input.py b/koswat/strategies/strategy_input.py index 8f551798..429c80e1 100644 --- a/koswat/strategies/strategy_input.py +++ b/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 diff --git a/koswat/strategies/strategy_reinforcement_input.py b/koswat/strategies/strategy_reinforcement_input.py new file mode 100644 index 00000000..84cb3e17 --- /dev/null +++ b/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 diff --git a/koswat/strategies/strategy_reinforcement_type_costs.py b/koswat/strategies/strategy_reinforcement_type_costs.py index 5f5088c5..190f807d 100644 --- a/koswat/strategies/strategy_reinforcement_type_costs.py +++ b/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: diff --git a/tests/cost_report/summary/test_koswat_summary_location_matrix_builder.py b/tests/cost_report/summary/test_koswat_summary_location_matrix_builder.py index a39b6db5..21cd67a0 100644 --- a/tests/cost_report/summary/test_koswat_summary_location_matrix_builder.py +++ b/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 + ) diff --git a/tests/strategies/conftest.py b/tests/strategies/conftest.py index f19f4d2d..f59eeb2d 100644 --- a/tests/strategies/conftest.py +++ b/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, ) diff --git a/tests/strategies/order_strategy/test_order_strategy.py b/tests/strategies/order_strategy/test_order_strategy.py index 83eb0bf3..51d49c05 100644 --- a/tests/strategies/order_strategy/test_order_strategy.py +++ b/tests/strategies/order_strategy/test_order_strategy.py @@ -1,3 +1,7 @@ +from copy import deepcopy + +import pytest + from koswat.dike_reinforcements.reinforcement_profile import ( CofferdamReinforcementProfile, PipingWallReinforcementProfile, @@ -38,15 +42,74 @@ def test_get_default_order_for_reinforcements(self): # 3. Verify expectations assert _result == _expected_result - def test_get_strategy_order_for_reinforcements( + def test_get_strategy_order_given_example_returns_default_order( self, + example_strategy_input: StrategyInput, ): # 1. Define test data. _expected_result = self._default_reinforcements _strategy = OrderStrategy() # 2. Run test. - _reinforcements = _strategy.get_strategy_order_for_reinforcements() + _reinforcements = _strategy.get_strategy_order_for_reinforcements( + example_strategy_input.strategy_reinforcements + ) + + # 3. Verify expectations + assert _reinforcements == _expected_result + assert _reinforcements[-1] == CofferdamReinforcementProfile + + @pytest.mark.parametrize( + "idx", range(len(_default_reinforcements) - 1), ids=_default_reinforcements[:-1] + ) + def test_get_strategy_order_increased_cost_filters_reinforcement( + self, + idx: int, + example_strategy_input: StrategyInput, + ): + # 1. Define test data. + # Increase the cost of the reinforcement at the given index + # to become more expensive than the next (more restrictive) reinforcement + # and will be filtered out. + example_strategy_input.strategy_reinforcements[idx].base_costs *= 20 + _expected_result = deepcopy(self._default_reinforcements) + _expected_result.remove(self._default_reinforcements[idx]) + + _strategy = OrderStrategy() + + # 2. Run test. + _reinforcements = _strategy.get_strategy_order_for_reinforcements( + example_strategy_input.strategy_reinforcements + ) + + # 3. Verify expectations + assert _reinforcements == _expected_result + assert _reinforcements[-1] == CofferdamReinforcementProfile + + @pytest.mark.parametrize( + "idx", + range(1, len(_default_reinforcements) - 1), + ids=_default_reinforcements[1:-1], + ) + def test_get_strategy_order_increased_surface_filters_reinforcement( + self, + idx: int, + example_strategy_input: StrategyInput, + ): + # 1. Define test data. + # Reduce the surface of the reinforcement at the given index + # to become less restrictive than the previous (cheaper) reinforcement + # and will be filtered out. + example_strategy_input.strategy_reinforcements[idx].ground_level_surface += 15 + _expected_result = deepcopy(self._default_reinforcements) + _expected_result.remove(self._default_reinforcements[idx]) + + _strategy = OrderStrategy() + + # 2. Run test. + _reinforcements = _strategy.get_strategy_order_for_reinforcements( + example_strategy_input.strategy_reinforcements + ) # 3. Verify expectations assert _reinforcements == _expected_result