Loaders¶

loaders ¶

This module contains loaders, used to load spatial data from different sources.

We want to unify loading from different data sources into a single interface. Thanks to this, we have a unified spatial data format, which makes it possible to feed them into any of the embedding methods available in this library.

Classes¶

Functions¶

srai.loaders.convert_to_features_gdf ¶

convert_to_features_gdf(
    geometry: Union[
        BaseGeometry, Iterable[BaseGeometry], gpd.GeoSeries, gpd.GeoDataFrame
    ],
    index_column: Optional[str] = None,
) -> gpd.GeoDataFrame

Convert any geometry to a features GeoDataFrame.

PARAMETER	DESCRIPTION
`geometry`	Geo objects to convert. TYPE: `Union[BaseGeometry, Iterable[BaseGeometry], GeoSeries, GeoDataFrame]`
`index_column`	Name of the column used to define the index. If None, will rename the existing index. Defaults to None. TYPE: `Optional[str]` DEFAULT: `None`

RETURNS	DESCRIPTION
`GeoDataFrame`	gpd.GeoDataFrame: Features gdf with proper index definition.

Source code in srai/geometry.py

def convert_to_features_gdf(
    geometry: Union[BaseGeometry, Iterable[BaseGeometry], gpd.GeoSeries, gpd.GeoDataFrame],
    index_column: Optional[str] = None,
) -> gpd.GeoDataFrame:
    """
    Convert any geometry to a features GeoDataFrame.

    Args:
        geometry (Union[BaseGeometry, Iterable[BaseGeometry], gpd.GeoSeries, gpd.GeoDataFrame]): Geo
            objects to convert.
        index_column (Optional[str], optional): Name of the column used to define the index.
            If None, will rename the existing index. Defaults to None.

    Returns:
        gpd.GeoDataFrame: Features gdf with proper index definition.
    """
    return _convert_to_internal_format(
        geometry=geometry, destination_index_name=FEATURES_INDEX, index_column=index_column
    )

srai.loaders.download_file ¶

download_file(
    url: str, fname: str, chunk_size: int = 1024, force_download: bool = True
) -> None

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`
`force_download`	Flag to force download even if file exists. TYPE: `bool` DEFAULT: `True`

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

Source code in srai/loaders/download.py

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

    Args:
        url (str): URL to download.
        fname (str): File name.
        chunk_size (str): Chunk size.
        force_download (bool): Flag to force download even if file exists.

    Source: https://gist.github.com/yanqd0/c13ed29e29432e3cf3e7c38467f42f51
    """
    if Path(fname).exists() and not force_download:
        warnings.warn("File exists. Skipping download.", stacklevel=1)
        return

    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,
    )
    resp.raise_for_status()
    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,
            disable=FORCE_TERMINAL,
        ) as bar,
    ):
        for data in resp.iter_content(chunk_size=chunk_size):
            size = file.write(data)
            bar.update(size)