feat: 595 adaptation run and combine losses and damages

ra2ce/analysis/adaptation/adaptation.py

@@ -27,65 +27,90 @@
     AdaptationOptionCollection,
 )
 from ra2ce.analysis.analysis_config_data.analysis_config_data import (
-    AnalysisConfigData,
     AnalysisSectionAdaptation,
 )
+from ra2ce.analysis.analysis_config_wrapper import AnalysisConfigWrapper
 from ra2ce.analysis.analysis_input_wrapper import AnalysisInputWrapper
 from ra2ce.analysis.damages.analysis_damages_protocol import AnalysisDamagesProtocol
 from ra2ce.network.graph_files.network_file import NetworkFile
 
 
 class Adaptation(AnalysisDamagesProtocol):
+    """
+    Execute the adaptation analysis.
+    For each adaptation option a damages and losses analysis is executed.
+    """
+
     analysis: AnalysisSectionAdaptation
     graph_file: NetworkFile
-    graph_file_hazard: NetworkFile
     input_path: Path
     output_path: Path
     adaptation_collection: AdaptationOptionCollection
 
     # TODO: add the proper protocol for the adaptation analysis.
     def __init__(
-        self, analysis_input: AnalysisInputWrapper, analysis_config: AnalysisConfigData
+        self,
+        analysis_input: AnalysisInputWrapper,
+        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.input_path = analysis_input.input_path
-        self.output_path = analysis_input.output_path
         self.adaptation_collection = AdaptationOptionCollection.from_config(
             analysis_config
         )
 
     def execute(self) -> GeoDataFrame:
         """
         Run the adaptation analysis.
+
+        Returns:
+            GeoDataFrame: The result of the adaptation analysis.
         """
         return self.calculate_bc_ratio()
 
     def run_cost(self) -> GeoDataFrame:
         """
-        Calculate the cost for all adaptation options.
+        Calculate the unit cost for all adaptation options.
+
+        Returns:
+            GeoDataFrame: The result of the cost calculation.
         """
-        # Open the network without hazard data
         _cost_gdf = deepcopy(self.graph_file.get_graph())
+
         for (
             _option,
             _cost,
-        ) in self.adaptation_collection.calculate_option_cost().items():
-            _cost_gdf[f"costs_{_option.id}"] = _cost
+        ) in self.adaptation_collection.calculate_options_cost().items():
+            _cost_gdf[f"{_option.id}_cost"] = _cost
+
+        # TODO: calculate link cost instead of unit cost
 
         return _cost_gdf
 
     def run_benefit(self) -> GeoDataFrame:
         """
-        Calculate the benefit for all adaptation options
+        Calculate the benefit for all adaptation options.
+
+        Returns:
+            GeoDataFrame: The result of the benefit calculation.
         """
-        return None
+        _benefit_gdf = deepcopy(self.graph_file.get_graph())
+
+        return self.adaptation_collection.calculation_options_impact(_benefit_gdf)
 
     def calculate_bc_ratio(self) -> GeoDataFrame:
         """
-        Calculate the benefit-cost ratio for all adaptation options
+        Calculate the benefit-cost ratio for all adaptation options.
+
+        Returns:
+            GeoDataFrame: The result of the benefit-cost ratio calculation.
         """
         _cost_gdf = self.run_cost()
         _benefit_gdf = self.run_benefit()
-        return None
+
+        # TODO: apply economic discounting
+        # TODO: calculate B/C ratio
+        # TODO: apply overlay
+
+        return _cost_gdf

ra2ce/analysis/adaptation/adaptation_option.py

@@ -20,15 +20,20 @@
 """
 from __future__ import annotations
 
-from copy import deepcopy
 from dataclasses import asdict, dataclass
-from pathlib import Path
 
+from geopandas import GeoDataFrame
+
+from ra2ce.analysis.adaptation.adaptation_option_analysis import (
+    AdaptationOptionAnalysis,
+)
 from ra2ce.analysis.analysis_config_data.analysis_config_data import (
     AnalysisSectionAdaptationOption,
-    AnalysisSectionDamages,
-    AnalysisSectionLosses,
 )
+from ra2ce.analysis.analysis_config_data.enums.analysis_damages_enum import (
+    AnalysisDamagesEnum,
+)
+from ra2ce.analysis.analysis_config_wrapper import AnalysisConfigWrapper
 
 
 @dataclass
@@ -39,71 +44,56 @@ class AdaptationOption:
     construction_interval: float
     maintenance_cost: float
     maintenance_interval: float
-    damages_root: Path
-    damages_config: AnalysisSectionDamages
-    losses_root: Path
-    losses_config: AnalysisSectionLosses
+    analyses: list[AdaptationOptionAnalysis]
+    analysis_config: AnalysisConfigWrapper
 
     def __hash__(self) -> int:
         return hash(self.id)
 
-    @property
-    def input_path(self) -> Path:
-        return self.damages_root.joinpath("input")
-
-    @property
-    def static_path(self) -> Path:
-        return self.damages_root.joinpath("static")
-
-    @property
-    def output_path(self) -> Path:
-        return self.damages_root.joinpath("output")
-
     @classmethod
     def from_config(
         cls,
-        root_path: Path,
+        analysis_config: AnalysisConfigWrapper,
         adaptation_option: AnalysisSectionAdaptationOption,
-        damages_section: AnalysisSectionDamages,
-        losses_section: AnalysisSectionLosses,
     ) -> AdaptationOption:
-        # Adjust path to the input files
-        def extend_path(analysis: str, input_path: Path) -> Path:
-            if not input_path:
-                return None
-            # Input is directory: add stuff at the end
-            if not (input_path.suffix):
-                return input_path.joinpath("input", adaptation_option.id, analysis)
-            return input_path.parent.joinpath(
-                "input", adaptation_option.id, analysis, input_path.name
-            )
+        """
+        Classmethod to create an AdaptationOption from an analysis configuration and an adaptation option.
+
+        Args:
+            analysis_config (AnalysisConfigWrapper): Analysis config input
+            adaptation_option (AnalysisSectionAdaptationOption): Adaptation option input
 
-        if not damages_section or not losses_section:
+        Raises:
+            ValueError: If damages and losses sections are not present in the analysis config data.
+
+        Returns:
+            AdaptationOption: The created adaptation option.
+        """
+        if (
+            not analysis_config.config_data.damages_list
+            or not analysis_config.config_data.losses_list
+        ):
             raise ValueError(
                 "Damages and losses sections are required to create an adaptation option."
             )
 
-        _damages_root = extend_path("damages", root_path)
-        _damages_section = deepcopy(damages_section)
-
-        _losses_root = extend_path("losses", root_path)
-        _losses_section = deepcopy(losses_section)
-        _losses_section.resilience_curves_file = extend_path(
-            "losses", losses_section.resilience_curves_file
-        )
-        _losses_section.traffic_intensities_file = extend_path(
-            "losses", losses_section.traffic_intensities_file
-        )
-        _losses_section.values_of_time_file = extend_path(
-            "losses", losses_section.values_of_time_file
-        )
+        # Create input for the analyses
+        _analyses = [
+            AdaptationOptionAnalysis.from_config(
+                analysis_config=analysis_config,
+                analysis_type=_analysis,
+                option_id=adaptation_option.id,
+            )
+            for _analysis in [
+                AnalysisDamagesEnum.DAMAGES,
+                analysis_config.config_data.adaptation.losses_analysis,
+            ]
+        ]
 
         return cls(
             **asdict(adaptation_option),
-            damages_root=_damages_root,
-            damages_config=_damages_section,
-            losses_root=_losses_root,
-            losses_config=_losses_section,
+            analyses=_analyses,
+            analysis_config=analysis_config,
         )
 
     def calculate_cost(self, time_horizon: float, discount_rate: float) -> float:
@@ -148,3 +138,21 @@ def calc_cost(cost: float, year: float) -> float:
                 _lifetime_cost += calc_cost(self.maintenance_cost, _maint_year)
 
         return _lifetime_cost
+
+    def calculate_impact(self, benefit_graph: GeoDataFrame) -> GeoDataFrame:
+        """
+        Calculate the impact of the adaptation option.
+
+        Returns:
+            float: The impact of the adaptation option.
+        """
+        for _analysis in self.analyses:
+            _result = _analysis.execute(self.analysis_config)
+            _col = _result.filter(regex=_analysis.result_col).columns[0]
+            benefit_graph[f"{self.id}_{_col}"] = _result[_col]
+
+        # Calculate the impact (summing the damages and losses values)
+        _option_cols = benefit_graph.filter(regex=f"{self.id}_").columns
+        benefit_graph[f"{self.id}_impact"] = benefit_graph[_option_cols].sum(axis=1)
+
+        return benefit_graph

ra2ce/analysis/adaptation/adaptation_option_analysis.py

@@ -0,0 +1,137 @@
+"""
+                    GNU GENERAL PUBLIC LICENSE
+                      Version 3, 29 June 2007
+
+    Risk Assessment and Adaptation for Critical Infrastructure (RA2CE).
+    Copyright (C) 2023 Stichting Deltares
+
+    This program is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    This program is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+
+    You should have received a copy of the GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""
+from __future__ import annotations
+
+from copy import deepcopy
+from dataclasses import dataclass
+
+from geopandas import GeoDataFrame
+
+from ra2ce.analysis.analysis_config_data.enums.analysis_damages_enum import (
+    AnalysisDamagesEnum,
+)
+from ra2ce.analysis.analysis_config_data.enums.analysis_losses_enum import (
+    AnalysisLossesEnum,
+)
+from ra2ce.analysis.analysis_config_wrapper import AnalysisConfigWrapper
+from ra2ce.analysis.analysis_input_wrapper import AnalysisInputWrapper
+from ra2ce.analysis.damages.damages import Damages
+from ra2ce.analysis.losses.losses_base import LossesBase
+from ra2ce.analysis.losses.multi_link_losses import MultiLinkLosses
+from ra2ce.analysis.losses.single_link_losses import SingleLinkLosses
+
+
+@dataclass
+class AdaptationOptionAnalysis:
+    analysis_class: type[Damages | LossesBase]
+    analysis_input: AnalysisInputWrapper
+    result_col: str
+
+    @staticmethod
+    def get_analysis(
+        analysis_type: AnalysisDamagesEnum | AnalysisLossesEnum,
+    ) -> tuple[type[Damages | LossesBase], str]:
+        """
+        Get the analysis class and the result column for the given analysis.
+
+        Args:
+            analysis (AnalysisDamagesEnum | AnalysisLossesEnum): The type of analysis.
+
+        Raises:
+            NotImplementedError: The analysis type is not implemented.
+
+        Returns:
+            tuple[type[Damages | LossesBase], str]: The analysis class and the result column.
+        """
+        if analysis_type == AnalysisDamagesEnum.DAMAGES:
+            return (Damages, "dam_")
+        elif analysis_type == AnalysisLossesEnum.SINGLE_LINK_LOSSES:
+            return (SingleLinkLosses, "vlh_.*_total")
+        elif analysis_type == AnalysisLossesEnum.MULTI_LINK_LOSSES:
+            return (MultiLinkLosses, "vlh_.*_total")
+        raise NotImplementedError(f"Analysis {analysis_type} not implemented")
+
+    @classmethod
+    def from_config(
+        cls,
+        analysis_config: AnalysisConfigWrapper,
+        analysis_type: AnalysisLossesEnum | AnalysisDamagesEnum,
+        option_id: str,
+    ) -> AdaptationOptionAnalysis:
+        """
+        Classmethod to create an AdaptationOptionAnalysis from an analysis configuration.
+
+        Args:
+            analysis_config (AnalysisConfigWrapper): The analysis configuration.
+            analysis_type (AnalysisLossesEnum | AnalysisDamagesEnum): The type of analysis.
+            option_id (str): The ID of the adaptation option.
+
+        Returns:
+            AdaptationOptionAnalysis: The created AdaptationOptionAnalysis.
+        """
+
+        # Need a deepcopy to avoid mixing up configs across analyses.
+        _analysis_config = deepcopy(analysis_config)
+        _analysis_config.config_data = (
+            _analysis_config.config_data.reroot_analysis_config(
+                analysis_type,
+                analysis_config.config_data.root_path.joinpath("input", option_id),
+            )
+        )
+
+        # Create analysis input
+        _analysis = _analysis_config.config_data.get_analysis(analysis_type)
+        if analysis_type == AnalysisDamagesEnum.DAMAGES:
+            _graph_file = None
+            _graph_file_hazard = analysis_config.graph_files.base_network_hazard
+        else:
+            _graph_file = analysis_config.graph_files.base_graph
+            _graph_file_hazard = analysis_config.graph_files.base_graph_hazard
+
+        _analysis_input = AnalysisInputWrapper.from_input(
+            analysis=_analysis,
+            analysis_config=_analysis_config,
+            graph_file=_graph_file,
+            graph_file_hazard=_graph_file_hazard,
+        )
+
+        # Create output object
+        _analysis_class, _result_col = cls.get_analysis(analysis_type)
+
+        return cls(
+            analysis_class=_analysis_class,
+            analysis_input=_analysis_input,
+            result_col=_result_col,
+        )
+
+    def execute(self, analysis_config: AnalysisConfigWrapper) -> GeoDataFrame:
+        """
+        Execute the analysis.
+
+        Args:
+            analysis_config (AnalysisConfigWrapper): The config for the analysis.
+
+        Returns:
+            DataFrame: The results of the analysis.
+        """
+        if self.analysis_class == Damages:
+            return self.analysis_class(self.analysis_input).execute()
+        return self.analysis_class(self.analysis_input, analysis_config).execute()