Merge pull request #271 from Deltares/feature/270-add-enums-for-analy…

…sis-config-settings

Feature/270 add enums for analysis config settings

ra2ce/analyses/analysis_config_data/analysis_config_data.py

@@ -27,6 +27,12 @@
 from pathlib import Path
 from typing import Optional
 
+from ra2ce.analyses.analysis_config_data.enums.analysis_direct_enum import (
+    AnalysisDirectEnum,
+)
+from ra2ce.analyses.analysis_config_data.enums.analysis_indirect_enum import (
+    AnalysisIndirectEnum,
+)
 from ra2ce.analyses.analysis_config_data.enums.weighing_enum import WeighingEnum
 from ra2ce.common.configuration.config_data_protocol import ConfigDataProtocol
 from ra2ce.graph.network_config_data.enums.aggregate_wl_enum import AggregateWlEnum
@@ -35,19 +41,12 @@
     OriginsDestinationsSection,
 )
 
-IndirectAnalysisNameList: list[str] = [
-    "single_link_redundancy",
-    "multi_link_redundancy",
-    "optimal_route_origin_destination",
-    "multi_link_origin_destination",
-    "optimal_route_origin_closest_destination",
-    "multi_link_origin_closest_destination",
-    "losses",
-    "single_link_losses",
-    "multi_link_losses",
-    "multi_link_isolated_locations",
-]
-DirectAnalysisNameList: list[str] = ["direct", "effectiveness_measures"]
+IndirectAnalysisNameList: list[str] = list(
+    map(str, AnalysisIndirectEnum.list_valid_options())
+)
+DirectAnalysisNameList: list[str] = list(
+    map(str, AnalysisDirectEnum.list_valid_options())
+)
 
 
 @dataclass
@@ -66,7 +65,6 @@ class AnalysisSectionBase:
     """
 
     name: str = ""
-    analysis: str = ""  # should be enum
     save_gpkg: bool = False
     save_csv: bool = False
 
@@ -77,6 +75,9 @@ class AnalysisSectionIndirect(AnalysisSectionBase):
     Reflects all possible settings that an indirect analysis section might contain.
     """
 
+    analysis: AnalysisIndirectEnum = field(
+        default_factory=lambda: AnalysisIndirectEnum.INVALID
+    )
     # general
     weighing: WeighingEnum = field(default_factory=lambda: WeighingEnum.NONE)
     loss_per_distance: str = ""
@@ -111,6 +112,9 @@ class AnalysisSectionDirect(AnalysisSectionBase):
     Reflects all possible settings that a direct analysis section might contain.
     """
 
+    analysis: AnalysisDirectEnum = field(
+        default_factory=lambda: AnalysisDirectEnum.INVALID
+    )
     # adaptation/effectiveness measures
     return_period: float = math.nan
     repair_costs: float = math.nan

ra2ce/analyses/analysis_config_data/analysis_config_data_reader.py

@@ -34,6 +34,12 @@
     IndirectAnalysisNameList,
     ProjectSection,
 )
+from ra2ce.analyses.analysis_config_data.enums.analysis_direct_enum import (
+    AnalysisDirectEnum,
+)
+from ra2ce.analyses.analysis_config_data.enums.analysis_indirect_enum import (
+    AnalysisIndirectEnum,
+)
 from ra2ce.analyses.analysis_config_data.enums.weighing_enum import WeighingEnum
 from ra2ce.common.configuration.ini_configuration_reader_protocol import (
     ConfigDataReaderProtocol,
@@ -97,6 +103,9 @@ def _get_analysis_section_indirect(
         self, section_name: str
     ) -> AnalysisSectionIndirect:
         _section = AnalysisSectionIndirect(**self._parser[section_name])
+        _section.analysis = AnalysisIndirectEnum.get_enum(
+            self._parser.get(section_name, "analysis", fallback=None)
+        )
         _section.save_gpkg = self._parser.getboolean(
             section_name, "save_gpkg", fallback=_section.save_gpkg
         )
@@ -187,6 +196,9 @@ def _get_analysis_section_indirect(
 
     def _get_analysis_section_direct(self, section_name: str) -> AnalysisSectionDirect:
         _section = AnalysisSectionDirect(**self._parser[section_name])
+        _section.analysis = AnalysisDirectEnum.get_enum(
+            self._parser.get(section_name, "analysis", fallback=None)
+        )
         _section.save_gpkg = self._parser.getboolean(
             section_name, "save_gpkg", fallback=_section.save_gpkg
         )

ra2ce/analyses/analysis_config_data/enums/analysis_direct_enum.py

@@ -0,0 +1,7 @@
+from ra2ce.configuration.ra2ce_enum_base import Ra2ceEnumBase
+
+
+class AnalysisDirectEnum(Ra2ceEnumBase):
+    DIRECT = 1
+    EFFECTIVENESS_MEASURES = 2
+    INVALID = 99

ra2ce/analyses/analysis_config_data/enums/analysis_indirect_enum.py

@@ -0,0 +1,15 @@
+from ra2ce.configuration.ra2ce_enum_base import Ra2ceEnumBase
+
+
+class AnalysisIndirectEnum(Ra2ceEnumBase):
+    SINGLE_LINK_REDUNDANCY = 1
+    MULTI_LINK_REDUNDANCY = 2
+    OPTIMAL_ROUTE_ORIGIN_DESTINATION = 3
+    MULTI_LINK_ORIGIN_DESTINATION = 4
+    OPTIMAL_ROUTE_ORIGIN_CLOSEST_DESTINATION = 5
+    MULTI_LINK_ORIGIN_CLOSEST_DESTINATION = 6
+    LOSSES = 7
+    SINGLE_LINK_LOSSES = 8
+    MULTI_LINK_LOSSES = 9
+    MULTI_LINK_ISOLATED_LOCATIONS = 10
+    INVALID = 99

ra2ce/analyses/analysis_config_wrapper.py

@@ -48,7 +48,7 @@ def initialize_output_dirs(self) -> None:
         """
         # Create the output folders
         for a in self.config_data.analyses:
-            output_path = self.config_data.output_path.joinpath(a.analysis)
+            output_path = self.config_data.output_path.joinpath(a.analysis.config_value)
             output_path.mkdir(parents=True, exist_ok=True)
 
     @classmethod

ra2ce/analyses/direct/analyses_direct.py

@@ -31,6 +31,9 @@
     AnalysisConfigData,
     AnalysisSectionDirect,
 )
+from ra2ce.analyses.analysis_config_data.enums.analysis_direct_enum import (
+    AnalysisDirectEnum,
+)
 from ra2ce.analyses.direct.cost_benefit_analysis import EffectivenessMeasures
 from ra2ce.analyses.direct.damage.manual_damage_functions import ManualDamageFunctions
 from ra2ce.analyses.direct.damage_calculation import (
@@ -72,18 +75,20 @@ def execute(self):
             )
             starttime = time.time()
 
-            if analysis.analysis == "direct":
+            if analysis.analysis == AnalysisDirectEnum.DIRECT:
                 gdf = self.road_damage(
                     analysis
                 )  # calls the coordinator for road damage calculation
 
-            elif analysis.analysis == "effectiveness_measures":
+            elif analysis.analysis == AnalysisDirectEnum.EFFECTIVENESS_MEASURES:
                 gdf = self.effectiveness_measures(analysis)
 
             else:
                 gdf = []
 
-            _output_path = self.config.output_path.joinpath(analysis.analysis)
+            _output_path = self.config.output_path.joinpath(
+                analysis.analysis.config_value
+            )
             if analysis.save_gpkg:
                 gpkg_path = _output_path.joinpath(
                     analysis.name.replace(" ", "_") + ".gpkg"
@@ -195,7 +200,7 @@ def effectiveness_measures(self, analysis: AnalysisSectionDirect):
         df_cba, costs_dict = em.cost_benefit_analysis(effectiveness_dict)
         df_cba.round(2).to_csv(
             self.config.output_path.joinpath(
-                analysis.analysis, "cost_benefit_analysis.csv"
+                analysis.analysis.config_value, "cost_benefit_analysis.csv"
             ),
             decimal=",",
             sep=";",
@@ -206,7 +211,9 @@ def effectiveness_measures(self, analysis: AnalysisSectionDirect):
         df_costs = em.calculate_strategy_costs(df, costs_dict)
         df_costs = df_costs.astype(float).round(2)
         df_costs.to_csv(
-            self.config.output_path.joinpath(analysis.analysis, "output_analysis.csv"),
+            self.config.output_path.joinpath(
+                analysis.analysis.config_value, "output_analysis.csv"
+            ),
             decimal=",",
             sep=";",
             index=False,

ra2ce/analyses/indirect/analyses_indirect.py

@@ -23,7 +23,6 @@
 import logging
 import time
 from pathlib import Path
-from typing import Any
 
 import geopandas as gpd
 import networkx as nx
@@ -38,6 +37,9 @@
     AnalysisConfigData,
     AnalysisSectionIndirect,
 )
+from ra2ce.analyses.analysis_config_data.enums.analysis_indirect_enum import (
+    AnalysisIndirectEnum,
+)
 from ra2ce.analyses.analysis_config_data.enums.weighing_enum import WeighingEnum
 from ra2ce.analyses.indirect.losses import Losses
 from ra2ce.analyses.indirect.origin_closest_destination import OriginClosestDestination
@@ -947,7 +949,8 @@ def multi_link_isolated_locations(
             # Save the output
             results_hz_roads.to_file(
                 self.config.output_path.joinpath(
-                    analysis.analysis, f"flooded_and_isolated_roads_{hazard_name}.gpkg"
+                    analysis.analysis.config_value,
+                    f"flooded_and_isolated_roads_{hazard_name}.gpkg",
                 )
             )
 
@@ -1062,7 +1065,9 @@ def execute(self):
             starttime = time.time()
             gdf = pd.DataFrame()
             opt_routes = None
-            _output_path = self.config.output_path.joinpath(analysis.analysis)
+            _output_path = self.config.output_path.joinpath(
+                analysis.analysis.config_value
+            )
 
             def _save_gpkg_analysis(
                 base_graph,
@@ -1085,13 +1090,16 @@ def _save_gpkg_analysis(
                 )
                 graph_to_gpkg(base_graph, gpkg_path_edges, gpkg_path_nodes)
 
-            if analysis.analysis == "single_link_redundancy":
+            if analysis.analysis == AnalysisIndirectEnum.SINGLE_LINK_REDUNDANCY:
                 g = self.graph_files.base_graph.get_graph()
                 gdf = self.single_link_redundancy(g, analysis)
-            elif analysis.analysis == "multi_link_redundancy":
+            elif analysis.analysis == AnalysisIndirectEnum.MULTI_LINK_REDUNDANCY:
                 g = self.graph_files.base_graph_hazard.get_graph()
                 gdf = self.multi_link_redundancy(g, analysis)
-            elif analysis.analysis == "optimal_route_origin_destination":
+            elif (
+                analysis.analysis
+                == AnalysisIndirectEnum.OPTIMAL_ROUTE_ORIGIN_DESTINATION
+            ):
                 g = self.graph_files.origins_destinations_graph.get_graph()
                 gdf = self.optimal_route_origin_destination(g, analysis)
 
@@ -1119,7 +1127,9 @@ def _save_gpkg_analysis(
                         (analysis.name.replace(" ", "_") + "_link_traffic.csv"),
                     )
                     route_traffic_df.to_csv(impact_csv_path, index=False)
-            elif analysis.analysis == "multi_link_origin_destination":
+            elif (
+                analysis.analysis == AnalysisIndirectEnum.MULTI_LINK_ORIGIN_DESTINATION
+            ):
                 g = self.graph_files.origins_destinations_graph_hazard.get_graph()
                 gdf = self.multi_link_origin_destination(g, analysis)
                 gdf_not_disrupted = self.optimal_route_origin_destination(g, analysis)
@@ -1155,11 +1165,17 @@ def _save_gpkg_analysis(
                     (analysis.name.replace(" ", "_") + "_impact_summary.csv"),
                 )
                 disruption_impact_df.to_csv(impact_csv_path, index=False)
-            elif analysis.analysis in ["single_link_losses", "multi_link_losses)"]:
+            elif analysis.analysis in [
+                AnalysisIndirectEnum.SINGLE_LINK_LOSSES,
+                AnalysisIndirectEnum.MULTI_LINK_LOSSES,
+            ]:
                 g = self.graph_files.base_graph_hazard.get_graph()
                 gdf = self.single_link_redundancy(g, analysis)
                 gdf = self.single_link_losses(gdf, analysis)
-            elif analysis.analysis == "optimal_route_origin_closest_destination":
+            elif (
+                analysis.analysis
+                == AnalysisIndirectEnum.OPTIMAL_ROUTE_ORIGIN_CLOSEST_DESTINATION
+            ):
                 analyzer = OriginClosestDestination(
                     self.config, analysis, self.graph_files, self.hazard_names_df
                 )
@@ -1186,7 +1202,10 @@ def _save_gpkg_analysis(
                     )
                     del opt_routes["geometry"]
                     opt_routes.to_csv(csv_path, index=False)
-            elif analysis.analysis == "multi_link_origin_closest_destination":
+            elif (
+                analysis.analysis
+                == AnalysisIndirectEnum.MULTI_LINK_ORIGIN_CLOSEST_DESTINATION
+            ):
                 analyzer = OriginClosestDestination(
                     self.config, analysis, self.graph_files, self.hazard_names_df
                 )
@@ -1268,12 +1287,14 @@ def _save_gpkg_analysis(
                         ),
                         index=False,
                     )
-            elif analysis.analysis == "losses":
+            elif analysis.analysis == AnalysisIndirectEnum.LOSSES:
                 gdf_in = self.graph_files.base_graph_hazard.get_graph()
                 losses = Losses(self.config, analysis)
                 df = losses.calculate_losses_from_table()
                 gdf = gdf_in.merge(df, how="left", on="LinkNr")
-            elif analysis.analysis == "multi_link_isolated_locations":
+            elif (
+                analysis.analysis == AnalysisIndirectEnum.MULTI_LINK_ISOLATED_LOCATIONS
+            ):
                 g = self.graph_files.base_graph_hazard.get_graph()
                 (gdf, df) = self.multi_link_isolated_locations(g, analysis)
 

ra2ce/configuration/ra2ce_enum_base.py

@@ -44,14 +44,24 @@ def is_valid(self) -> bool:
             return False
         return True
 
-    def list_valid_options(self) -> list[Ra2ceEnumBase]:
+    @classmethod
+    def list_valid_options(cls) -> list[Ra2ceEnumBase]:
         """
         List the enum options as allowed in the config.
 
         Returns:
             list[str | None]: Concatenated options, separated by ", "
         """
-        return [_enum for _enum in type(self)][:-1]
+        return [_enum for _enum in cls][:-1]
+
+    def __str__(self) -> str:
+        """
+        Overwrite the default __str__
+
+        Returns:
+            str: Value as in the config
+        """
+        return str(self.config_value)
 
     @property
     def config_value(self) -> str | None:

tests/analyses/analysis_config_data/test_analysis_config_data.py

@@ -8,6 +8,12 @@
     IndirectAnalysisNameList,
     ProjectSection,
 )
+from ra2ce.analyses.analysis_config_data.enums.analysis_direct_enum import (
+    AnalysisDirectEnum,
+)
+from ra2ce.analyses.analysis_config_data.enums.analysis_indirect_enum import (
+    AnalysisIndirectEnum,
+)
 from tests import test_results
 
 
@@ -22,16 +28,22 @@ def test_initialize(self):
     def valid_config(self) -> AnalysisConfigData:
         _config = AnalysisConfigData(project=ProjectSection())
         for _indirect in IndirectAnalysisNameList:
-            _config.analyses.append(AnalysisSectionIndirect(analysis=_indirect))
+            _config.analyses.append(
+                AnalysisSectionIndirect(
+                    analysis=AnalysisIndirectEnum.get_enum(_indirect)
+                )
+            )
         for _direct in DirectAnalysisNameList:
-            _config.analyses.append(AnalysisSectionDirect(analysis=_direct))
+            _config.analyses.append(
+                AnalysisSectionDirect(analysis=AnalysisDirectEnum.get_enum(_direct))
+            )
         yield _config
 
     def test_indirect(self, valid_config: AnalysisConfigData):
         # 1. Define test data
 
         # 2. Run test
-        _indirect = [_config.analysis for _config in valid_config.indirect]
+        _indirect = [_config.analysis.config_value for _config in valid_config.indirect]
 
         # 3. Verify expectations
         assert all(item in _indirect for item in IndirectAnalysisNameList)
@@ -40,7 +52,7 @@ def test_direct(self, valid_config: AnalysisConfigData):
         # 1. Define test data
 
         # 2. Run test
-        _direct = [_config.analysis for _config in valid_config.direct]
+        _direct = [_config.analysis.config_value for _config in valid_config.direct]
 
         # 3. Verify expectations
         assert all(item in _direct for item in DirectAnalysisNameList)