smart_geocubes.core.backend

Write specific backends.

Classes:

• DownloadBackend –

  Base class for download backends.

DownloadBackend

DownloadBackend(
    repo: Repository, f: Callable[[PatchIndex], Dataset]
)

Bases: ABC

Base class for download backends.

Initialize the ThreadedBackend.

Parameters:

• repo

  (Repository) –

  The icechunk repository.

• f

  (callable[[PatchIndex], Dataset]) –

  A function that takes a PatchIndex and returns an xr.Dataset. This should be implemented by the specific source backend.

Methods:

• assert_created –

  Assert that the datacube exists in the storage.

• close –

  Close the backend.

• loaded_patches –

  Get a list of all loaded patch ids.

• open_xarray –

  Open the xarray datacube in read-only mode.

• open_zarr –

  Open the zarr datacube in read-only mode.

• submit –

  Submit a patch download request to the backend.

Attributes:

• created (bool) –

  Check if the datacube already exists in the storage.

Source code in src/smart_geocubes/core/backend.py

def __init__(self, repo: icechunk.Repository, f: Callable[[PatchIndex], xr.Dataset]):
    """Initialize the ThreadedBackend.

    Args:
        repo (icechunk.Repository): The icechunk repository.
        f (callable[[PatchIndex], xr.Dataset]): A function that takes a PatchIndex and returns an xr.Dataset.
            This should be implemented by the specific source backend.

    """
    self.repo = repo
    self.download_from_source = f

    # For debugging purposes
    self._events: list[_Event] = []

created property

created: bool

Check if the datacube already exists in the storage.

Returns:

• bool ( bool ) –

  True if the datacube already exists in the storage.

assert_created

assert_created(session: Session | None = None)

Assert that the datacube exists in the storage.

Raises:

• FileNotFoundError –

  If the datacube does not exist.

Source code in src/smart_geocubes/core/backend.py

def assert_created(self, session: icechunk.Session | None = None):
    """Assert that the datacube exists in the storage.

    Raises:
        FileNotFoundError: If the datacube does not exist.

    """
    if session is None:
        session = self.repo.readonly_session("main")
    if sync(session.store.is_empty("")):
        msg = "Datacube does not exist."
        " Please use the `create` method or pass `create=True` to `load`."
        logger.error(msg)
        raise FileNotFoundError(msg)

close

close() -> bool

Close the backend.

Returns:

• bool ( bool ) –

  True if the backend was closed successfully, False otherwise.

Source code in src/smart_geocubes/core/backend.py

def close(self) -> bool:
    """Close the backend.

    Returns:
        bool: True if the backend was closed successfully, False otherwise.

    """
    # Base implementation does nothing
    return True

loaded_patches

loaded_patches(session: Session | None = None) -> list[str]

Get a list of all loaded patch ids.

Returns:

• list[str] –

  list[str]: A list of all loaded patch ids.

Source code in src/smart_geocubes/core/backend.py

def loaded_patches(self, session: icechunk.Session | None = None) -> list[str]:
    """Get a list of all loaded patch ids.

    Returns:
        list[str]: A list of all loaded patch ids.

    """
    zcube = self.open_zarr(session)
    loaded_patches = zcube.attrs.get("loaded_patches", [])
    return loaded_patches

open_xarray

open_xarray(session: Session | None = None) -> xr.Dataset

Open the xarray datacube in read-only mode.

Returns:

• Dataset –

  xr.Dataset: The xarray datacube.

Source code in src/smart_geocubes/core/backend.py

def open_xarray(self, session: icechunk.Session | None = None) -> xr.Dataset:
    """Open the xarray datacube in read-only mode.

    Returns:
        xr.Dataset: The xarray datacube.

    """
    if session is None:
        session = self.repo.readonly_session("main")
    self.assert_created(session)
    xcube = xr.open_zarr(session.store, mask_and_scale=False, consolidated=False).set_coords("spatial_ref")
    return xcube

open_zarr

open_zarr(session: Session | None = None) -> zarr.Group

Open the zarr datacube in read-only mode.

Returns:

• Group –

  zarr.Group: The zarr datacube.

Source code in src/smart_geocubes/core/backend.py

def open_zarr(self, session: icechunk.Session | None = None) -> zarr.Group:
    """Open the zarr datacube in read-only mode.

    Returns:
        zarr.Group: The zarr datacube.

    """
    if session is None:
        session = self.repo.readonly_session("main")
    self.assert_created(session)
    zcube = zarr.open(store=session.store, mode="r")
    return zcube

submit abstractmethod

submit(idx: PatchIndex | list[PatchIndex])

Submit a patch download request to the backend.

Parameters:

• idx

  (PatchIndex | list[PatchIndex]) –

  The index or multiple indices of the patch(es) to download.

Source code in src/smart_geocubes/core/backend.py

@abc.abstractmethod
def submit(self, idx: PatchIndex | list[PatchIndex]):
    """Submit a patch download request to the backend.

    Args:
        idx (PatchIndex | list[PatchIndex]): The index or multiple indices of the patch(es) to download.

    """
    pass