Gui updates + Ground Floor Height from NSI

.gitignore

@@ -121,6 +121,8 @@ venv.bak/
 cache
 /examples/FIAT_database
 hydromt_fiat/data/damage_functions/flooding/AllDDF_HAZUS_fractions.xlsx
+hydromt_fiat/data/damage_functions/flooding/Hazus_IWR_curves.csv
+hydromt_fiat/data/damage_functions/flooding/Hazus_IWR_curves.xlsx
 examples/data/update_ground_floor_height/test_update_ground_floor_height_points
 examples/data/update_ground_floor_height/test_update_ground_floor_height_polygons
 examples/data/building_footprints/fiat_model_bfs

hydromt_fiat/api/data_types.py

@@ -1,5 +1,5 @@
 from enum import Enum
-from typing_extensions import Optional, TypedDict, Union
+from typing_extensions import Optional, TypedDict, Union, List
 
 from pydantic import BaseModel
 
@@ -53,36 +53,26 @@ class ModelIni(BaseModel):
     crs: str
 
 
-class ExposureVectorIni(BaseModel):
+class VulnerabilityIni(BaseModel):
+    vulnerability_fn: str
+    vulnerability_identifiers_and_linking_fn: str
+    unit: Units
+    functions_mean: Union[str, list]
+    functions_max: Union[str, list, None]
+    step_size: Union[float, None]
+
+
+class ExposureBuildingsIni(BaseModel):
     asset_locations: str
     occupancy_type: str
     max_potential_damage: str
     ground_floor_height: str
     unit: Units
     extraction_method: ExtractionMethod
-
-
-class HazardIni(BaseModel):
-    hazard_map_fn: str
-    hazard_type: str
-    return_period: Optional[Union[int, None]] = None
-    crs: Optional[str] = None
-    no_data: Optional[int] = -9999
-    var: Optional[Union[str, None]] = None
-    chunks: Optional[Union[int, str]] = "auto"
-    no_data: Optional[int] = -9999
-    var: Optional[Union[str, None]] = None
-    chunks: Optional[Union[int, str]] = "auto"
-
-
-class VulnerabilityIni(BaseModel):
-    vulnerability_fn: str
-    link_table: str
-    units: Units
+    damage_types : Union[List[str], None]
 
 
 class ConfigIni(BaseModel):
     setup_config: ModelIni
-    setup_hazard: HazardIni
     setup_vulnerability: VulnerabilityIni
-    setup_exposure_buildings: ExposureVectorIni
+    setup_exposure_buildings: ExposureBuildingsIni

hydromt_fiat/api/exposure_vm.py

@@ -12,7 +12,7 @@
     DataCatalogEntry,
     DataType,
     Driver,
-    ExposureVectorIni,
+    ExposureBuildingsIni,
     ExtractionMethod,
     Units,
 )
@@ -22,17 +22,19 @@ class ExposureViewModel:
     def __init__(
         self, database: IDatabase, data_catalog: DataCatalog, logger: logging.Logger
     ):
-        self.exposure_model = ExposureVectorIni(
+        self.exposure_model = ExposureBuildingsIni(
             asset_locations="",
             occupancy_type="",
             max_potential_damage=-999,
             ground_floor_height=-999,
             unit=Units.m.value,
             extraction_method=ExtractionMethod.centroid.value,
+            damage_types=["structure", "content"]
         )
         self.database: IDatabase = database
         self.data_catalog: DataCatalog = data_catalog
         self.logger: logging.Logger = logger
+        self.exposure: ExposureVector = None
 
     def create_interest_area(self, **kwargs: str):
         fpath = kwargs.get("fpath")
@@ -57,7 +59,6 @@ def set_asset_locations_source(
     ):
         if input_source == "NSI":
             # NSI is already defined in the data catalog
-            # Add NSI to the configuration file
             self.exposure_model.asset_locations = input_source
             self.exposure_model.occupancy_type = input_source
             self.exposure_model.max_potential_damage = input_source
@@ -66,28 +67,28 @@ def set_asset_locations_source(
 
             # Download NSI from the database
             region = self.data_catalog.get_geodataframe("area_of_interest")
-            exposure = ExposureVector(
+            self.exposure = ExposureVector(
                 data_catalog=self.data_catalog,
                 logger=self.logger,
                 region=region,
                 crs=crs,
             )
 
-            exposure.setup_from_single_source(
+            self.exposure.setup_buildings_from_single_source(
                 input_source,
                 self.exposure_model.ground_floor_height,
                 "centroid",  # TODO: MAKE FLEXIBLE
             )
             primary_object_types = (
-                exposure.exposure_db["Primary Object Type"].unique().tolist()
+                self.exposure.exposure_db["Primary Object Type"].unique().tolist()
             )
             secondary_object_types = (
-                exposure.exposure_db["Secondary Object Type"].unique().tolist()
+                self.exposure.exposure_db["Secondary Object Type"].unique().tolist()
             )
-            exposure.set_exposure_geoms_from_xy()
+            gdf = self.exposure.get_full_gdf(self.exposure.exposure_db)
 
             return (
-                exposure.exposure_geoms[0],
+                gdf,
                 primary_object_types,
                 secondary_object_types,
             )
@@ -110,13 +111,6 @@ def set_asset_locations_source(
             print(catalog_entry)
         # write to data catalog
 
-    def create_extraction_map(self, *args):
-        # TODO: implement callback
-        # if no exceptions, then self.exposure_model.extraction_method = args[0]
-        # else if
-        # make backend call to api with arguments to set extraction method per object:
-        # create first with default method. Then get uploaded or drawn area and merge with default methid
-        # save file to database
-        # change self.exposure_model.extraction_method to file
-        ...
-        # change self.exposure_model.extraction_method to file
+    def setup_extraction_method(self, extraction_method):
+        if self.exposure:
+            self.exposure.setup_extraction_method(extraction_method)

hydromt_fiat/api/hydromt_fiat_vm.py

@@ -7,58 +7,42 @@
 from hydromt_fiat.api.data_types import ConfigIni
 from hydromt_fiat.api.dbs_controller import LocalDatabase
 from hydromt_fiat.api.exposure_vm import ExposureViewModel
-from hydromt_fiat.api.hazard_vm import HazardViewModel
 from hydromt_fiat.api.model_vm import ModelViewModel
 from hydromt_fiat.api.vulnerability_vm import VulnerabilityViewModel
 from hydromt_fiat.fiat import FiatModel
 from hydromt.log import setuplog
 
 
-class Singleton(object):
-    _instance = None
-
-    def __new__(cls, *args: Any, **kwargs: Any):
-        if not isinstance(cls._instance, cls):
-            cls._instance = object.__new__(cls)
-        return cls._instance
-
-
-class HydroMtViewModel(Singleton):
-    is_initialized: bool = False
+class HydroMtViewModel:
     data_catalog: DataCatalog
     database: LocalDatabase
 
     def __init__(
         self,
         database_path: str,
         catalog_path: Union[List, str],
-        hydromt_fiat_path: str = None,
+        hydromt_fiat_path: str,
     ):
-        if not self.__class__.is_initialized:
-            database_path = Path(database_path)
-
-            HydroMtViewModel.database = LocalDatabase.create_database(database_path)
-            HydroMtViewModel.data_catalog = DataCatalog(catalog_path)
+        database_path = Path(database_path)
 
-            if hydromt_fiat_path is not None:
-                logger = setuplog("hydromt_fiat", log_level=10)
-                self.fiat_model = FiatModel(
-                    data_libs=catalog_path,
-                    root=hydromt_fiat_path,
-                    mode="w+",
-                    logger=logger,
-                )
+        HydroMtViewModel.database = LocalDatabase.create_database(database_path)
+        HydroMtViewModel.data_catalog = DataCatalog(catalog_path)
 
-                self.model_vm = ModelViewModel()
-                self.exposure_vm = ExposureViewModel(
-                    HydroMtViewModel.database, HydroMtViewModel.data_catalog, logger
-                )
-                self.vulnerability_vm = VulnerabilityViewModel(
-                    HydroMtViewModel.database, HydroMtViewModel.data_catalog, logger
-                )
-                self.hazard_vm = HazardViewModel()
+        logger = setuplog("hydromt_fiat", log_level=10)
+        self.fiat_model = FiatModel(
+            data_libs=catalog_path,
+            root=hydromt_fiat_path,
+            mode="w+",
+            logger=logger,
+        )
 
-            self.__class__.is_initialized = True
+        self.model_vm = ModelViewModel()
+        self.exposure_vm = ExposureViewModel(
+            HydroMtViewModel.database, HydroMtViewModel.data_catalog, logger
+        )
+        self.vulnerability_vm = VulnerabilityViewModel(
+            HydroMtViewModel.database, HydroMtViewModel.data_catalog, logger
+        )
 
     def clear_database(self):
         # TODO: delete database after hydromt_fiat has run
@@ -71,7 +55,6 @@ def save_data_catalog(self):
     def build_config_ini(self):
         config_ini = ConfigIni(
             setup_config=self.model_vm.config_model,
-            setup_hazard=self.hazard_vm.hazard_model,
             setup_vulnerability=self.vulnerability_vm.vulnerability_model,
             setup_exposure_buildings=self.exposure_vm.exposure_model,
         )
@@ -81,7 +64,17 @@ def build_config_ini(self):
         with open(database_path / "config.ini", "wb") as f:
             tomli_w.dump(config_ini.dict(exclude_none=True), f)
 
+    def read(self):
+        self.fiat_model.read()
+
     def run_hydromt_fiat(self):
+        config_ini = ConfigIni(
+            setup_config=self.model_vm.config_model,
+            setup_vulnerability=self.vulnerability_vm.vulnerability_model,
+            setup_exposure_buildings=self.exposure_vm.exposure_model,
+        )
         region = self.data_catalog.get_geodataframe("area_of_interest")
-        self.fiat_model.build(region={"geom": region}, opt=ConfigIni.dict())
+        self.fiat_model.build(region={"geom": region}, opt=config_ini.dict())
         self.fiat_model.write()
+
+

hydromt_fiat/api/vulnerability_vm.py

@@ -12,15 +12,29 @@ def __init__(
         self, database: IDatabase, data_catalog: DataCatalog, logger: logging.Logger
     ):
         self.vulnerability_model = VulnerabilityIni(
-            vulnerability_fn="", link_table="", units=Units.m.value
+            vulnerability_fn="",
+            vulnerability_identifiers_and_linking_fn="",
+            unit=Units.ft.value,
+            functions_mean="default",
         )
         self.database: IDatabase = database
         self.data_catalog: DataCatalog = data_catalog
         self.logger: logging.Logger = logger
-        self.linking_standard_colnames = ["Name", "Link"]
+        self.linking_standard_colnames = [
+            "FIAT Damage Function Name",
+            "Exposure Link",
+        ]  # add "Damage Type" ?
         self.linking_colnames = list()
         self.linking_list = list()
 
+    def get_default_curves_linking_file(self):
+        return self.data_catalog.get_dataframe("default_hazus_iwr_linking")
+
+    def add_vulnerability_curves_to_model(self, vulnerability_fn, link_table):
+        self.vulnerability_model.vulnerability_fn = vulnerability_fn
+        self.vulnerability_model.vulnerability_identifiers_and_linking_fn = link_table
+        print("Damage curves added")
+
     def get_hazus_curves(self):
         self.logger.info("Getting Hazus curves")
         df_vulnerability = self.data_catalog.get_dataframe("hazus_vulnerability_curves")
@@ -45,7 +59,10 @@ def save_linking_table(self):
         df = pd.DataFrame(
             columns=self.linking_standard_colnames, data=self.linking_list
         )
-        self.vulnerability_model.link_table = str(
+        self.vulnerability_model.vulnerability_identifiers_and_linking_fn = str(
             self.database.drive / "linking_table.csv"
         )
-        df.to_csv(self.vulnerability_model.link_table, index=False)
+        df.to_csv(
+            self.vulnerability_model.vulnerability_identifiers_and_linking_fn,
+            index=False,
+        )

hydromt_fiat/data/attribute_linking/NSI_attributes_to_FIAT.json

@@ -5,8 +5,8 @@
     "Secondary Object Type": "occtype",
     "Max Potential Damage: Structure": "val_struct",
     "Max Potential Damage: Content": "val_cont",
+    "Ground Floor Height": "found_ht",
     "Ground Elevation": "ground_elv",
-    "X Coordinate": "x",
-    "Y Coordinate": "y",
+    "geometry": "geometry",
     "Aggregation Label: Census Block": "cbfips"
 }

hydromt_fiat/fiat.py

@@ -318,7 +318,7 @@ def setup_exposure_buildings(
         """
         self.exposure = ExposureVector(self.data_catalog, self.logger, self.region)
 
-        if asset_locations == occupancy_type == max_potential_damage:
+        if asset_locations == occupancy_type == max_potential_damage == ground_floor_height:
             # The source for the asset locations, occupancy type and maximum potential
             # damage is the same, use one source to create the exposure data.
             self.exposure.setup_buildings_from_single_source(

hydromt_fiat/workflows/exposure_vector.py

@@ -183,16 +183,22 @@ def setup_buildings_from_single_source(
         if len(self.exposure_db.index) != len(set(self.exposure_db["Object ID"])):
             self.exposure_db["Object ID"] = range(1, len(self.exposure_db.index) + 1)
 
-        self.setup_ground_floor_height(ground_floor_height)
+        # Set the ground floor height if not yet set
+        if "Ground Floor Height" not in self.exposure_db.columns:
+            self.setup_ground_floor_height(ground_floor_height)
 
         # Set the extraction method
         self.setup_extraction_method(extraction_method)
 
-        # Set the geoms from the X and Y coordinates
-        self.set_exposure_geoms_from_xy()
+        # Set the exposure_geoms
+        self.set_exposure_geoms(gpd.GeoDataFrame(self.exposure_db[["Object ID", "geometry"]], crs=self.crs))
 
         # Set the name to the geom_names
         self.set_geom_names("buildings")
+
+        # Remove the geometry column from the exposure_db
+        if "geometry" in self.exposure_db:
+            del self.exposure_db["geometry"]
 
     def setup_roads(
         self,

tests/test_SVI_exposure.py

@@ -55,7 +55,7 @@
                 "asset_locations": "NSI",
                 "occupancy_type": "NSI",
                 "max_potential_damage": "NSI",
-                "ground_floor_height": 1,
+                "ground_floor_height": "NSI",
                 "unit": "ft",
             },
             "setup_social_vulnerability_index": {

tests/test_integration.py

@@ -55,7 +55,7 @@
                 "asset_locations": "NSI",
                 "occupancy_type": "NSI",
                 "max_potential_damage": "NSI",
-                "ground_floor_height": 1,
+                "ground_floor_height": "NSI",
                 "unit": "ft",
             },
             "setup_hazard": {

tests/test_vulnerability_exposure.py

@@ -56,7 +56,7 @@
                 "occupancy_type": "NSI",
                 "max_potential_damage": "NSI",
                 "damage_types": ["structure"],
-                "ground_floor_height": 1,
+                "ground_floor_height": "NSI",
                 "unit": "ft",
             },
         },