tilecache

darts_utils.tilecache

Caching functionality for xarray datasets.

logger module-attribute

logger = logging.getLogger(
    __name__.replace("darts_", "darts.")
)

XarrayCacheManager

XarrayCacheManager(
    cache_dir: str | pathlib.Path | None = None,
)

Manager for caching xarray datasets.

Example

    def process_tile(tile_id: str):
        # Initialize cache manager
        preprocess_cache = Path("preprocess_cache")
        cache_manager = XarrayCacheManager(preprocess_cache)

        def create_tile():
            # Your existing tile creation logic goes here
            return create_tile(...)  # Replace with actual implementation

        # Get cached tile or create and cache it
        tile = cache_manager.get_or_create(
            identifier=tile_id,
            creation_func=create_tile
        )

        return tile

Initialize the cache manager.

Parameters:

• cache_dir (str | pathlib.Path | None, default: None ) –

  Directory path for caching files

Source code in darts-utils/src/darts_utils/tilecache.py

def __init__(self, cache_dir: str | Path | None = None):
    """Initialize the cache manager.

    Args:
        cache_dir (str | Path | None): Directory path for caching files

    """
    self.cache_dir = Path(cache_dir) if isinstance(cache_dir, str) else cache_dir

cache_dir instance-attribute

cache_dir = (
    pathlib.Path(
        darts_utils.tilecache.XarrayCacheManager(cache_dir)
    )
    if isinstance(
        darts_utils.tilecache.XarrayCacheManager(cache_dir),
        str,
    )
    else darts_utils.tilecache.XarrayCacheManager(cache_dir)
)

exists

exists(identifier: str) -> bool

Check if a cached Dataset exists.

Parameters:

• identifier (str) –

  Unique identifier for the cached file

Returns:

• bool ( bool ) –

  True if the Dataset exists in cache, False otherwise

Source code in darts-utils/src/darts_utils/tilecache.py

def exists(self, identifier: str) -> bool:
    """Check if a cached Dataset exists.

    Args:
        identifier (str): Unique identifier for the cached file

    Returns:
        bool: True if the Dataset exists in cache, False otherwise

    """
    if not self.cache_dir:
        return False

    cache_path = self.cache_dir / f"{identifier}.nc"
    return cache_path.exists()

get_or_create

get_or_create(
    identifier: str,
    creation_func: callable,
    force: bool,
    *args: tuple[typing.Any, ...],
    **kwargs: dict[str, typing.Any],
) -> xarray.Dataset

Get cached Dataset or create and cache it if it doesn't exist.

Parameters:

• identifier (str) –

  Unique identifier for the cached file

• creation_func (callable) –

  Function to create the Dataset if not cached

• force (bool) –

  If True, forces reprocessing even if cached

• *args (tuple[typing.Any, ...], default: () ) –

  Arguments to pass to creation_func

• **kwargs (dict[str, typing.Any], default: {} ) –

  Keyword arguments to pass to creation_func

Returns:

• xarray.Dataset –

  xr.Dataset: The Dataset (either loaded from cache or newly created)

Source code in darts-utils/src/darts_utils/tilecache.py

def get_or_create(
    self,
    identifier: str,
    creation_func: callable,
    force: bool,
    *args: tuple[Any, ...],
    **kwargs: dict[str, Any],
) -> xr.Dataset:
    """Get cached Dataset or create and cache it if it doesn't exist.

    Args:
        identifier (str): Unique identifier for the cached file
        creation_func (callable): Function to create the Dataset if not cached
        force (bool): If True, forces reprocessing even if cached
        *args: Arguments to pass to creation_func
        **kwargs: Keyword arguments to pass to creation_func

    Returns:
        xr.Dataset: The Dataset (either loaded from cache or newly created)

    """
    logger.debug(f"Checking cache for {identifier} ({force=})")
    cached_dataset = None if force else self.load_from_cache(identifier)
    logger.debug(f"Cache hit: {cached_dataset is not None}")
    if cached_dataset is not None:
        return cached_dataset

    dataset = creation_func(*args, **kwargs)
    if cached_dataset is None:
        self.save_to_cache(dataset, identifier)
    return dataset

load_from_cache

load_from_cache(identifier: str) -> xarray.Dataset | None

Load a Dataset from cache if it exists.

Parameters:

• identifier (str) –

  Unique identifier for the cached file

Returns:

• xarray.Dataset | None –

  xr.Dataset | None: Dataset if found in cache, otherwise None

Source code in darts-utils/src/darts_utils/tilecache.py

def load_from_cache(self, identifier: str) -> xr.Dataset | None:
    """Load a Dataset from cache if it exists.

    Args:
        identifier (str): Unique identifier for the cached file

    Returns:
        xr.Dataset | None: Dataset if found in cache, otherwise None

    """
    if not self.cache_dir:
        return None

    cache_path = self.cache_dir / f"{identifier}.nc"
    if not cache_path.exists():
        return None
    dataset = xr.open_dataset(cache_path, engine="h5netcdf").set_coords("spatial_ref")
    return dataset

save_to_cache

save_to_cache(
    dataset: xarray.Dataset, identifier: str
) -> bool

Save a Dataset to cache.

Parameters:

• dataset (xarray.Dataset) –

  Dataset to cache

• identifier (str) –

  Unique identifier for the cached file

Returns:

• bool ( bool ) –

  Success of operation

Source code in darts-utils/src/darts_utils/tilecache.py

def save_to_cache(self, dataset: xr.Dataset, identifier: str) -> bool:
    """Save a Dataset to cache.

    Args:
        dataset (xr.Dataset): Dataset to cache
        identifier (str): Unique identifier for the cached file

    Returns:
        bool: Success of operation

    """
    if not self.cache_dir:
        return False

    self.cache_dir.mkdir(exist_ok=True, parents=True)
    cache_path = self.cache_dir / f"{identifier}.nc"
    logger.debug(f"Caching {identifier=} to {cache_path}")
    dataset.to_netcdf(cache_path, engine="h5netcdf")
    return True