Skip to content

Commit

Permalink
Browse files Browse the repository at this point in the history
  • Loading branch information
frederique-hub committed Dec 15, 2023
1 parent b996e6b commit d854cdf
Show file tree
Hide file tree
Showing 5 changed files with 61 additions and 39 deletions.
2 changes: 1 addition & 1 deletion hydromt_fiat/api/data_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down
65 changes: 35 additions & 30 deletions hydromt_fiat/api/exposure_vm.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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
Expand All @@ -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

Expand Down Expand Up @@ -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] = [
Expand Down
18 changes: 16 additions & 2 deletions hydromt_fiat/api/hydromt_fiat_vm.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
3 changes: 2 additions & 1 deletion hydromt_fiat/workflows/equity_data.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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)
Expand Down
12 changes: 7 additions & 5 deletions hydromt_fiat/workflows/social_vulnerability_index.py
Original file line number Diff line number Diff line change
Expand Up @@ -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',
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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)
Expand Down

0 comments on commit d854cdf

Please sign in to comment.