physrisk.kernel

Submodules

physrisk.kernel.assets

class physrisk.kernel.assets.FuelKind(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Biomass = 1

Coal = 2

Cogeneration = 3

Gas = 4

Geothermal = 5

Hydro = 6

Nuclear = 7

Oil = 8

Other = 9

Petcoke = 10

Solar = 11

Storage = 12

Waste = 13

WaveAndTidal = 14

Wind = 15

class physrisk.kernel.assets.CoolingKind(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Dry = 1

OnceThrough = 2

Recirculating = 3

class physrisk.kernel.assets.TurbineKind(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Gas = 1

Steam = 2

class physrisk.kernel.assets.Asset(latitude: float | None = None, longitude: float | None = None, wkt_geometry: str | None = None, buffer: float = 0.0, id: str | None = None, **kwargs)

Bases: object

Class holding information about an asset that is used by vulnerability models.

Generic vulnerability models

Generic vulnerability models cover a significant number of use-cases. For example, inundation vulnerability models for a wide variety of assets may apply vulnerability curves which are looked up from the different attributes of the asset (e.g. occupancy identifier and properties such as number of storeys).

Asset class sub-types

There may be cases however where vulnerability models have specificities associated with the class of the asset. For example, thermal power generating assets might require a model for estimating efficiency loss as air temperature rises. Such a model and its inputs are quite specific to the asset class and it can be convenient to define a vulnerability model to encapsulate that logic. To this end, a number of asset classes, sub-types of Asset as defined to facilitate plugging in of different vulnerability models.

Adherence to conventions

In general asset attributes should follow the Open Exposure Data (OED) standard. OED and NACE codes contribute to naming of asset sub-types although as noted above, sub-types are defined at the level of granularity to facilitate the plugging in of different models defined in code.

Asset ‘archetypes’ for sectorial models

Calculations might have detailed asset information from which precise vulnerability functions can be obtained, but sometimes - for example in the case of sectorial calculations - information might be limited to the broad category of the asset. This can be specified via type (aka asset type as distinct from asset class) and location attributes of assets.

Current Asset sub-type classes are:

AgricultureAsset ConstructionAsset ManufacturingAsset OilGasAsset PowerGeneratingAsset RealEstateAsset ThermalPowerGeneratingAsset TransportationAsset UtilityAsset

It is emphasized that the sub-types exist to facilitate development of plug-in models which can each deal with a large number of types of asset.

static buffered_geometry(geometry: BaseGeometry, buffer: float) → BaseGeometry

Relatively simple buffer, suitable for small buffers in metres around lat/lon points.

Parameters:

geometry (BaseGeometry) – Geometry in EPSG:4326.
buffer (float) – Buffer in metres.

Returns:

Buffered geometry in EPSG:4326.

Return type:

BaseGeometry

class physrisk.kernel.assets.OEDAsset(latitude: float | None = None, longitude: float | None = None, occupancy_code: int = 1000, wkt_geometry: str | None = None, buffer: float = 0.0, number_of_storeys: int = -1, basement: int = 0, construction_code: int = 5000, first_floor_height: float = 0.305, **kwargs): Bases: Asset

class physrisk.kernel.assets.SimpleTypeLocationAsset(*, location: str | None = None, type: str | None = None, **kwargs): Bases: Asset

class physrisk.kernel.assets.AgricultureAsset(**kwargs): Bases: OEDAsset, SimpleTypeLocationAsset

class physrisk.kernel.assets.ConstructionAsset(**kwargs): Bases: OEDAsset, SimpleTypeLocationAsset

class physrisk.kernel.assets.IndustrialActivity(**kwargs)

Bases: OEDAsset, SimpleTypeLocationAsset

To be deprecated. Preferred model is that loss for Assets is calculated both on the asset value (i.e. loss from damage) and the revenue-generation associated with the Asset. That is, revenue-generation is not separated out as it is here.

class physrisk.kernel.assets.ManufacturingAsset(**kwargs): Bases: OEDAsset, SimpleTypeLocationAsset

class physrisk.kernel.assets.OilGasAsset(**kwargs): Bases: OEDAsset, SimpleTypeLocationAsset

class physrisk.kernel.assets.PowerGeneratingAsset(capacity: float | None = None, **kwargs)

Bases: OEDAsset, SimpleTypeLocationAsset

get_protection_return_period(hazard_type: type) → float | None

class physrisk.kernel.assets.RealEstateAsset(**kwargs): Bases: OEDAsset, SimpleTypeLocationAsset

class physrisk.kernel.assets.HasStandardOfProtection(*args, **kwargs)

Bases: Protocol

get_protection_return_period(hazard_type: type) → float | None

class physrisk.kernel.assets.ThermalPowerGeneratingAsset(*, type: str | None = None, location: str | None = None, capacity: float | None = None, **kwargs)

Bases: PowerGeneratingAsset

get_inundation_protection_return_period() → float

get_protection_return_period(hazard_type: type) → float | None

class physrisk.kernel.assets.TestAsset(latitude: float | None = None, longitude: float | None = None, wkt_geometry: str | None = None, buffer: float = 0.0, id: str | None = None, **kwargs): Bases: Asset

class physrisk.kernel.assets.TransportationAsset(**kwargs): Bases: OEDAsset, SimpleTypeLocationAsset

class physrisk.kernel.assets.UtilityAsset(**kwargs): Bases: OEDAsset, SimpleTypeLocationAsset

Bases: Asset

capacity: float | None = None

hub_height: float | None = None

cut_in_speed: float | None = None

cut_out_speed: float | None = None

fixed_base: bool | None = True

rotor_diameter: float | None = None

physrisk.kernel.assets.all_asset_types()

physrisk.kernel.calculation

physrisk.kernel.calculation.get_default_hazard_model() → HazardModel

physrisk.kernel.calculation.placeholder_models() → Sequence[VulnerabilityModelBase]

physrisk.kernel.calculation.default_vulnerability_models() → Dict[type, Sequence[VulnerabilityModelBase]]: Base set of programmatic models; other models are added on top of these There is a specific treatment for power generating assets and real estate assets.

physrisk.kernel.calculation.alternate_default_vulnerability_models_scores() → Dict[type, Sequence[VulnerabilityModelBase]]: A vulnerability models set that combines loss-based and exposure-based scores.

physrisk.kernel.calculation.get_default_risk_measure_calculators() → Dict[Type[Asset], RiskMeasureCalculator]: For asset-level risk measure, define the measure calculators to use.

class physrisk.kernel.calculation.DefaultMeasuresFactory(*args, **kwargs)

Bases: RiskMeasuresFactory

Factory class for selecting appropriate risk measure calculators based on the use case.

asset_calculators(use_case_id: str) → Dict[Type[Asset], RiskMeasureCalculator]: Get the appropriate risk measure calculators based on the use case identifier.

portfolio_calculator(use_case_id: str) → PortfolioRiskMeasureCalculator

physrisk.kernel.curve

physrisk.kernel.curve.add_x_value_to_curve(x, curve_x, curve_y): Add an x value to a curve, interpolated from the existing curve. curve_x and curve_y are the curve x and y values. curve_x is sorted non-decreasing. This function may be used to align curves and bins.

physrisk.kernel.curve.to_exceedance_curve(bin_edges, probs): An exceedance curve gives the probability that the random variable is greater than the value, a type of cumulative probability.

physrisk.kernel.curve.process_bin_edges_and_probs(bin_edges, probs, range_fraction=0.05)

physrisk.kernel.curve.process_bin_edges_for_graph(bin_edges, range_fraction=0.05): Process infinitessimal (zero width) bins for graph display. We make width 5% of range or 1/4 the width to the next bin edge, whichever is smaller

class physrisk.kernel.curve.ExceedanceCurve(probs: List[float] | ndarray, values: List[float] | ndarray)

Bases: object

A point on an exceedance curve comprises an value, v, and a probability, p. p is the probability that the random variable >= v, e.g. an event occurs with event value (e.g. intensity) >= v.

probs

values

add_value_point(value): Add a point to the curve with specified value and exceedance probability determined from existing curve by linear interpolation.

get_value(prob)

get_probability_bins(include_last: bool = False)

Convert from exceedance (cumulative) probability to bins of constant probability density. This is equivalent to the assumption of linear interpolation of exceedance points.

\[p^\text{b}_i = p^\text{e}_{i + 1} - p^\text{e}_i\]

Returns:: The contiguous bin lower and upper values, probabilities of each bin. If value_bins is of length n then there are n-1 bins and n-1 probabilities
Return type:: value_bins (ndarray), probs

get_samples(uniforms): Return value, v, for each probability p in uniforms such that p is the probability that the random variable < v.

physrisk.kernel.events

physrisk.kernel.events.calculate_cumulative_probs(bins_lower: ndarray, bins_upper: ndarray, probs: ndarray)

physrisk.kernel.events.sample_from_cumulative_probs(values: ndarray, cum_probs: ndarray, uniforms: ndarray)

class physrisk.kernel.events.MultivariateDistribution(*args, **kwargs)

Bases: Protocol

inv_cumulative_marginal_probs(cum_probs: ndarray)

class physrisk.kernel.events.EmpiricalMultivariateDistribution(bins_lower: ndarray, bins_upper: ndarray, probs: ndarray)

Bases: MultivariateDistribution

Stores an N dimensional empirical probability density function.

inv_cumulative_marginal_probs(cum_probs: ndarray)

Calculate inverse cumulative probabilities for each of the N marginal probability distributions. By definition, this is the vectorized form of get_inv_cumulative_marginal_prob, vectorized as a performance optimization.

Parameters:

cum_probs (np.ndarray) – Cumulative probabilities (N, P), P being number of samples.
axis (int) – Specifies the axis of the N events.

physrisk.kernel.events.event_samples(impacts_bins: ndarray, probs: List[ndarray], nb_events: int, nb_samples: int)

physrisk.kernel.events.find(elements: ndarray, value): In case we need a specific formulation…

physrisk.kernel.events.event_samples_numba(impacts_bins: ndarray, probs: List[ndarray], nb_events: int, nb_samples: int)

class physrisk.kernel.events.CumulativeProb(*args, **kwargs)

Bases: CumulativeProb

class_type = jitclass.CumulativeProb#75ac1982fd10<values:array(float64, 1d, A),cum_probs:array(float64, 1d, A)>

physrisk.kernel.exposure

class physrisk.kernel.exposure.Category(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

LOWEST = 1

LOW = 2

MEDIUM = 3

HIGH = 4

HIGHEST = 5

NODATA = 6

class physrisk.kernel.exposure.Bounds(category: str, lower: float, upper: float)

Bases: object

Category applies if lower <= value < upper

category: str

lower: float

upper: float

class physrisk.kernel.exposure.AssetExposureResult(hazard_categories: Dict[type, Tuple[physrisk.kernel.exposure.Category, float, str]])

Bases: object

hazard_categories: Dict[type, Tuple[Category, float, str]]

class physrisk.kernel.exposure.ExposureMeasure(*args, **kwargs)

Bases: DataRequester

abstract get_exposures(asset: Asset, data_responses: Sequence[HazardDataResponse]) → Dict[type, Tuple[Category, float, str]]

class physrisk.kernel.exposure.JupterExposureMeasure

Bases: ExposureMeasure

get_data_requests(asset: Asset, *, scenario: str, year: int) → Sequence[HazardDataRequest]

get_exposures(asset: Asset, data_responses: Sequence[HazardDataResponse])

get_exposure_bins()

bounds_to_lookup(bounds: Sequence[Bounds])

physrisk.kernel.exposure.calculate_exposures(assets: List[Asset], hazard_model: HazardModel, exposure_measure: ExposureMeasure, scenario: str, year: int) → Dict[Asset, AssetExposureResult]

physrisk.kernel.financial_model

class physrisk.kernel.financial_model.FinancialDataProvider(*args, **kwargs)

Bases: Protocol

Financial data sufficient to convert impacts into financial quantities that can be aggregated.

Damage, specified as a fraction of total insurable value (TIV), is multiplied by TIV in order to compute the cost of asset restoration in the currency specified – which can then be aggregated across assets.

Disruption, specified as a fraction of annual revenue, is multiplied by the portion of the annual revenue that is attributable to that asset: that which would be lost if the asset were completely non-productive.

Dealing with productivity loss is more complex, as a decrease in worker productivity does not necessarily result in a proportionate loss in revenue. It is assumed, however, that productivity loss models (e.g. related to heat), provide figures that are already adjusted for this effect, and can be multiplied by annual revenue in order to aggregate.

What might this adjustment look like? Possibly the simplest approach is to account for the fact that labour costs are only a fraction of value-added, which is itself a fraction of revenue, i.e.:

Revenue = Intermediate inputs + Value added Value added = Labour costs + EBITDA

revenue_attributable_to_asset(asset: Asset, currency: str) → float

Annual revenue that would be lost if the asset were completely non-productive.

Parameters:

asset (Asset) – Asset.
currency (str) – Currency (3-letter code).

Returns:

Revenue attributable to the asset in specified currency.

Return type:

float

total_insurable_value(asset: Asset, currency: str) → float

Total insurable value of the asset.

Parameters:

asset (Asset) – Asset.
currency (str) – Currency (3-letter code).

Returns:

Total insurable value of the asset in specified currency.

Return type:

float

class physrisk.kernel.financial_model.FinancialModel(*args, **kwargs)

Bases: Protocol

“Financial Model using a FinancialDataProvider as source of information.

property financial_data_provider: FinancialDataProvider

Get the financial data provider.

Returns:: Financial data provider.
Return type:: FinancialDataProvider

frac_damage_to_restoration_cost_and_revenue_loss(asset: Asset, impact: ndarray, currency: str) → tuple[ndarray, ndarray]

Convert damage, specified as a fraction of total insurable value, to cost of asset restoration and loss of revenue during downtime.

Parameters:

asset (Asset) – Asset.
impact (np.ndarray) – Damage as a fraction of total insurable value.
currency (str) – Currency (3-letter code).

Returns:

Tuple containing:

Cost of asset restoration in specified currency.
Annual loss of revenue from downtime in specified currency.

Return type:

tuple[np.ndarray, np.ndarray]

frac_disruption_to_revenue_loss(asset: Asset, impact: ndarray, year: int, currency: str)

Convert disruption, specified as a fraction of annual revenue, to loss of revenue.

Parameters:

asset (Asset) – Asset.
impact (np.ndarray) – Disruption as a fraction of annual revenue.
year (int) – Year of disruption.
currency (str) – Currency (3-letter code).

class physrisk.kernel.financial_model.DefaultFinancialModel(data_provider: FinancialDataProvider, downtime_config: Sequence[DowntimeConfigItem])

Bases: FinancialModel

“Financial Model using a FinancialDataProvider as source of information.

property financial_data_provider: FinancialDataProvider

Get the financial data provider.

Returns:: Financial data provider.
Return type:: FinancialDataProvider

frac_damage_to_restoration_cost_and_revenue_loss(asset: Asset, impact: ndarray, currency: str)

Convert damage, specified as a fraction of total insurable value, to cost of asset restoration and loss of revenue during downtime.

Parameters:

asset (Asset) – Asset.
impact (np.ndarray) – Damage as a fraction of total insurable value.
currency (str) – Currency (3-letter code).

Returns:

Tuple containing:

Cost of asset restoration in specified currency.
Annual loss of revenue from downtime in specified currency.

Return type:

tuple[np.ndarray, np.ndarray]

frac_disruption_to_revenue_loss(asset: Asset, impact: ndarray, year: int, currency: str)

Convert disruption, specified as a fraction of annual revenue, to loss of revenue.

Parameters:

asset (Asset) – Asset.
impact (np.ndarray) – Disruption as a fraction of annual revenue.
year (int) – Year of disruption.
currency (str) – Currency (3-letter code).

physrisk.kernel.hazard_event_distrib

class physrisk.kernel.hazard_event_distrib.HazardEventDistrib(event_type: type[Hazard], intensity_bins: List[float] | ndarray, prob: List[float] | ndarray, path: List[str], units: str)

Bases: object

Intensity distribution of a hazard event (e.g. inundation depth, wind speed etc), specific to an asset – that is, at the location of the asset.

intensity_bins()

to_exceedance_curve()

property intensity_bin_edges: ndarray

property prob: ndarray

property path: List[str]

property units: str

class physrisk.kernel.hazard_event_distrib.EmptyHazardEventDistrib: Bases: HazardEventDistrib

physrisk.kernel.hazard_model

class physrisk.kernel.hazard_model.HazardDataRequest(hazard_type: Type[Hazard], longitude: float, latitude: float, *, indicator_id: str, scenario: str, year: int, hint: HazardDataHint | None = None, geometry: BaseGeometry | None = None, buffer: int | None = None)

Bases: object

Request for hazard data. The event_type determines whether the hazard is acute or chronic. An acute hazard is an event and the response will therefore comprise hazard intensities for the different event return periods. A chronic hazard on the other hand is a shift in a climate parameter and the parameter value is returned.

hazard_type

longitude

latitude

indicator_id

scenario

year

hint

geometry

buffer

group_key(): Key used to group EventDataRequests into batches.

class physrisk.kernel.hazard_model.HazardDataResponse(*args, **kwargs)

Bases: Protocol

path: str

units: str

class physrisk.kernel.hazard_model.HazardDataFailedResponse(err: Exception | None = None, reason: str | None = None): Bases: HazardDataResponse

class physrisk.kernel.hazard_model.HazardEventDataResponse(return_periods: ndarray, intensities: ndarray, units: str = 'default', path: str = 'unknown')

Bases: HazardDataResponse

Response to HazardDataRequest for acute hazards.

class physrisk.kernel.hazard_model.HazardParameterDataResponse(parameters: ndarray, param_defns: ndarray = array(0.0078125), units: str = 'default', path: str = 'unknown')

Bases: HazardDataResponse

Response to HazardDataRequest.

property parameter: float

Convenience function to return single parameter.

Returns:: Single parameter.
Return type:: float

class physrisk.kernel.hazard_model.HazardModel

Bases: ABC

Hazard model. The model accepts a set of HazardDataRequests and returns the corresponding HazardDataResponses.

abstract get_hazard_data(requests: Sequence[HazardDataRequest]) → Mapping[HazardDataRequest, HazardDataResponse]

Process the hazard indicator data requests and return responses.

Parameters:: requests (Sequence[HazardDataRequest]) – Hazard indicator data requests.
Returns:: Responses for all hazard indicator data requests.
Return type:: Mapping[HazardDataRequest, HazardDataResponse]

get_hazard_events(requests: Sequence[HazardDataRequest]) → Mapping[HazardDataRequest, HazardDataResponse]: Deprecated: this has been renamed to get_hazard_data.

class physrisk.kernel.hazard_model.Tile(x, y, z)

Bases: NamedTuple

x: int: Alias for field number 0

y: int: Alias for field number 1

z: int: Alias for field number 2

class physrisk.kernel.hazard_model.HazardImageCreator(*args, **kwargs)

Bases: Protocol

Alongside a HazardModel, it is possible to define a class that can generate image tiles.

Creates an image Tile for display on maps.

Parameters:

resource_id (str) – Unique identifier of the resource.
scenario (str) – Scenario ID.
year (int) – Year for future scenarios.
format (str, optional) – Output format. Defaults to “PNG”.
colormap (str, optional) – Colormap ID. Defaults to “heating”.
tile (Optional[Tile], optional) – Tile for which image is requested. Defaults to None.
min_value (Optional[float], optional) – Value of colormap minimum. Defaults to None.
max_value (Optional[float], optional) – Value of colormap maximum. Defaults to None.
index_value (Optional[str | float], optional) – Value of the non-spatial ‘index’ dimension. Defaults to None.

get_info(resource_id: str, scenario: str, year: int) → Tuple[Sequence[Any], Sequence[Any], str, str, int | None]

Provides additional image information required to create an image.

Parameters:

resource_id (str) – Unique identifier of the resource.
scenario (str) – Scenario ID.
year (int) – Year for future scenarios.

Returns:

All non-spatial (‘index’) coordinate values, non-spatial (‘index’) coordinate values for which map data exists, display name, units, maximum zoom level.

Return type:

Tuple[Sequence[Any], Sequence[Any], str, str, Optional[int]]

class physrisk.kernel.hazard_model.HazardModelFactory(*args, **kwargs)

Bases: Protocol

hazard_model(interpolation: str | None = None, provider_max_requests: Dict[str, int] = {}, interpolate_years: bool = False) → HazardModel

Create a HazardModel instance based on a number of options.

Parameters:

interpolation (str) – Interpolation type to use for sub-pixel raster interpolation (where
models). (this is supported by hazard)
provider_max_requests (Dict[str, int]) – The maximum permitted number of permitted
providers. (requests to external)
interpolate_years (bool) – If True, apply linear interpolation for hazard requests where exact match is not present.

image_creator() → HazardImageCreator

class physrisk.kernel.hazard_model.DataSource(*args, **kwargs): Bases: Protocol

class physrisk.kernel.hazard_model.CompositeHazardModel(hazard_models: Dict[type, HazardModel])

Bases: HazardModel

Hazard Model that uses other models to process EventDataRequests.

get_hazard_data(requests: Sequence[HazardDataRequest]) → Mapping[HazardDataRequest, HazardDataResponse]

Process the hazard indicator data requests and return responses.

Parameters:: requests (Sequence[HazardDataRequest]) – Hazard indicator data requests.
Returns:: Responses for all hazard indicator data requests.
Return type:: Mapping[HazardDataRequest, HazardDataResponse]

physrisk.kernel.hazards

class physrisk.kernel.hazards.IndicatorData(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

EVENT = 1

PARAMETERS = 2

class physrisk.kernel.hazards.HazardKind(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

ACUTE = 1

CHRONIC = 2

UNKNOWN = 3

class physrisk.kernel.hazards.Hazard

Bases: object

kind = 3

indicator_data: Dict[str, IndicatorData] = {}

physrisk.kernel.hazards.hazard_kind(hazard_type: Type[Hazard])

physrisk.kernel.hazards.indicator_data(hazard_type: Type[Hazard], indicator_id: str)

class physrisk.kernel.hazards.ChronicHeat

Bases: Hazard

kind = 2

class physrisk.kernel.hazards.Inundation

Bases: Hazard

kind = 1

indicator_data: Dict[str, IndicatorData] = {'flood_depth': IndicatorData.EVENT, 'flood_sop': IndicatorData.PARAMETERS}

class physrisk.kernel.hazards.AirTemperature: Bases: ChronicHeat

class physrisk.kernel.hazards.CoastalInundation: Bases: Inundation

class physrisk.kernel.hazards.ChronicWind

Bases: Hazard

kind = 2

class physrisk.kernel.hazards.CombinedInundation

Bases: Hazard

kind = 2

class physrisk.kernel.hazards.Drought

Bases: Hazard

kind = 2

class physrisk.kernel.hazards.Fire

Bases: Hazard

kind = 1

indicator_data: Dict[str, IndicatorData] = {'daily_probability_fwi20': IndicatorData.PARAMETERS, 'fire_probability': IndicatorData.PARAMETERS}

class physrisk.kernel.hazards.Hail

Bases: Hazard

kind = 1

indicator_data: Dict[str, IndicatorData] = {'days/above/5cm': IndicatorData.PARAMETERS}

class physrisk.kernel.hazards.PluvialInundation: Bases: Inundation

class physrisk.kernel.hazards.Precipitation

Bases: Hazard

kind = 2

class physrisk.kernel.hazards.RiverineInundation: Bases: Inundation

class physrisk.kernel.hazards.WaterRisk

Bases: Hazard

kind = 2

class physrisk.kernel.hazards.WaterTemperature: Bases: ChronicHeat

class physrisk.kernel.hazards.Wind

Bases: Hazard

kind = 1

class physrisk.kernel.hazards.Subsidence

Bases: Hazard

kind = 2

physrisk.kernel.hazards.all_hazards()

physrisk.kernel.hazards.hazard_class(name: str) → Type[Hazard]

class physrisk.kernel.hazards.Landslide

Bases: Hazard

kind = 1

class physrisk.kernel.hazards.Earthquake

Bases: Hazard

kind = 1

physrisk.kernel.impact

class physrisk.kernel.impact.ImpactKey(asset, hazard_type, scenario, key_year)

Bases: NamedTuple

asset: Asset: Alias for field number 0

hazard_type: type[Hazard]: Alias for field number 1

scenario: str: Alias for field number 2

key_year: int | None: Alias for field number 3

class physrisk.kernel.impact.AssetImpactResult(impact: physrisk.kernel.impact_distrib.ImpactDistrib, vulnerability: physrisk.kernel.vulnerability_distrib.VulnerabilityDistrib | None = None, event: physrisk.kernel.hazard_event_distrib.HazardEventDistrib | None = None, hazard_data: Sequence[physrisk.kernel.hazard_model.HazardDataResponse] | None = None)

Bases: object

impact: ImpactDistrib

vulnerability: VulnerabilityDistrib | None = None

event: HazardEventDistrib | None = None

hazard_data: Sequence[HazardDataResponse] | None = None

physrisk.kernel.impact.calculate_impacts(assets: Iterable[Asset], hazard_model: HazardModel, vulnerability_models: VulnerabilityModels, *, scenarios: Sequence[str], years: Sequence[int]) → Dict[ImpactKey, List[AssetImpactResult]]: Calculate asset level impacts.

class physrisk.kernel.impact.ScenarioYear(scenario, key_year)

Bases: NamedTuple

scenario: str: Alias for field number 0

key_year: int | None: Alias for field number 1

physrisk.kernel.impact_distrib

class physrisk.kernel.impact_distrib.ImpactType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

damage = 1

disruption = 2

class physrisk.kernel.impact_distrib.ImpactDistrib(hazard_type: Type[Hazard], impact_bin_edges: Sequence[float] | ndarray, probabilities: Sequence[float] | ndarray, hazard_indicator_id: str, impact_type: ImpactType = ImpactType.damage, path: Sequence[str] = [])

Bases: object

Impact distributions specific to an asset.

impact_bins_explicit()

mean_impact()

standard_deviation()

semi_standard_deviation()

to_exceedance_curve()

property hazard_type: type

property hazard_indicator_id: str

property impact_bin_edges: ndarray

property impact_type

property probabilities: ndarray

property path: Sequence[str]

class physrisk.kernel.impact_distrib.EmptyReason(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: int, Enum

NO_VULNERABILITY = 2

NO_DATA = 1

EXCEPTION = 3

class physrisk.kernel.impact_distrib.EmptyImpactDistrib(empty_reason: EmptyReason = EmptyReason.NO_VULNERABILITY): Bases: ImpactDistrib

class physrisk.kernel.impact_distrib.PlaceholderImpactDistrib: Bases: ImpactDistrib

physrisk.kernel.risk

class physrisk.kernel.risk.BatchId(scenario, key_year)

Bases: NamedTuple

scenario: str: Alias for field number 0

key_year: int | None: Alias for field number 1

class physrisk.kernel.risk.QuantityType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

DAMAGE = 'damage'

REVENUE_LOSS = 'revenue_loss'

COSTS = 'costs'

TIV = 'tiv'

REVENUE = 'revenue'

class physrisk.kernel.risk.RiskQuantityKey(quantity, agg_id, hazard_type)

Bases: NamedTuple

quantity: QuantityType | None: Alias for field number 0

agg_id: str | None: Alias for field number 1

hazard_type: type[Hazard] | None: Alias for field number 2

class physrisk.kernel.risk.Quantity(values: numpy.ndarray[tuple[Any, ...], numpy.dtype[numpy.floating[Any]]], percentiles: numpy.ndarray[tuple[Any, ...], numpy.dtype[numpy.floating[Any]]], percentile_values: numpy.ndarray[tuple[Any, ...], numpy.dtype[numpy.floating[Any]]], mean: float)

Bases: object

values: ndarray[tuple[Any, ...], dtype[floating[Any]]]

percentiles: ndarray[tuple[Any, ...], dtype[floating[Any]]]

percentile_values: ndarray[tuple[Any, ...], dtype[floating[Any]]]

mean: float

class physrisk.kernel.risk.MeasureKey(asset: Asset | None, scenario: str, year: int | None, hazard_type: Type[Hazard] | None, hazard_indicator_id: str | None = None)

Bases: NamedTuple

Key for risk measures: asset can be None for measures aggregated over assets (i.e. portfolio-level); the hazard_type can be None for measures aggregated over hazard types; scenario is required; year should be None for the historical scenario; hazard_indicator_id can be used to distinguish between different hazard indicators for the same hazard type, e.g. for ChronicHeat, multiple hazard indicators may be used to capture different mechanisms of impact.

asset: Asset | None: Alias for field number 0

scenario: str: Alias for field number 1

year: int | None: Alias for field number 2

hazard_type: Type[Hazard] | None: Alias for field number 3

hazard_indicator_id: str | None: Alias for field number 4

class physrisk.kernel.risk.Measure(score: int, measure_0: float, definition: physrisk.api.v1.impact_req_resp.ScoreBasedRiskMeasureDefinition, measure_1: float | None = None)

Bases: object

score: int

measure_0: float

definition: ScoreBasedRiskMeasureDefinition

measure_1: float | None = None

class physrisk.kernel.risk.PortfolioRiskMeasureCalculator(*args, **kwargs)

Bases: Protocol

Class to calculate portfolio-level score-based risk measures, either from a set of asset-level score-based risk measures or from portfolio-level

get_definition(hazard_type: type[Hazard] | None = None) → ScoreBasedRiskMeasureDefinition

calculate_risk_measures(asset_level_measures: dict[MeasureKey, Measure] = {}, portfolio_quantities: dict[RiskQuantityKey, Quantity] = {}) → dict[MeasureKey, Measure]

asset_level_measures_required() → bool

portfolio_quantities_required() → bool

class physrisk.kernel.risk.NullAssetBasedPortfolioRiskMeasureCalculator

Bases: PortfolioRiskMeasureCalculator

Calculates portfolio score-based risk measures from asset-level score-based risk measures only.

get_definition(hazard_type: Type[Hazard] | None = None)

calculate_risk_measures(asset_level_measures: dict[MeasureKey, Measure] = {}, portfolio_quantities: dict[RiskQuantityKey, Quantity] = {}) → dict[MeasureKey, Measure]

asset_level_measures_required() → bool

portfolio_quantities_required() → bool

class physrisk.kernel.risk.RiskModel(hazard_model: HazardModel, vulnerability_models: VulnerabilityModels)

Bases: object

Base class for a risk model.

That is, a calculation of risk that makes use of hazard and vulnerability models).

calculate_risk_measures(assets: Sequence[Asset], prosp_scens: Sequence[str], years: Sequence[int]): Calculate risk measures for a set of assets, scenarios, and years.

class physrisk.kernel.risk.RiskMeasureCalculator(*args, **kwargs)

Bases: Protocol

calc_measure(hazard_type: Type[Hazard], base_impacts: Sequence[AssetImpactResult], impacts: Sequence[AssetImpactResult]) → Measure | dict[str | None, Measure]

Calculate the Measure (score-based risk measure) for the hazard, given the set of base (i.e. historical) and future asset-level impacts. Most often there may be a single impact for a given type of hazard, but in general there can be multiple corresponding to different vulnerability models.

Parameters:

hazard_type (Type[Hazard]) – Hazard type.
base_impacts (Sequence[AssetImpactResult]) – Historical asset-level impacts.
impacts (Sequence[AssetImpactResult]) – Future asset-level impacts. The historical
model (and future impacts are aligned by the generating vulnerability)
required. (allowing these to be zipped and compared as)

Returns:

Single score-based risk measures, or a dictionary of score-based risk measures keyed by a hazard indicator ID, for cases where drill-down by hazard indicator is desired.

Return type:

Measure | dict[str, Measure]

get_definition(hazard_type: Type[Hazard], hazard_indicator_id: str | None = None) → ScoreBasedRiskMeasureDefinition

supported_hazards() → Set[type]

aggregate_risk_measures(measures: Dict[MeasureKey, Measure], assets: Sequence[Asset], prosp_scens: Sequence[str], years: Sequence[int]) → Dict[MeasureKey, Measure]

The RiskMeasureCalculator can aggregate child hazards into parent hazards or proxy one hazard to another. If no aggregation or proxying is needed, the measures input is returned unchanged.

Parameters:

measures (Dict[MeasureKey, Measure]) – Score-based risk measures.
prosp_scens (Sequence[str]) – Requested prospective scenarios.
years (Sequence[int]) – Requested prospective years.

Returns:

Aggregated or proxied score-based risk measures.

Return type:

Dict[MeasureKey, Measure]

class physrisk.kernel.risk.RiskMeasuresFactory(*args, **kwargs)

Bases: Protocol

Protocol for selecting risk measure calculators.

asset_calculators(use_case_id: str) → Dict[Type[Asset], RiskMeasureCalculator]: Get risk measure calculators for asset types.

Args:

use_case_id (Optional[str]): Optional use case ID to filter calculators.

portfolio_calculator(use_case_id: str) → PortfolioRiskMeasureCalculator

class physrisk.kernel.risk.AssetLevelRiskModel(hazard_model: ~physrisk.kernel.hazard_model.HazardModel, vulnerability_models: ~physrisk.kernel.vulnerability_model.VulnerabilityModels, measure_calculators: ~typing.Dict[type[~physrisk.kernel.assets.Asset], ~physrisk.kernel.risk.RiskMeasureCalculator], portfolio_measure_calculator: ~physrisk.kernel.risk.PortfolioRiskMeasureCalculator = <physrisk.kernel.risk.NullAssetBasedPortfolioRiskMeasureCalculator object>)

Bases: RiskModel

Risk model that calculates risk measures at the asset level for various assets.

calculate_impacts(assets: Sequence[Asset], prosp_scens: Sequence[str], years: Sequence[int])

populate_measure_definitions(assets: Sequence[Asset], hazard_type_indicators: dict[type[Hazard], set[str]] = {}) → tuple[dict[type[Hazard], list[str]], dict[ScoreBasedRiskMeasureDefinition, str], dict[tuple[type[Hazard], str], list[str]]]

calculate_risk_measures(assets: Sequence[Asset], scenarios: Sequence[str], years: Sequence[int])

Calculate risk measures for a set of assets, scenarios, and years, according to the selected method calculation.

For the Default Method: The impact of the historical scenario is chosen as the base impact, and risk measures are calculated using the calc_measure function defined in the RealEstateToyRiskMeasures class. This method performs calculations differently depending on whether the hazard is chronic heat or another type. The difference between the two methods is that calc_measure_cooling uses mean impacts for calculations, while calc_measure_acute uses exceedance curves. In both cases, a Measure object is returned, which contains a score (REDFLAG, HIGH, MEDIUM, LOW), measures_0 (future_loss), and a definition.

Args:

assets (Sequence[Asset]): List of assets. scenarios (Sequence[str]): List of prospective scenarios. years (Sequence[int]): List of years for the calculations.

Return:

Tuple[
Dict[ImpactKey, List[AssetImpactResult]], Dict[MeasureKey, Measure]

]: A tuple containing:

A dictionary mapping asset and hazard type tuples to impact results.

A dictionary mapping MeasureKeys to calculated measures.

physrisk.kernel.vulnerability_distrib

class physrisk.kernel.vulnerability_distrib.VulnerabilityDistrib(hazard_type: type[Hazard], intensity_bins: List[float] | ndarray, impact_bins: List[float] | ndarray, prob_matrix: List[List[float]] | ndarray, hazard_indicator_id: str)

Bases: object

Vulnerability distribution as a discrete matrix.

property event_type: type

property hazard_type: type

property hazard_indicator_id: str

intensity_bin_bounds()

impact_bin_bounds()

property impact_bins: ndarray

property intensity_bins: ndarray

property prob_matrix: ndarray

class physrisk.kernel.vulnerability_distrib.EmptyVulnerabilityDistrib: Bases: VulnerabilityDistrib

physrisk.kernel.vulnerability_matrix_provider

class physrisk.kernel.vulnerability_matrix_provider.Distribution(mean, std_dev): Bases: object

class physrisk.kernel.vulnerability_matrix_provider.VulnMatrixProvider(intensity_bin_centres: ndarray, impact_cdfs: Sequence[Callable[[ndarray], ndarray]])

Bases: object

intensity_bin_centres

impact_cdfs

to_prob_matrix(impact_bin_edges: ndarray) → ndarray

Return probability matrix, p with dimension (number intensity bins, number impact bins) where p[i, j] is the conditional probability that given the intensity falls in bin i, the impact is in bin j.

Parameters:: impact_bin_edges (Iterable[float]) – Bin edges of the impact bins.
Returns:: Probability matrix.
Return type:: np.ndarray

physrisk.kernel.vulnerability_model

physrisk.kernel.vulnerability_model.repeat(num_times)

physrisk.kernel.vulnerability_model.applies_to_events(event_types)

physrisk.kernel.vulnerability_model.applies_to_assets(asset_types)

physrisk.kernel.vulnerability_model.get_vulnerability_curves_from_resource(id: str) → VulnerabilityCurves

physrisk.kernel.vulnerability_model.delta_cdf(y)

physrisk.kernel.vulnerability_model.checked_beta_distrib(mean, std, scaling_factor=1.0)

physrisk.kernel.vulnerability_model.beta_distrib(mean, std, scaling_factor)

class physrisk.kernel.vulnerability_model.DataRequester(*args, **kwargs)

Bases: Protocol

get_data_requests(asset: Asset, *, scenario: str, year: int) → HazardDataRequest | Sequence[HazardDataRequest]

class physrisk.kernel.vulnerability_model.EventBased(*args, **kwargs)

Bases: Protocol

impact_samples(asset: Asset, data_responses: Sequence[HazardDataResponse]) → ndarray

class physrisk.kernel.vulnerability_model.VulnerabilityModelBase(indicator_id: str, hazard_type: type[Hazard], impact_type: ImpactType)

Bases: ABC, DataRequester

abstract get_data_requests(asset: Asset, *, scenario: str, year: int) → HazardDataRequest | Sequence[HazardDataRequest]

Provide the hazard event data requests.

It will return one or more HazardDataRequest as required to calculate the VulnerabilityDistrib and HazardEventDistrib for the asset.

abstract get_impact(asset: Asset, hazard_data: Sequence[HazardDataResponse]) → ImpactDistrib

class physrisk.kernel.vulnerability_model.VulnerabilityModels(*args, **kwargs)

Bases: Protocol

vuln_model_for_asset_of_type(type: Type[Asset]) → Sequence[VulnerabilityModelBase]

Return for a given asset type the vulnerability models for each hazard required.

Returns:: Dict[type, Sequence[VulnerabilityModelBase]]
Return type:: Vulnerability models.s

class physrisk.kernel.vulnerability_model.VulnerabilityModelsFactory(*args, **kwargs)

Bases: Protocol

vulnerability_models(hazard_scope: dict[type[Hazard], set[str] | None] | None = None) → VulnerabilityModels

Create a VulnerabilityModels instance, that can based on a number of options.

Although no options used at present, implemented this way in order to add in future (e.g. to allow a request to specify preferred methodology).

Returns:: VulnerabilityModels
Return type:: Instance that provides vulnerability models for asset types.

class physrisk.kernel.vulnerability_model.DictBasedVulnerabilityModels(models: Dict[Type[Asset], Sequence[VulnerabilityModelBase]])

Bases: VulnerabilityModels

vuln_model_for_asset_of_type(type: Type[Asset])

Return for a given asset type the vulnerability models for each hazard required.

Returns:: Dict[type, Sequence[VulnerabilityModelBase]]
Return type:: Vulnerability models.s

class physrisk.kernel.vulnerability_model.VulnerabilityModelAcuteBase(indicator_id: str, hazard_type: type, impact_type: ImpactType)

Bases: VulnerabilityModelBase

Models generate the VulnerabilityDistrib and HazardEventDistrib of an Asset.

abstract get_distributions(asset: Asset, event_data_responses: Sequence[HazardDataResponse]) → Tuple[VulnerabilityDistrib, HazardEventDistrib]

Return distributions for asset: VulnerabilityDistrib and HazardEventDistrib.

The hazard event data is used to do this.

Args:

asset: the asset. event_data_responses: the responses to the requests made by get_data_requests, in the same order.

get_impact(asset: Asset, data_responses: Sequence[HazardDataResponse])

get_impact_details(asset: Asset, data_responses: Sequence[HazardDataResponse]) → Tuple[ImpactDistrib, VulnerabilityDistrib, HazardEventDistrib]: Return impact distribution along with vulnerability and hazard event distributions used to infer this.

Args:

asset: the asset. event_data_responses: the responses to the requests made by get_data_requests, in the same order.

class physrisk.kernel.vulnerability_model.VulnerabilityModel(*, indicator_id: str = '', hazard_type: type, impact_type: ImpactType, impact_bin_edges, buffer: int | None = None)

Bases: VulnerabilityModelAcuteBase

A vulnerability model that requires only specification of distributions of impacts for given intensities, by implementing get_impact_curve.

get_data_requests(asset: Asset, *, scenario: str, year: int) → HazardDataRequest | Sequence[HazardDataRequest]

Provide the hazard event data requests.

It will return one or more HazardDataRequest as required to calculate the VulnerabilityDistrib and HazardEventDistrib for the asset.

get_distributions(asset: Asset, event_data_responses: Sequence[HazardDataResponse]) → Tuple[VulnerabilityDistrib, HazardEventDistrib]

Return distributions for asset: VulnerabilityDistrib and HazardEventDistrib.

The hazard event data is used to do this.

Args:

asset: the asset. event_data_responses: the responses to the requests made by get_data_requests, in the same order.

abstract get_impact_curve(intensity_bin_centres: ndarray, asset: Asset) → VulnMatrixProvider

Define a VulnMatrixProvider.

The VulnMatrixProvider returns probabilities of specified impact bins for the intensity bin centres.

class physrisk.kernel.vulnerability_model.CurveBasedVulnerabilityModel(*, indicator_id: str = '', hazard_type: type, impact_type: ImpactType, impact_bin_edges, buffer: int | None = None)

Bases: VulnerabilityModel

get_impact_curve(intensity_bin_centres: ndarray, asset: Asset) → VulnMatrixProvider

Define a VulnMatrixProvider.

The VulnMatrixProvider returns probabilities of specified impact bins for the intensity bin centres.

abstract get_vulnerability_curve(asset: Asset) → VulnerabilityCurve

class physrisk.kernel.vulnerability_model.DeterministicVulnerabilityModel(*, hazard_type: type, damage_curve_intensities: Sequence[float], damage_curve_impacts: Sequence[float], indicator_id: str, impact_type: ImpactType, buffer: int | None = None)

Bases: VulnerabilityModelAcuteBase

get_data_requests(asset: Asset, *, scenario: str, year: int) → HazardDataRequest | Sequence[HazardDataRequest]

Provide the hazard event data requests.

It will return one or more HazardDataRequest as required to calculate the VulnerabilityDistrib and HazardEventDistrib for the asset.

get_distributions(asset: Asset, event_data_responses: Sequence[HazardDataResponse]) → Tuple[VulnerabilityDistrib, HazardEventDistrib]

Return distributions for asset: VulnerabilityDistrib and HazardEventDistrib.

The hazard event data is used to do this.

Args:

asset: the asset. event_data_responses: the responses to the requests made by get_data_requests, in the same order.

physrisk.kernel

Submodules

physrisk.kernel.assets

Generic vulnerability models

Asset class sub-types

Adherence to conventions

Asset ‘archetypes’ for sectorial models

physrisk.kernel.calculation

physrisk.kernel.curve

physrisk.kernel.events

physrisk.kernel.exposure

physrisk.kernel.financial_model

physrisk.kernel.hazard_event_distrib

physrisk.kernel.hazard_model

physrisk.kernel.hazards

physrisk.kernel.impact

physrisk.kernel.impact_distrib

physrisk.kernel.risk

Args:

Args:

Return:

physrisk.kernel.vulnerability_distrib

physrisk.kernel.vulnerability_matrix_provider

physrisk.kernel.vulnerability_model

Args:

Args:

Args:

Args:

Module contents