Utils

Module containing different utility functions.

Those are either used internally by other modules, or can be used to simplify spatial data processing.

download_file

Download a file with progress bar.

PARAMETER	DESCRIPTION
url	URL to download. TYPE: str
fname	File name. TYPE: str
chunk_size	Chunk size. TYPE: str DEFAULT: 1024

Source: https://gist.github.com/yanqd0/c13ed29e29432e3cf3e7c38467f42f51

Source code in srai/utils/download.py

def download_file(url: str, fname: str, chunk_size: int = 1024) -> None:
    """
    Download a file with progress bar.

    Args:
        url (str): URL to download.
        fname (str): File name.
        chunk_size (str): Chunk size.

    Source: https://gist.github.com/yanqd0/c13ed29e29432e3cf3e7c38467f42f51
    """
    Path(fname).parent.mkdir(parents=True, exist_ok=True)
    resp = requests.get(
        url,
        headers={"User-Agent": "SRAI Python package (https://github.com/kraina-ai/srai)"},
        stream=True,
    )
    total = int(resp.headers.get("content-length", 0))
    with open(fname, "wb") as file, tqdm(
        desc=fname.split("/")[-1],
        total=total,
        unit="iB",
        unit_scale=True,
        unit_divisor=1024,
    ) as bar:
        for data in resp.iter_content(chunk_size=chunk_size):
            size = file.write(data)
            bar.update(size)

geocode_to_region_gdf

Geocode a query to the regions_gdf unified format.

This functions is a wrapper around the ox.geocode_to_gdf[1] function from the osmnx library. For parameters description look into the source documentation.

References

1. https://osmnx.readthedocs.io/en/stable/osmnx.html#osmnx.geocoder.geocode_to_gdf

Source code in srai/utils/geocode.py

def geocode_to_region_gdf(
    query: Union[str, List[str], Dict[str, Any]], by_osmid: bool = False
) -> gpd.GeoDataFrame:
    """
    Geocode a query to the `regions_gdf` unified format.

    This functions is a wrapper around the `ox.geocode_to_gdf`[1] function from the `osmnx` library.
    For parameters description look into the source documentation.

    References:
        1. https://osmnx.readthedocs.io/en/stable/osmnx.html#osmnx.geocoder.geocode_to_gdf
    """
    geocoded_gdf = ox.geocode_to_gdf(query=query, by_osmid=by_osmid, which_result=None)
    regions_gdf = (
        geocoded_gdf[["display_name", "geometry"]]
        .rename(columns={"display_name": REGIONS_INDEX})
        .set_index(REGIONS_INDEX)
    )
    return regions_gdf

buffer_geometry

Buffer geometry by a given radius in meters.

Projects geometry into azimuthal projection before applying buffer and then changes values back to WGS84 coordinates.

Doesn't work with polygons covering the whole earth (from -180 to 180 longitude).

PARAMETER	DESCRIPTION
geometry	Geometry to buffer. TYPE: BaseGeometry
meters	Radius distance in meters. TYPE: float

RETURNS	DESCRIPTION
BaseGeometry	Buffered geometry. TYPE: BaseGeometry

Source code in srai/utils/geometry.py

def buffer_geometry(geometry: BaseGeometry, meters: float) -> BaseGeometry:
    """
    Buffer geometry by a given radius in meters.

    Projects geometry into azimuthal projection before applying buffer and then changes values
    back to WGS84 coordinates.

    Doesn't work with polygons covering the whole earth (from -180 to 180 longitude).

    Args:
        geometry (BaseGeometry): Geometry to buffer.
        meters (float): Radius distance in meters.

    Returns:
        BaseGeometry: Buffered geometry.
    """
    _lon, _lat = geometry.centroid.coords[0]

    aeqd_proj = pyproj.Proj(proj="aeqd", ellps="WGS84", datum="WGS84", lat_0=_lat, lon_0=_lon)
    wgs84_proj = pyproj.Proj(proj="latlong", ellps="WGS84")

    wgs84_to_aeqd = pyproj.Transformer.from_proj(wgs84_proj, aeqd_proj, always_xy=True).transform
    aeqd_to_wgs84 = pyproj.Transformer.from_proj(aeqd_proj, wgs84_proj, always_xy=True).transform

    projected_geometry = shapely_transform(wgs84_to_aeqd, geometry)
    bufferred_projected_geometry = projected_geometry.buffer(meters)

    return shapely_transform(aeqd_to_wgs84, bufferred_projected_geometry)

flatten_geometry

Flatten all geometries into a list of BaseGeometries.

Source code in srai/utils/geometry.py

def flatten_geometry(geometry: BaseGeometry) -> List[BaseGeometry]:
    """Flatten all geometries into a list of BaseGeometries."""
    if isinstance(geometry, BaseMultipartGeometry):
        geometries: List[BaseGeometry] = (
            seq([flatten_geometry(sub_geom) for sub_geom in geometry.geoms]).flatten().to_list()
        )
        return geometries
    return [geometry]

flatten_geometry_series

Flatten all geometries from a series into a list of BaseGeometries.

Source code in srai/utils/geometry.py

def flatten_geometry_series(geometry_series: gpd.GeoSeries) -> List[BaseGeometry]:
    """Flatten all geometries from a series into a list of BaseGeometries."""
    geometries: List[BaseGeometry] = (
        seq([flatten_geometry(geometry) for geometry in geometry_series]).flatten().to_list()
    )
    return geometries

remove_interiors

Close polygon holes by limitation to the exterior ring.

PARAMETER	DESCRIPTION
polygon	Polygon to close. TYPE: Polygon

RETURNS	DESCRIPTION
Polygon	Closed polygon. TYPE: Polygon

Source code in srai/utils/geometry.py

def remove_interiors(polygon: Polygon) -> Polygon:
    """
    Close polygon holes by limitation to the exterior ring.

    Args:
        polygon (Polygon): Polygon to close.

    Returns:
        Polygon: Closed polygon.
    """
    if polygon.interiors:
        return Polygon(list(polygon.exterior.coords))
    return polygon

merge_disjointed_gdf_geometries

Merges geometries from a GeoDataFrame into a single MultiPolygon.

Input geometries are expected to be disjointed.

PARAMETER	DESCRIPTION
gdf	GeoDataFrame with geometries to merge. TYPE: gpd.GeoDataFrame

RETURNS	DESCRIPTION
MultiPolygon	Merged polygon TYPE: MultiPolygon

Source code in srai/utils/merge.py

def merge_disjointed_gdf_geometries(gdf: gpd.GeoDataFrame) -> MultiPolygon:
    """
    Merges geometries from a GeoDataFrame into a single MultiPolygon.

    Input geometries are expected to be disjointed.

    Args:
        gdf (gpd.GeoDataFrame): GeoDataFrame with geometries to merge.

    Returns:
        MultiPolygon: Merged polygon
    """
    return merge_disjointed_polygons(list(gdf.geometry))

merge_disjointed_polygons

Merges all polygons into a single MultiPolygon.

Input polygons are expected to be disjointed.

PARAMETER	DESCRIPTION
polygons	List of polygons to merge TYPE: List[Union[Polygon, MultiPolygon]]

RETURNS	DESCRIPTION
MultiPolygon	Merged polygon TYPE: MultiPolygon

Source code in srai/utils/merge.py

def merge_disjointed_polygons(polygons: List[Union[Polygon, MultiPolygon]]) -> MultiPolygon:
    """
    Merges all polygons into a single MultiPolygon.

    Input polygons are expected to be disjointed.

    Args:
        polygons (List[Union[Polygon, MultiPolygon]]): List of polygons to merge

    Returns:
        MultiPolygon: Merged polygon
    """
    single_polygons = []
    for geom in polygons:
        if type(geom) is Polygon:
            single_polygons.append(geom)
        else:
            single_polygons.extend(geom.geoms)
    return MultiPolygon(single_polygons)