598 adaptation calculate impact adaptation option for time horizon

ra2ce/analysis/adaptation/adaptation_option.py

@@ -139,7 +139,9 @@ def calc_cost(cost: float, year: float) -> float:
 
         return _lifetime_cost
 
-    def calculate_impact(self, benefit_graph: GeoDataFrame) -> GeoDataFrame:
+    def calculate_impact(
+        self, benefit_graph: GeoDataFrame, net_present_value_factor: float
+    ) -> GeoDataFrame:
         """
         Calculate the impact of the adaptation option.
 
@@ -157,4 +159,7 @@ def calculate_impact(self, benefit_graph: GeoDataFrame) -> GeoDataFrame:
         _option_cols = benefit_graph.filter(regex=f"{self.id}_").columns
         benefit_graph[f"{self.id}_impact"] = benefit_graph[_option_cols].sum(axis=1)
 
+        # convert event impact into time-horizon impact
+        benefit_graph[f"{self.id}_impact"] *= net_present_value_factor
+
         return benefit_graph

ra2ce/analysis/adaptation/adaptation_option_collection.py

@@ -22,6 +22,7 @@
 
 from dataclasses import dataclass, field
 
+import numpy as np
 from geopandas import GeoDataFrame
 
 from ra2ce.analysis.adaptation.adaptation_option import AdaptationOption
@@ -91,6 +92,19 @@ def from_config(
 
         return _collection
 
+    def get_net_present_value_factor(self) -> float:
+        """
+        Calculate the net present value factor for the entire time horizon. To be multiplied to the event impact to
+        obtain the net present value.
+        """
+        _years_array = np.arange(0, self.time_horizon)
+        _frequency_per_year = (
+            self.initial_frequency + _years_array * self.climate_factor
+        )
+        _discount = (1 + self.discount_rate) ** _years_array
+        _ratio = _frequency_per_year / _discount
+        return _ratio.sum()
+
     def calculate_options_unit_cost(self) -> dict[AdaptationOption, float]:
         """
         Calculate the unit cost for all adaptation options.
@@ -116,7 +130,11 @@ def calculation_options_impact(self, benefit_graph: GeoDataFrame) -> GeoDataFram
         Returns:
             NetworkFile: The calculated impact of all adaptation options.
         """
+        net_present_value_factor = self.get_net_present_value_factor()
+
         for _option in self.all_options:
-            benefit_graph = _option.calculate_impact(benefit_graph)
+            benefit_graph = _option.calculate_impact(
+                benefit_graph, net_present_value_factor
+            )
 
         return benefit_graph

tests/analysis/adaptation/conftest.py

@@ -166,6 +166,8 @@ def get_losses_section(analysis: AnalysisLossesEnum) -> AnalysisSectionLosses:
         adaptation_options=AdaptationOptionCases.config_cases,
         discount_rate=0.025,
         time_horizon=20,
+        climate_factor=0.00036842,
+        initial_frequency=0.01,
     )
 
     _analysis_data = AnalysisConfigData(

tests/analysis/adaptation/test_adaptation_option_collection.py

@@ -60,3 +60,22 @@ def test_calculate_options_unit_cost(
         # 3. Verify expectations.
         assert isinstance(_result, dict)
         assert all(_option in _result for _option in _collection.adaptation_options)
+
+    def test_calculate_correct_get_net_present_value_factor(
+        self,
+        valid_adaptation_config: tuple[AnalysisInputWrapper, AnalysisConfigWrapper],
+    ):
+        # 1. Define test data.
+        _config_wrapper = valid_adaptation_config[1]
+        assert isinstance(_config_wrapper, AnalysisConfigWrapper)
+        _collection = AdaptationOptionCollection.from_config(_config_wrapper)
+
+        # 2. Run test.
+        _result = _collection.get_net_present_value_factor()
+
+        # 3. Verify expectations.
+        assert isinstance(_result, float)
+        assert _result == pytest.approx(0.2109011023, rel=1e-9)
+
+
+