Merge pull request #230 from Deltares/feature/220-refactor-analysis-c…

…onfig-reader

feature: refactor analysis config reader

ra2ce/analyses/analysis_config_data/analysis_config_data.py

@@ -22,24 +22,158 @@
 
 from __future__ import annotations
 
+import math
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Optional
+
 from ra2ce.common.configuration.config_data_protocol import ConfigDataProtocol
+from ra2ce.graph.network_config_data.network_config_data import (
+    NetworkConfigData,
+    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"]
+
+
+@dataclass
+class ProjectSection:
+    """
+    Reflects all possible settings that a project section might contain.
+    """
+
+    name: str = ""
+
+
+@dataclass
+class AnalysisSectionBase:
+    """
+    Reflects all common settings that direct and indirect analysis sections might contain.
+    """
+
+    name: str = ""
+    analysis: str = ""  # should be enum
+    save_gpkg: bool = False
+    save_csv: bool = False
+
+
+@dataclass
+class AnalysisSectionIndirect(AnalysisSectionBase):
+    """
+    Reflects all possible settings that an indirect analysis section might contain.
+    """
 
+    # general
+    weighing: str = ""  # should be enum
+    loss_per_distance: str = ""
+    loss_type: str = ""  # should be enum
+    disruption_per_category: str = ""
+    traffic_cols: list[str] = field(default_factory=list)
+    # losses
+    duration_event: float = math.nan
+    duration_disruption: float = math.nan
+    fraction_detour: float = math.nan
+    fraction_drivethrough: float = math.nan
+    rest_capacity: float = math.nan
+    maximum_jam: float = math.nan
+    partofday: str = ""
+    # accessiblity analyses
+    aggregate_wl: str = ""  # should be enum
+    threshold: float = math.nan
+    threshold_destinations: float = math.nan
+    uniform_duration: float = math.nan
+    gdp_percapita: float = math.nan
+    equity_weight: str = ""
+    calculate_route_without_disruption: bool = False
+    buffer_meters: float = math.nan
+    threshold_locations: float = math.nan
+    category_field_name: str = ""
+    save_traffic: bool = False
 
+
+@dataclass
+class AnalysisSectionDirect(AnalysisSectionBase):
+    """
+    Reflects all possible settings that a direct analysis section might contain.
+    """
+
+    # adaptation/effectiveness measures
+    return_period: float = math.nan
+    repair_costs: float = math.nan
+    evaluation_period: float = math.nan
+    interest_rate: float = math.nan
+    climate_factor: float = math.nan
+    climate_period: float = math.nan
+    # road damage
+    damage_curve: str = ""
+    event_type: str = ""  # should be enum
+    risk_calculation: str = ""  # should be enum
+    create_table: bool = False
+    file_name: Optional[Path] = None
+
+
+@dataclass
 class AnalysisConfigData(ConfigDataProtocol):
-    @classmethod
-    def from_dict(cls, dict_values: dict) -> AnalysisConfigData:
-        _new_analysis_ini_config_data = cls()
-        _new_analysis_ini_config_data.update(**dict_values)
-        return _new_analysis_ini_config_data
+    """
+    Reflects all config data from analysis.ini with defaults set.
+    Additionally some attributes from the network config are added for completeness (files, origins_destinations, network, hazard_names)
+    """
+
+    root_path: Optional[Path] = None
+    input_path: Optional[Path] = None
+    output_path: Optional[Path] = None
+    static_path: Optional[Path] = None
+    project: ProjectSection = field(default_factory=lambda: ProjectSection())
+    analyses: list[AnalysisSectionBase] = field(default_factory=list)
+    files: Optional[dict[str, Path]] = field(default_factory=dict)
+    origins_destinations: Optional[OriginsDestinationsSection] = field(
+        default_factory=lambda: OriginsDestinationsSection()
+    )
+    network: Optional[NetworkConfigData] = field(
+        default_factory=lambda: NetworkConfigData()
+    )
+    hazard_names: Optional[list[str]] = field(default_factory=list)
+
+    @property
+    def direct(self) -> list[AnalysisSectionDirect]:
+        """
+        Get all direct analyses from config.
+
+        Returns:
+            list[AnalysisSectionDirect]: List of all direct analyses.
+        """
+        return list(
+            filter(lambda x: isinstance(x, AnalysisSectionDirect), self.analyses)
+        )
+
+    @property
+    def indirect(self) -> list[AnalysisSectionIndirect]:
+        """
+        Get all indirect analyses from config.
+
+        Returns:
+            list[AnalysisSectionIndirect]: List of all indirect analyses.
+        """
+        return list(
+            filter(lambda x: isinstance(x, AnalysisSectionIndirect), self.analyses)
+        )
 
 
 class AnalysisConfigDataWithNetwork(AnalysisConfigData):
-    @classmethod
-    def from_dict(cls, dict_values: dict) -> AnalysisConfigDataWithNetwork:
-        return super().from_dict(dict_values)
+    pass
 
 
 class AnalysisConfigDataWithoutNetwork(AnalysisConfigData):
-    @classmethod
-    def from_dict(cls, dict_values: dict) -> AnalysisConfigDataWithoutNetwork:
-        return super().from_dict(dict_values)
+    pass

ra2ce/analyses/analysis_config_data/analysis_config_data_validator_with_network.py

@@ -16,10 +16,10 @@ def validate(self) -> ValidationReport:
         _base_report = AnalysisConfigDataValidatorWithoutNetwork(
             self._config
         ).validate()
-        _output_network_dir = self._config.get("output", None)
+        _output_network_dir = self._config.output_path
         if (
             not _output_network_dir
-            or not (_output_network_dir / "network.ini").is_file()
+            or not (_output_network_dir.joinpath("network.ini")).is_file()
         ):
             _base_report.error(
                 f"The configuration file 'network.ini' is not found at {_output_network_dir}."

ra2ce/analyses/analysis_config_data/analysis_config_data_validator_without_network.py

@@ -20,30 +20,21 @@
 """
 
 
+import math
 from pathlib import Path
+from typing import Any
 
 from ra2ce.analyses.analysis_config_data.analysis_config_data import (
     AnalysisConfigDataWithoutNetwork,
+    IndirectAnalysisNameList,
+    DirectAnalysisNameList,
 )
 from ra2ce.common.validation.ra2ce_validator_protocol import Ra2ceIoValidator
 from ra2ce.common.validation.validation_report import ValidationReport
 from ra2ce.graph.network_config_data.network_config_data_validator import (
     NetworkDictValues,
 )
 
-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"]
 AnalysisNetworkDictValues = NetworkDictValues | {
     "analysis": IndirectAnalysisNameList + DirectAnalysisNameList
 }
@@ -53,40 +44,29 @@ class AnalysisConfigDataValidatorWithoutNetwork(Ra2ceIoValidator):
     def __init__(self, config_data: AnalysisConfigDataWithoutNetwork) -> None:
         self._config = config_data
 
-    def _validate_road_types(self, road_type_value: str) -> ValidationReport:
-        _road_types_report = ValidationReport()
-        if not road_type_value:
-            return _road_types_report
-        _expected_road_types = AnalysisNetworkDictValues["road_types"]
-        _road_type_value_list = road_type_value.replace(" ", "").split(",")
-        for road_type in _road_type_value_list:
-            if road_type not in _expected_road_types:
-                _road_types_report.error(
-                    f"Wrong road type is configured ({road_type}), has to be one or multiple of: {_expected_road_types}"
-                )
-        return _road_types_report
-
-    def _validate_files(
-        self, header: str, path_value_list: list[Path]
-    ) -> ValidationReport:
-        # Value should be none or a list of paths, because it already
-        # checked that it's not none, we can assume it's a list of Paths.
-        _files_report = ValidationReport()
-        if not path_value_list:
-            return _files_report
-        for path_value in path_value_list:
-            if not path_value.is_file():
-                _files_report.error(
-                    f"Wrong input to property [ {header} ], file does not exist: {path_value}"
-                )
-                _files_report.error(
-                    f"If no file is needed, please insert value - None - for property - {header} -"
-                )
-        return _files_report
+    def _validate_header(self, header: Any) -> ValidationReport:
+        _report = ValidationReport()
+
+        if isinstance(header, list):
+            for _item in header:
+                _report.merge(self._validate_header(_item))
+        else:
+            for key, value in header.__dict__.items():
+                if not value:
+                    continue
+                if key not in AnalysisNetworkDictValues.keys():
+                    continue
+                _expected_values_list = AnalysisNetworkDictValues[key]
+                if value not in _expected_values_list:
+                    _report.error(
+                        f"Wrong input to property [ {key} ], has to be one of: {_expected_values_list}"
+                    )
+
+        return _report
 
     def _validate_headers(self, required_headers: list[str]) -> ValidationReport:
         _report = ValidationReport()
-        _available_keys = self._config.keys()
+        _available_keys = self._config.__dict__.keys()
 
         def _check_header(header: str) -> None:
             if header not in _available_keys:
@@ -100,47 +80,27 @@ def _check_header(header: str) -> None:
 
         # check if properties have correct input
         # TODO: Decide whether also the non-used properties must be checked or those are not checked
-        # TODO: Decide how to check for multiple analyses (analysis1, analysis2, etc)
 
         for header in required_headers:
             # Now check the parameters per configured item.
-            for key, value in self._config[header].items():
-                if key not in AnalysisNetworkDictValues.keys():
-                    continue
-                _expected_values_list = AnalysisNetworkDictValues[key]
-                if "file" in _expected_values_list:
-                    # Value should be none or a list of paths, because it already
-                    # checked that it's not none, we can assume it's a list of Paths.
-                    _report.merge(self._validate_files(key, value))
-                    continue
-
-                if key == "road_types":
-                    _report.merge(self._validate_road_types(value))
-                    continue
-
-                if value not in _expected_values_list:
-                    _report.error(
-                        f"Wrong input to property [ {key} ], has to be one of: {_expected_values_list}"
-                    )
+            _attr = getattr(self._config, header)
+            if not _attr:
+                continue
+            else:
+                _report.merge(self._validate_header(_attr))
 
         if not _report.is_valid():
             _report.error(
-                "There are inconsistencies in the *.ini file. Please consult the log file for more information: {}".format(
-                    self._config["root_path"]
-                    / "data"
-                    / self._config["project"]["name"]
-                    / "output"
-                    / "RA2CE.log"
+                "There are inconsistencies in the *.ini file. Please consult the log file for more information: {}.".format(
+                    self._config.output_path.joinpath("RA2CE.log")
                 )
             )
 
         return _report
 
     def validate(self) -> ValidationReport:
         _report = ValidationReport()
-        _required_headers = ["project"]
-        # Analysis are marked as [analysis1], [analysis2], ...
-        _required_headers.extend([a for a in self._config.keys() if "analysis" in a])
+        _required_headers = ["project", "analyses"]
 
         _report.merge(self._validate_headers(_required_headers))
         return _report