physrisk.data

Subpackages

• physrisk.data.static
  • Subpackages
    • physrisk.data.static.example_portfolios
      • Module contents
    • physrisk.data.static.hazard
      • Module contents
    • physrisk.data.static.vulnerability
      • Module contents
  • Submodules
  • physrisk.data.static.world
  • Module contents

Submodules

physrisk.data.colormap_provider

physrisk.data.colormap_provider.colormap(id: str)

physrisk.data.geotiff_reader

physrisk.data.geotiff_reader.zarr_read(path, longitudes, latitudes)

A version that uses Zarr rather than GDAL. Typically faster than GDAL / rasterio.

physrisk.data.geotiff_reader.dataset_read_bounded(dataset, longitudes, latitudes, window_half_width=0.01)

physrisk.data.geotiff_reader.dataset_read_points(dataset, longitudes, latitudes, window_half_width=0.01)

physrisk.data.geotiff_reader.dataset_read_windows(dataset, longitudes, latitudes, window_half_width=0.01)

physrisk.data.geotiff_reader.file_read_bounded(path, longitudes, latitudes, window_half_width=0.01)

physrisk.data.geotiff_reader.file_read_points(path, longitudes, latitudes)

physrisk.data.hazard_data_provider

class physrisk.data.hazard_data_provider.HazardDataHint(path: str | None = None)

Bases: object

Requestors of hazard data may provide a hint which may be taken into account by the Hazard Model. A hazard resource path can be specified which uniquely defines the hazard resource; otherwise the resource is inferred from the indicator_id.

path: str | None = None

group_key()

class physrisk.data.hazard_data_provider.ScenarioPaths(years: Sequence[int], path: Callable[[int], str])

Bases: object

For a given Hazard Resource and scenario, gives the available years and function to generate the path for each year. If no years are available (because the scenario cannot be matched), this will be an empty list.

years: Sequence[int]

path: Callable[[int], str]

class physrisk.data.hazard_data_provider.ResourcePaths(resource_path: str, scenarios: Dict[str, ScenarioPaths], units: str)

Bases: object

Provides for a single HazardResource the mappings of scenarios to the ScenarioPaths. The ScenarioPaths for each scenario will give the available years and a function to get the array path for a year. Units (inferred from the inventory) are also provided as these are needed in addition to path information to derive HazardDataResponses.

resource_path: str

scenarios: Dict[str, ScenarioPaths]

units: str

class physrisk.data.hazard_data_provider.SourcePaths(*args, **kwargs)

Bases: Protocol

hazard_types() → List[Type[Hazard]]

Lists the available hazard types.

Returns:

Available hazard types.

Return type:

List[Type[Hazard]]

resource_paths(hazard_type: Type[Hazard], indicator_id: str, scenarios: Sequence[str], hint: HazardDataHint | None = None) → List[ResourcePaths]

Provides a list of ResourcePaths. Each item in the list provides the paths for a different HazardResource. The ResourcePaths will be tried in order. If a latitude/longitude is out of bounds of one HazardResource

Parameters:

• hazard_type (Type[Hazard]) – Hazard type.

• indicator_id (str) – Hazard indicator identifier.

• scenarios (Sequence[str]) – Scenario identifiers.

• hint (Optional[HazardDataHint], optional) – Hint to be applied to select path. Generally only

• None. (makes sense if there is just a single scenario. Defaults to)

Returns:

List of ResourcePaths to be tried in order.

Return type:

List[ResourcePaths]

scenario_paths_for_id(resource_id: str, scenarios: Sequence[str], map: bool = False, map_zoom: int | None = None) → Dict[str, ScenarioPaths]

Returns the ScenarioPaths when a unique ID is specified.

exception physrisk.data.hazard_data_provider.DataSourcingError

Bases: Exception

class physrisk.data.hazard_data_provider.ScenarioYear(scenario, year)

Bases: NamedTuple

scenario: str

Alias for field number 0

year: int

Alias for field number 1

class physrisk.data.hazard_data_provider.ScenarioYearRes(scenario, year, resource_index)

Bases: NamedTuple

scenario: str

Alias for field number 0

year: int

Alias for field number 1

resource_index: int | None

Alias for field number 2

class physrisk.data.hazard_data_provider.ScenarioYearResult(values: numpy.ndarray, indices: numpy.ndarray, indices_length: numpy.ndarray, coverage_mask: numpy.ndarray, units: str, paths: numpy.ndarray)

Bases: object

values: ndarray

indices: ndarray

indices_length: ndarray

coverage_mask: ndarray

units: str

paths: ndarray

class physrisk.data.hazard_data_provider.WeightedSum(weights: List[Tuple[physrisk.data.hazard_data_provider.ScenarioYear, float]])

Bases: object

weights: List[Tuple[ScenarioYear, float]]

class physrisk.data.hazard_data_provider.HazardDataProvider(hazard_type: Type[Hazard], source_paths: SourcePaths, *, store: MutableMapping | None = None, zarr_reader: ZarrReader | None = None, interpolation: str | None = 'floor', historical_year: int = 2025)

Bases: ABC

async get_data_cascading(longitudes: ndarray, latitudes: ndarray, *, indicator_id: str, scenarios: Sequence[str], years: Sequence[int], hint: HazardDataHint | None = None, buffer: int | None = None, interpolate_years: bool = False)

Returns data for set of latitude and longitudes.

Parameters:

• longitudes (np.ndarray) – Longitudes.

• latitudes (np.ndarray) – Latitudes.

• indicator_id (str) – Hazard Indicator ID.

• scenarios (Sequence[str]) – Identifier of scenario, e.g. ssp585 (SSP 585), rcp8p5 (RCP 8.5).

• years (Sequence[int]) – Projection years, e.g. [2050, 2080].

• hint (Optional[HazardDataHint], optional) – Hint. Defaults to None.

• buffer (Optional[int], optional) – _description_. Buffer around each point.

• interpolate_years (bool, optional) – If True, interpolate between years. Defaults to False.

Returns:

Results.

Return type:

Dict[ScenarioYear, ScenarioYearResult]

async get_scenarios_and_years(resource_index: int, coverage: ndarray, longitudes: ndarray, latitudes: ndarray, indicator_id: str, resource_paths: ResourcePaths, years: Sequence[int], buffer: int | None, interpolate_years: bool)

Get data for all scenarios and years using just a single HazardResource as the source. The importance of this is that interpolation of years is assumed to be feasible within the same resource as this is a single model (with consistent meaning of the values).

async get_single_item(item: ScenarioYear, latitudes: ndarray, longitudes: ndarray, buffer: int | None, path: str)

physrisk.data.image_creator

class physrisk.data.image_creator.ImageCreator(inventory: Inventory, source_paths: SourcePaths, reader: ZarrReader, historical_year: int = 2025)

Bases: HazardImageCreator

Convert small arrays into images for map display. Intended for arrays <~1500x1500 (otherwise, recommended to use Mapbox tiles - or similar).

create_image(resource_id: str, scenario: str, year: int, format='PNG', colormap: str = 'heating', tile: Tile | None = None, min_value: float | None = None, max_value: float | None = None, index_value: str | float | None = None)

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]]

to_file(filename: str, path: str, format='PNG', colormap: str = 'heating', min_value: float | None = None, max_value: float | None = None)

Create image for path specified and save as file.

Parameters:

• filename (str) – Filename.

• path (str) – Path to array.

• format (str, optional) – Image format. Defaults to “PNG”.

• colormap (str, optional) – Colormap name. Defaults to “heating”.

• min_value (Optional[float], optional) – Min value. Defaults to None.

• max_value (Optional[float], optional) – Max value. Defaults to None.

static to_rgba(data: ndarray, get_colors: Callable[[int], List[int]], min_value: float | None = None, max_value: float | None = None, nodata_lower: float | None = None, nodata_upper: float | None = None, nodata_bin_transparent: bool = False, min_bin_transparent: bool = False) → ndarray

Convert the data to an RGBA image using values provided by get_colors. We are particular about min and max values, ensuring that these get their own indices from the colormap. Thee rules are: 0: value is nodata 1: value <= min_value 2: min_value < value < (max_value - min_value) / 253 254: (max_value - min_value) / 253 <= value < max_value 255 is >= max_value

Parameters:

• data (np.ndarray) – Two dimensional array.

• get_colors (Callable[[int], Tuple[int, int, int]]) – When passed an integer index in range 0:256, returns RGB components as integers in range 0:256.

• min_value (Optional[float]) – Minimum value. Defaults to None.

• max_value (Optional[float]) – Maximum value. Defaults to None.

• nodata_lower (Optional[float], optional) – If supplied, values smaller than or equal to nodata_lower threshold are considered nodata. Defaults to None.

• nodata_upper (Optional[float], optional) – If supplied, values larger than or equal to nodata_upper threshold are considered nodata. Defaults to None.

• nodata_bin_transparent (bool, optional) – If True make no_data bin transparent. Defaults to False.

• min_bin_transparent (bool, optional) – If True make min_bin transparent. Defaults to False.

Returns:

RGBA array.

Return type:

np.ndarray

physrisk.data.image_creator.get_data(reader: ZarrReader, path: str)

physrisk.data.inventory

class physrisk.data.inventory.Inventory(hazard_resources: Iterable[HazardResource])

Bases: object

json_ordered()

class physrisk.data.inventory.EmbeddedInventory(include_api_based: bool = False)

Bases: Inventory

Load up inventory embedded in file src/physrisk/data/static/hazard/inventory.json. This file is automatically generated by the hazard repo. In general the inventory of hazard resources used by physrisk is a combination of embedded and non-embedded inventories

colormaps()

Color maps. Key can be identical to a model identifier or more descriptive (if shared by many models).

api_based_resources()

physrisk.data.inventory.alphanumeric(text)

Return alphanumeric hash from supplied string.

physrisk.data.inventory.base36encode(number, alphabet='0123456789abcdefghijklmnopqrstuvwxyz')

Converts an integer to a base36 string.

physrisk.data.inventory.expand(resources: List[HazardResource]) → List[HazardResource]

physrisk.data.inventory_reader

class physrisk.data.inventory_reader.HazardModels(*, resources: List[HazardResource])

Bases: BaseModel

resources: List[HazardResource]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class physrisk.data.inventory_reader.InventoryReader(*, get_env: ~typing.Callable[[str, str | None], str] = <function get_env>, fs: ~fsspec.spec.AbstractFileSystem | None = None, base_path: ~fsspec.spec.AbstractFileSystem | None = None)

Bases: object

read(path: str) → List[HazardResource]

Read inventory at path provided and return HazardModels.

read_description_markdown(paths: List[str]) → Dict[str, str]

Read description markdown at path provided.

read_json(path: str) → str

Read inventory at path provided and return json.

append(path: str, hazard_models: Iterable[HazardResource])

physrisk.data.pregenerated_hazard_model

class physrisk.data.pregenerated_hazard_model.PregeneratedHazardModel(hazard_data_providers: Dict[Type[Hazard], HazardDataProvider], interpolate_years: bool = False)

Bases: HazardModel

Hazard event model that processes requests using EventProviders.

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]

log_response_issues(responses: Dict[HazardDataRequest, HazardDataResponse])

class physrisk.data.pregenerated_hazard_model.ZarrHazardModel(*, source_paths: SourcePaths, reader: ZarrReader | None = None, store=None, interpolation='floor', interpolate_years: bool = False)

Bases: PregeneratedHazardModel

physrisk.data.zarr_reader

physrisk.data.zarr_reader.get_env(key: str, default: str | None = None) → str

class physrisk.data.zarr_reader.ZarrReader(store: ~typing.MutableMapping | None = None, path_provider: ~typing.Callable[[...], str] | None = None, get_env: ~typing.Callable[[str, str | None], str] = <function get_env>)

Bases: object

Reads hazard event data from Zarr files, including OSC-format-specific attributes.

all_data(set_id: str)

ls(path: str)

classmethod create_s3_zarr_store(get_env: ~typing.Callable[[str, str | None], str] = <function get_env>)

get_curves(set_id: str, longitudes: ndarray | Sequence[float], latitudes: ndarray | Sequence[float], interpolation='floor')

Get intensity curve for each latitude and longitude coordinate pair.

Parameters:

• set_id – string or tuple representing data set, converted into path by path_provider.

• longitudes – list of longitudes.

• latitudes – list of latitudes.

• interpolation – interpolation method, “floor”, “linear”, “max” or “min”.

Returns:

numpy array of intensity (no. coordinate pairs, no. return periods). return_periods: return periods in years.

Return type:

curves

in_bounds(set_id: str, longitudes: ndarray | Sequence[float], latitudes: ndarray | Sequence[float])

get_index_values(z: Array) → Tuple[List[Any], str]

get_max_curves(set_id: str, shapes: Sequence[Polygon], interpolation: str = 'floor')

Get maximal intensity curve for a given geometry.

Parameters:

• set_id – string or tuple representing data set, converted into path by path_provider.

• shapes – list of shapely.Polygon.

• interpolation – interpolation method, “floor”, “linear”, “max” or “min”.

Returns:

numpy array of maximum intensity on the grid for a given geometry (no. coordinate pairs, no. return periods). return_periods: return periods in years. units: units.

Return type:

curves_max

get_max_curves_on_grid(set_id, longitudes, latitudes, interpolation='floor', delta_km=1.0, n_grid=5)

Get maximal intensity curve for a grid around a given latitude and longitude coordinate pair. It is almost equivalent to:

self.get_max_curves
(
    set_id,
    [
        Polygon(
            (
                (x - 0.5 * delta_deg, y - 0.5 * delta_deg),
                (x - 0.5 * delta_deg, y + 0.5 * delta_deg),
                (x + 0.5 * delta_deg, y + 0.5 * delta_deg),
                (x + 0.5 * delta_deg, y - 0.5 * delta_deg)
            )
        )
        for x, y in zip(longitudes, latitudes)
    ]
    interpolation
)

Parameters:

• set_id – string or tuple representing data set, converted into path by path_provider.

• longitudes – list of longitudes.

• latitudes – list of latitudes.

• interpolation – interpolation method, “floor”, “linear”, “max” or “min”.

• delta_km – linear distance in kilometres of the side of the square grid surrounding a given position.

• n_grid – number of grid points along the latitude and longitude dimensions used for calculating the maximal value.

Returns:

numpy array of maximum intensity on the grid for a given coordinate pair (no. coordinate pairs, no. return periods). return_periods: return periods in years.

Return type:

curves_max

Module contents