format,doc: cleans up code, adds docstrings

Author: spjuhel

climada/trajectories/impact_calc_strat.py

@@ -124,33 +124,25 @@ def calculate_residual_or_risk_transfer_impact_matrix(
         scipy.sparse.csr_matrix
             The adjusted impact matrix, either residual or risk transfer.
 
-        Example
-        -------
-        >>> calc_residual_or_risk_transf_imp_mat(imp_mat, attachment=100, cover=500, calc_residual=True)
-        Residual impact matrix with applied risk layer adjustments.
         """
+
         if risk_transf_attach and risk_transf_cover:
-            # Make a copy of the impact matrix
             imp_mat = copy.deepcopy(imp_mat)
             # Calculate the total impact per event
             total_at_event = imp_mat.sum(axis=1).A1
             # Risk layer at event
             transfer_at_event = np.minimum(
                 np.maximum(total_at_event - risk_transf_attach, 0), risk_transf_cover
             )
-            # Resiudal impact
             residual_at_event = np.maximum(total_at_event - transfer_at_event, 0)
 
             # Calculate either the residual or transfer impact matrix
             # Choose the denominator to rescale the impact values
             if calc_residual:
-                # Rescale the impact values
                 numerator = residual_at_event
             else:
-                # Rescale the impact values
                 numerator = transfer_at_event
 
-            # Rescale the impact values
             rescale_impact_values = np.divide(
                 numerator,
                 total_at_event,

climada/trajectories/interpolation.py

@@ -1,8 +1,30 @@
+"""
+This file is part of CLIMADA.
+
+Copyright (C) 2017 ETH Zurich, CLIMADA contributors listed in AUTHORS.
+
+CLIMADA 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, version 3.
+
+CLIMADA 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 CLIMADA. If not, see <https://www.gnu.org/licenses/>.
+
+---
+
+This modules implements different sparce matrices interpolation approaches.
+
+"""
+
 import logging
 from abc import ABC, abstractmethod
 
 import numpy as np
-from scipy.sparse import lil_matrix
+from scipy.sparse import csr_matrix, lil_matrix
 
 LOGGER = logging.getLogger(__name__)
 
@@ -21,7 +43,7 @@ def interpolate(self, imp_E0, imp_E1, time_points: int):
         try:
             return self.interpolate_imp_mat(imp_E0, imp_E1, time_points)
         except ValueError as e:
-            if str(e) == "inconsistent shape":
+            if str(e) == "inconsistent shapes":
                 raise ValueError(
                     "Interpolation between impact matrices of different shapes"
                 )
@@ -61,7 +83,7 @@ def interpolate_sm(mat_start, mat_end, time, time_points):
             # Perform the linear interpolation
             mat_interpolated = mat_start + ratio * (mat_end - mat_start)
 
-            return mat_interpolated
+            return csr_matrix(mat_interpolated)
 
         LOGGER.debug(f"imp0: {imp0.imp_mat.data[0]}, imp1: {imp1.imp_mat.data[0]}")
         return [
@@ -70,19 +92,11 @@ def interpolate_sm(mat_start, mat_end, time, time_points):
         ]
 
 
-class ExponentialInterpolation:
+class ExponentialInterpolation(InterpolationStrategy):
     """Exponential interpolation strategy."""
 
     def interpolate(self, imp_E0, imp_E1, time_points: int):
-        try:
-            return self.interpolate_imp_mat(imp_E0, imp_E1, time_points)
-        except ValueError as e:
-            if str(e) == "inconsistent shape":
-                raise ValueError(
-                    "Interpolation between impact matrices of different shapes"
-                )
-            else:
-                raise e
+        return self.interpolate_imp_mat(imp_E0, imp_E1, time_points)
 
     @staticmethod
     def interpolate_imp_mat(imp0, imp1, time_points):
@@ -119,15 +133,9 @@ def interpolate_sm(mat_start, mat_end, time, time_points):
             # Convert back to the original domain using the exponential function
             mat_interpolated = np.exp(log_mat_interpolated)
 
-            return lil_matrix(mat_interpolated)
+            return csr_matrix(mat_interpolated)
 
         return [
             interpolate_sm(imp0.imp_mat, imp1.imp_mat, time, time_points)
             for time in range(time_points)
         ]
-
-
-# Example usage
-# Assuming imp0 and imp1 are instances of ImpactCalc with imp_mat attributes as sparse matrices
-# interpolator = ExponentialInterpolation()
-# interpolated_matrices = interpolator.interpolate(imp0, imp1, 100)

climada/trajectories/risk_trajectory.py

@@ -23,7 +23,6 @@
 import logging
 
 import matplotlib.pyplot as plt
-import numpy as np
 import pandas as pd
 
 from climada.entity.disc_rates.base import DiscRates

climada/trajectories/riskperiod.py

@@ -41,29 +41,63 @@
 
 
 def lazy_property(method):
+    # This function is used as a decorator for properties
+    # that require "heavy" computation and are not always needed
+    # if the property is none, it uses the corresponding computation method
+    # and stores the result in the corresponding private attribute
     attr_name = f"_{method.__name__}"
 
     @property
     def _lazy(self):
         if getattr(self, attr_name) is None:
-            meas_n = self.measure.name if self.measure else "no_measure"
-            LOGGER.debug(
-                f"Computing {method.__name__} for {self._snapshot0.date}-{self._snapshot1.date} with {meas_n}."
-            )
+            # meas_n = self.measure.name if self.measure else "no_measure"
+            # LOGGER.debug(
+            #     f"Computing {method.__name__} for {self._snapshot0.date}-{self._snapshot1.date} with {meas_n}."
+            # )
             setattr(self, attr_name, method(self))
         return getattr(self, attr_name)
 
     return _lazy
 
 
 class CalcRiskPeriod:
-    """Handles the computation of impacts for a risk period."""
+    """Handles the computation of impacts for a risk period.
+
+    This object handles the interpolations and computations of risk metrics in
+    between two given snapshots, along a DateTime index build on
+    `interval_freq` or `time_points`.
+
+    Attributes
+    ----------
+
+    date_idx: pd.DateTimeIndex
+        The date index for the different interpolated points between the two snapshots
+    interpolation_strategy: InterpolationStrategy, optional
+        The approach used to interpolate impact matrices in between the two snapshots, linear by default.
+    impact_computation_strategy: ImpactComputationStrategy, optional
+        The method used to calculate the impact from the (Haz,Exp,Vul) of the two snapshots.
+        Defaults to ImpactCalc
+    risk_transf_attach: float, optional
+        The attachement of risk transfer to apply. Defaults to None.
+    risk_transf_cover: float, optional
+        The cover of risk transfer to apply. Defaults to None.
+    calc_residual: bool, optional
+        A boolean stating whether the residual (True) or transfered risk (False) is retained when doing
+        the risk transfer. Defaults to False.
+    measure: Measure, optional
+        The measure to apply to both snapshots. Defaults to None.
+
+    Notes
+    -----
+
+    This class is intended for internal computation. Users should favor `RiskTrajectory` objects.
+    """
 
     def __init__(
         self,
         snapshot0: Snapshot,
         snapshot1: Snapshot,
-        interval_freq: str | None = "YS",
+        interval_freq: str | None = "AS-JAN",
         time_points: int | None = None,
         interpolation_strategy: InterpolationStrategy | None = None,
         impact_computation_strategy: ImpactComputationStrategy | None = None,
@@ -99,6 +133,7 @@ def _reset_impact_data(self):
         self._imp_mats_E0, self._imp_mats_E1 = None, None
         self._per_date_eai_H0, self._per_date_eai_H1 = None, None
         self._per_date_aai_H0, self._per_date_aai_H1 = None, None
+        self._eai_gdf = None
         self._per_date_return_periods_H0, self._per_date_return_periods_H1 = None, None
 
     @staticmethod
@@ -146,12 +181,18 @@ def _set_date_idx(
             periods=points,
             freq=freq,  # type: ignore
             name=name,
+            normalize=True,
         )
         if periods is not None and len(ret) != periods:
             raise ValueError(
                 "Number of periods and frequency given to date_range are inconsistant"
             )
 
+        if pd.infer_freq(ret) != freq:
+            LOGGER.debug(
+                f"Given interval frequency ( {pd.infer_freq(ret)} ) and infered interval frequency differ ( {freq} )."
+            )
+
         return ret
 
     @property
@@ -171,7 +212,7 @@ def date_idx(self, value, /):
         if not isinstance(value, pd.DatetimeIndex):
             raise ValueError("Not a DatetimeIndex")
 
-        self._date_idx = value
+        self._date_idx = value.normalize()
         self._time_points = len(self.date_idx)
         self._interval_freq = pd.infer_freq(self.date_idx)
         self._prop_H1 = np.linspace(0, 1, num=self.time_points)

climada/trajectories/snapshot.py

@@ -46,9 +46,10 @@ class Snapshot:
     Notes
     -----
 
-    The object creates copies of the exposure hazard and impact function set.
+    The object creates deep copies of the exposure hazard and impact function set.
 
-    To create a snapshot with a measure use Snapshot.apply_measure(measure).
+    To create a snapshot with a measure, create a snapshot `snap` without
+    the measure and call `snap.apply_measure(measure)`, which returns a new Snapshot object.
     """
 
     def __init__(
@@ -97,6 +98,18 @@ def _convert_to_date(date_arg) -> datetime.date:
             raise TypeError("date_arg must be an int, str, or datetime.date")
 
     def apply_measure(self, measure: Measure):
+        """Create a new snapshot from a measure
+
+        This methods creates a new `Snapshot` object by applying a measure on
+        the current one.
+
+        Parameters
+        ----------
+        measure : Measure
+            The measure to be applied to the snapshot
+
+        """
+
         LOGGER.debug(f"Applying measure {measure.name} on snapshot {id(self)}")
         exp_new, impfset_new, haz_new = measure.apply(
             self.exposure, self.impfset, self.hazard