Skip to content

Commit

Permalink
chore: Refactor object_model/tipping_point.py and interface/tipping_p…
Browse files Browse the repository at this point in the history
…oints.py

Refactor the `object_model/tipping_point.py` and `interface/tipping_points.py` files to improve code organization and naming consistency. This includes renaming the `TipPointModel` class to `TippingPointModel` in both files.
  • Loading branch information
dumontgoulart committed Jun 19, 2024
1 parent 3ab3964 commit b44fba8
Show file tree
Hide file tree
Showing 2 changed files with 29 additions and 35 deletions.
6 changes: 3 additions & 3 deletions flood_adapt/object_model/interface/tipping_points.py
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ class TippingPointOperator(str, Enum):
less = "less"


class TipPointModel(BaseModel):
class TippingPointModel(BaseModel):
"""BaseModel describing the expected variables and data types of a Tipping Point analysis object"""

name: str
Expand All @@ -68,11 +68,11 @@ class TipPointModel(BaseModel):
projection: str
sealevelrise: list[float] # could be a numpy array too
tipping_point_metric: list[tuple[TippingPointMetrics, float, TippingPointOperator]]
status: Optional[TippingPointStatus] = TippingPointStatus.not_reached
status: TippingPointStatus = TippingPointStatus.not_reached


class ITipPoint(ABC):
attrs: TipPointModel
attrs: TippingPointModel
database_input_path: Union[str, os.PathLike]
results_path: Union[str, os.PathLike]
scenarios: pd.DataFrame
Expand Down
58 changes: 26 additions & 32 deletions flood_adapt/object_model/tipping_point.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,14 +12,15 @@
import tomli_w
from fiat_toolbox.metrics_writer.fiat_read_metrics_file import MetricsFileReader

from flood_adapt.dbs_controller import Database
from flood_adapt.object_model.interface.tipping_points import (
TipPointModel,
TippingPointModel,
ITipPoint,
TippingPointStatus,
)
from flood_adapt.object_model.scenario import Scenario
from flood_adapt.object_model.io.unitfulvalue import UnitfulLength, UnitTypesLength
from flood_adapt.api.startup import read_database
from flood_adapt.api.static import read_database

"""
This script implements a Tipping Point model to analyze the impact of sea level rise (SLR)
Expand All @@ -45,33 +46,29 @@
class TippingPoint(ITipPoint):
"""Class holding all information related to tipping points analysis"""

attrs: TipPointModel
database_input_path: Union[str, os.PathLike]

def __init__(self, database_input_path: Union[str, os.PathLike]):
"""Initiation function when object is created through file or dict options"""
self.database_input_path = Path(database_input_path)
self.site_toml_path = (
Path(database_input_path).parent / "static" / "site" / "site.toml"
)

def init_object_model(self):
"""Create input and output folders for the tipping point"""

self.results_path = Path(self.database_input_path).parent.joinpath(
"output", "Scenarios", self.attrs.name
)

# create an input baseline folder for the scenarios
if not (self.database_input_path / "scenarios" / self.attrs.name).exists():
(self.database_input_path / "scenarios" / self.attrs.name).mkdir()
self.save(
self.database_input_path
/ "scenarios"
/ self.attrs.name
/ f"{self.attrs.name}.toml"
)
return self
# def init_object_model(self):
# """Create input and output folders for the tipping point"""
# self.results_path = Path(self.database_input_path).parent.joinpath(
# "output", "Scenarios", self.attrs.name
# )

# # create an input baseline folder for the scenarios
# if not (self.database_input_path / "scenarios" / self.attrs.name).exists():
# (self.database_input_path / "scenarios" / self.attrs.name).mkdir()
# self.save(
# self.database_input_path
# / "scenarios"
# / self.attrs.name
# / f"{self.attrs.name}.toml"
# )
# return self

def slr_projections(self, slr):
"""Create projections for sea level rise value"""
Expand All @@ -90,7 +87,7 @@ def slr_projections(self, slr):

def create_tp_scenarios(self):
"""Create scenarios for each sea level rise value"""
self.init_object_model()
# self.init_object_model()
# TODO: commenting out because now we are creating all scenarios, check it later to see how we deal with redundant scenarios
# self.check_scenarios()
# self.has_run = self.has_run_check()
Expand All @@ -99,15 +96,12 @@ def create_tp_scenarios(self):
self.slr_projections(slr)

# crete scenarios for each SLR value
str_rpl = str(slr).replace(".", "")
scenarios = {
f"slr_"
+ str(slr).replace(".", ""): {
"name": f"slr_" + str(slr).replace(".", ""),
f"slr_{str_rpl}": {
"name": f"slr_{str_rpl}",
"event": self.attrs.event_set,
# get a string from slr removing the dot
"projection": self.attrs.projection
+ "_slr"
+ str(slr).replace(".", ""),
"projection": f"{self.attrs.projection}_slr{str_rpl}",
"strategy": self.attrs.strategy,
}
for slr in self.attrs.sealevelrise
Expand Down Expand Up @@ -263,7 +257,7 @@ def load_file(
obj = TippingPoint(database_input_path)
with open(filepath, mode="rb") as fp:
toml = tomli.load(fp)
obj.attrs = TipPointModel.model_validate(toml)
obj.attrs = TippingPointModel.model_validate(toml)
return obj

def load_dict(
Expand All @@ -272,7 +266,7 @@ def load_dict(
"""create risk event from toml file"""

obj = TippingPoint(database_input_path)
obj.attrs = TipPointModel.model_validate(dct)
obj.attrs = TippingPointModel.model_validate(dct)
return obj

def save(self, filepath: Union[str, os.PathLike]):
Expand Down

0 comments on commit b44fba8

Please sign in to comment.