553 Adding files to Database when importing extrenal files

flood_adapt/api/events.py

@@ -7,16 +7,13 @@
 
 from flood_adapt.dbs_controller import Database
 from flood_adapt.object_model.hazard.event.event import Event
-from flood_adapt.object_model.hazard.event.event_factory import EventFactory
-from flood_adapt.object_model.hazard.event.historical_nearshore import (
+from flood_adapt.object_model.hazard.event.event_factory import (
+    EventFactory,
+    HistoricalHurricane,
     HistoricalNearshore,
-)
-from flood_adapt.object_model.interface.events import (
+    HistoricalOffshore,
     IEvent,
-    IHistoricalHurricane,
-    IHistoricalNearshore,
-    IHistoricalOffshore,
-    ISynthetic,
+    Synthetic,
 )
 from flood_adapt.object_model.io.unitfulvalue import UnitTypesLength
 
@@ -31,11 +28,11 @@ def get_event(name: str) -> IEvent:
 
 
 def get_event_mode(name: str) -> str:
-    filename = Database().events.get_database_path() / f"{name}" / f"{name}.toml"
+    filename = Database().events.input_path / f"{name}" / f"{name}.toml"
     return Event.get_mode(filename)
 
 
-def create_synthetic_event(attrs: dict[str, Any]) -> ISynthetic:
+def create_synthetic_event(attrs: dict[str, Any]) -> Synthetic:
     """Create a synthetic event object from a dictionary of attributes.
 
     Parameters
@@ -51,7 +48,7 @@ def create_synthetic_event(attrs: dict[str, Any]) -> ISynthetic:
     return EventFactory.get_event("Synthetic").load_dict(attrs)
 
 
-def create_historical_nearshore_event(attrs: dict[str, Any]) -> IHistoricalNearshore:
+def create_historical_nearshore_event(attrs: dict[str, Any]) -> HistoricalNearshore:
     """Create a historical nearshore event object from a dictionary of attributes.
 
     Parameters
@@ -67,7 +64,7 @@ def create_historical_nearshore_event(attrs: dict[str, Any]) -> IHistoricalNears
     return EventFactory.get_event("Historical_nearshore").load_dict(attrs)
 
 
-def create_historical_offshore_event(attrs: dict[str, Any]) -> IHistoricalOffshore:
+def create_historical_offshore_event(attrs: dict[str, Any]) -> HistoricalOffshore:
     """Create a historical offshore event object from a dictionary of attributes.
 
     Parameters
@@ -83,7 +80,7 @@ def create_historical_offshore_event(attrs: dict[str, Any]) -> IHistoricalOffsho
     return EventFactory.get_event("Historical_offshore").load_dict(attrs)
 
 
-def create_historical_hurricane_event(attrs: dict[str, Any]) -> IHistoricalHurricane:
+def create_historical_hurricane_event(attrs: dict[str, Any]) -> HistoricalHurricane:
     """Create a historical hurricane event object from a dictionary of attributes.
 
     Parameters
@@ -99,7 +96,7 @@ def create_historical_hurricane_event(attrs: dict[str, Any]) -> IHistoricalHurri
     return EventFactory.get_event("Historical_hurricane").load_dict(attrs)
 
 
-def save_event_toml(event: IEvent) -> None:
+def save_event(event: IEvent) -> None:
     Database().events.save(event)
 
 
@@ -152,3 +149,7 @@ def plot_wind(event: IEvent, input_wind_df: pd.DataFrame = None) -> str:
 
 def save_cyclone_track(event: IEvent, track: TropicalCyclone):
     Database().write_cyc(event, track)
+
+
+def get_cyclone_track_by_index(index: int) -> TropicalCyclone:
+    return Database().cyclone_track_database.get_track(index)

flood_adapt/api/measures.py

@@ -15,7 +15,7 @@
 from flood_adapt.object_model.interface.measures import (
     IMeasure,
 )
-from flood_adapt.object_model.interface.site import ISite
+from flood_adapt.object_model.interface.site import Site
 
 
 def get_measures() -> dict[str, Any]:
@@ -35,29 +35,26 @@ def create_measure(attrs: dict[str, Any], type: str = None) -> IMeasure:
         Dictionary of attributes for the measure.
     type : str
         Type of measure to create.
-    database : IDatabase, optional
-        Database to use for creating the measure, by default None
 
     Returns
     -------
     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
-
     if type == "elevate_properties":
-        return Elevate.load_dict(attrs, database_path)
+        return Elevate.load_dict(attrs)
     elif type == "buyout_properties":
-        return Buyout.load_dict(attrs, database_path)
+        return Buyout.load_dict(attrs)
     elif type == "floodproof_properties":
-        return FloodProof.load_dict(attrs, database_path)
+        return FloodProof.load_dict(attrs)
     elif type in ["floodwall", "thin_dam", "levee"]:
-        return FloodWall.load_dict(attrs, database_path)
+        return FloodWall.load_dict(attrs)
     elif type in ["pump", "culvert"]:
-        return Pump.load_dict(attrs, database_path)
+        return Pump.load_dict(attrs)
     elif type in ["water_square", "total_storage", "greening"]:
-        return GreenInfrastructure.load_dict(attrs, database_path)
+        return GreenInfrastructure.load_dict(attrs)
+    else:
+        raise ValueError(f"Invalid measure type: {type}")
 
 
 def save_measure(measure: IMeasure) -> None:
@@ -77,7 +74,7 @@ def copy_measure(old_name: str, new_name: str, new_description: str) -> None:
 
 
 # Green infrastructure
-def calculate_polygon_area(gdf: gpd.GeoDataFrame, site: ISite) -> float:
+def calculate_polygon_area(gdf: gpd.GeoDataFrame, site: Site) -> float:
     return GreenInfrastructure.calculate_polygon_area(gdf=gdf, site=site)
 
 

flood_adapt/api/output.py

@@ -66,11 +66,7 @@ def get_obs_point_timeseries(name: str) -> gpd.GeoDataFrame:
             f"Scenario {name} has not been run. Please run the scenario first."
         )
 
-    output_path = (
-        Database()
-        .scenarios.get_database_path(get_input_path=False)
-        .joinpath(hazard.name)
-    )
+    output_path = Database().scenarios.output_path.joinpath(hazard.name)
     gdf = Database().static.get_obs_points()
     gdf["html"] = [
         str(output_path.joinpath("Flooding", f"{station}_timeseries.html"))
@@ -106,9 +102,7 @@ def get_infographic(name: str) -> str:
         )
 
     config_path = database.static_path.joinpath("templates", "infographics")
-    output_path = database.scenarios.get_database_path(get_input_path=False).joinpath(
-        impact.name
-    )
+    output_path = database.scenarios.output_path.joinpath(impact.name)
     metrics_outputs_path = output_path.joinpath(f"Infometrics_{impact.name}.csv")
 
     infographic_path = InforgraphicFactory.create_infographic_file_writer(
@@ -142,13 +136,9 @@ def get_infometrics(name: str) -> pd.DataFrame:
         If the metrics file does not exist.
     """
     # Create the infographic path
-    metrics_path = (
-        Database()
-        .scenarios.get_database_path(get_input_path=False)
-        .joinpath(
-            name,
-            f"Infometrics_{name}.csv",
-        )
+    metrics_path = Database().scenarios.output_path.joinpath(
+        name,
+        f"Infometrics_{name}.csv",
     )
 
     # Check if the file exists

flood_adapt/api/scenarios.py

@@ -15,7 +15,7 @@ def get_scenario(name: str) -> IScenario:
 
 
 def create_scenario(attrs: dict[str, Any]) -> IScenario:
-    return Scenario.load_dict(attrs, Database().input_path)
+    return Scenario.load_dict(attrs)
 
 
 def save_scenario(scenario: IScenario) -> tuple[bool, str]:

flood_adapt/api/strategies.py

@@ -15,7 +15,7 @@ def get_strategy(name: str) -> IStrategy:
 
 
 def create_strategy(attrs: dict[str, Any]) -> IStrategy:
-    return Strategy.load_dict(attrs, Database().input_path)
+    return Strategy.load_dict(attrs)
 
 
 def save_strategy(strategy: IStrategy) -> None:

flood_adapt/database_builder/create_database.py

@@ -25,9 +25,8 @@
 from flood_adapt.api.static import read_database
 from flood_adapt.api.strategies import create_strategy, save_strategy
 from flood_adapt.log import FloodAdaptLogging
-from flood_adapt.object_model.interface.site import Obs_pointModel, SlrModel
+from flood_adapt.object_model.interface.site import Obs_pointModel, Site, SlrModel
 from flood_adapt.object_model.io.unitfulvalue import UnitfulDischarge, UnitfulLength
-from flood_adapt.object_model.site import Site
 
 config_path = None
 

flood_adapt/dbs_classes/dbs_benefit.py

@@ -6,9 +6,7 @@
 
 
 class DbsBenefit(DbsTemplate):
-    _type = "benefit"
-    _folder_name = "benefits"
-    _object_model_class = Benefit
+    _object_class = Benefit
 
     def save(self, benefit: IBenefit, overwrite: bool = False):
         """Save a benefit object in the database.
@@ -54,10 +52,7 @@ def delete(self, name: str, toml_only: bool = False):
         super().delete(name, toml_only=toml_only)
 
         # Delete output if edited
-        output_path = (
-            self._database.benefits.get_database_path(get_input_path=False) / name
-        )
-
+        output_path = self.output_path / name
         if output_path.exists():
             shutil.rmtree(output_path, ignore_errors=True)
 
@@ -78,10 +73,6 @@ def edit(self, benefit: IBenefit):
         super().edit(benefit)
 
         # Delete output if edited
-        output_path = (
-            self._database.benefits.get_database_path(get_input_path=False)
-            / benefit.attrs.name
-        )
-
+        output_path = self.output_path / benefit.attrs.name
         if output_path.exists():
             shutil.rmtree(output_path, ignore_errors=True)

flood_adapt/dbs_classes/dbs_event.py

@@ -1,20 +1,17 @@
-import shutil
 from pathlib import Path
 from typing import Any
 
 from flood_adapt.dbs_classes.dbs_template import DbsTemplate
 from flood_adapt.object_model.hazard.event.event import Event
 from flood_adapt.object_model.hazard.event.event_factory import EventFactory
 from flood_adapt.object_model.hazard.event.eventset import EventSet
-from flood_adapt.object_model.interface.events import IEvent, Mode
+from flood_adapt.object_model.interface.events import Mode
 
 
 class DbsEvent(DbsTemplate):
-    _type = "event"
-    _folder_name = "events"
-    _object_model_class = Event
+    _object_class = Event
 
-    def get(self, name: str) -> IEvent:
+    def get(self, name: str) -> Event:
         """Return an event object.
 
         Parameters
@@ -28,11 +25,13 @@ def get(self, name: str) -> IEvent:
             event object
         """
         # Get event path
-        event_path = self._path / f"{name}" / f"{name}.toml"
+        event_path = self.input_path / f"{name}" / f"{name}.toml"
 
         # Check if the object exists
         if not Path(event_path).is_file():
-            raise ValueError(f"{self._type.capitalize()} '{name}' does not exist.")
+            raise ValueError(
+                f"{self._object_class.class_name} '{name}' does not exist."
+            )
 
         # Load event
         mode = Event.get_mode(event_path)
@@ -44,7 +43,7 @@ def get(self, name: str) -> IEvent:
         elif mode == Mode.risk:
             return EventSet.load_file(event_path)
 
-    def list_objects(self) -> dict[str, Any]:
+    def list_objects(self) -> dict[str, list[Any]]:
         """Return a dictionary with info on the events that currently exist in the database.
 
         Returns
@@ -53,48 +52,11 @@ def list_objects(self) -> dict[str, Any]:
             Includes 'name', 'description', 'path' and 'last_modification_date' info
         """
         events = self._get_object_list()
-        objects = [self._database.events.get(name) for name in events["name"]]
+        objects = [self.get(name) for name in events["name"]]
         events["description"] = [obj.attrs.description for obj in objects]
         events["objects"] = objects
         return events
 
-    def copy(self, old_name: str, new_name: str, new_description: str):
-        """Copy (duplicate) an existing object, and give it a new name.
-
-        Parameters
-        ----------
-        old_name : str
-            name of the existing measure
-        new_name : str
-            name of the new measure
-        new_description : str
-            description of the new measure
-        """
-        # Check if the provided old_name is valid
-        if old_name not in self.list_objects()["name"]:
-            raise ValueError(f"'{old_name}' {self._type} does not exist.")
-
-        # First do a get and change the name and description
-        copy_object = self.get(old_name)
-        copy_object.attrs.name = new_name
-        copy_object.attrs.description = new_description
-
-        # After changing the name and description, receate the model to re-trigger the validators
-        copy_object.attrs = type(copy_object.attrs)(**copy_object.attrs.dict())
-
-        # Then a save. Checking whether the name is already in use is done in the save function
-        self.save(copy_object)
-
-        # Then save all the accompanied files
-        src = self._path / old_name
-        dest = self._path / new_name
-
-        EXCLUDE = [".spw", ".toml"]
-        for file in src.glob("*"):
-            if file.suffix in EXCLUDE:
-                continue
-            shutil.copy(file, dest / file.name)
-
     def _check_standard_objects(self, name: str) -> bool:
         """Check if an event is a standard event.
 
@@ -109,10 +71,10 @@ def _check_standard_objects(self, name: str) -> bool:
             True if the event is a standard event, False otherwise
         """
         # Check if event is a standard event
-        if self._database.site.attrs.standard_objects.events:
-            if name in self._database.site.attrs.standard_objects.events:
-                return True
-
+        if self._database.site.attrs.standard_objects:
+            if self._database.site.attrs.standard_objects.events:
+                if name in self._database.site.attrs.standard_objects.events:
+                    return True
         return False
 
     def check_higher_level_usage(self, name: str) -> list[str]:
@@ -129,10 +91,7 @@ def check_higher_level_usage(self, name: str) -> list[str]:
             list of scenarios that use the event
         """
         # Get all the scenarios
-        scenarios = [
-            self._database.scenarios.get(name)
-            for name in self._database.scenarios.list_objects()["name"]
-        ]
+        scenarios = self._database.scenarios.list_objects()["objects"]
 
         # Check if event is used in a scenario
         used_in_scenario = [