feat: 659 convert damage classes into dataclasses; add a reader for d…

…amage curves (#656)

ra2ce/analysis/damages/damage_calculation/damage_network_base.py

@@ -28,6 +28,9 @@
 from scipy.interpolate import interp1d
 
 from ra2ce.analysis.analysis_config_data.enums.damage_curve_enum import DamageCurveEnum
+from ra2ce.analysis.damages.damage_functions.manual_damage_functions import (
+    ManualDamageFunctions,
+)
 from ra2ce.analysis.damages.damages_lookup import LookUp as lookup
 from ra2ce.analysis.damages.damages_lookup import dataframe_lookup
 from ra2ce.analysis.damages.damages_utils import (
@@ -56,7 +59,11 @@ def __init__(
         # TODO: also create constructors of the children of this class
 
     @abstractmethod
-    def main(self, damage_function: DamageCurveEnum, manual_damage_functions):
+    def main(
+        self,
+        damage_function: DamageCurveEnum,
+        manual_damage_functions: ManualDamageFunctions,
+    ):
         """
         Controller for doing the EAD calculation
 
@@ -169,29 +176,34 @@ def remove_unclassified_road_types_from_mask(self):
             self._gdf_mask = df.loc[~(df["road_type"] == "none")]
 
     ### Damage handlers
-    def calculate_damage_manual_functions(self, events, manual_damage_functions):
-        """
-        Arguments:
-            *events* (list) : list of events (or return periods) to iterate over, these should match the hazard column names
-            *manual_damage_functions* (RA2CE ManualDamageFunctions object) :
+    def calculate_damage_manual_functions(
+        self, events: list[str], manual_damage_functions: ManualDamageFunctions
+    ) -> None:
         """
+        Calculate the damage using the manual damage functions
 
+        Args:
+            events (list[str]): list of events (or return periods) to iterate over, these should match the hazard column names
+            manual_damage_functions (ManualDamageFunctions): The manual damage functions object
+        """
         # Todo: Dirty fixes, these should be read from the init
         hazard_prefix = "F"
 
         # dataframe to carry out the damage calculation #todo: this is a bit dirty
         df = self._gdf_mask
 
-        assert manual_damage_functions is not None, "No damage functions were loaded"
+        assert (
+            len(manual_damage_functions.damage_functions) > 0
+        ), "No damage functions were loaded"
 
-        for _loaded_func in manual_damage_functions.loaded:
+        for _damage_func in manual_damage_functions.damage_functions.values():
             # Add max damage values to df
-            df = _loaded_func.add_max_damage(df, _loaded_func.prefix)
+            df = _damage_func.add_max_damage(df, _damage_func.prefix)
             for event in events:
                 # Add apply interpolator objects
                 event_prefix = event
-                df = _loaded_func.calculate_damage(
-                    df, _loaded_func.prefix, hazard_prefix, event_prefix
+                df = _damage_func.calculate_damage(
+                    df, _damage_func.prefix, hazard_prefix, event_prefix
                 )
 
         # Only transfer the final results to the damage column
@@ -201,7 +213,7 @@ def calculate_damage_manual_functions(self, events, manual_damage_functions):
             "Damage calculation with the manual damage functions was succesfull."
         )
 
-    def calculate_damage_HZ(self, events):
+    def calculate_damage_HZ(self, events: list[str]) -> None:
         """
         Arguments:
            *events* (list) = list of events (or return periods) to iterate over, these should match the hazard column names
@@ -265,7 +277,7 @@ def calculate_damage_HZ(self, events):
             "calculate_damage_HZ(): Damage calculation with the Huizinga damage functions was successful"
         )
 
-    def calculate_damage_OSdaMage(self, events):
+    def calculate_damage_OSdaMage(self, events: list[str]) -> None:
         """Damage calculation with the OSdaMage functions"""
 
         def interpolate_damage(row, representative_damage_percentage):
@@ -416,7 +428,7 @@ def interpolate_damage(row, representative_damage_percentage):
         )
 
     ### Utils handlers
-    def create_mask(self):
+    def create_mask(self) -> None:
         """
         #Create a mask of only the dataframes with hazard data (to speed-up damage calculations)
         effect: *self._gdf_mask* = mask of only the rows with hazard data
@@ -434,8 +446,6 @@ def create_mask(self):
             column_names.remove("geometry")
         self._gdf_mask = self._gdf_mask[column_names]
 
-    def replace_none_with_nan(self):
-        import numpy as np
-
+    def replace_none_with_nan(self) -> None:
         dam_cols = [c for c in self.gdf.columns if c.startswith("dam_")]
         self.gdf[dam_cols] = self.gdf[dam_cols].fillna(value=np.nan)

ra2ce/analysis/damages/damage_calculation/damage_network_return_periods.py

@@ -67,9 +67,9 @@ def __init__(
 
     @classmethod
     def construct_from_csv(
-        cls, path: Path, representative_damage_percentage: float, sep: str = ";"
+        cls, csv_path: Path, representative_damage_percentage: float, sep: str
     ):
-        road_gdf = pd.read_csv(path, sep=sep)
+        road_gdf = pd.read_csv(csv_path, sep=sep)
         val_cols = [
             c for c in road_gdf.columns if c.startswith("F_")
         ]  # Find everything starting with 'F'

ra2ce/analysis/damages/damage/damage_fraction_uniform.py → ra2ce/analysis/damages/damage_functions/damage_fraction_uniform.py

@@ -19,36 +19,43 @@
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 
+from __future__ import annotations
 
 import logging
+from dataclasses import dataclass
 from pathlib import Path
 
 import pandas as pd
+from scipy.interpolate import interp1d
 
 
+@dataclass(kw_only=True)
 class DamageFractionUniform:
     """
     Uniform: assuming the same curve for
     each road type and lane numbers and any other metadata
 
-
     self.raw_data (pd.DataFrame) : Raw data from the csv file
     self.data (pd.DataFrame) : index = hazard severity (e.g. flood depth); column 0 = damage fraction
-
     """
 
-    def __init__(self, name=None, hazard_unit=None):
-        self.name = name
-        self.hazard_unit = hazard_unit
-        self.interpolator = None
+    name: str
+    hazard_unit: str
+    data: pd.DataFrame
+    origin_path: Path
+    interpolator: interp1d = None
 
-    def from_csv(self, path: Path, sep=",") -> None:
+    def __post_init__(self):
+        self._convert_hazard_severity_unit("m")
+
+    @classmethod
+    def from_csv(cls, csv_path: Path, sep: str) -> DamageFractionUniform:
         """Construct object from csv file. Damage curve name is inferred from filename
 
         Arguments:
-            *path* (Path) : Path to the csv file
+            *csv_path* (Path) : Path to the csv file
             *sep* (str) : csv seperator
-            *output_unit* (str) : desired output unit (default = 'm')
+            *output_unit* (str) : desired output unit
 
         The CSV file should have the following structure:
          - column 1: hazard severity
@@ -71,23 +78,25 @@ def from_csv(self, path: Path, sep=",") -> None:
 
 
         """
-        self.name = path.stem
-        self.raw_data = pd.read_csv(path, index_col=0, sep=sep)
-        self.origin_path = path  # to track the original path from which the object was constructed; maybe also date?
+        _name = csv_path.stem
+        _raw_data = pd.read_csv(csv_path, index_col=0, sep=sep)
+        _origin_path = csv_path  # to track the original path from which the object was constructed; maybe also date?
 
         # identify unit and drop from data
-        self.hazard_unit = self.raw_data.index[0]
-        self.data = self.raw_data.drop(
-            self.hazard_unit
+        _hazard_unit = _raw_data.index[0]
+        _data = _raw_data.drop(
+            _hazard_unit
         )  # Todo: This could also be a series instead of DataFrame
 
         # convert data to floats
-        self.data = self.data.astype("float")
-        self.data.index = self.data.index.astype("float")
+        _data = _data.astype("float")
+        _data.index = _data.index.astype("float")
 
-        self.convert_hazard_severity_unit()
+        return cls(
+            name=_name, hazard_unit=_hazard_unit, data=_data, origin_path=_origin_path
+        )
 
-    def convert_hazard_severity_unit(self, desired_unit="m") -> None:
+    def _convert_hazard_severity_unit(self, desired_unit: str) -> None:
         """Converts hazard severity values to a different unit
         Arguments:
             self.hazard_unit - implicit (string)
@@ -98,34 +107,34 @@ def convert_hazard_severity_unit(self, desired_unit="m") -> None:
         """
         if desired_unit == self.hazard_unit:
             logging.info(
-                "Damage units are already in the desired format {}".format(desired_unit)
+                "Damage units are already in the desired format %s", desired_unit
             )
             return None
 
         if self.hazard_unit == "cm" and desired_unit == "m":
             scaling_factor = 1 / 100
             self.data.index = self.data.index * scaling_factor
             logging.info(
-                "Hazard severity from {} data was scaled by a factor {}, to convert from {} to {}".format(
-                    self.origin_path, scaling_factor, self.hazard_unit, desired_unit
-                )
+                "Hazard severity from %s data was scaled by a factor %s, to convert from %s to %s",
+                self.origin_path,
+                scaling_factor,
+                self.hazard_unit,
+                desired_unit,
             )
             self.damage_unit = desired_unit
             return None
         else:
             logging.warning(
-                "Hazard severity scaling from {} to {} is not  supported".format(
-                    self.hazard_unit, desired_unit
-                )
+                "Hazard severity scaling from %s to %s is not  supported",
+                self.hazard_unit,
+                desired_unit,
             )
             return None
 
     def create_interpolator(self):
         """Create interpolator object from loaded data
         sets result to self.interpolator (Scipy interp1d)
         """
-        from scipy.interpolate import interp1d
-
         x_values = self.data.index.values
         y_values = self.data.values[:, 0]