Feat/618 adaptation add option to adaptation config to calculate cost based on fraction segment exposed

ra2ce/analysis/adaptation/adaptation.py

@@ -43,7 +43,7 @@ class Adaptation(AnalysisBase, AnalysisDamagesProtocol):
     """
 
     analysis: AnalysisSectionAdaptation
-    graph_file: NetworkFile
+    graph_file_hazard: NetworkFile
     input_path: Path
     output_path: Path
     adaptation_collection: AdaptationOptionCollection
@@ -55,7 +55,6 @@ def __init__(
         analysis_config: AnalysisConfigWrapper,
     ):
         self.analysis = analysis_input.analysis
-        self.graph_file = analysis_input.graph_file
         self.graph_file_hazard = analysis_input.graph_file_hazard
         self.adaptation_collection = AdaptationOptionCollection.from_config(
             analysis_config
@@ -78,13 +77,14 @@ def execute(self) -> AnalysisResultWrapper:
     def run_cost(self) -> GeoDataFrame:
         """
         Calculate the link cost for all adaptation options.
+        The unit cost is multiplied by the length of the link.
+        If the hazard fraction cost is enabled, the cost is multiplied by the fraction of the link that is impacted.
 
-        Returns:
-            GeoDataFrame: The result of the cost calculation.
         Returns:
             GeoDataFrame: The result of the cost calculation.
         """
-        _orig_gdf = self.graph_file.get_graph()
+        _orig_gdf = self.graph_file_hazard.get_graph()
+        _fraction_col = _orig_gdf.filter(regex="EV.*_fr").columns[0]
 
         _cost_gdf = GeoDataFrame()
         for (
@@ -94,6 +94,9 @@ def run_cost(self) -> GeoDataFrame:
             _cost_gdf[f"{_option.id}_cost"] = _orig_gdf.apply(
                 lambda x, cost=_cost: x["length"] * cost, axis=1
             )
+            # Only calculate the cost for the impacted fraction of the links.
+            if self.analysis.hazard_fraction_cost:
+                _cost_gdf[f"{_option.id}_cost"] *= _orig_gdf[_fraction_col]
 
         return _cost_gdf
 

ra2ce/analysis/analysis_config_data/analysis_config_data.py

@@ -152,11 +152,12 @@ class AnalysisSectionAdaptation(AnalysisSectionBase):
     analysis: AnalysisEnum = AnalysisEnum.ADAPTATION
     losses_analysis: AnalysisLossesEnum = AnalysisLossesEnum.SINGLE_LINK_LOSSES
     # Economical settings
-    discount_rate: float = 0.0
     time_horizon: float = 0.0
+    discount_rate: float = 0.0
     # Hazard settings
-    climate_factor: float = 0.0
     initial_frequency: float = 0.0
+    climate_factor: float = 0.0
+    hazard_fraction_cost: bool = False
     # First option is the no adaptation option
     adaptation_options: list[AnalysisSectionAdaptationOption] = field(
         default_factory=list

ra2ce/analysis/analysis_config_data/analysis_config_data_reader.py

@@ -236,6 +236,11 @@ def _get_adaptation_option(
         ) = AnalysisLossesEnum.get_enum(
             self._parser.get(section_name, "losses_analysis", fallback=None)
         )
+        _section.hazard_fraction_cost = self._parser.getboolean(
+            section_name,
+            "hazard_fraction_costs",
+            fallback=_section.hazard_fraction_cost,
+        )
 
         _adaptation_options = list(
             _adaptation_option

ra2ce/analysis/analysis_factory.py

@@ -222,7 +222,7 @@ def get_adaptation_analysis(
         _analysis_input = AnalysisInputWrapper.from_input(
             analysis=analysis,
             analysis_config=analysis_config,
-            graph_file=analysis_config.graph_files.base_network,
+            graph_file_hazard=analysis_config.graph_files.base_network_hazard,
         )
 
         if analysis.analysis == AnalysisEnum.ADAPTATION:

tests/analysis/adaptation/conftest.py

@@ -64,7 +64,7 @@ class AdaptationOptionCases:
             maintenance_interval=3.0,
         ),
     ]
-    total_cost: list[float] = [0.0, 97800589.027952, 189253296.099491]
+    total_cost: list[float] = [0.0, 10073869.180362, 19493880.004279]
     total_benefit: list[float] = [0.0, 0.0, 0.0]
     cases: list[tuple[AnalysisSectionAdaptationOption, tuple[float, float]]] = list(
         zip(config_cases, zip(total_cost, total_benefit))
@@ -167,10 +167,11 @@ def get_losses_section(analysis: AnalysisLossesEnum) -> AnalysisSectionLosses:
         name="Adaptation",
         losses_analysis=AnalysisLossesEnum.MULTI_LINK_LOSSES,
         adaptation_options=AdaptationOptionCases.config_cases,
-        discount_rate=0.025,
         time_horizon=20,
-        climate_factor=0.00036842,
+        discount_rate=0.025,
         initial_frequency=0.01,
+        climate_factor=0.00036842,
+        hazard_fraction_cost=True,
     )
 
     _analysis_data = AnalysisConfigData(

tests/test_data/acceptance_test_data/analyses.ini

@@ -53,10 +53,11 @@ save_csv = True
 name = adaptation module
 analysis = adaptation
 losses_analysis = single_link_losses
-discount_rate = 0.05
 time_horizon = 10
-climate_factor = 0.2
+discount_rate = 0.05
 initial_frequency = 0.2
+climate_factor = 0.2
+hazard_fraction_cost = True
 save_gpkg = True
 save_csv = True