Tipping points

flood_adapt/api/tipping_points.py

@@ -0,0 +1,43 @@
+from typing import Any
+
+from flood_adapt.dbs_controller import Database
+from flood_adapt.object_model.interface.tipping_points import ITipPoint
+from flood_adapt.object_model.tipping_point import TippingPoint
+
+
+def get_tipping_points() -> dict[str, Any]:
+    # sorting and filtering either with PyQt table or in the API
+    return Database().tipping_points.list_objects()
+
+
+def get_tipping_point(name: str) -> ITipPoint:
+    return Database().tipping_points.get(name)
+
+
+def create_tipping_point(attrs: dict[str, Any]) -> ITipPoint:
+    return TippingPoint.load_dict(attrs)
+
+
+def save_tipping_point(tipping_point: ITipPoint) -> None:
+    Database().tipping_points.save(tipping_point)
+
+
+def edit_tipping_point(tipping_point: ITipPoint) -> None:
+    Database().tipping_points.edit(tipping_point)
+
+
+def delete_tipping_point(name: str) -> None:
+    Database().tipping_points.delete(name)
+
+
+def create_tipping_point_scenarios(name: str) -> None:
+    Database().tipping_points.get(name).create_tp_scenarios()
+
+
+def run_tipping_point(name: str) -> None:
+    Database().tipping_points.get(name).run_tp_scenarios()
+
+
+def check_scenario_has_run(name: str) -> None:
+    scenario_ran = Database().tipping_points.get(name).check_scenario_has_run()
+    return scenario_ran

flood_adapt/dbs_classes/dbs_tipping_point.py

@@ -0,0 +1,71 @@
+import shutil
+
+from flood_adapt.dbs_classes.dbs_template import DbsTemplate
+from flood_adapt.object_model.interface.tipping_points import ITipPoint
+from flood_adapt.object_model.tipping_point import TippingPoint
+
+
+class DbsTippingPoint(DbsTemplate):
+    _type = "tipping_point"
+    _folder_name = "tipping_points"
+    _object_model_class = TippingPoint
+
+    def save(self, tipping_point: ITipPoint, overwrite: bool = False):
+        """Save a tipping point object in the database.
+
+        Parameters
+        ----------
+        tipping_point : ITipPoint
+            object of tipping point type
+        overwrite : bool, optional
+            whether to overwrite existing tipping point with same name, by default False
+
+        Raises
+        ------
+        ValueError
+            Raise error if name is already in use. Names of tipping points should be unique.
+        """
+        # Save the tipping point
+        super().save(tipping_point, overwrite=overwrite)
+
+    def delete(self, name: str, toml_only: bool = False):
+        """Delete an already existing tipping point in the database.
+
+        Parameters
+        ----------
+        name : str
+            name of the tipping point
+        toml_only : bool, optional
+            whether to only delete the toml file or the entire folder. If the folder is empty after deleting the toml,
+            it will always be deleted. By default False
+        """
+        # First delete the tipping point
+        super().delete(name, toml_only=toml_only)
+
+        # Delete output if edited
+        output_path = (
+            self._database.tipping_points.get_database_path(get_input_path=False) / name
+        )
+
+        if output_path.exists():
+            shutil.rmtree(output_path)
+
+    def edit(self, tipping_point: ITipPoint):
+        """Edit an already existing tipping point in the database.
+
+        Parameters
+        ----------
+        tipping_point : ITipPoint
+            object of tipping point type
+        """
+        # Edit the tipping point
+        super().edit(tipping_point)
+
+        # Delete output if edited
+        output_path = (
+            self._database.tipping_points.get_database_path(get_input_path=False)
+            / tipping_point.attrs.name
+        )
+
+        if output_path.exists():
+            shutil.rmtree(output_path)

flood_adapt/dbs_controller.py

@@ -23,6 +23,7 @@
 from flood_adapt.dbs_classes.dbs_scenario import DbsScenario
 from flood_adapt.dbs_classes.dbs_static import DbsStatic
 from flood_adapt.dbs_classes.dbs_strategy import DbsStrategy
+from flood_adapt.dbs_classes.dbs_tipping_point import DbsTippingPoint
 from flood_adapt.integrator.sfincs_adapter import SfincsAdapter
 from flood_adapt.log import FloodAdaptLogging
 from flood_adapt.object_model.hazard.event.event_factory import EventFactory
@@ -63,6 +64,7 @@ class Database(IDatabase):
     _measures: DbsMeasure
     _projections: DbsProjection
     _benefits: DbsBenefit
+    _tipping_points: DbsTippingPoint
 
     def __new__(cls, *args, **kwargs):
         if not cls._instance:  # Singleton pattern
@@ -133,6 +135,7 @@ def __init__(
         self._measures = DbsMeasure(self)
         self._projections = DbsProjection(self)
         self._benefits = DbsBenefit(self)
+        self._tipping_points = DbsTippingPoint(self)
 
         # Delete any unfinished/crashed scenario output
         self.cleanup()
@@ -195,6 +198,10 @@ def projections(self) -> DbsProjection:
     def benefits(self) -> DbsBenefit:
         return self._benefits
 
+    @property
+    def tipping_points(self) -> DbsTippingPoint:
+        return self._tipping_points
+
     def interp_slr(self, slr_scenario: str, year: float) -> float:
         r"""Interpolate SLR value and reference it to the SLR reference year from the site toml.
 
@@ -750,6 +757,7 @@ def update(self) -> None:
         self.strategies = self._strategies.list_objects()
         self.scenarios = self._scenarios.list_objects()
         self.benefits = self._benefits.list_objects()
+        self.tipping_points = self._tipping_points.list_objects()
 
     def get_outputs(self) -> dict[str, Any]:
         """Return a dictionary with info on the outputs that currently exist in the database.

flood_adapt/object_model/benefit.py

@@ -606,4 +606,4 @@ def save(self, filepath: Union[str, os.PathLike]):
             path for saving the toml file
         """
         with open(filepath, "wb") as f:
-            tomli_w.dump(self.attrs.dict(exclude_none=True), f)
+            tomli_w.model_dump(self.attrs, f, exclude_none=True)

flood_adapt/object_model/interface/tipping_points.py

@@ -0,0 +1,102 @@
+import os
+from abc import ABC, abstractmethod
+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."""
+
+    # based on what I have found in floodadapt - but can be changed
+    FloodedAll = "FloodedAll"
+    FloodedLowVulnerability = "FloodedLowVulnerability"
+    FloodedHighVulnerability = "FloodedHighVulnerability"
+    TotalDamageEvent = "TotalDamageEvent"
+    TotalResDamageEvent = "TotalResDamageEvent"
+    ResidentialMinorCount = "ResidentialMinorCount"
+    ResidentialMajorCount = "ResidentialMajorCount"
+    ResidentialDestroyedCount = "ResidentialDestroyedCount"
+    CommercialCount = "CommercialCount"
+    CommercialMinorCount = "CommercialMinorCount"
+    CommercialMajorCount = "CommercialMajorCount"
+    CommercialDestroyedCount = "CommercialDestroyedCount"
+    HealthCount = "HealthCount"
+    HealthMinorCount = "HealthMinorCount"
+    HealthMajorCount = "HealthMajorCount"
+    HealthDestroyedCount = "HealthDestroyedCount"
+    SchoolsCount = "SchoolsCount"
+    SchoolsMinorCount = "SchoolsMinorCount"
+    SchoolsMajorCount = "SchoolsMajorCount"
+    SchoolsDestroyedCount = "SchoolsDestroyedCount"
+    EmergencyCount = "EmergencyCount"
+    EmergencyMinorCount = "EmergencyMinorCount"
+    EmergencyMajorCount = "EmergencyMajorCount"
+    EmergencyDestroyedCount = "EmergencyDestroyedCount"
+    DisplacedLowVulnerability = "DisplacedLowVulnerability"
+    DisplacedHighVulnerability = "DisplacedHighVulnerability"
+    SlightlyFloodedRoads = "SlightlyFloodedRoads"
+    MinorFloodedRoads = "MinorFloodedRoads"
+    MajorFloodedRoads = "MajorFloodedRoads"
+    FullyFloodedRoads = "FullyFloodedRoads"
+
+
+class TippingPointStatus(str, Enum):
+    """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."""
+
+    greater = "greater"
+    less = "less"
+
+
+class TippingPointModel(BaseModel):
+    """BaseModel describing the expected variables and data types of a Tipping Point analysis object."""
+
+    name: str
+    description: Optional[str] = ""
+    strategy: str
+    event_set: str
+    projection: str
+    sealevelrise: list[float]  # could be a numpy array too
+    tipping_point_metric: list[tuple[TippingPointMetrics, float, TippingPointOperator]]
+    status: TippingPointStatus = TippingPointStatus.not_reached
+    scenarios: list[str] = []
+
+
+class ITipPoint(ABC):
+    attrs: TippingPointModel
+    database_input_path: Union[str, os.PathLike]
+    results_path: Union[str, os.PathLike]
+    scenarios: pd.DataFrame
+    has_run: bool = False
+
+    @staticmethod
+    @abstractmethod
+    def load_file(filepath: Union[str, os.PathLike]):
+        """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."""
+        ...
+
+    @abstractmethod
+    def save(self, filepath: Union[str, os.PathLike]):
+        """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."""
+        ...

flood_adapt/object_model/scenario.py

@@ -62,7 +62,7 @@ def load_dict(data: dict[str, Any], database_input_path: os.PathLike):
     def save(self, filepath: Union[str, os.PathLike]):
         """Save Scenario to a toml file."""
         with open(filepath, "wb") as f:
-            tomli_w.dump(self.attrs.dict(exclude_none=True), f)
+            tomli_w.dump(self.attrs.model_dump(exclude_none=True), f)
 
     def run(self):
         """Run direct impact models for the scenario."""