snail package

snail - the spatial networks impact assessment library

Subpackages

• snail.core package
  • get_cell_indices()
  • split_linestring()
  • split_polygon()
• snail.data package
  • Subpackages
    • snail.data.damage_curves package

Submodules

snail.cli module

snail.cli.process(args)

snail process command

snail.cli.snail(args=None)

snail command

snail.cli.split(args)

snail split command

snail.damage_library module

Utilities to access packaged damage curve data.

The library distributes a curated subset of the Nirandjan et al. (2023) Physical Vulnerability Database. The upstream work is licensed under CC-BY 4.0; see snail/data/damage_curves for the packaged metadata and curve samples.

The helper functions defined here make it easy to enumerate available curves, filter by hazard or infrastructure attributes, and instantiate PiecewiseLinearDamageCurve objects ready for use in damage modelling.

class snail.damage_library.DamageCurveMetadata(curve_id: str, hazard_type: str, hazard_name: str, sector: str, sheet_name: str, intensity_metric: str, intensity_axis: str, intensity_unit: str | None, exposed_element: str, additional_characteristics: str | None, curve_type: str | None, curve_characteristics: str | None, damage_states: str | None, cost_feature: str | None, uncertainty_range: str | None, derivation_methodology: str | None, geographical_application: str | None, readily_available: str | None, source: str, source_details: str | None, original_id: str | None)

Bases: object

Metadata describing a packaged damage curve.

additional_characteristics: str | None

cost_feature: str | None

curve_characteristics: str | None

curve_id: str

curve_type: str | None

damage_states: str | None

derivation_methodology: str | None

exposed_element: str

geographical_application: str | None

hazard_name: str

hazard_type: str

intensity_axis: str

intensity_metric: str

intensity_unit: str | None

original_id: str | None

readily_available: str | None

sector: str

sheet_name: str

source: str

source_details: str | None

uncertainty_range: str | None

snail.damage_library.available_curves(hazard: str | None = None, sector: str | None = None, exposed_element: str | None = None, curve_type: str | None = None) → DataFrame

Return metadata for available curves, with optional filters.

snail.damage_library.get_metadata(curve_id: str) → DamageCurveMetadata

Return the metadata record for a specific curve.

snail.damage_library.load_curve(curve_id: str)

Instantiate a PiecewiseLinearDamageCurve from the packaged dataset.

snail.damages module

Damage assessment

class snail.damages.DamageCurve

Bases: ABC

A damage curve

Methods

damage_fraction()

Evaluate damage fraction for exposure to a given hazard intensity

damage_fraction() → array

Evaluate damage fraction for exposure to a given hazard intensity

class snail.damages.PiecewiseLinearDamageCurve(curve: DataFrame[PiecewiseLinearDamageCurveSchema])

Bases: DamageCurve

A piecewise-linear damage curve

Methods

clip_curve_data(intensity, damage)	Clip damage curve values to valid 0-1 damage range
damage_fraction(exposure)	Evaluate damage fraction for exposure to a given hazard intensity
from_csv(fname[, intensity_col, damage_col, ...])	Read a damage curve from a CSV file.
from_excel(fname[, sheet_name, ...])	Read a damage curve from an Excel file.
interpolate(a, b, factor)	Interpolate damage values between two curves
plot([ax])	Plot a line chart of the damage curve
scale_x(x)	Scale intensity by a factor, x
scale_y(y)	Scale damage by a factor, y
translate_x(x)	Translate intensity by a factor, x
translate_y(y)	Translate damage by a factor, y

static clip_curve_data(intensity, damage)

Clip damage curve values to valid 0-1 damage range

damage: Series[float]

damage_fraction(exposure: array) → array

Evaluate damage fraction for exposure to a given hazard intensity

classmethod from_csv(fname, intensity_col='intensity', damage_col='damage_ratio', comment='#', **kwargs)

Read a damage curve from a CSV file.

By default, the CSV should have columns named “intensity” and “damage_ratio”, with any additional header lines commented out by “#”.

Any additional keyword arguments are passed through to pandas.read_csv

Parameters:

fnamestr, path object or file-like object

intensity_colstr, default “intensity”

Column name to read hazard intensity values

damage_colstr, default “damage_ratio”

Column name to read damage values

commentstr, default “#”

Indicates remainder of the line in the CSV should not be parsed. If found at the beginning of a line, the line will be ignored altogether.

kwargs

see pandas.read_csv documentation

Returns:

PiecewiseLinearDamageCurve

classmethod from_excel(fname, sheet_name=0, intensity_col='intensity', damage_col='damage_ratio', comment='#', **kwargs)

Read a damage curve from an Excel file.

By default, the file should have columns named “intensity” and “damage_ratio”, with any additional header lines commented out by “#”.

Any additional keyword arguments are passed through to pandas.read_excel

Parameters:

fnamestr, path object or file-like object

sheet_namestr or int

Strings are used for sheet names. Integers are used in zero-indexed sheet positions (chart sheets do not count as a sheet position).

intensity_colstr, default “intensity”

Column name to read hazard intensity values

damage_colstr, default “damage_ratio”

Column name to read damage values

commentstr, default “#”

Indicates remainder of the line in the CSV should not be parsed. If found at the beginning of a line, the line will be ignored altogether.

kwargs

see pandas.read_csv documentation

Returns:

PiecewiseLinearDamageCurve

intensity: Series[float]

classmethod interpolate(a, b, factor: float)

Interpolate damage values between two curves

` new_curve_damage = a_damage + ((b_damage - a_damage) * factor) `

Parameters:

aPiecewiseLinearDamageCurve

bPiecewiseLinearDamageCurve

factorfloat

Interpolation factor, used to calculate the new curve

Returns:

PiecewiseLinearDamageCurve

plot(ax=None)

Plot a line chart of the damage curve

scale_x(x: float)

Scale intensity by a factor, x

scale_y(y: float)

Scale damage by a factor, y

translate_x(x: float)

Translate intensity by a factor, x

translate_y(y: float)

Translate damage by a factor, y

class snail.damages.PiecewiseLinearDamageCurveSchema(*args, **kwargs)

Bases: DataFrameModel

Methods

empty(*_args)	Create an empty DataFrame with the schema of this model.
example(**kwargs)	Generate an example of this data model specification.
from_json(source)	Load a schema from JSON.
from_yaml(yaml_schema)	Load a schema from YAML.
get_metadata()	Provide metadata for columns and schema level
pydantic_validate(schema_model)	Verify that the input is a compatible dataframe model.
strategy(**kwargs)	Create a data synthesis strategy.
to_json([target])	Convert this model's schema to JSON.
to_json_schema()	Serialize schema metadata into json-schema format.
to_schema()	Create DataFrameSchema from the DataFrameModel.
to_yaml([stream])	Convert this model's schema to YAML.
validate(check_obj[, head, tail, sample, ...])	Validate a DataFrame based on the schema specification.

Config
build_schema_

class Config

Bases: BaseConfig

Attributes:

description

dtype

from_format

from_format_kwargs

metadata

multiindex_name

multiindex_unique

title

to_format

to_format_buffer

to_format_kwargs

unique

name: str | None = 'PiecewiseLinearDamageCurveSchema'

name of schema

damage: Series[float] = 'damage'

intensity: Series[float] = 'intensity'

snail.intersection module

class snail.intersection.GridDefinition(crs: str, width: int, height: int, transform: Tuple[float])

Bases: object

Store a raster transform and CRS

A note on transform - these six numbers define the transform from i,j cell index (column/row) coordinates in the rectangular grid to x,y geographic coordinates, in the coordinate reference system of the input and output files. They effectively form the first two rows of a 3x3 matrix:

| x |   | a  b  c | | i |
| y | = | d  e  f | | j |
| 1 |   | 0  0  1 | | 1 |

In cases without shear or rotation, a and e define scaling or grid cell size, while c and f define the offset or grid upper-left corner:

| x_scale 0       x_offset |
| 0       y_scale y_offset |
| 0       0       1        |

Methods

from_extent(xmin, ymin, xmax, ymax, ...)	GridDefinition for a given extent, cell size and CRS
from_raster(fname)	GridDefinition for a raster file (readable by rasterio)
from_rasterio_dataset(dataset)	GridDefinition for a rasterio dataset

crs: str

classmethod from_extent(xmin: float, ymin: float, xmax: float, ymax: float, cell_width: float, cell_height: float, crs)

GridDefinition for a given extent, cell size and CRS

classmethod from_raster(fname)

GridDefinition for a raster file (readable by rasterio)

classmethod from_rasterio_dataset(dataset)

GridDefinition for a rasterio dataset

height: int

transform: Tuple[float]

width: int

snail.intersection.aggregate_values_to_grid(splits: GeoDataFrame, value_column: str, grid: GridDefinition, index_i: str = 'index_i', index_j: str = 'index_j', fill_value: float = 0.0, dtype=None) → ndarray

Aggregate split-geometry attributes onto a raster-shaped array.

Parameters:

splits

GeoDataFrame containing split geometries and raster index columns.

value_column

Name of the column to aggregate per cell (e.g. length_km).

grid

A GridDefinition that defines the raster bounds.

index_i, index_j

Column names storing raster column (i) and row (j) indices.

fill_value

Initial fill value for cells without observations.

dtype

Optional dtype for the resulting array. Defaults to promoting the column dtype with the fill value dtype.

snail.intersection.apply_indices(features: GeoDataFrame, grid: GridDefinition, index_i='index_i', index_j='index_j') → GeoDataFrame

snail.intersection.generate_grid_boxes(grid: GridDefinition)

Generate all the box polygons for a grid

snail.intersection.get_indices(geom, grid: GridDefinition, index_i='index_i', index_j='index_j') → Series

Given a geometry, find the cell index (i, j) of its midpoint for the enclosing grid.

N.B. There is no checking whether a geometry spans more than one cell.

snail.intersection.get_raster_values_for_splits(splits: DataFrame, data: ndarray, index_i: str = 'index_i', index_j: str = 'index_j') → Series

For each split geometry, lookup the relevant raster value.

Cell indices must have been previously calculated and stored as index_i and index_j.

N.B. This will pass through no data values from the raster (no filtering).

Parameters:

splits: pandas.DataFrame

Table of features, each with cell indices to look up raster pixel. Indices must be stored under columns with names referenced by index_i and index_j.

data: numpy.ndarray

Raster data (2D array)

index_i: str

Column name for i-indices

index_j: str

Column name for j-indices

Returns:

pd.Series

Series of raster values, with same row indexing as df.

snail.intersection.idx_to_ij(idx: int, width: int, height: int) → Tuple[int]

snail.intersection.ij_to_idx(ij: Tuple[int], width: int, height: int)

snail.intersection.prepare_linestrings(features: GeoDataFrame) → GeoDataFrame

snail.intersection.prepare_points(features: GeoDataFrame) → GeoDataFrame

Prepare points for splitting

snail.intersection.prepare_polygons(features: GeoDataFrame) → GeoDataFrame

snail.intersection.split_features_for_rasters(features: GeoDataFrame, grids: List[GridDefinition], split_func: Callable)

snail.intersection.split_linestrings(linestring_features: GeoDataFrame, grid: GridDefinition) → GeoDataFrame

Split linestrings along a grid

snail.intersection.split_points(points: GeoDataFrame, grid: GridDefinition) → GeoDataFrame

Split points along a grid

This is a no-op, written for equivalence when processing multiple geometry types.

snail.intersection.split_polygons(polygon_features: GeoDataFrame, grid: GridDefinition) → GeoDataFrame

Split polygons along a grid

snail.intersection.split_polygons_experimental(polygon_features: GeoDataFrame, grid: GridDefinition) → GeoDataFrame

Split polygons along a grid

Experimental implementation of split_polygons, possibly fast/incorrect with some inputs.

snail.io module

snail.io.associate_raster_files(splits, rasters)

Read values from a list of raster files for a set of indexed split geometries

Parameters:

splits: pandas.DataFrame

split geometries with raster indices in columns named “i_{grid_id}”, “j_{grid_id}” for each grid_id in rasters

rasters: pandas.DataFrame

table of raster metadata with columns: key, grid_id, path, bands

Returns:

pandas.DataFrame

split geometries with raster data values at indexed locations

snail.io.extend_rasters_metadata(rasters: DataFrame) → Tuple[DataFrame, List[GridDefinition]]

snail.io.read_features(path, layer=None)

snail.io.read_raster_band_data(fname: str, band_number: int = 1) → ndarray

snail.io.read_raster_metadata(path) → Tuple[GridDefinition, Tuple[int]]

snail.io.read_rasters(rasters)

snail.io.write_grid_to_raster(array: ndarray, output_path, transform, crs, *, nodata=None, dtype=None, driver: str = 'GTiff', compress: str = 'lzw', **profile_kwargs)

Write a 2D NumPy array to a single-band raster using rasterio.

snail.routing module

snail.routing.shortest_paths(sources, destinations, graph, weight)

Compute all shortest paths from an ensemble of sources to an ensemble of destinations.

Positional arguments: sources – list of source node ids (string or int). destinations – list of destination node ids (string or int). graph: igraph.Graph instance representing the network. weight – Edge attribute according to which paths should be weighted (string)

Returns: A list of tuples (source, destination) A list of list of edge ids corresponding to shortest paths. For each (source, destination) pair, their is either 0, 1 or several shortest paths.

snail.tqdm_standin module

snail.tqdm_standin.tqdm_standin(iterator, *_, **__)

Alternative to tqdm, with no progress bar - ignore any arguments after the first