diff --git a/hydromt_fiat/api/data_types.py b/hydromt_fiat/api/data_types.py index bbe2a94a..c4c8ad46 100644 --- a/hydromt_fiat/api/data_types.py +++ b/hydromt_fiat/api/data_types.py @@ -104,7 +104,7 @@ class RoadVulnerabilitySettings(BaseModel): class ExposureRoadsSettings(BaseModel): roads_fn: str road_types: Union[List[str], bool] - road_damage: Union[str, int] + road_damage: int unit: Units diff --git a/hydromt_fiat/api/exposure_vm.py b/hydromt_fiat/api/exposure_vm.py index a805bcb2..aa886482 100644 --- a/hydromt_fiat/api/exposure_vm.py +++ b/hydromt_fiat/api/exposure_vm.py @@ -60,7 +60,7 @@ def set_asset_locations_source( ground_floor_height: str, fiat_key_maps: Optional[Dict[str, str]] = None, crs: Union[str, int] = None, - ): + ) -> None: if source == "NSI": # NSI is already defined in the data catalog self.exposure_buildings_model = ExposureBuildingsSettings( @@ -72,35 +72,6 @@ def set_asset_locations_source( extraction_method=ExtractionMethod.centroid.value, damage_types=["structure", "content"], ) - - # Download NSI from the database - region = self.data_catalog.get_geodataframe("area_of_interest") - self.exposure = ExposureVector( - data_catalog=self.data_catalog, - logger=self.logger, - region=region, - crs=crs, - ) - - self.exposure.setup_buildings_from_single_source( - source, - ground_floor_height, - "centroid", - ) - primary_object_types = ( - self.exposure.exposure_db["Primary Object Type"].unique().tolist() - ) - secondary_object_types = ( - self.exposure.exposure_db["Secondary Object Type"].unique().tolist() - ) - gdf = self.exposure.get_full_gdf(self.exposure.exposure_db) - - return ( - gdf, - primary_object_types, - secondary_object_types, - ) - elif source == "file" and fiat_key_maps is not None: # maybe save fiat_key_maps file in database # make calls to backend to derive file meta info such as crs, data type and driver @@ -119,6 +90,19 @@ def set_asset_locations_source( print(catalog_entry) # write to data catalog + def get_object_types(self): + if self.exposure: + primary_object_types = ( + self.exposure.exposure_db["Primary Object Type"].unique().tolist() + ) + secondary_object_types = ( + self.exposure.exposure_db["Secondary Object Type"].unique().tolist() + ) + return ( + primary_object_types, + secondary_object_types, + ) + def set_asset_data_source(self, source): self.exposure_buildings_model.asset_locations = source @@ -159,6 +143,27 @@ def set_ground_elevation(self, source: Union[int, float, None, str]): source=source ) + def set_roads_settings( + self, + road_types: List[str] = [ + "motorway", + "motorway_link", + "trunk", + "trunk_link", + "primary", + "primary_link", + "secondary", + "secondary_link", + ], + ): + self.exposure_roads_model = ExposureRoadsSettings( + roads_fn="OSM", + road_types=road_types, + road_damage=1, + unit=Units.ft.value, + ) + + def get_osm_roads( self, road_types: List[str] = [ diff --git a/hydromt_fiat/api/hydromt_fiat_vm.py b/hydromt_fiat/api/hydromt_fiat_vm.py index cfb744fe..9a86f07a 100644 --- a/hydromt_fiat/api/hydromt_fiat_vm.py +++ b/hydromt_fiat/api/hydromt_fiat_vm.py @@ -106,5 +106,19 @@ def run_hydromt_fiat(self): config_yaml = self.build_config_yaml() region = self.data_catalog.get_geodataframe("area_of_interest") self.fiat_model.build(region={"geom": region}, opt=config_yaml.dict()) - updated_exposure_output = gpd.GeoDataFrame(self.fiat_model.exposure.exposure_db.merge(self.exposure_vm.exposure.exposure_geoms[0], on ="Object ID"), geometry="geometry") - return updated_exposure_output + + exposure_db = self.fiat_model.exposure.exposure_db + if "setup_exposure_buildings" in config_yaml.dict() and "setup_exposure_roads" not in config_yaml.dict(): + # Only buildings are set up + buildings_gdf = self.fiat_model.exposure.get_full_gdf(exposure_db) + return buildings_gdf, None + elif "setup_exposure_buildings" in config_yaml.dict() and "setup_exposure_roads" in config_yaml.dict(): + # Buildings and roads are set up + full_gdf = self.fiat_model.exposure.get_full_gdf(exposure_db) + buildings_gdf = full_gdf.loc[full_gdf["Primary Object Type"] != "roads"] + roads_gdf = full_gdf.loc[full_gdf["Primary Object Type"] != "roads"] + return buildings_gdf, roads_gdf + elif "setup_exposure_buildings" not in config_yaml.dict() and "setup_exposure_roads" in config_yaml.dict(): + # Only roads are set up + roads_gdf = self.fiat_model.exposure.get_full_gdf(exposure_db) + return None, roads_gdf diff --git a/hydromt_fiat/workflows/equity_data.py b/hydromt_fiat/workflows/equity_data.py index 98cb5bcd..e91a53dd 100644 --- a/hydromt_fiat/workflows/equity_data.py +++ b/hydromt_fiat/workflows/equity_data.py @@ -9,6 +9,7 @@ from zipfile import ZipFile from pathlib import Path from typing import List +from hydromt_fiat.workflows.social_vulnerability_index import list_of_states class EquityData: @@ -54,7 +55,7 @@ def set_up_state_code(self, state_abbreviation: List[str]): states_done = [] for state in state_abbreviation: if state not in states_done: - self.logger.info(f"The states for which census data will be downloaded is (it's an abbreviation): {state}") + self.logger.info(f"The states for which census data will be downloaded is: {list_of_states(inverted=False)[state]}") state_obj = getattr(states, state) self.state_fips.append(state_obj.fips) states_done.append(state) diff --git a/hydromt_fiat/workflows/social_vulnerability_index.py b/hydromt_fiat/workflows/social_vulnerability_index.py index 72980132..53a08b6b 100644 --- a/hydromt_fiat/workflows/social_vulnerability_index.py +++ b/hydromt_fiat/workflows/social_vulnerability_index.py @@ -11,7 +11,7 @@ from pathlib import Path from typing import List -def list_of_states(): +def list_of_states(inverted: bool=True): states = { 'AK': 'Alaska', 'AL': 'Alabama', @@ -66,9 +66,11 @@ def list_of_states(): 'WY': 'Wyoming' } - states_inverted = {value: key for key, value in states.items()} - - return states_inverted + if inverted: + states_inverted = {value: key for key, value in states.items()} + return states_inverted + else: + return states class SocialVulnerabilityIndex: def __init__(self, data_catalog: DataCatalog, logger: Logger, save_folder: str): @@ -144,7 +146,7 @@ def set_up_state_code(self, state_abbreviation: List[str]): states_done = [] for state in state_abbreviation: if state not in states_done: - self.logger.info(f"The states for which census data will be downloaded is (it's an abbreviation): {state}") + self.logger.info(f"The states for which census data will be downloaded is: {list_of_states(inverted=False)[state]}") state_obj = getattr(states, state) self.state_fips.append(state_obj.fips) states_done.append(state)