veriflow.cache.cache#

Zarr cache implementation.

Classes

`ZarrCache`(config)	The Zarr cache.
`ZarrCacheConfig`(*[, read_write_mode, ...])	Configuration for the veriflow cache.
`CacheRequest`(*, variables, stations[, ...])	Determine what requested data is missing from the cache.
`DataRequest`(*, requested, cached)	Helper class to track requested and cached data and what data is missing from the cache.

class veriflow.cache.cache.ZarrCache(config)[source]#

The Zarr cache.

In veriflow, you can cache datasets fetched from any datasource using a Zarr store. The cache enables incremental appends of new data along a single dimension (e.g. time, lead time, or station), and can be materialized on both a local filesystem and remote object storage (e.g. S3). An example use-case for using the cache is the operational application of veriflow, where new forecast data is generated and added to the cache on a daily basis, or when working with multiple people on a centralized verification project, where high-speed access to a shared cache is desired.

For now, the cache is only used to store datasets fetched from other datasources, and is not used to cache computation results.

The cache is configured in the veriflow.configuration.base.GeneralInfoConfig.cache field. All caching logic is handled in the veriflow.datasources.base.BaseDatasource.get_data() method, which consults the cache configuration and uses this ZarrCache class to read/write cached datasets as needed.

The current cache can handle the following scenarios: 1. Historical data with missing time steps, variables, or stations. 2. Forecast data with missing forecast reference times, lead times, variables, or stations

The current implementation has the following limitations: 1. When data is requested that is missing in the cache along more than one dimension, the cache will not retrieve data from the cache, but will instead fetch all data from the datasource. 2. The current cache does not (yet) support caching of computation results. 3. Large Zarr files on local filesystems may be slow to open due to the overhead of opening many small files. This is a known limitation of Zarr and is not specific to this implementation. However, using consolidated metadata (consolidated=True) can help mitigate this issue by reducing the number of files that need to be opened. However, datasets with a size of below 1TB should be fine. See: zarr-developers/zarr-python#86. Also note that caching is optional and can always be turned off.

The layout of the Zarr cache is as follows, where each source has its own subgroup under the datasets group. The source name is identical to the source field in the datasource configuration.

example_cache.zarr/
├── datasets/
│   ├── observations/
│   │   ├── temperature/
│   │   ├── precipitation/
│   │   └── discharge/
│   └── forecasts_model_a
│   │   ├── temperature/
│   │   ├── precipitation/
│   │   └── discharge/
│   └── forecasts_model_b
│   │   ├── temperature/
│   │   ├── precipitation/
│   │   └── discharge/

A flowchart of the caching logic is shown below, which is implemented in the veriflow.datasources.base.BaseDatasource.get_data() method.

        flowchart TD
    A([get_data]) --> B{Cache configured and<br/>data type cacheable?}
    B -- no --> F[Fetch all data<br/>from datasource]
    B -- yes --> C[Open cached dataset<br/>from the Zarr store]
    C --> D{Any data cached<br/>for this source?}
    D -- no --> F
    D -- yes --> E[Build CacheRequest and<br/>determine missing dimensions]
    E --> G{How many dimensions<br/>are missing?}
    G -- none --> H["Load requested data from cache<br/>(full hit)"]
    G -- one --> I["Fetch missing, append to cache,<br/>read back dataset"]
    G -- many --> F
    F --> W[(Write to cache<br/>if writable)]
    I --> W
    H --> R([Return dataset])
    W --> R

Parameters:: config (ZarrCacheConfig)

get_dataset(source)[source]#

Open a dataset from the cache.

Parameters:: source (str)
Return type:: Dataset | None

append(new_dataset, source, append_dim=None)[source]#

Incrementally add new_dataset to the cache along a single dimension.

Only the coordinate index of the existing store is read to skip values that are already cached; the cached data itself is never loaded into memory, keeping the cache lazy-friendly. New coordinate values are appended along dim (append_dim), new variables are added in place (mode="a"), and an empty store is created from scratch.

Parameters:

new_dataset (Dataset)
source (str)
append_dim (Literal[StandardDim.station, StandardDim.forecast_reference_time, StandardDim.lead_time, StandardDim.time] | None)

Return type:

None

property is_remote: bool#: Return True if path looks like a remote/fsspec URL (e.g. s3://).

property is_writable: bool#: Return True if the cache is writable.

property sources: list[str]#: Return the list of sources (subgroups) cached in the datasets group.

clear(source=None)[source]#

Delete a Zarr archive from either a local filesystem or a remote MinIO server.

If source is specified, only the corresponding subgroup is deleted.

Parameters:: source (str | None)
Return type:: None

class veriflow.cache.cache.ZarrCacheConfig(*, read_write_mode=ReadWriteMode.read, path, auth_config=None, storage_options=None, consolidated=None)[source]#

Configuration for the veriflow cache.

The cache will be materialized as a Zarr store on the local filesystem or remote object storage (e.g. S3). When configured, the cache will be used to store and retrieve datasets from any datasource. For example: when requesting forecast data from a datasource, and part of the data is already cached, the cache will be used to retrieve the cached data and only the missing data will be fetched from the datasource.

Parameters:

read_write_mode (ReadWriteMode)
path (Annotated[str, MinLen(min_length=1)])
auth_config (S3AuthConfig | None)
storage_options (dict[str, str] | None)
consolidated (bool | None)

read_write_mode: ReadWriteMode#

path: Annotated[str, FieldInfo(annotation=NoneType, required=True, description="Path to a single Zarr store. Local filesystem path (absolute or relative) or a remote URL such as 's3://bucket/key/store.zarr'.", metadata=[MinLen(min_length=1)])]#

auth_config: Annotated[S3AuthConfig | None, FieldInfo(annotation=NoneType, required=False, default=None, description="Authentication configuration for remote stores. Only consulted when 'path' points to an 's3://' location. When the path is remote and this is left unset, credentials are loaded automatically from S3_-prefixed environment variables, so configuring 'auth_config: {}' in YAML is not required.")]#

storage_options: Annotated[dict[str, str] | None, FieldInfo(annotation=NoneType, required=False, default=None, description="Additional storage_options forwarded to xr.open_zarr. Merged on top of the options derived from 'auth_config'. Use this for advanced fsspec / s3fs settings not exposed by S3AuthConfig.")]#

consolidated: Annotated[bool | None, FieldInfo(annotation=NoneType, required=False, default=None, description="Whether to use consolidated metadata when opening the store. Forwarded to xr.open_zarr. Default ('None') lets xarray auto-detect.")]#

is_remote_path()[source]#

Return True if path looks like a remote/fsspec URL (e.g. s3://).

Return type:: bool

validate_cache_path_accessible()[source]#

Check that a local cache dir exists, or initialize S3 auth for remote paths.

Return type:: Self

class veriflow.cache.cache.CacheRequest(*, variables, stations, time_period=None, forecast_reference_time_period=None, lead_times=None)[source]#

Determine what requested data is missing from the cache.

Handles both historical requests (time_period) and forecast requests (forecast_reference_time_period + lead_times); fields that do not apply stay None.

Parameters:

variables (DataRequest)
stations (DataRequest)
time_period (DataRequest | None)
forecast_reference_time_period (DataRequest | None)
lead_times (DataRequest | None)

variables: DataRequest#

stations: DataRequest#

time_period: DataRequest | None#

forecast_reference_time_period: DataRequest | None#

lead_times: DataRequest | None#

property missing_count: int#: Return the number of dimensions along which data is missing from the cache.

property missing_dims: list[str]#: Return the dimensions along which data is missing from the cache.

split_config(datasource)[source]#

Split the datasource into a datasource-fetch and a cache-fetch datasource.

Only splits when exactly one dimension is missing from the cache; otherwise returns None so that all data is fetched from the datasource.

Parameters:: datasource (BaseDatasource)
Return type:: tuple[BaseDatasource, BaseDatasource] | None

class veriflow.cache.cache.DataRequest(*, requested, cached)[source]#

Helper class to track requested and cached data and what data is missing from the cache.

Parameters:

requested (set[str] | TimePeriod | LeadTimes | None)
cached (set[str] | TimePeriod | LeadTimes | None)

requested: set[str] | TimePeriod | LeadTimes | None#

cached: set[str] | TimePeriod | LeadTimes | None#

validate_requested_and_cached()[source]#

Validate that the requested and cached values are of the same type.

Return type:: Self

static find_missing_and_available_time_period(requested, cached)[source]#

Find the missing time period between the requested and cached time periods.

This helper function is used to determine what time period needs to be fetched from the datasource given what is already available in the cache. The returned tuple contains the missing time period and the available time period, respectively.

The requested period R is compared against the cached period C on the time axis. The five possible cases and their outcomes are shown below (# marks the extent of each period):

time --->

(1) R inside C  ->  full hit
    R      ####
    C   ##########
    missing:   (none)          available: R

(2) R and C disjoint
    R   ####
    C            ####
    missing:   R               available: (none)

(3) C inside R
    R   ##########
    C      ####
    missing:   R               available: (none)

(4) left overlap (R starts before C)
    R   ######
    C       ######
    missing:   R.start .. C.start
    available: C.start .. R.end

(5) right overlap (R starts inside C)
    R       ######
    C   ######
    missing:   C.end .. R.end
    available: R.start .. C.end

Parameters:

requested (TimePeriod)
cached (TimePeriod)

Return type:

tuple[TimePeriod | None, TimePeriod | None]

property get_missing_and_available: tuple[set[str] | TimePeriod | LeadTimes | None, set[str] | TimePeriod | LeadTimes | None]#: Return the missing and available data given what is cached.