`climate_ref_core.dataset_registry` #

Data registries for non-published reference data

These data are placeholders until these data have been added to obs4MIPs. The CMIP7 Assessment Fas Track REF requires that reference datasets are openly licensed before it is included in any published data catalogs.

`DatasetRegistryManager` #

A collection of reference datasets registries

The REF requires additional reference datasets in addition to obs4MIPs data which can be downloaded via ESGF. Each provider may have different sets of reference data that are needed. These are provider-specific datasets are datasets not yet available in obs4MIPs, or are post-processed from obs4MIPs.

A dataset registry consists of a file that contains a list of files and checksums, in combination with a base URL that is used to fetch the files. Pooch is used within the DataRegistry to manage the caching, downloading and validation of the files.

All datasets that are registered here are expected to be openly licensed and freely available.

Source code in packages/climate-ref-core/src/climate_ref_core/dataset_registry.py

class DatasetRegistryManager:
    """
    A collection of reference datasets registries

    The REF requires additional reference datasets
    in addition to obs4MIPs data which can be downloaded via ESGF.
    Each provider may have different sets of reference data that are needed.
    These are provider-specific datasets are datasets not yet available in obs4MIPs,
    or are post-processed from obs4MIPs.

    A dataset registry consists of a file that contains a list of files and checksums,
    in combination with a base URL that is used to fetch the files.
    [Pooch](https://www.fatiando.org/pooch/latest/) is used within the DataRegistry
    to manage the caching, downloading and validation of the files.

    All datasets that are registered here are expected to be openly licensed and freely available.
    """

    def __init__(self) -> None:
        self._registries: dict[str, RegistryEntry] = {}

    def __getitem__(self, item: str) -> pooch.Pooch:
        """
        Get a registry by name
        """
        return self._registries[item].registry

    def entry(self, item: str) -> RegistryEntry:
        """
        Get the full registry entry (registry, source type and use-case) by name
        """
        return self._registries[item]

    def keys(self) -> list[str]:
        """
        Get the list of registry names
        """
        return list(self._registries.keys())

    @staticmethod
    @staticmethod
    def _migrate_cache(
        registry: pooch.Pooch,
        legacy_cache_dirs: list[pathlib.Path],
    ) -> None:
        """
        Migrate cached files from legacy cache directories to the current location.

        For each file in the registry,
        if it does not already exist at the
        current cache path but is found in one of the legacy directories,
        move it to the new location.

        Parameters
        ----------
        registry
            The Pooch registry whose cache may need migrating.
        legacy_cache_dirs
            Directories where files may have been cached previously.
        """
        new_root: pathlib.Path = registry.abspath  # type: ignore[attr-defined]

        for key in registry.registry:
            new_path = new_root / key
            if new_path.exists():
                continue

            for legacy_dir in legacy_cache_dirs:
                old_path = legacy_dir / key
                if old_path.exists():
                    # Registration happens at import time, so a read-only cache must not
                    # stop the package from importing.
                    try:
                        new_path.parent.mkdir(parents=True, exist_ok=True)
                        logger.info(f"Migrating cached file {key}: {old_path} -> {new_path}")
                        shutil.move(str(old_path), str(new_path))
                    except OSError as exc:
                        logger.warning(f"Could not migrate cached file {key} to {new_path}: {exc}")
                        return
                    break

    def register(  # noqa: PLR0913
        self,
        name: str,
        base_url: str,
        package: str,
        resource: str,
        cache_name: str | None = None,
        version: str | None = None,
        legacy_cache_dirs: list[pathlib.Path] | None = None,
        source_type: SourceDatasetType | None = None,
        use_case: RegistryUseCase = RegistryUseCase.support,
    ) -> None:
        """
        Register a new dataset registry

        This will create a new Pooch registry and add it to the list of registries.
        This is typically used by a provider to register a new collections of datasets at runtime.

        Parameters
        ----------
        name
            Name of the registry

            This is used to identify the registry
        base_url
            Commmon URL prefix for the files
        package
            Name of the package containing the registry resource.
        resource
            Name of the resource in the package that contains a list of files and checksums.

            This must be formatted in a way that is expected by pooch.
        version
            The version of the data.

            Changing the version will invalidate the cache and force a re-download of the data.
        cache_name
            Name to use to generate the cache directory.

            This defaults to the value of `name` if not provided.
        legacy_cache_dirs
            Previous cache directories to migrate files from.

            If provided, any files that exist in a legacy directory but not in
            the current cache will be moved to the new location. This avoids
            re-downloading data after a cache layout change.
        source_type
            The target source type used to select the parsing adapter.

            Required for ``reference`` registries.
            Leave as ``None`` for ``support`` registries,
            or for registries whose contents span more than one source type
            (which cannot be catalogued by registry-driven ingest at all).
        use_case
            Whether the registry's contents are catalog-ingestable (``reference``)
            or fetch-only (``support``).

            Defaults to ``support`` so that an unannotated registry is never
            catalogued by registry-driven ingest.
        """
        if cache_name is None:
            cache_name = name

        cache_path = resolve_cache_dir(cache_name)

        # Before v0.13.0 everything was cached directly under
        # pooch.os_cache("climate_ref") with no per-registry subdirectory.
        # Always include that as a legacy location so files are migrated
        # automatically, regardless of whether the caller passes extra dirs.
        default_legacy = pathlib.Path(pooch.os_cache("climate_ref"))
        all_legacy_dirs = [default_legacy]
        if legacy_cache_dirs:
            all_legacy_dirs.extend(legacy_cache_dirs)

        registry = pooch.create(
            path=cache_path,
            base_url=base_url,
            version=version,
            retry_if_failed=10,
        )
        # pooch reads the manifest eagerly, so the temporary path only has to outlive the call.
        with PackagedResource(package, resource).as_path() as manifest:
            registry.load_registry(manifest)

        if cache_path != default_legacy:
            self._migrate_cache(registry, all_legacy_dirs)

        self._registries[name] = RegistryEntry(
            registry=registry,
            source_type=source_type,
            use_case=use_case,
        )

`getitem(item)` #

Get a registry by name

Source code in packages/climate-ref-core/src/climate_ref_core/dataset_registry.py

def __getitem__(self, item: str) -> pooch.Pooch:
    """
    Get a registry by name
    """
    return self._registries[item].registry

`entry(item)` #

Get the full registry entry (registry, source type and use-case) by name

Source code in packages/climate-ref-core/src/climate_ref_core/dataset_registry.py

def entry(self, item: str) -> RegistryEntry:
    """
    Get the full registry entry (registry, source type and use-case) by name
    """
    return self._registries[item]

`keys()` #

Get the list of registry names

Source code in packages/climate-ref-core/src/climate_ref_core/dataset_registry.py

def keys(self) -> list[str]:
    """
    Get the list of registry names
    """
    return list(self._registries.keys())

`register(name, base_url, package, resource, cache_name=None, version=None, legacy_cache_dirs=None, source_type=None, use_case=RegistryUseCase.support)` #

Register a new dataset registry

This will create a new Pooch registry and add it to the list of registries. This is typically used by a provider to register a new collections of datasets at runtime.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the registry This is used to identify the registry	required
`base_url`	`str`	Commmon URL prefix for the files	required
`package`	`str`	Name of the package containing the registry resource.	required
`resource`	`str`	Name of the resource in the package that contains a list of files and checksums. This must be formatted in a way that is expected by pooch.	required
`version`	`str \| None`	The version of the data. Changing the version will invalidate the cache and force a re-download of the data.	`None`
`cache_name`	`str \| None`	Name to use to generate the cache directory. This defaults to the value of `name` if not provided.	`None`
`legacy_cache_dirs`	`list[Path] \| None`	Previous cache directories to migrate files from. If provided, any files that exist in a legacy directory but not in the current cache will be moved to the new location. This avoids re-downloading data after a cache layout change.	`None`
`source_type`	`SourceDatasetType \| None`	The target source type used to select the parsing adapter. Required for `reference` registries. Leave as `None` for `support` registries, or for registries whose contents span more than one source type (which cannot be catalogued by registry-driven ingest at all).	`None`
`use_case`	`RegistryUseCase`	Whether the registry's contents are catalog-ingestable (`reference`) or fetch-only (`support`). Defaults to `support` so that an unannotated registry is never catalogued by registry-driven ingest.	`support`

Source code in packages/climate-ref-core/src/climate_ref_core/dataset_registry.py

def register(  # noqa: PLR0913
    self,
    name: str,
    base_url: str,
    package: str,
    resource: str,
    cache_name: str | None = None,
    version: str | None = None,
    legacy_cache_dirs: list[pathlib.Path] | None = None,
    source_type: SourceDatasetType | None = None,
    use_case: RegistryUseCase = RegistryUseCase.support,
) -> None:
    """
    Register a new dataset registry

    This will create a new Pooch registry and add it to the list of registries.
    This is typically used by a provider to register a new collections of datasets at runtime.

    Parameters
    ----------
    name
        Name of the registry

        This is used to identify the registry
    base_url
        Commmon URL prefix for the files
    package
        Name of the package containing the registry resource.
    resource
        Name of the resource in the package that contains a list of files and checksums.

        This must be formatted in a way that is expected by pooch.
    version
        The version of the data.

        Changing the version will invalidate the cache and force a re-download of the data.
    cache_name
        Name to use to generate the cache directory.

        This defaults to the value of `name` if not provided.
    legacy_cache_dirs
        Previous cache directories to migrate files from.

        If provided, any files that exist in a legacy directory but not in
        the current cache will be moved to the new location. This avoids
        re-downloading data after a cache layout change.
    source_type
        The target source type used to select the parsing adapter.

        Required for ``reference`` registries.
        Leave as ``None`` for ``support`` registries,
        or for registries whose contents span more than one source type
        (which cannot be catalogued by registry-driven ingest at all).
    use_case
        Whether the registry's contents are catalog-ingestable (``reference``)
        or fetch-only (``support``).

        Defaults to ``support`` so that an unannotated registry is never
        catalogued by registry-driven ingest.
    """
    if cache_name is None:
        cache_name = name

    cache_path = resolve_cache_dir(cache_name)

    # Before v0.13.0 everything was cached directly under
    # pooch.os_cache("climate_ref") with no per-registry subdirectory.
    # Always include that as a legacy location so files are migrated
    # automatically, regardless of whether the caller passes extra dirs.
    default_legacy = pathlib.Path(pooch.os_cache("climate_ref"))
    all_legacy_dirs = [default_legacy]
    if legacy_cache_dirs:
        all_legacy_dirs.extend(legacy_cache_dirs)

    registry = pooch.create(
        path=cache_path,
        base_url=base_url,
        version=version,
        retry_if_failed=10,
    )
    # pooch reads the manifest eagerly, so the temporary path only has to outlive the call.
    with PackagedResource(package, resource).as_path() as manifest:
        registry.load_registry(manifest)

    if cache_path != default_legacy:
        self._migrate_cache(registry, all_legacy_dirs)

    self._registries[name] = RegistryEntry(
        registry=registry,
        source_type=source_type,
        use_case=use_case,
    )

`RegistryEntry` #

A dataset registry together with its ingest metadata.

Source code in packages/climate-ref-core/src/climate_ref_core/dataset_registry.py

@frozen
class RegistryEntry:
    """
    A dataset registry together with its ingest metadata.
    """

    registry: pooch.Pooch
    """The underlying Pooch registry."""

    source_type: SourceDatasetType | None
    """
    The target source type used to select the parsing adapter.

    Required for ``reference`` registries.
    ``None`` for ``support`` registries, since fetch-only content is never parsed.
    """

    use_case: RegistryUseCase
    """Whether the registry's contents are catalog-ingestable or fetch-only."""

`registry` `instance-attribute` #

The underlying Pooch registry.

`source_type` `instance-attribute` #

The target source type used to select the parsing adapter.

Required for reference registries. None for support registries, since fetch-only content is never parsed.

`use_case` `instance-attribute` #

Whether the registry's contents are catalog-ingestable or fetch-only.

`RegistryUseCase` #

Bases: Enum

Whether a registry's contents are catalog-ingestable or fetch-only.

Source code in packages/climate-ref-core/src/climate_ref_core/dataset_registry.py

class RegistryUseCase(enum.Enum):
    """
    Whether a registry's contents are catalog-ingestable or fetch-only.
    """

    reference = "reference"
    """
    Contents are catalog-ingestable reference data.

    The registry declares exactly one target source type,
    which is used to select the parsing adapter during ingest.
    """

    support = "support"
    """
    Contents are fetch-only support files (e.g. region masks, recipes).

    These are never entered into the catalog.
    """

`reference = 'reference'` `class-attribute` `instance-attribute` #

Contents are catalog-ingestable reference data.

The registry declares exactly one target source type, which is used to select the parsing adapter during ingest.

`support = 'support'` `class-attribute` `instance-attribute` #

Contents are fetch-only support files (e.g. region masks, recipes).

These are never entered into the catalog.

`fetch_all_files(registry, name, output_dir, symlink=False, verify=True)` #

Fetch all files associated with a pooch registry and write them to an output directory.

Pooch fetches, caches and validates up to four downloaded files concurrently. Subsequent calls to this function will not refetch any previously downloaded files. The number of simultaneous threads can be overridden using the REF_DATASET_FETCH_WORKERS environment variable.

Parameters:

Name	Type	Description	Default
`registry`	`Pooch`	Pooch directory containing a set of files that should be fetched.	required
`name`	`str`	Name of the registry.	required
`output_dir`	`Path \| None`	The root directory to write the files to. The directory will be created if it doesn't exist, and matching files will be overwritten. If no directory is provided, the files will be fetched from the remote server, but not copied anywhere.	required
`symlink`	`bool`	If True, symlink all files to this directory. Otherwise, perform a copy.	`False`
`verify`	`bool`	If True, verify the checksums of the local files against the registry.	`True`

Source code in packages/climate-ref-core/src/climate_ref_core/dataset_registry.py

def fetch_all_files(
    registry: pooch.Pooch,
    name: str,
    output_dir: pathlib.Path | None,
    symlink: bool = False,
    verify: bool = True,
) -> None:
    """
    Fetch all files associated with a pooch registry and write them to an output directory.

    Pooch fetches, caches and validates up to four downloaded files concurrently.
    Subsequent calls to this function will not refetch any previously downloaded files.
    The number of simultaneous threads can be overridden using the
    ``REF_DATASET_FETCH_WORKERS`` environment variable.

    Parameters
    ----------
    registry
        Pooch directory containing a set of files that should be fetched.
    name
        Name of the registry.
    output_dir
        The root directory to write the files to.

        The directory will be created if it doesn't exist,
        and matching files will be overwritten.

        If no directory is provided, the files will be fetched from the remote server,
        but not copied anywhere.
    symlink
        If True, symlink all files to this directory.
        Otherwise, perform a copy.
    verify
        If True, verify the checksums of the local files against the registry.
    """
    if output_dir:
        output_dir.mkdir(parents=True, exist_ok=True)

    def fetch_one(key: str) -> None:
        fetch_file = registry.fetch(key)
        expected_hash = registry.registry[key]
        if not isinstance(expected_hash, str) or not expected_hash:  # pragma: no cover
            raise ValueError(f"Expected a hash for {key} but got {expected_hash}")

        if output_dir is None:
            # Fetching was the whole job: the file is in the cache and there is nowhere to link it.
            return

        linked_file = output_dir / key
        linked_file.parent.mkdir(parents=True, exist_ok=True)
        if not linked_file.exists():  # pragma: no cover
            if symlink:
                logger.info(f"Linking {key} to {linked_file}")

                os.symlink(fetch_file, linked_file)
            else:
                logger.info(f"Copying {key} to {linked_file}")
                shutil.copy(fetch_file, linked_file)
        else:
            logger.info(f"File {linked_file} already exists. Skipping.")
        if verify:
            _verify_hash_matches(linked_file, expected_hash)

    keys = list(registry.registry)
    # Pooch creates missing cache directories without exist_ok=True.
    # Prepare them before starting workers so files with a shared parent cannot race in mkdir.
    for key in keys:
        (registry.abspath / key).parent.mkdir(parents=True, exist_ok=True)  # type: ignore[attr-defined]

    with ThreadPoolExecutor(max_workers=_MAX_FETCH_WORKERS) as executor:
        futures = [executor.submit(fetch_one, key) for key in keys]
        for future in track(
            as_completed(futures),
            total=len(futures),
            description=f"Fetching {name} data",
        ):
            future.result()

`iter_reference_registries(manager)` #

Yield the (registry, source_type) pairs for all catalog-ingestable registries.

Registries whose use_case is not reference, or whose source_type is None (e.g. registries that mix source types and so cannot be catalogued as a whole), are skipped. A later ingest driver can pass the yielded source_type to :func:climate_ref.datasets.get_dataset_adapter to select the parsing adapter.

Parameters:

Name	Type	Description	Default
`manager`	`DatasetRegistryManager`	The registry manager to read entries from.	required

Returns:

Type	Description
`Iterator[tuple[Pooch, SourceDatasetType]]`	An iterator of `(registry, source_type)` pairs.

Source code in packages/climate-ref-core/src/climate_ref_core/dataset_registry.py

def iter_reference_registries(
    manager: DatasetRegistryManager,
) -> Iterator[tuple[pooch.Pooch, SourceDatasetType]]:
    """
    Yield the ``(registry, source_type)`` pairs for all catalog-ingestable registries.

    Registries whose ``use_case`` is not ``reference``, or whose ``source_type`` is
    ``None`` (e.g. registries that mix source types and so cannot be catalogued as a whole),
    are skipped.
    A later ingest driver can pass the yielded ``source_type`` to
    :func:`climate_ref.datasets.get_dataset_adapter` to select the parsing adapter.

    Parameters
    ----------
    manager
        The registry manager to read entries from.

    Returns
    -------
    :
        An iterator of ``(registry, source_type)`` pairs.
    """
    for name in manager.keys():
        entry = manager.entry(name)
        if entry.use_case is RegistryUseCase.reference and entry.source_type is not None:
            yield entry.registry, entry.source_type

`validate_registry_cache(registry, name)` #

Validate that all files in a registry are cached and have correct checksums.

Parameters:

Name	Type	Description	Default
`registry`	`Pooch`	Pooch registry to validate.	required
`name`	`str`	Name of the registry (for error messages).	required

Returns:

Type	Description
`list[str]`	List of error messages for any validation failures. Empty list if all files are valid.

Source code in packages/climate-ref-core/src/climate_ref_core/dataset_registry.py

def validate_registry_cache(
    registry: pooch.Pooch,
    name: str,
) -> list[str]:
    """
    Validate that all files in a registry are cached and have correct checksums.

    Parameters
    ----------
    registry
        Pooch registry to validate.
    name
        Name of the registry (for error messages).

    Returns
    -------
    list[str]
        List of error messages for any validation failures.
        Empty list if all files are valid.
    """
    errors: list[str] = []

    for key in registry.registry.keys():
        expected_hash = registry.registry[key]
        if not isinstance(expected_hash, str) or not expected_hash:  # pragma: no cover
            errors.append(f"{name}: No hash defined for {key}")
            continue

        # Check if file exists in cache
        cached_path = registry.abspath / key  # type: ignore[attr-defined]
        if not cached_path.exists():
            errors.append(f"{name}: File not cached: {key}")
            continue

        # Verify checksum
        try:
            _verify_hash_matches(cached_path, expected_hash)
        except ValueError as e:
            errors.append(f"{name}: {e}")

    return errors

climate_ref_core.dataset_registry #

DatasetRegistryManager #

__getitem__(item) #

entry(item) #

keys() #

register(name, base_url, package, resource, cache_name=None, version=None, legacy_cache_dirs=None, source_type=None, use_case=RegistryUseCase.support) #

RegistryEntry #

registry instance-attribute #

source_type instance-attribute #

use_case instance-attribute #

RegistryUseCase #

reference = 'reference' class-attribute instance-attribute #

support = 'support' class-attribute instance-attribute #

fetch_all_files(registry, name, output_dir, symlink=False, verify=True) #

iter_reference_registries(manager) #

validate_registry_cache(registry, name) #

`climate_ref_core.dataset_registry` #

`DatasetRegistryManager` #

`getitem(item)` #

`entry(item)` #

`keys()` #

`register(name, base_url, package, resource, cache_name=None, version=None, legacy_cache_dirs=None, source_type=None, use_case=RegistryUseCase.support)` #

`RegistryEntry` #

`registry` `instance-attribute` #

`source_type` `instance-attribute` #

`use_case` `instance-attribute` #

`RegistryUseCase` #

`reference = 'reference'` `class-attribute` `instance-attribute` #

`support = 'support'` `class-attribute` `instance-attribute` #

`fetch_all_files(registry, name, output_dir, symlink=False, verify=True)` #

`iter_reference_registries(manager)` #

`validate_registry_cache(registry, name)` #