Custom vulnerability functions for asset-level impact calculations

In this walk-through, custom vulnerability functions are defined and used to perform the asset-level impact calculations. The portfolio is identical to the one previously calculated; the difference is only in the config-based vulnerability.

There is no option exposed to inject a custom vulnerability config via the API at the moment, hence the library is used.

We start by defining a portfolio and an acute vulnerability function for Wind.

# pip install nbformat pandas plotly requests

from io import StringIO
import json
from typing import NamedTuple
from dependency_injector import providers
from physrisk.container import Container
from physrisk.vulnerability_models.vulnerability import VulnerabilityModelsFactory
from physrisk.vulnerability_models.config_based_impact_curves import (
    VulnerabilityConfigItem,
    config_items_from_csv,
    config_items_to_df,
)
import plotly.io
from plotly.subplots import make_subplots

plotly.io.renderers.default = "notebook"

Define a portfolio. The “type” attribute can be used to define an arbitrary scheme to capture categories of assets, matching the vulnerability configuration.

portfolio = {
    "items": [
        {
            "asset_class": "RealEstateAsset",
            "type": "Buildings/Industrial",
            "location": "Asia",
            "latitude": 24.0426,
            "longitude": 91.0158,
        },
        {
            "asset_class": "RealEstateAsset",
            "type": "Buildings/Industrial",
            "location": "Asia",
            "latitude": 22.6588,
            "longitude": 90.3373,
        },
    ]
}
request = {
    "assets": portfolio,
    "include_asset_level": True,
    "include_calc_details": True,
    "include_measures": True,
    "years": [2050],
    "scenario": "ssp585",
}

Configuration can be defined programmatically, but is also intended to be maintained in CSV format. Here we write the config line to CSV and read back again. In practise, a file of config lines can be maintained.

config_items = [
    VulnerabilityConfigItem(
        hazard_class="Wind",
        asset_class="RealEstateAsset",
        asset_identifier="type=Buildings/Industrial,location=Asia",
        indicator_id="max_speed",
        indicator_units="m/s",
        impact_id="damage",
        impact_units=None,
        curve_type="indicator/piecewise_linear",
        points_x=[20, 25, 30, 35, 40, 45, 50, 55, 60, 65, 70, 75, 80, 85, 90, 110],
        points_y=[
            0,
            0,
            0,
            0.03,
            0.11,
            0.23,
            0.37,
            0.52,
            0.63,
            0.71,
            0.78,
            0.83,
            0.87,
            0.89,
            0.92,
            1.0,
        ],
        activation_of_points_x=None,
        cap_of_points_x=None,
        cap_of_points_y=None,
    )
]

csv_config = config_items_to_df(config_items).to_csv(index=False)
print(csv_config)

config_items = config_items_from_csv(StringIO(csv_config))

hazard_class,asset_class,asset_identifier,indicator_id,indicator_units,impact_id,impact_units,curve_type,points_x,points_y,points_z,cap_of_points_x,cap_of_points_y,activation_of_points_x
Wind,RealEstateAsset,"type=Buildings/Industrial,location=Asia",max_speed,m/s,damage,,indicator/piecewise_linear,"[20.0, 25.0, 30.0, 35.0, 40.0, 45.0, 50.0, 55.0, 60.0, 65.0, 70.0, 75.0, 80.0, 85.0, 90.0, 110.0]","[0.0, 0.0, 0.0, 0.03, 0.11, 0.23, 0.37, 0.52, 0.63, 0.71, 0.78, 0.83, 0.87, 0.89, 0.92, 1.0]",,,,

Having defined the configuration the calculation is run. Dependency injection is used to apply the custom configuration.

# we make use of dependency injection to inject a vulnerability model that accepts the configuration.
container = Container()
container.override_providers(
    vulnerability_models_factory=providers.Factory(
        VulnerabilityModelsFactory, config=config_items
    )
)
requester = container.requester()
response = json.loads(
    requester.get(request_id="get_asset_impact", request_dict=request)
)

asset0_impacts = response["asset_impacts"][1]["impacts"]


class Key(NamedTuple):
    hazard_type: str
    scenario_id: str
    year: str


asset0_impact_dict = {}
for i in asset0_impacts:
    key = i["key"]
    asset0_impact_dict[Key(key["hazard_type"], key["scenario_id"], key["year"])] = i

hazard_types = set(k.hazard_type for k in asset0_impact_dict.keys())
wind_impact_histo = asset0_impact_dict[Key("Wind", "historical", "None")]
wind_impact_ssp585 = asset0_impact_dict[Key("Wind", "ssp585", "2050")]

exceedance_histo = wind_impact_histo["impact_exceedance"]
exceedance_ssp585 = wind_impact_ssp585["impact_exceedance"]

fig1 = make_subplots(rows=1, cols=1)

fig1.add_scatter(
    x=exceedance_histo["exceed_probabilities"],
    y=exceedance_histo["values"],
    name="baseline wind",
    row=1,
    col=1,
)
fig1.add_scatter(
    x=exceedance_ssp585["exceed_probabilities"],
    y=exceedance_ssp585["values"],
    name="wind SSP585",
    row=1,
    col=1,
)
fig1.update_xaxes(
    title="Annual exceedance probability",
    title_font={"size": 14},
    row=1,
    col=1,
    type="log",
    autorange="reversed",
)
fig1.update_yaxes(
    title="Damage as fraction of insurable value", title_font={"size": 14}, row=1, col=1
)