Running FloodAdapt on Linux

.gitignore

@@ -166,3 +166,6 @@ cython_debug/
 .vscode/launch.json
 /tests/test_database
 /tests/system
+
+# Pixi
+.pixi

flood_adapt/api/scenarios.py

@@ -18,7 +18,7 @@ def create_scenario(attrs: dict[str, Any]) -> IScenario:
     return Scenario.load_dict(attrs, Database().input_path)
 
 
-def save_scenario(scenario: IScenario) -> (bool, str):
+def save_scenario(scenario: IScenario) -> tuple[bool, str]:
     """Save the scenario to the Database().
 
     Parameters

flood_adapt/config.py

@@ -1,167 +1,85 @@
-import os
 from pathlib import Path
-from typing import Union
+from platform import system
+from typing import Any, Dict, Union
 
 import tomli
-
-
-def set_database_root(database_root: Path, overwrite: bool = True) -> None:
-    """
-    Set the database root path.
-
-    Parameters
-    ----------
-    database_root : Path
-        The absolute new database_root path.
-    overwrite : bool, optional
-        If False, it will only be set if it is not already set.
-
-    Returns
-    -------
-    None
-
-    Raises
-    ------
-    ValueError
-        If the provided database root is not a valid directory.
-    """
-    if database_root == get_database_root():
-        return
-    abs_database_root = Path(database_root).resolve()
-    if not Path(abs_database_root).is_dir():
-        raise ValueError(f"{abs_database_root} is not a valid database root directory")
-
-    if get_database_root() is None:
-        os.environ["DATABASE_ROOT"] = str(abs_database_root)
-        print(f"database_root set: {abs_database_root}")
-    elif overwrite:
-        print(f"database_root overwritten: {abs_database_root}")
-        os.environ["DATABASE_ROOT"] = str(abs_database_root)
-
-
-def set_system_folder(system_folder: Path, overwrite: bool = True) -> None:
-    """
-    Set the system folder path.
-
-    Parameters
-    ----------
-    system_folder : Path
-        The new system folder path.
-    overwrite : bool, optional
-        If False, it will only be set if it is not already set.
-
-    Returns
-    -------
-    None
-
-    Raises
-    ------
-    ValueError
-        If the provided system folder is not a valid directory.
-    """
-    if system_folder == get_system_folder():
-        return
-    abs_system_folder = Path(system_folder).resolve()
-    if not Path(abs_system_folder).is_dir():
-        raise ValueError(f"{abs_system_folder} is not a valid system folder directory")
-
-    if get_system_folder() is None:
-        os.environ["SYSTEM_FOLDER"] = str(abs_system_folder)
-        print(f"system_folder set: {abs_system_folder}")
-    elif overwrite:
-        print(f"system_folder overwritten: {abs_system_folder}")
-        os.environ["SYSTEM_FOLDER"] = str(abs_system_folder)
-
-
-def set_database_name(database_name: str, overwrite: bool = True) -> None:
-    """
-    Set the database_name.
-
-    Parameters
-    ----------
-    database_name : str
-        The new database name.
-    overwrite : bool, optional
-        If False, it will only be set if it is not already set.
-
-    Returns
-    -------
-    None
-
-    Raises
-    ------
-    ValueError
-        If DATABASE_ROOT is not set or if the provided database_name is not a valid directory in DATABASE_ROOT.
-    """
-    if database_name == get_database_name():
-        return
-    db_root = get_database_root()
-    if db_root is None:
-        raise ValueError(
-            "DATABASE_ROOT is not set, set it before setting DATABASE_NAME\n"
-        )
-
-    full_database_path = Path(db_root, database_name)
-    if not full_database_path.is_dir():
-        raise ValueError(f"{full_database_path} is not a valid directory\n")
-
-    if get_database_name() is None:
-        os.environ["DATABASE_NAME"] = str(database_name)
-        print(f"database_name set: {database_name}")
-    elif overwrite:
-        print(f"database_name overwritten: {database_name}")
-        os.environ["DATABASE_NAME"] = str(database_name)
-
-
-def get_database_root() -> Union[Path, None]:
-    """
-    Get the root directory for the database.
-
-    Returns
-    -------
-    Path or None
-        The path to the root of the database if the DATABASE_ROOT environment variable is set,
-        None otherwise.
-    """
-    if os.environ.get("DATABASE_ROOT", None):
-        return Path(os.environ["DATABASE_ROOT"])
-
-
-def get_system_folder() -> Union[Path, None]:
-    """
-    Get the system folder path.
-
-    Returns
-    -------
-    Path or None
-        The system folder path if the SYSTEM_FOLDER environment variable is set, otherwise None.
-    """
-    if os.environ.get("SYSTEM_FOLDER", None):
-        return Path(os.environ["SYSTEM_FOLDER"])
-
-
-def get_database_name() -> Union[str, None]:
-    """
-    Get the database name.
-
-    Returns
-    -------
-    str or None
-        The database name if the DATABASE_NAME environment variable is set, otherwise None.
-    """
-    return os.environ.get("DATABASE_NAME", None)
-
-
-def parse_config(config_path: Path, overwrite: bool = True) -> dict:
-    """
-    Parse the configuration file and return the parsed configuration dictionary.
+from pydantic import Field, computed_field, model_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_ignore_empty=True, validate_default=True
+    )  # empty env uses default
+
+    database_root: Path = Field(default=Path(__file__).parents[2] / "Database")
+    system_folder: Path
+    database_name: str = Field(default="default")
+    sfincs_path: Path
+    fiat_path: Path
+
+    @computed_field
+    @property
+    def database_path(self) -> Path:
+        return self.database_root / self.database_name
+
+    @model_validator(mode="before")
+    def set_defaults_if_missing(cls, data: Any) -> "Settings":
+        if isinstance(data, dict):
+            cls._validate_system_folder(data)
+            cls._validate_fiat_path(data)
+            cls._validate_sfincs_path(data)
+        return data
+
+    @classmethod
+    def _validate_system_folder(cls, data: Dict[str, Any]):
+        """Set system_folder path if not set."""
+        field_name: str = "system_folder"
+        system_folder: Any = data.get(field_name)
+        if system_folder is None:
+            # base system folder on database root.
+            database_root: Union[Path, str] = (
+                data.get("database_root") or cls.model_fields["database_root"].default
+            )
+            database_root: Path = Path(database_root)
+            data[field_name] = database_root / "system"
+
+    @classmethod
+    def _validate_sfincs_path(cls, data: Dict[str, Any]):
+        """Set SFINCS path if not set."""
+        field_name: str = "sfincs_path"
+        sfincs_path: Any = data.get(field_name)
+        if sfincs_path is None:
+            system_folder: Path = data.get("system_folder")
+            if system() == "Windows":
+                data[field_name] = system_folder / "sfincs.exe"
+            else:
+                data[field_name] = system_folder / "sfincs"
+
+    @classmethod
+    def _validate_fiat_path(cls, data: Any) -> Path:
+        """Set FIAT path if not set."""
+        field_name: str = "fiat_path"
+        fiat_path: Any = data.get(field_name)
+        if fiat_path is None:
+            system_folder: Path = data.get("system_folder")
+            if system() == "Windows":
+                data[field_name] = system_folder / "fiat" / "fiat.exe"
+            else:
+                data[field_name] = system_folder / "fiat" / "fiat"
+
+
+settings = Settings()
+
+
+def parse_config(config_path: Path) -> Settings:
+    """
+    Parse the configuration file and return the parsed settings.
 
     Parameters
     ----------
     config_path : Path
         The path to the configuration file.
-    overwrite : bool, optional
-        Flag indicating whether to overwrite existing configuration values, defaults to True.
 
     Returns
     -------
@@ -176,78 +94,4 @@ def parse_config(config_path: Path, overwrite: bool = True) -> dict:
     with open(config_path, "rb") as f:
         config = tomli.load(f)
 
-    config_base_dir = config_path.parent
-
-    try:
-        # Parse the config file
-        if "DATABASE_ROOT" not in config:
-            raise ValueError(f"DATABASE_ROOT not found in {config_path}")
-        database_root = config_base_dir / config["DATABASE_ROOT"]
-        set_database_root(database_root, overwrite=overwrite)
-
-        if "SYSTEM_FOLDER" not in config:
-            raise ValueError(f"SYSTEM_FOLDER not found in {config_path}")
-        system_folder = config_base_dir / config["SYSTEM_FOLDER"]
-        set_system_folder(system_folder, overwrite=overwrite)
-
-        if "DATABASE_NAME" not in config:
-            raise ValueError(f"DATABASE_NAME not found in {config_path}")
-        set_database_name(config["DATABASE_NAME"], overwrite=overwrite)
-
-        if overwrite:
-            print(f"Configuration loaded from {config_path}")
-
-    except ValueError as e:
-        full_error = f"""
-        {e}
-        Error parsing configuration toml file: {config_path}
-        Please make sure the file is formatted correctly and contains the required fields.
-        """
-        raise ValueError(full_error)
-
-    return config
-
-
-def parse_user_input(
-    database_root=None, system_folder=None, database_name=None, overwrite=True
-) -> None:
-    """
-    Parse the user input and set the corresponding configuration values.
-
-    Parameters
-    ----------
-    database_root : str, optional
-        The absolute path to the root directory of the database.
-    system_folder : str, optional
-        The absolute path to the system folder.
-    database_name : str, optional
-        The name of the database.
-    overwrite : bool, optional
-        Whether to overwrite existing configuration values. Default is True.
-
-    Returns
-    -------
-    None
-    """
-    # Set database_root if given
-    if database_root is not None:
-        set_database_root(database_root, overwrite=overwrite)
-
-    # Set system folder if given
-    if system_folder is not None:
-        set_system_folder(system_folder, overwrite=overwrite)
-
-    # Set database_name if given
-    if database_name is not None:
-        set_database_name(database_name, overwrite=overwrite)
-
-    if any(val is not None for val in [database_root, system_folder, database_name]):
-        print("Parsed user input successfully")
-
-
-def main() -> None:
-    pass
-
-
-if __name__ == "__main__":
-    main()
+    return Settings(config)

flood_adapt/object_model/benefit.py

@@ -202,7 +202,7 @@ def cba(self):
         scenarios = self.scenarios.copy(deep=True)
         scenarios["EAD"] = None
 
-        results_path = self.database_input_path.parent.joinpath("output", "Scenarios")
+        results_path = self.database_input_path.parent.joinpath("output", "scenarios")
 
         # Get metrics per scenario
         for index, scenario in scenarios.iterrows():
@@ -280,7 +280,7 @@ def cba(self):
 
     def cba_aggregation(self):
         """Zonal Benefits analysis for the different aggregation areas."""
-        results_path = self.database_input_path.parent.joinpath("output", "Scenarios")
+        results_path = self.database_input_path.parent.joinpath("output", "scenarios")
         # Get years of interest
         year_start = self.attrs.current_situation.year
         year_end = self.attrs.future_year

flood_adapt/object_model/direct_impacts.py

@@ -15,7 +15,7 @@
 from fiat_toolbox.spatial_output.aggregation_areas import AggregationAreas
 from fiat_toolbox.spatial_output.points_to_footprint import PointsToFootprints
 
-import flood_adapt.config as FloodAdapt_config
+from flood_adapt.config import settings
 from flood_adapt.integrator.fiat_adapter import FiatAdapter
 from flood_adapt.log import FloodAdaptLogging
 from flood_adapt.object_model.direct_impact.impact_strategy import ImpactStrategy
@@ -189,7 +189,7 @@ def preprocess_fiat(self):
                 / self.scenario.projection
                 / self.socio_economic_change.attrs.new_development_shapefile
             )
-            dem = self.database.static_path / "Dem" / self.site_info.attrs.dem.filename
+            dem = self.database.static_path / "dem" / self.site_info.attrs.dem.filename
             aggregation_areas = [
                 self.database.static_path / aggr.file
                 for aggr in self.site_info.attrs.fiat.aggregation
@@ -251,20 +251,12 @@ def preprocess_fiat(self):
         del fa
 
     def run_fiat(self):
-        if not FloodAdapt_config.get_system_folder():
-            raise ValueError(
-                """
-                SYSTEM_FOLDER environment variable is not set. Set it by calling FloodAdapt_config.set_system_folder() and provide the path.
-                The path should be a directory containing folders with the model executables
-                """
-            )
-        fiat_exec = FloodAdapt_config.get_system_folder() / "fiat" / "fiat.exe"
-
         with cd(self.fiat_path):
             with open(self.fiat_path.joinpath("fiat.log"), "a") as log_handler:
                 process = subprocess.run(
-                    f'"{fiat_exec}" run settings.toml',
+                    f'"{settings.fiat_path}" run settings.toml',
                     stdout=log_handler,
+                    env={},  # need environment variables from runtime hooks
                     check=True,
                     shell=True,
                 )