From 8f57e19a7016e4a7144e03efb1fb3f010b52ff91 Mon Sep 17 00:00:00 2001 From: Henrique Goulart <45360568+dumontgoulart@users.noreply.github.com> Date: Thu, 11 Jul 2024 16:21:40 +0200 Subject: [PATCH] Refactor object_model/tipping_point.py and interface/tipping_points.py --- .../object_model/interface/tipping_points.py | 18 +++---- flood_adapt/object_model/tipping_point.py | 35 ++++++------- .../test_object_model/test_tipping_points.py | 51 ++++++++++++++++++- 3 files changed, 77 insertions(+), 27 deletions(-) diff --git a/flood_adapt/object_model/interface/tipping_points.py b/flood_adapt/object_model/interface/tipping_points.py index 804a48303..0478944aa 100644 --- a/flood_adapt/object_model/interface/tipping_points.py +++ b/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,7 +44,7 @@ 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" @@ -52,14 +52,14 @@ class TippingPointStatus(str, Enum): 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.""" ... diff --git a/flood_adapt/object_model/tipping_point.py b/flood_adapt/object_model/tipping_point.py index 8f56bd51e..79c1f5fa5 100644 --- a/flood_adapt/object_model/tipping_point.py +++ b/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,13 +198,13 @@ 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) @@ -211,13 +212,13 @@ def load_file(filepath: Union[str, Path]) -> "TippingPoint": 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 = { diff --git a/tests/test_object_model/test_tipping_points.py b/tests/test_object_model/test_tipping_points.py index a107d489d..4f38177ee 100644 --- a/tests/test_object_model/test_tipping_points.py +++ b/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",