feat: 592 adaptation create class adaptation_option_collection (#609)

ra2ce/analysis/adaptation/adaptation_option.py

@@ -0,0 +1,86 @@
+"""
+                    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
+
+import math
+from copy import deepcopy
+from dataclasses import dataclass
+from pathlib import Path
+
+from ra2ce.analysis.analysis_config_data.analysis_config_data import (
+    AnalysisSectionAdaptationOption,
+    AnalysisSectionDamages,
+    AnalysisSectionLosses,
+)
+
+
+@dataclass
+class AdaptationOption:
+    id: str
+    name: str
+    construction_cost: float
+    maintenance_interval: float
+    maintenance_cost: float
+    damages_config: AnalysisSectionDamages = None
+    losses_config: AnalysisSectionLosses = None
+
+    @classmethod
+    def from_config(
+        cls,
+        adaptation_option: AnalysisSectionAdaptationOption,
+        damages_section: AnalysisSectionDamages,
+        losses_section: AnalysisSectionLosses,
+    ) -> AdaptationOption:
+        # Adjust path to the input files
+        def extend_path(analysis: str, input_path: Path | None) -> Path | None:
+            if not input_path:
+                return None
+            return input_path.parent.joinpath(
+                "input", adaptation_option.id, analysis, input_path.name
+            )
+
+        if not damages_section or not losses_section:
+            raise ValueError(
+                "Damages and losses sections are required to create an adaptation option."
+            )
+
+        _damages_section = deepcopy(damages_section)
+
+        _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
+        )
+
+        return cls(
+            id=adaptation_option.id,
+            name=adaptation_option.name,
+            construction_cost=adaptation_option.construction_cost,
+            maintenance_interval=adaptation_option.maintenance_interval,
+            maintenance_cost=adaptation_option.maintenance_cost,
+            damages_config=_damages_section,
+            losses_config=_losses_section,
+        )

ra2ce/analysis/adaptation/adaptation_option_collection.py

@@ -0,0 +1,87 @@
+"""
+                    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 dataclasses import dataclass, field
+
+from ra2ce.analysis.adaptation.adaptation_option import AdaptationOption
+from ra2ce.analysis.analysis_config_data.analysis_config_data import AnalysisConfigData
+from ra2ce.analysis.analysis_config_data.enums.analysis_damages_enum import (
+    AnalysisDamagesEnum,
+)
+
+
+@dataclass
+class AdaptationOptionCollection:
+    """
+    Collection of adaptation options with all their related properties.
+    """
+
+    discount_rate: float = 0.0
+    time_horizon: float = 0.0
+    vat: float = 0.0
+    climate_factor: float = 0.0
+    initial_frequency: float = 0.0
+    all_options: list[AdaptationOption] = field(default_factory=list)
+
+    @property
+    def reference_option(self) -> AdaptationOption:
+        if not self.all_options:
+            return None
+        return self.all_options[0]
+
+    @property
+    def adaptation_options(self) -> list[AdaptationOption]:
+        if len(self.all_options) < 2:
+            return []
+        return self.all_options[1:]
+
+    @classmethod
+    def from_config(
+        cls,
+        analysis_config_data: AnalysisConfigData,
+    ) -> AdaptationOptionCollection:
+        if not analysis_config_data.adaptation:
+            raise ValueError("No adaptation section found in the analysis config data.")
+        _collection = cls(
+            discount_rate=analysis_config_data.adaptation.discount_rate,
+            time_horizon=analysis_config_data.adaptation.time_horizon,
+            vat=analysis_config_data.adaptation.vat,
+            climate_factor=analysis_config_data.adaptation.climate_factor,
+            initial_frequency=analysis_config_data.adaptation.initial_frequency,
+        )
+
+        _damages_analysis = analysis_config_data.get_analysis(
+            AnalysisDamagesEnum.DAMAGES
+        )
+        _losses_analysis = analysis_config_data.get_analysis(
+            analysis_config_data.adaptation.losses_analysis
+        )
+        for _config_option in analysis_config_data.adaptation.adaptation_options:
+            _collection.all_options.append(
+                AdaptationOption.from_config(
+                    _config_option,
+                    _damages_analysis,
+                    _losses_analysis,
+                )
+            )
+
+        return _collection

ra2ce/analysis/analysis_config_data/analysis_config_data.py

@@ -156,10 +156,6 @@ class AnalysisSectionAdaptation(AnalysisSectionBase):
     vat: float = 0.0
     climate_factor: float = 0.0
     initial_frequency: float = 0.0
-    # The option to not implement any adaptation measure
-    no_adaptation_option: AnalysisSectionAdaptationOption = field(
-        default_factory=lambda: AnalysisSectionAdaptationOption()
-    )
     adaptation_options: list[AnalysisSectionAdaptationOption] = field(
         default_factory=list
     )
@@ -224,17 +220,29 @@ def losses_list(self) -> list[AnalysisSectionLosses]:
         )
 
     @property
-    def adaptation(self) -> AnalysisSectionAdaptation:
+    def adaptation(self) -> AnalysisSectionAdaptation | None:
         """
         Get the adaptation analysis from config.
 
         Returns:
             AnalysisSectionAdaptation: Adaptation analysis.
         """
         return next(
-            filter(lambda x: isinstance(x, AnalysisSectionAdaptation), self.analyses)
+            filter(lambda x: isinstance(x, AnalysisSectionAdaptation), self.analyses),
+            None,
         )
 
+    def get_analysis(
+        self, analysis: AnalysisEnum | AnalysisDamagesEnum | AnalysisLossesEnum
+    ) -> AnalysisSectionBase | None:
+        """
+        Get a certain analysis from config.
+
+        Returns:
+            AnalysisSectionBase: The analysis.
+        """
+        return next(filter(lambda x: x.analysis == analysis, self.analyses), None)
+
     @staticmethod
     def get_data_output(ini_file: Path) -> Path:
         return ini_file.parent.joinpath("output")

ra2ce/analysis/analysis_config_data/analysis_config_data_reader.py

@@ -243,11 +243,7 @@ def _get_adaptation_option(
             for _adaptation_option in self._parser.sections()
             if "adaptationoption" in _adaptation_option
         )
-        if len(_adaptation_options) > 0:
-            _section.no_adaptation_option = _get_adaptation_option(
-                _adaptation_options[0]
-            )
-        for _adaptation_option in _adaptation_options[1:]:
+        for _adaptation_option in _adaptation_options:
             _section.adaptation_options.append(
                 _get_adaptation_option(_adaptation_option)
             )

tests/analysis/adaptation/conftest.py

@@ -0,0 +1,85 @@
+from shutil import copytree, rmtree
+from typing import Iterator
+
+import pytest
+
+from ra2ce.analysis.analysis_config_data.analysis_config_data import (
+    AnalysisConfigData,
+    AnalysisSectionAdaptation,
+    AnalysisSectionAdaptationOption,
+    AnalysisSectionDamages,
+    AnalysisSectionLosses,
+)
+from ra2ce.analysis.analysis_config_data.enums.analysis_damages_enum import (
+    AnalysisDamagesEnum,
+)
+from ra2ce.analysis.analysis_config_data.enums.analysis_enum import AnalysisEnum
+from ra2ce.analysis.analysis_config_data.enums.analysis_losses_enum import (
+    AnalysisLossesEnum,
+)
+from tests import test_data, test_results
+
+
+@pytest.fixture(name="valid_adaptation_config")
+def _get_valid_adaptation_config_fixture(
+    request: pytest.FixtureRequest,
+) -> Iterator[AnalysisConfigData]:
+    _adaptation_options = ["AO0", "AO1", "AO2"]
+    _root_path = test_results.joinpath(request.node.name, "adaptation")
+    _input_path = _root_path.joinpath("input")
+    _static_path = _root_path.joinpath("static")
+    _output_path = _root_path.joinpath("output")
+
+    # Create the input files
+    if _root_path.exists():
+        rmtree(_root_path)
+
+    _input_path.mkdir(parents=True)
+    for _option in _adaptation_options:
+        _ao_path = _input_path.joinpath(_option)
+        copytree(test_data.joinpath("adaptation", "input"), _ao_path)
+    copytree(test_data.joinpath("adaptation", "static"), _static_path)
+
+    # Create the config
+    # - damages
+    _damages_section = AnalysisSectionDamages(
+        analysis=AnalysisDamagesEnum.DAMAGES,
+    )
+    # - losses
+    _losses_section = AnalysisSectionLosses(
+        analysis=AnalysisLossesEnum.SINGLE_LINK_LOSSES,
+        resilience_curves_file=_root_path.joinpath(
+            "damage_functions", "resilience_curves.csv"
+        ),
+        traffic_intensities_file=_root_path.joinpath(
+            "damage_functions", "traffic_intensities.csv"
+        ),
+        values_of_time_file=_root_path.joinpath(
+            "damage_functions", "values_of_time.csv"
+        ),
+    )
+    # - adaptation
+    _adaptation_collection = []
+    for i, _option in enumerate(_adaptation_options):
+        _adaptation_collection.append(
+            AnalysisSectionAdaptationOption(
+                id=_option,
+                name=f"Option {i}",
+                construction_cost=1000.0,
+                maintenance_interval=5.0,
+                maintenance_cost=100.0,
+            )
+        )
+    _adaptation_section = AnalysisSectionAdaptation(
+        analysis=AnalysisEnum.ADAPTATION,
+        losses_analysis=AnalysisLossesEnum.SINGLE_LINK_LOSSES,
+        adaptation_options=_adaptation_collection,
+    )
+
+    yield AnalysisConfigData(
+        root_path=_root_path,
+        input_path=_input_path,
+        static_path=_static_path,
+        output_path=_output_path,
+        analyses=[_damages_section, _losses_section, _adaptation_section],
+    )

tests/analysis/adaptation/test_adaptation_option.py

@@ -0,0 +1,60 @@
+import pytest
+
+from ra2ce.analysis.adaptation.adaptation_option import AdaptationOption
+from ra2ce.analysis.analysis_config_data.analysis_config_data import (
+    AnalysisConfigData,
+    AnalysisSectionAdaptation,
+)
+from ra2ce.analysis.analysis_config_data.enums.analysis_damages_enum import (
+    AnalysisDamagesEnum,
+)
+from ra2ce.analysis.analysis_config_data.enums.analysis_losses_enum import (
+    AnalysisLossesEnum,
+)
+
+
+class TestAdaptationOption:
+    def test_from_config(self, valid_adaptation_config: AnalysisConfigData):
+        # 1. Define test data.
+        _orig_path = valid_adaptation_config.losses_list[0].resilience_curves_file
+        _expected_path = _orig_path.parent.joinpath(
+            "input",
+            valid_adaptation_config.adaptation.adaptation_options[0].id,
+            "losses",
+            _orig_path.name,
+        )
+
+        # 2. Run test.
+        _option = AdaptationOption.from_config(
+            adaptation_option=valid_adaptation_config.adaptation.adaptation_options[0],
+            damages_section=valid_adaptation_config.get_analysis(
+                AnalysisDamagesEnum.DAMAGES
+            ),
+            losses_section=valid_adaptation_config.get_analysis(
+                AnalysisLossesEnum.SINGLE_LINK_LOSSES
+            ),
+        )
+
+        # 3. Verify expectations.
+        assert isinstance(_option, AdaptationOption)
+        assert _option.id == "AO0"
+        assert _option.damages_config.analysis == AnalysisDamagesEnum.DAMAGES
+        assert _option.losses_config.analysis == AnalysisLossesEnum.SINGLE_LINK_LOSSES
+        assert _option.losses_config.resilience_curves_file == _expected_path
+
+    def test_from_config_no_damages_losses_raises(self):
+        # 1. Define test data.
+        _config = AnalysisConfigData()
+
+        # 2. Run test.
+        with pytest.raises(ValueError) as _exc:
+            AdaptationOption.from_config(
+                adaptation_option=AnalysisSectionAdaptation(),
+                damages_section=None,
+                losses_section=None,
+            )
+
+        # 3. Verify expectations.
+        assert _exc.match(
+            "Damages and losses sections are required to create an adaptation option."
+        )