refactor(adapter,object-model): Add adapter interfaces and refactor SficnsAdapter to use it. Also refactor events to use modular forcings

flood_adapt/api/benefits.py

@@ -3,47 +3,47 @@
 import geopandas as gpd
 import pandas as pd
 
-from flood_adapt.dbs_controller import Database
+import flood_adapt.dbs_controller as db
 from flood_adapt.object_model.benefit import Benefit
 from flood_adapt.object_model.interface.benefits import IBenefit
 
 
 def get_benefits() -> dict[str, Any]:
     # sorting and filtering either with PyQt table or in the API
-    return Database().benefits.list_objects()
+    return db.Database().benefits.list_objects()
 
 
 def get_benefit(name: str) -> IBenefit:
-    return Database().benefits.get(name)
+    return db.Database().benefits.get(name)
 
 
 def create_benefit(attrs: dict[str, Any]) -> IBenefit:
-    return Benefit.load_dict(attrs, Database().input_path)
+    return Benefit.load_dict(attrs, db.Database().input_path)
 
 
 def save_benefit(benefit: IBenefit) -> None:
-    Database().benefits.save(benefit)
+    db.Database().benefits.save(benefit)
 
 
 def edit_benefit(benefit: IBenefit) -> None:
-    Database().benefits.edit(benefit)
+    db.Database().benefits.edit(benefit)
 
 
 def delete_benefit(name: str) -> None:
-    Database().benefits.delete(name)
+    db.Database().benefits.delete(name)
 
 
 def check_benefit_scenarios(benefit: IBenefit) -> pd.DataFrame:
-    return Database().check_benefit_scenarios(benefit)
+    return db.Database().check_benefit_scenarios(benefit)
 
 
 def create_benefit_scenarios(benefit: IBenefit):
-    Database().create_benefit_scenarios(benefit)
+    db.Database().create_benefit_scenarios(benefit)
 
 
 def run_benefit(name: Union[str, list[str]]) -> None:
-    Database().run_benefit(name)
+    db.Database().run_benefit(name)
 
 
 def get_aggregation_benefits(name: str) -> dict[gpd.GeoDataFrame]:
-    return Database().get_aggregation_benefits(name)
+    return db.Database().get_aggregation_benefits(name)

flood_adapt/api/events.py

@@ -5,150 +5,94 @@
 import pandas as pd
 from cht_cyclones.tropical_cyclone import TropicalCyclone
 
-from flood_adapt.dbs_controller import Database
-from flood_adapt.object_model.hazard.event.event import Event
+import flood_adapt.dbs_controller as db
 from flood_adapt.object_model.hazard.event.event_factory import EventFactory
-from flood_adapt.object_model.hazard.event.historical_nearshore import (
-    HistoricalNearshore,
-)
-from flood_adapt.object_model.interface.events import (
-    IEvent,
-    IHistoricalHurricane,
-    IHistoricalNearshore,
-    IHistoricalOffshore,
-    ISynthetic,
-)
+from flood_adapt.object_model.hazard.event.historical import HistoricalEvent
+from flood_adapt.object_model.hazard.interface.events import IEvent, IEventModel
 from flood_adapt.object_model.io.unitfulvalue import UnitTypesLength
 
 
 def get_events() -> dict[str, Any]:
     # use PyQt table / sorting and filtering either with PyQt table or in the API
-    return Database().events.list_objects()
+    return db.Database().events.list_objects()
 
 
 def get_event(name: str) -> IEvent:
-    return Database().events.get(name)
+    return db.Database().events.get(name)
 
 
 def get_event_mode(name: str) -> str:
-    filename = Database().events.get_database_path() / f"{name}" / f"{name}.toml"
-    return Event.get_mode(filename)
+    filename = db.Database().events.get_database_path() / f"{name}" / f"{name}.toml"
+    return EventFactory.get_mode(filename)
 
 
-def create_synthetic_event(attrs: dict[str, Any]) -> ISynthetic:
-    """Create a synthetic event object from a dictionary of attributes.
+def create_event(attrs: dict[str, Any] | IEventModel) -> IEvent:
+    """Create a event object from a dictionary of attributes.
 
     Parameters
     ----------
-    attrs : dict[str, Any]
+    attrs : IEventModel [str, Any]
         Dictionary of attributes
 
     Returns
     -------
-    Synthetic
-        Synthetic event object
+    IEvent
+        Depending on attrs.template an event object.
+        Can be of type: Synthetic, Historical_nearshore, Historical_offshore, or Historical_hurricane.
     """
-    return EventFactory.get_event("Synthetic").load_dict(attrs)
-
-
-def create_historical_nearshore_event(attrs: dict[str, Any]) -> IHistoricalNearshore:
-    """Create a historical nearshore event object from a dictionary of attributes.
-
-    Parameters
-    ----------
-    attrs : dict[str, Any]
-        Dictionary of attributes
-
-    Returns
-    -------
-    HistoricalNearshore
-        Historical nearshore event object
-    """
-    return EventFactory.get_event("Historical_nearshore").load_dict(attrs)
-
-
-def create_historical_offshore_event(attrs: dict[str, Any]) -> IHistoricalOffshore:
-    """Create a historical offshore event object from a dictionary of attributes.
-
-    Parameters
-    ----------
-    attrs : dict[str, Any]
-        Dictionary of attributes
-
-    Returns
-    -------
-    HistoricalNearshore
-        Historical offshore event object
-    """
-    return EventFactory.get_event("Historical_offshore").load_dict(attrs)
-
-
-def create_historical_hurricane_event(attrs: dict[str, Any]) -> IHistoricalHurricane:
-    """Create a historical hurricane event object from a dictionary of attributes.
-
-    Parameters
-    ----------
-    attrs : dict[str, Any]
-        Dictionary of attributes
-
-    Returns
-    -------
-    HistoricalHurricane
-        Historical hurricane event object
-    """
-    return EventFactory.get_event("Historical_hurricane").load_dict(attrs)
+    return EventFactory.load_dict(attrs)
 
 
 def save_event_toml(event: IEvent) -> None:
-    Database().events.save(event)
+    db.Database().events.save(event)
 
 
 def save_timeseries_csv(name: str, event: IEvent, df: pd.DataFrame) -> None:
-    Database().write_to_csv(name, event, df)
+    db.Database().write_to_csv(name, event, df)
 
 
 def edit_event(event: IEvent) -> None:
-    Database().events.edit(event)
+    db.Database().events.edit(event)
 
 
 def delete_event(name: str) -> None:
-    Database().events.delete(name)
+    db.Database().events.delete(name)
 
 
 def copy_event(old_name: str, new_name: str, new_description: str) -> None:
-    Database().events.copy(old_name, new_name, new_description)
+    db.Database().events.copy(old_name, new_name, new_description)
 
 
 def download_wl_data(
     station_id, start_time, end_time, units: UnitTypesLength, source: str, file=None
 ) -> pd.DataFrame:
-    return HistoricalNearshore.download_wl_data(
+    return HistoricalEvent.download_wl_data(
         station_id, start_time, end_time, units, source, file
     )
 
 
 def read_csv(csvpath: Union[str, os.PathLike]) -> pd.DataFrame:
-    return Event.read_csv(csvpath)
+    return IEvent.read_csv(csvpath)
 
 
 def plot_wl(event: IEvent, input_wl_df: pd.DataFrame = None) -> str:
-    return Database().plot_wl(event, input_wl_df)
+    return db.Database().plot_wl(event, input_wl_df)
 
 
 def plot_river(
     event: IEvent,
     input_river_df: list[pd.DataFrame],
 ) -> str:
-    return Database().plot_river(event, input_river_df)
+    return db.Database().plot_river(event, input_river_df)
 
 
 def plot_rainfall(event: IEvent, input_rainfall_df: pd.DataFrame = None) -> str:
-    return Database().plot_rainfall(event, input_rainfall_df)
+    return db.Database().plot_rainfall(event, input_rainfall_df)
 
 
 def plot_wind(event: IEvent, input_wind_df: pd.DataFrame = None) -> str:
-    return Database().plot_wind(event, input_wind_df)
+    return db.Database().plot_wind(event, input_wind_df)
 
 
 def save_cyclone_track(event: IEvent, track: TropicalCyclone):
-    Database().write_cyc(event, track)
+    db.Database().write_cyc(event, track)

flood_adapt/api/measures.py

@@ -3,7 +3,7 @@
 import geopandas as gpd
 import pandas as pd
 
-from flood_adapt.dbs_controller import Database
+import flood_adapt.dbs_controller as db
 from flood_adapt.object_model.direct_impact.measure.buyout import Buyout
 from flood_adapt.object_model.direct_impact.measure.elevate import Elevate
 from flood_adapt.object_model.direct_impact.measure.floodproof import FloodProof
@@ -19,11 +19,11 @@
 
 
 def get_measures() -> dict[str, Any]:
-    return Database().measures.list_objects()
+    return db.Database().measures.list_objects()
 
 
 def get_measure(name: str) -> IMeasure:
-    return Database().measures.get(name)
+    return db.Database().measures.get(name)
 
 
 def create_measure(attrs: dict[str, Any], type: str = None) -> IMeasure:
@@ -44,7 +44,7 @@ def create_measure(attrs: dict[str, Any], type: str = None) -> IMeasure:
         Measure object.
     """
     # If a database is provided, use it to set the input path for the measure. Otherwise, set it to None.
-    database_path = Database().input_path
+    database_path = db.Database().input_path
 
     if type == "elevate_properties":
         return Elevate.load_dict(attrs, database_path)
@@ -61,19 +61,19 @@ def create_measure(attrs: dict[str, Any], type: str = None) -> IMeasure:
 
 
 def save_measure(measure: IMeasure) -> None:
-    Database().measures.save(measure)
+    db.Database().measures.save(measure)
 
 
 def edit_measure(measure: IMeasure) -> None:
-    Database().measures.edit(measure)
+    db.Database().measures.edit(measure)
 
 
 def delete_measure(name: str) -> None:
-    Database().measures.delete(name)
+    db.Database().measures.delete(name)
 
 
 def copy_measure(old_name: str, new_name: str, new_description: str) -> None:
-    Database().measures.copy(old_name, new_name, new_description)
+    db.Database().measures.copy(old_name, new_name, new_description)
 
 
 # Green infrastructure
@@ -90,4 +90,4 @@ def calculate_volume(
 
 
 def get_green_infra_table(measure_type: str) -> pd.DataFrame:
-    return Database().static.get_green_infra_table(measure_type)
+    return db.Database().static.get_green_infra_table(measure_type)

flood_adapt/api/output.py

@@ -6,40 +6,40 @@
 from fiat_toolbox.infographics.infographics_factory import InforgraphicFactory
 from fiat_toolbox.metrics_writer.fiat_read_metrics_file import MetricsFileReader
 
-from flood_adapt.dbs_controller import Database
+import flood_adapt.dbs_controller as db
 
 
 def get_outputs() -> dict[str, Any]:
     # sorting and filtering either with PyQt table or in the API
-    return Database().get_outputs()
+    return db.Database().get_outputs()
 
 
 def get_topobathy_path() -> str:
-    return Database().get_topobathy_path()
+    return db.Database().get_topobathy_path()
 
 
 def get_index_path() -> str:
-    return Database().get_index_path()
+    return db.Database().get_index_path()
 
 
 def get_depth_conversion() -> float:
-    return Database().get_depth_conversion()
+    return db.Database().get_depth_conversion()
 
 
 def get_max_water_level(name: str, rp: int = None) -> np.array:
-    return Database().get_max_water_level(name, rp)
+    return db.Database().get_max_water_level(name, rp)
 
 
 def get_fiat_footprints(name: str) -> gpd.GeoDataFrame:
-    return Database().get_fiat_footprints(name)
+    return db.Database().get_fiat_footprints(name)
 
 
 def get_aggregation(name: str) -> dict[gpd.GeoDataFrame]:
-    return Database().get_aggregation(name)
+    return db.Database().get_aggregation(name)
 
 
 def get_roads(name: str) -> gpd.GeoDataFrame:
-    return Database().get_roads(name)
+    return db.Database().get_roads(name)
 
 
 def get_obs_point_timeseries(name: str) -> gpd.GeoDataFrame:
@@ -58,7 +58,7 @@ def get_obs_point_timeseries(name: str) -> gpd.GeoDataFrame:
         The HTML strings of the water level timeseries
     """
     # Get the direct_impacts objects from the scenario
-    hazard = Database().scenarios.get(name).direct_impacts.hazard
+    hazard = db.Database().scenarios.get(name).direct_impacts.hazard
 
     # Check if the scenario has run
     if not hazard.has_run_check():
@@ -67,11 +67,11 @@ def get_obs_point_timeseries(name: str) -> gpd.GeoDataFrame:
         )
 
     output_path = (
-        Database()
+        db.Database()
         .scenarios.get_database_path(get_input_path=False)
         .joinpath(hazard.name)
     )
-    gdf = Database().static.get_obs_points()
+    gdf = db.Database().static.get_obs_points()
     gdf["html"] = [
         str(output_path.joinpath("Flooding", f"{station}_timeseries.html"))
         for station in gdf.name
@@ -96,7 +96,7 @@ def get_infographic(name: str) -> str:
         The HTML string of the infographic.
     """
     # Get the direct_impacts objects from the scenario
-    database = Database()
+    database = db.Database()
     impact = database.scenarios.get(name).direct_impacts
 
     # Check if the scenario has run
@@ -143,7 +143,7 @@ def get_infometrics(name: str) -> pd.DataFrame:
     """
     # Create the infographic path
     metrics_path = (
-        Database()
+        db.Database()
         .scenarios.get_database_path(get_input_path=False)
         .joinpath(
             name,