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.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)