Skip to content

base

Base class for OSM loaders.

OSMLoader

Bases: Loader, ABC

Abstract class for loaders.

load(area, tags)

abstractmethod

Load data for a given area.

PARAMETER DESCRIPTION
area

Shapely geometry with the area of interest.

TYPE: Union[BaseGeometry, Iterable[BaseGeometry], GeoSeries, GeoDataFrame]

tags

OSM tags filter.

TYPE: Union[OsmTagsFilter, GroupedOsmTagsFilter]

RETURNS DESCRIPTION
GeoDataFrame

gpd.GeoDataFrame: GeoDataFrame with the downloaded data.

Source code in srai/loaders/osm_loaders/_base.py
@abc.abstractmethod
def load(
    self,
    area: Union[BaseGeometry, Iterable[BaseGeometry], gpd.GeoSeries, gpd.GeoDataFrame],
    tags: Union[OsmTagsFilter, GroupedOsmTagsFilter],
) -> gpd.GeoDataFrame:  # pragma: no cover
    """
    Load data for a given area.

    Args:
        area (Union[BaseGeometry, Iterable[BaseGeometry], gpd.GeoSeries, gpd.GeoDataFrame]):
            Shapely geometry with the area of interest.
        tags (Union[OsmTagsFilter, GroupedOsmTagsFilter]): OSM tags filter.

    Returns:
        gpd.GeoDataFrame: GeoDataFrame with the downloaded data.
    """
    raise NotImplementedError

prepare_area_gdf_for_loader(area)

Prepare an area for the loader.

Loader expects a GeoDataFrame input, but users shouldn't be limited by this requirement. All Shapely geometries will by transformed into GeoDataFrame with proper CRS.

PARAMETER DESCRIPTION
area

Area to be parsed into GeoDataFrame.

TYPE: Union[BaseGeometry, Iterable[BaseGeometry], GeoSeries, GeoDataFrame]

RETURNS DESCRIPTION
GeoDataFrame

gpd.GeoDataFrame: Sanitized GeoDataFrame.

Source code in srai/loaders/osm_loaders/_base.py
def prepare_area_gdf_for_loader(
    area: Union[BaseGeometry, Iterable[BaseGeometry], gpd.GeoSeries, gpd.GeoDataFrame],
) -> gpd.GeoDataFrame:
    """
    Prepare an area for the loader.

    Loader expects a GeoDataFrame input, but users shouldn't be limited by this requirement.
    All Shapely geometries will by transformed into GeoDataFrame with proper CRS.

    Args:
        area (Union[BaseGeometry, Iterable[BaseGeometry], gpd.GeoSeries, gpd.GeoDataFrame]):
            Area to be parsed into GeoDataFrame.

    Returns:
        gpd.GeoDataFrame: Sanitized GeoDataFrame.
    """
    if isinstance(area, gpd.GeoDataFrame):
        # Return a GeoDataFrame with changed CRS
        return area.to_crs(WGS84_CRS)
    elif isinstance(area, gpd.GeoSeries):
        # Create a GeoDataFrame with GeoSeries
        return gpd.GeoDataFrame(geometry=area, crs=WGS84_CRS)
    elif isinstance(area, Iterable):
        # Create a GeoSeries with a list of geometries
        return prepare_area_gdf_for_loader(gpd.GeoSeries(area, crs=WGS84_CRS))
    # Wrap a single geometry with a list
    return prepare_area_gdf_for_loader([area])