Skip to content

Commit

Permalink
Vulnerability configuration generation and documentation. (#364)
Browse files Browse the repository at this point in the history 
* Move vulnerability config building into physrisk itself.

Signed-off-by: Joe Moorhouse <5102656+joemoorhouse@users.noreply.github.com>

* Lint.

Signed-off-by: Joe Moorhouse <5102656+joemoorhouse@users.noreply.github.com>

* Fix bug running with tox.

Signed-off-by: Joe Moorhouse <5102656+joemoorhouse@users.noreply.github.com>

---------

Signed-off-by: Joe Moorhouse <5102656+joemoorhouse@users.noreply.github.com>
  • Loading branch information
@joemoorhouse
joemoorhouse authored Oct 20, 2024
1 parent 6477185 commit 7be265b
Show file tree
Hide file tree
Showing 6 changed files with 264 additions and 215 deletions.
6 changes: 6 additions & 0 deletions docs/references.bib
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
@techreport{huizinga2017global,
title={Global flood depth-damage functions: Methodology and the database with guidelines},
author={Huizinga, Jan and De Moel, Hans and Szewczyk, Wojciech},
year={2017},
institution={Joint Research Centre}
}
13 changes: 13 additions & 0 deletions docs/user_guide/vulnerability/vulnerability.rst
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,3 +9,16 @@ Vulnerability configuration is generated programmatically from a number of diffe

vulnerability_config
vulnerability_uncertainty


Vulnerability function sources
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

Flood
-----
.. toctree::
:maxdepth: 2

vulnerability_functions/inundation_jrc/onboard

.. bibliography::
157 changes: 90 additions & 67 deletions docs/user_guide/vulnerability/vulnerability_functions/inundation_jrc/onboard.ipynb
View file

Large diffs are not rendered by default.

141 changes: 141 additions & 0 deletions src/physrisk/vulnerability_models/configuration/config_builders.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,141 @@
from pathlib import Path
from typing import List, Optional, Protocol
from fsspec import AbstractFileSystem
from fsspec.implementations.local import LocalFileSystem
import numpy as np
import pandas as pd

from physrisk.kernel.assets import RealEstateAsset
from physrisk.kernel.hazards import (
CoastalInundation,
PluvialInundation,
RiverineInundation,
)
from physrisk.kernel.impact_distrib import ImpactType
from physrisk.vulnerability_models.config_based_impact_curves import (
VulnerabilityConfigItem,
config_items_to_csv,
)


def build_all_config(source_dir: Optional[Path] = None):
"""Build all of the configuration."""
config_items: list[VulnerabilityConfigItem] = []
source_dir = vulnerability_onboarding_dir() if source_dir is None else source_dir
config_builders = [ConfigBuilderJRCFlood(source_dir=source_dir)]
for builder in config_builders:
config_items += builder.build_config()
config_items_to_csv(
config_items, vulnerability_config_dir() / "candidate_vulnerability_config.csv"
)


def vulnerability_onboarding_dir():
"""Path of the vulnerability on-boarding directory."""
path = (
Path(__file__).parents[4]
/ "docs"
/ "user_guide"
/ "vulnerability"
/ "vulnerability_functions"
)
return path


def vulnerability_config_dir():
"""Path of the vulnerability configuration directory."""
path = Path(__file__).parents[2] / "data" / "static" / "vulnerability"
return path


class ConfigBuilder(Protocol):
def build_config(self) -> List[VulnerabilityConfigItem]: ...


class ConfigBuilderBase:
def __init__(self, source_dir: Path, fs: Optional[AbstractFileSystem] = None):
"""_summary_
Args:
source_dir (str): Full path to directory containing any source files.
Defaults to None, in which case the default on-boarding directory in the
repo is used.
fs (Optional[AbstractFileSystem], optional): Instance of AbstractFileSystem to use to
access source_dir e.g. S3FileSystem.
Defaults to None, in which case a LocalFileSystem is assumed.
"""
self.fs = fs if fs is not None else LocalFileSystem()
self.source_dir = source_dir


class ConfigBuilderJRCFlood(ConfigBuilderBase, ConfigBuilder):
def build_config(self):
"""Create vulnerability config for flood based on the EU JRC global depth-damage functions.
This replaces the notebook implementation and sets the pattern for code-based generation of the config.
"""

# raw data is actually a reformatting of the original Excel spreadsheet to facilitate loading.
raw_data = self.source_dir / "inundation_jrc" / "raw.csv"
df = pd.read_csv(raw_data)

# consistent with physrisk continent definition
location_mappings = {
"Europe": "Europe",
"North America": "North America",
"Central & South America": "South America",
"Asia": "Asia",
"Africa": "Africa",
"Oceania": "Oceania",
"Global": "Global",
}
type_mappings = {
"Residential buildings": "Buildings/Residential",
"Commercial buildings": "Buildings/Commercial",
"Industrial buildings": "Buildings/Industrial",
}

config_items: List[VulnerabilityConfigItem] = []
for mapping in type_mappings:
type_df = df[df["Type"] == mapping]
flood_depth = type_df["Flood depth [m]"].to_numpy()
for location in location_mappings:
# whether zero depth is considered really zero or a flood event with smallest depth
zero_as_minimum = True if location == "North America" else False
# for North America, the 0 depth damage is for flooding of any depth. We consider that a 1 cm inundation.
depth = (
np.concatenate([[0, 0.01], flood_depth[1:]])
if zero_as_minimum
else flood_depth
)

mean = type_df[location + "_Mean"].to_numpy()
std = type_df[location + "_Std"].to_numpy()
mean = np.concatenate([[0], mean]) if zero_as_minimum else mean
std = np.concatenate([[0], std]) if zero_as_minimum else std
if np.any(np.isnan(mean)):
mean = []
if np.any(np.isnan(std)):
std = []
if not any(mean):
continue
config_items.append(
VulnerabilityConfigItem(
hazard_class=",".join(
[
CoastalInundation.__name__,
PluvialInundation.__name__,
RiverineInundation.__name__,
]
),
asset_class=RealEstateAsset.__name__,
asset_identifier=f"type={type_mappings[mapping]},location={location_mappings[location]}",
indicator_id="flood_depth",
indicator_units="metres",
impact_id=str(ImpactType.damage.name),
curve_type="indicator/piecewise_linear",
points_x=list(depth),
points_y=list(mean),
# points_z=std, # omit for now: until implemented
)
)
return config_items
41 changes: 0 additions & 41 deletions tests/conftest.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -100,47 +100,6 @@ def hazard_dir():
return hazard.absolute()


@pytest.fixture(scope="session")
def vulnerability_onboarding_dir():
"""Location of the vulnerability on-boarding directory..
Returns:
str: Directory path.
"""
path = (
Path(__file__).parents[1]
/ "docs"
/ "user_guide"
/ "vulnerability"
/ "vulnerability_functions"
)
return path


@pytest.fixture(scope="session")
def vulnerability_config_dir():
"""Location of the vulnerability on-boarding directory..
Returns:
str: Directory path.
"""
path = (
Path(__file__).parents[1]
/ "src"
/ "physrisk"
/ "data"
/ "static"
/ "vulnerability"
)
return path


# def assert_expected(result: str, func_name: str, update_expected: bool):
# result_dict, expected_dict = get_result_expected(result, func_name, update_expected)
# diffs = DeepDiff(expected_dict, result_dict)
# assert len(diffs.affected_paths) == 0


def get_result_expected(result: str, func_name: str, update_expected: bool):
path = (
pathlib.Path(__file__).parent
Expand Down
121 changes: 14 additions & 107 deletions tests/vulnerability_models/config_generate_test.py
View file
Original file line number Diff line number Diff line change
@@ -1,109 +1,16 @@
from pathlib import Path
import numpy as np
import pandas as pd

from physrisk.kernel.assets import RealEstateAsset
from physrisk.kernel.hazards import (
CoastalInundation,
PluvialInundation,
RiverineInundation,
)
from physrisk.kernel.impact_distrib import ImpactType
from physrisk.vulnerability_models.config_based_impact_curves import (
VulnerabilityConfigItem,
config_items_to_csv,
)


def test_generate_all_config(vulnerability_onboarding_dir, vulnerability_config_dir):
"""Generate all vulnerability config. This is done via test because:
- this ensures that the config remains traceable to source and
- in general the process is not compute-intensive.
Args:
vulnerability_onboarding_dir (Path): Path to vulnerability on-boarding data directory.
"""
config_items = generate_all_config(vulnerability_onboarding_dir)
config_items_to_csv(
config_items, vulnerability_config_dir / "candidate_vulnerability_config.csv"
from physrisk.vulnerability_models.configuration.config_builders import build_all_config


def test_generate_all_config():
# if this is run against the library, e.g. via tox, docs data will not be available
source_dir = (
Path(__file__).parents[2]
/ "docs"
/ "user_guide"
/ "vulnerability"
/ "vulnerability_functions"
/ "vulnerability_functions"
)


def generate_all_config(vulnerability_onboarding_dir: Path):
config_items: list[VulnerabilityConfigItem] = []
config_items += flood_config_jrc(vulnerability_onboarding_dir)
return config_items


def flood_config_jrc(vulnerability_onboarding_dir: Path):
"""Create vulnerability config for flood based on the EU JRC global depth-damage functions.
This replaces the notebook implementation and sets the pattern for code-based generation of the config.
Args:
vulnerability_onboarding_dir (_type_): Path to vulnerability on-boarding data directory.
"""
# raw data is actually a reformatting of the original Excel spreadsheet to facilitate loading.
raw_data = vulnerability_onboarding_dir / "inundation_jrc" / "raw.csv"
df = pd.read_csv(raw_data)

# consistent with physrisk continent definition
location_mappings = {
"Europe": "Europe",
"North America": "North America",
"Central & South America": "South America",
"Asia": "Asia",
"Africa": "Africa",
"Oceania": "Oceania",
"Global": "Global",
}
type_mappings = {
"Residential buildings": "Buildings/Residential",
"Commercial buildings": "Buildings/Commercial",
"Industrial buildings": "Buildings/Industrial",
}

config_items: list[VulnerabilityConfigItem] = []
for mapping in type_mappings:
type_df = df[df["Type"] == mapping]
flood_depth = type_df["Flood depth [m]"].to_numpy()
for location in location_mappings:
# whether zero depth is considered really zero or a flood event with smallest depth
zero_as_minimum = True if location == "North America" else False
# for North America, the 0 depth damage is for flooding of any depth. We consider that a 1 cm inundation.
depth = (
np.concatenate([[0, 0.01], flood_depth[1:]])
if zero_as_minimum
else flood_depth
)

mean = type_df[location + "_Mean"].to_numpy()
std = type_df[location + "_Std"].to_numpy()
mean = np.concatenate([[0], mean]) if zero_as_minimum else mean
std = np.concatenate([[0], std]) if zero_as_minimum else std
if np.any(np.isnan(mean)):
mean = []
if np.any(np.isnan(std)):
std = []
if not any(mean):
continue
config_items.append(
VulnerabilityConfigItem(
hazard_class=",".join(
[
CoastalInundation.__name__,
PluvialInundation.__name__,
RiverineInundation.__name__,
]
),
asset_class=RealEstateAsset.__name__,
asset_identifier=f"type={type_mappings[mapping]},location={location_mappings[location]}",
indicator_id="flood_depth",
indicator_units="metres",
impact_id=str(ImpactType.damage.name),
curve_type="indicator/piecewise_linear",
points_x=list(depth),
points_y=list(mean),
# points_z=std, # omit for now: until implemented
)
)
return config_items
if source_dir.exists():
build_all_config(source_dir)

0 comments on commit 7be265b

Please sign in to comment.