Refactor object_model/tipping_point.py and interface/tipping_points.py

flood_adapt/object_model/interface/tipping_points.py

@@ -1,14 +1,14 @@
 import os
 from abc import ABC, abstractmethod
-from typing import Any, Optional, Union
 from enum import Enum
+from typing import Any, Optional, Union
 
 import pandas as pd
 from pydantic import BaseModel
 
 
 class TippingPointMetrics(str, Enum):
-    """class describing the accepted input for the variable metric_type in TippingPoint"""
+    """class describing the accepted input for the variable metric_type in TippingPoint."""
 
     # based on what I have found in floodadapt - but can be changed
     FloodedAll = "FloodedAll"
@@ -44,22 +44,22 @@ class TippingPointMetrics(str, Enum):
 
 
 class TippingPointStatus(str, Enum):
-    """class describing the accepted input for the variable metric_type in TippingPoint"""
+    """class describing the accepted input for the variable metric_type in TippingPoint."""
 
     reached = "reached"
     not_reached = "not_reached"
     completed = "completed"
 
 
 class TippingPointOperator(str, Enum):
-    """class describing the accepted input for the variable operator in TippingPoint"""
+    """class describing the accepted input for the variable operator in TippingPoint."""
 
     greater = "greater"
     less = "less"
 
 
 class TippingPointModel(BaseModel):
-    """BaseModel describing the expected variables and data types of a Tipping Point analysis object"""
+    """BaseModel describing the expected variables and data types of a Tipping Point analysis object."""
 
     name: str
     description: Optional[str] = ""
@@ -82,21 +82,21 @@ class ITipPoint(ABC):
     @staticmethod
     @abstractmethod
     def load_file(filepath: Union[str, os.PathLike]):
-        """get Tipping Point attributes from toml file"""
+        """Get Tipping Point attributes from toml file."""
         ...
 
     @staticmethod  # copping from benefits.py
     @abstractmethod
     def load_dict(data: dict[str, Any]):
-        """get Tipping Point attributes from an object, e.g. when initialized from GUI"""
+        """Get Tipping Point attributes from an object, e.g. when initialized from GUI."""
         ...
 
     @abstractmethod
     def save(self, filepath: Union[str, os.PathLike]):
-        """save Tipping Point attributes to a toml file"""
+        """Save Tipping Point attributes to a toml file."""
         ...
 
     @abstractmethod
     def check_scenarios_exist(self) -> pd.DataFrame:
-        """Check which scenarios are needed for this tipping point calculation and if they have already been created"""
+        """Check which scenarios are needed for this tipping point calculation and if they have already been created."""
         ...

flood_adapt/object_model/tipping_point.py

@@ -41,10 +41,10 @@
 
 
 class TippingPoint(ITipPoint):
-    """Class holding all information related to tipping points analysis"""
+    """Class holding all information related to tipping points analysis."""
 
     def __init__(self):
-        """Initiation function when object is created through file or dict options"""
+        """Initiate function when object is created through file or dict options."""
         self.site_toml_path = Path(Database().static_path) / "site" / "site.toml"
         self.results_path = Database().output_path / "tipping_points"
         self.scenarios = {}
@@ -64,7 +64,7 @@ def create_tp_obj(self):
         return self
 
     def slr_projections(self, slr):
-        """Create projections for sea level rise value"""
+        """Create projections for sea level rise value."""
         new_projection_name = self.attrs.projection + "_slr" + str(slr).replace(".", "")
         proj = Database().projections.get(self.attrs.projection)
         proj.attrs.physical_projection.sea_level_rise = UnitfulLength(
@@ -87,7 +87,7 @@ def check_scenarios_exist(self, scenario_obj):
         return db_list
 
     def create_tp_scenarios(self):
-        """Create scenarios for each sea level rise value inside the tipping_point folder"""
+        """Create scenarios for each sea level rise value inside the tipping_point folder."""
         self.create_tp_obj()
         # create projections based on SLR values
         for i, slr in enumerate(self.attrs.sealevelrise):
@@ -135,20 +135,22 @@ def create_tp_scenarios(self):
         )  # for later when we have a database_tp: TODO: Database().tipping_points.save(self)
 
     def run_tp_scenarios(self):
-        """Run all scenarios to determine tipping points"""
+        """Run all scenarios to determine tipping points."""
         for name, scenario in self.scenarios.items():
             scenario_obj = scenario["object"]
-            # check if scenario has been run, if yes skip it
+
+            if self.attrs.status == TippingPointStatus.reached:
+                self.scenarios[name]["tipping point reached"] = True
+                continue
+
             if not self.scenario_has_run(scenario_obj):
                 scenario_obj.run()
 
-            # if the status is reached, save the SLR and the metric value
+            # Check the tipping point status
             if self.check_tipping_point(scenario_obj):
                 self.attrs.status = TippingPointStatus.reached
                 self.scenarios[name]["tipping point reached"] = True
-                break
             else:
-                self.attrs.status = TippingPointStatus.not_reached
                 self.scenarios[name]["tipping point reached"] = False
 
         tp_path = self.results_path.joinpath(self.attrs.name)
@@ -175,8 +177,7 @@ def scenario_has_run(self, scenario_obj):
         return False
 
     def check_tipping_point(self, scenario: Scenario):
-        """Load results and check if the tipping point is reached"""
-        # already optimised for multiple metrics
+        """Load results and check if the tipping point is reached."""
         info_df = pd.read_csv(
             scenario.init_object_model().direct_impacts.results_path.joinpath(
                 f"Infometrics_{scenario.direct_impacts.name}.csv"
@@ -197,27 +198,27 @@ def check_tipping_point(self, scenario: Scenario):
         )
 
     def evaluate_tipping_point(self, current_value, threshold, operator):
-        """Compare current value with threshold for tipping point"""
+        """Compare current value with threshold for tipping point."""
         operations = {"greater": lambda x, y: x >= y, "less": lambda x, y: x <= y}
         return operations[operator](current_value, threshold)
 
     ### standard functions ###
     def load_file(filepath: Union[str, Path]) -> "TippingPoint":
-        """Create risk event from toml file"""
+        """Create risk event from toml file."""
         obj = TippingPoint()
         with open(filepath, mode="rb") as fp:
             toml = tomli.load(fp)
         obj.attrs = TippingPointModel.model_validate(toml)
         return obj
 
     def load_dict(dct: Union[str, Path]) -> "TippingPoint":
-        """Create risk event from toml file"""
+        """Create risk event from toml file."""
         obj = TippingPoint()
         obj.attrs = TippingPointModel.model_validate(dct)
         return obj
 
     def save(self, filepath: Union[str, os.PathLike]):
-        """Save tipping point to a toml file"""
+        """Save tipping point to a toml file."""
         with open(filepath, "wb") as f:
             tomli_w.dump(self.attrs.model_dump(exclude_none=True), f)
 
@@ -239,11 +240,11 @@ def __eq__(self, other):
     from flood_adapt.config import set_system_folder
 
     database = read_database(
-        rf"C:\\Users\\morenodu\\OneDrive - Stichting Deltares\\Documents\\GitHub\\Database",
+        r"C:\\Users\\morenodu\\OneDrive - Stichting Deltares\\Documents\\GitHub\\Database",
         "charleston_test",
     )
     set_system_folder(
-        rf"C:\\Users\\morenodu\\OneDrive - Stichting Deltares\\Documents\\GitHub\\Database\\system"
+        r"C:\\Users\\morenodu\\OneDrive - Stichting Deltares\\Documents\\GitHub\\Database\\system"
     )
 
     tp_dict = {

tests/test_object_model/test_tipping_points.py

@@ -1,6 +1,10 @@
+from pathlib import Path
+
 import pytest
 
-from flood_adapt.object_model.tipping_point import TippingPoint
+from flood_adapt.dbs_controller import Database
+from flood_adapt.object_model.scenario import Scenario
+from flood_adapt.object_model.tipping_point import TippingPoint, TippingPointStatus
 
 
 class TestTippingPoints:
@@ -25,6 +29,11 @@ def created_tp_scenarios(self, tp_dict):
         test_point.create_tp_scenarios()
         return test_point
 
+    @pytest.fixture()
+    def run_tp_scenarios(self, created_tp_scenarios):
+        created_tp_scenarios.run_tp_scenarios()
+        return created_tp_scenarios
+
     def test_createTippingPoints_scenariosAlreadyExist_notDuplicated(
         self, test_db, tp_dict
     ):
@@ -37,6 +46,27 @@ def test_run_scenarios(self, test_db, created_tp_scenarios):
         created_tp_scenarios.run_tp_scenarios()
         assert created_tp_scenarios is not None
 
+    def test_slr_projections_creation(self, test_db, tp_dict):
+        test_point = TippingPoint.load_dict(tp_dict)
+        for slr in test_point.attrs.sealevelrise:
+            test_point.slr_projections(slr)
+            projection_path = (
+                Path(Database().input_path)
+                / "projections"
+                / f"{test_point.attrs.projection}_slr{str(slr).replace('.', '')}"
+                / f"{test_point.attrs.projection}_slr{str(slr).replace('.', '')}.toml"
+            )
+            assert projection_path.exists()
+
+    def test_scenario_tippingpoint_reached(self, test_db, run_tp_scenarios):
+        for name, scenario in run_tp_scenarios.scenarios.items():
+            assert (
+                "tipping point reached" in scenario
+            ), f"Key 'tipping point reached' not found in scenario: {name}"
+            assert isinstance(
+                scenario["tipping point reached"], bool
+            ), f"Value for 'tipping point reached' is not boolean in scenario: {name}"
+
 
 class TestTippingPointInvalidInputs:
     @pytest.mark.parametrize(
@@ -73,6 +103,25 @@ def test_load_dict_with_invalid_inputs(self, invalid_tp_dict):
         with pytest.raises(ValueError):
             TippingPoint.load_dict(invalid_tp_dict)
 
+    def test_edge_cases_empty_sealevelrise(self, test_db):
+        tp_dict = {
+            "name": "tipping_point_test",
+            "description": "",
+            "event_set": "extreme12ft",
+            "strategy": "no_measures",
+            "projection": "current",
+            "sealevelrise": [],
+            "tipping_point_metric": [
+                ("TotalDamageEvent", 110974525.0, "greater"),
+                ("FullyFloodedRoads", 2305, "greater"),
+            ],
+        }
+        test_point = TippingPoint.load_dict(tp_dict)
+        test_point.create_tp_scenarios()
+        assert (
+            len(test_point.scenarios) == 0
+        ), "Scenarios should not be created for empty sealevelrise list"
+
 
 # database = read_database(
 #     rf"C:\\Users\\morenodu\\OneDrive - Stichting Deltares\\Documents\\GitHub\\FloodAdapt-Database",