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

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 (
@@ -168,8 +171,10 @@ 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):
+    ### Damage handlers: TODO: move to dataclass ManualDamageFunctions
+    def calculate_damage_manual_functions(
+        self, events, manual_damage_functions: ManualDamageFunctions
+    ):
         """
         Arguments:
             *events* (list) : list of events (or return periods) to iterate over, these should match the hazard column names
@@ -182,16 +187,18 @@ def calculate_damage_manual_functions(self, events, manual_damage_functions):
         # 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

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

@@ -19,30 +19,34 @@
     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 = None
+    hazard_unit: str = None
+    data: pd.DataFrame = None
+    origin_path: Path = None
+    interpolator: interp1d = None
 
-    def from_csv(self, path: Path, sep=",") -> None:
+    @classmethod
+    def from_csv(cls, path: Path, sep=",") -> DamageFractionUniform:
         """Construct object from csv file. Damage curve name is inferred from filename
 
         Arguments:
@@ -71,23 +75,28 @@ 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 = path.stem
+        _raw_data = pd.read_csv(path, index_col=0, sep=sep)
+        _origin_path = 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()
+        _damage_fraction = cls(
+            name=_name, hazard_unit=_hazard_unit, data=_data, origin_path=_origin_path
+        )
+        _damage_fraction._convert_hazard_severity_unit()
 
-    def convert_hazard_severity_unit(self, desired_unit="m") -> None:
+        return _damage_fraction
+
+    def _convert_hazard_severity_unit(self, desired_unit="m") -> None:
         """Converts hazard severity values to a different unit
         Arguments:
             self.hazard_unit - implicit (string)
@@ -124,8 +133,6 @@ 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]
 

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

@@ -18,56 +18,38 @@
     You should have received a copy of the GNU General Public License
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
-
-import logging
+from dataclasses import dataclass
 from pathlib import Path
 
 import pandas as pd
 
-from ra2ce.analysis.damages.damage.damage_fraction_uniform import DamageFractionUniform
-from ra2ce.analysis.damages.damage.max_damage import MaxDamageByRoadTypeByLane
+from ra2ce.analysis.damages.damage_functions.damage_fraction_uniform import (
+    DamageFractionUniform,
+)
+from ra2ce.analysis.damages.damage_functions.max_damage import MaxDamage
 
 
+@dataclass(kw_only=True)
 class DamageFunctionByRoadTypeByLane:
     """
     A damage function that has different max damages per road type, but a uniform damage_fraction curve
 
-
     The attributes need to be of the type:
-    self.max_damage (MaxDamage_byRoadType_byLane)
-    self.damage_fraction (DamageFractionHazardSeverityUniform)
-
+    self.max_damage (MaxDamage)
+    self.damage_fraction (DamageFractionUniform)
+    name (str)
     """
 
-    def __init__(
-        self,
-        max_damage: MaxDamageByRoadTypeByLane = None,
-        damage_fraction: DamageFractionUniform = None,
-        name: str = "",
-        hazard: str = "flood",
-        type: str = "depth_damage",
-        infra_type: str = "road",
-    ):
-        # Construct using the parent class __init__
-        self.name = name
-        self.hazard = hazard
-        self.type = type
-        self.infra_type = infra_type
-        self.max_damage = max_damage  # Should be a MaxDamage object
-        self.damage_fraction = (
-            damage_fraction  # Should be a DamageFractionHazardSeverity object
-        )
-        self.prefix = None  # Should be two characters long at maximum
-
-    def set_prefix(self):
-        self.prefix = self.name[0:2]
-        logging.info(
-            "The prefix: '{}' refers to curve name '{}' in the results".format(
-                self.prefix, self.name
-            )
-        )
+    max_damage: MaxDamage = None
+    damage_fraction: DamageFractionUniform = None
+    name: str = None
+
+    @property
+    def prefix(self) -> str:
+        return self.name[0:2]
 
-    def from_input_folder(self, folder_path: Path):
+    @classmethod
+    def from_input_folder(cls, name, folder_path: Path):
         """Construct a set of damage functions from csv files located in the folder_path
 
         Arguments:
@@ -95,21 +77,18 @@ def find_unique_csv_file(folder_path: Path, part_of_filename: str) -> Path:
             return result[0]
 
         # Load the max_damage object
-        max_damage = MaxDamageByRoadTypeByLane()
         max_dam_path = find_unique_csv_file(folder_path, "max_damage")
-        max_damage.from_csv(max_dam_path, sep=";")
-
-        self.max_damage = max_damage
+        max_damage = MaxDamage.from_csv(max_dam_path, sep=";")
 
         # Load the damage fraction function
         # search in the folder for something *damage_fraction
-        damage_fraction = DamageFractionUniform()
         dam_fraction_path = find_unique_csv_file(folder_path, "hazard_severity")
-        damage_fraction.from_csv(dam_fraction_path, sep=";")
-        self.damage_fraction = damage_fraction
+        damage_fraction = DamageFractionUniform.from_csv(dam_fraction_path, sep=";")
 
         damage_fraction.create_interpolator()
 
+        return cls(max_damage=max_damage, damage_fraction=damage_fraction, name=name)
+
     # Todo: these two below functions are maybe better implemented at a lower level?
     def add_max_damage(self, df: pd.DataFrame, prefix: str = None):
         """ "Ads the max damage value to the dataframe"""

ra2ce/analysis/damages/damage_functions/manual_damage_functions.py

@@ -0,0 +1,40 @@
+"""
+                    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 dataclasses import dataclass, field
+
+from ra2ce.analysis.damages.damage_functions.damage_function_road_type_lane import (
+    DamageFunctionByRoadTypeByLane,
+)
+
+
+@dataclass(kw_only=True)
+class ManualDamageFunctions:
+    """ "
+    This class keeps an overview of the manual damage functions
+
+    Default behaviour is to find, load and apply all available functions
+    At 22 sept 2022: only implemented workflow for DamageFunction_by_RoadType_by_Lane
+    """
+
+    damage_functions: dict[str, DamageFunctionByRoadTypeByLane] = field(
+        default_factory=dict
+    )