climate_ref.datasets

Dataset handling utilities

CMIP6DatasetAdapter

Bases: FinaliseableDatasetAdapterMixin, DatasetAdapter

Adapter for CMIP6 datasets

Source code in packages/climate-ref/src/climate_ref/datasets/cmip6.py

class CMIP6DatasetAdapter(FinaliseableDatasetAdapterMixin, DatasetAdapter):
    """
    Adapter for CMIP6 datasets
    """

    dataset_cls = CMIP6Dataset
    slug_column = "instance_id"

    columns_requiring_finalisation = frozenset(
        {
            "branch_method",
            "branch_time_in_child",
            "branch_time_in_parent",
            "experiment",
            "grid",
            "long_name",
            "nominal_resolution",
            "parent_activity_id",
            "parent_experiment_id",
            "parent_source_id",
            "parent_time_units",
            "parent_variant_label",
            "product",
            "realm",
            "source_type",
            "standard_name",
            "sub_experiment",
            "sub_experiment_id",
            "time_units",
            "calendar",
            "units",
            "vertical_levels",
        }
    )

    dataset_specific_metadata = (
        "activity_id",
        "branch_method",
        "branch_time_in_child",
        "branch_time_in_parent",
        "experiment",
        "experiment_id",
        "frequency",
        "grid",
        "grid_label",
        "institution_id",
        "nominal_resolution",
        "parent_activity_id",
        "parent_experiment_id",
        "parent_source_id",
        "parent_time_units",
        "parent_variant_label",
        "product",
        "realm",
        "source_id",
        "source_type",
        "sub_experiment",
        "sub_experiment_id",
        "table_id",
        "variable_id",
        "variant_label",
        "member_id",
        "vertical_levels",
        "version",
        # Variable identifiers
        "standard_name",
        "long_name",
        "units",
        # Time metadata
        "time_units",
        "calendar",
        "finalised",
        slug_column,
    )

    file_specific_metadata = ("start_time", "end_time", "path")

    version_metadata = "version"
    # See https://wcrp-cmip.github.io/WGCM_Infrastructure_Panel/Papers/CMIP6_global_attributes_filenames_CVs_v6.2.7.pdf
    # under "Directory structure template"
    dataset_id_metadata = (
        "activity_id",
        "institution_id",
        "source_id",
        "experiment_id",
        "member_id",
        "table_id",
        "variable_id",
        "grid_label",
    )

    def __init__(self, n_jobs: int = 1, config: Config | None = None):
        self.n_jobs = n_jobs
        self.config = config or Config.default()

    def get_complete_parser(self) -> DatasetParsingFunction:
        """
        Return the complete parser that opens files to extract full CMIP6 metadata.

        Returns
        -------
        :
            Complete CMIP6 parsing function
        """
        return parse_cmip6_complete

    def _post_finalise_fixes(self, datasets: pd.DataFrame) -> pd.DataFrame:
        """
        Apply CMIP6-specific fixes after finalisation.

        Parameters
        ----------
        datasets
            DataFrame with finalised metadata

        Returns
        -------
        :
            DataFrame with fixes applied
        """
        return _apply_fixes(datasets)

    def get_parsing_function(self) -> DatasetParsingFunction:
        """
        Get the parsing function for CMIP6 datasets based on configuration

        The parsing function used is determined by the `cmip6_parser` configuration value:
        - "drs": Use the DRS parser (default)
        - "complete": Use the complete parser that extracts all available metadata

        Returns
        -------
        :
            The appropriate parsing function based on configuration
        """
        parser_type = self.config.cmip6_parser
        if parser_type == "complete":
            logger.info("Using complete CMIP6 parser")
            return parse_cmip6_complete
        else:
            logger.info(f"Using DRS CMIP6 parser (config value: {parser_type})")
            return parse_cmip6_drs

    def _enrich_parsed_catalog(self, datasets: pd.DataFrame) -> pd.DataFrame:
        """
        Apply CMIP6 post-parse enrichment to a raw catalog DataFrame.

        Shared by :meth:`find_local_datasets` (whole-tree) and  :meth:`iter_local_datasets` (chunked)
        so behaviour stays identical.

        The caller owns ``datasets`` and the result,
        so we mutate in place to avoid an extra full-table copy in :func:`build_instance_id`.
        """
        if "init_year" in datasets.columns:
            datasets = datasets.drop(["init_year"], axis=1)

        cal = datasets["calendar"] if "calendar" in datasets.columns else "standard"
        if "start_time" in datasets.columns:
            datasets["start_time"] = parse_cftime_dates(datasets["start_time"], cal)
        if "end_time" in datasets.columns:
            datasets["end_time"] = parse_cftime_dates(datasets["end_time"], cal)

        drs_items = [*self.dataset_id_metadata, self.version_metadata]
        datasets = build_instance_id(datasets, drs_items, prefix="CMIP6", copy=False)

        missing_columns = set(self.dataset_specific_metadata + self.file_specific_metadata) - set(
            datasets.columns
        )
        for column in missing_columns:
            datasets[column] = pd.NA

        # TODO: Replace with a standalone package that contains metadata fixes for CMIP6 datasets
        return _apply_fixes(datasets)

    def find_local_datasets(self, file_or_directory: Path) -> pd.DataFrame:
        """
        Generate a data catalog from the specified file or directory

        Each dataset may contain multiple files, which are represented as rows in the data catalog.
        Each dataset has a unique identifier, which is in `slug_column`.

        Parameters
        ----------
        file_or_directory
            File or directory containing the datasets

        Returns
        -------
        :
            Data catalog containing the metadata for the dataset
        """
        parsing_function = self.get_parsing_function()

        datasets = build_catalog(
            paths=[str(file_or_directory)],
            parsing_func=parsing_function,
            include_patterns=["*.nc"],
            depth=10,
            n_jobs=self.n_jobs,
        )

        return self._enrich_parsed_catalog(datasets)

    def iter_local_datasets(
        self, file_or_directory: Path, chunk_size: int = 10_000
    ) -> Iterator[pd.DataFrame]:
        """
        Stream the data catalog in chunks to bound peak memory.

        Discovery walks the tree once, but parsing and DataFrame construction
        happen ``chunk_size`` files at a time. Chunks flush at directory
        boundaries so files belonging to the same dataset (which share a DRS
        version directory) stay together in a single chunk.

        Parameters
        ----------
        file_or_directory
            Root of the CMIP6 archive (or a single file) to ingest.
        chunk_size
            Soft target for the number of files per chunk. Increasing this
            trades higher peak memory for fewer per-chunk overheads.

        Yields
        ------
        :
            Catalog DataFrames, each containing metadata for one chunk of files.
            Empty chunks are skipped.
        """
        parsing_function = self.get_parsing_function()

        for raw_chunk in iter_built_catalogs(
            paths=[str(file_or_directory)],
            parsing_func=parsing_function,
            include_patterns=["*.nc"],
            depth=10,
            n_jobs=self.n_jobs,
            chunk_size=chunk_size,
        ):
            enriched = self._enrich_parsed_catalog(raw_chunk)
            if enriched.empty:
                continue
            yield enriched

find_local_datasets(file_or_directory)

Generate a data catalog from the specified file or directory

Each dataset may contain multiple files, which are represented as rows in the data catalog. Each dataset has a unique identifier, which is in slug_column.

Parameters:

Name	Type	Description	Default
file_or_directory	Path	File or directory containing the datasets	required

Returns:

Type	Description
DataFrame	Data catalog containing the metadata for the dataset

Source code in packages/climate-ref/src/climate_ref/datasets/cmip6.py

def find_local_datasets(self, file_or_directory: Path) -> pd.DataFrame:
    """
    Generate a data catalog from the specified file or directory

    Each dataset may contain multiple files, which are represented as rows in the data catalog.
    Each dataset has a unique identifier, which is in `slug_column`.

    Parameters
    ----------
    file_or_directory
        File or directory containing the datasets

    Returns
    -------
    :
        Data catalog containing the metadata for the dataset
    """
    parsing_function = self.get_parsing_function()

    datasets = build_catalog(
        paths=[str(file_or_directory)],
        parsing_func=parsing_function,
        include_patterns=["*.nc"],
        depth=10,
        n_jobs=self.n_jobs,
    )

    return self._enrich_parsed_catalog(datasets)

get_complete_parser()

Return the complete parser that opens files to extract full CMIP6 metadata.

Returns:

Type	Description
DatasetParsingFunction	Complete CMIP6 parsing function

Source code in packages/climate-ref/src/climate_ref/datasets/cmip6.py

def get_complete_parser(self) -> DatasetParsingFunction:
    """
    Return the complete parser that opens files to extract full CMIP6 metadata.

    Returns
    -------
    :
        Complete CMIP6 parsing function
    """
    return parse_cmip6_complete

get_parsing_function()

Get the parsing function for CMIP6 datasets based on configuration

The parsing function used is determined by the cmip6_parser configuration value: - "drs": Use the DRS parser (default) - "complete": Use the complete parser that extracts all available metadata

Returns:

Type	Description
DatasetParsingFunction	The appropriate parsing function based on configuration

Source code in packages/climate-ref/src/climate_ref/datasets/cmip6.py

def get_parsing_function(self) -> DatasetParsingFunction:
    """
    Get the parsing function for CMIP6 datasets based on configuration

    The parsing function used is determined by the `cmip6_parser` configuration value:
    - "drs": Use the DRS parser (default)
    - "complete": Use the complete parser that extracts all available metadata

    Returns
    -------
    :
        The appropriate parsing function based on configuration
    """
    parser_type = self.config.cmip6_parser
    if parser_type == "complete":
        logger.info("Using complete CMIP6 parser")
        return parse_cmip6_complete
    else:
        logger.info(f"Using DRS CMIP6 parser (config value: {parser_type})")
        return parse_cmip6_drs

iter_local_datasets(file_or_directory, chunk_size=10000)

Stream the data catalog in chunks to bound peak memory.

Discovery walks the tree once, but parsing and DataFrame construction happen chunk_size files at a time. Chunks flush at directory boundaries so files belonging to the same dataset (which share a DRS version directory) stay together in a single chunk.

Parameters:

Name	Type	Description	Default
file_or_directory	Path	Root of the CMIP6 archive (or a single file) to ingest.	required
chunk_size	int	Soft target for the number of files per chunk. Increasing this trades higher peak memory for fewer per-chunk overheads.	10000

Yields:

Type	Description
DataFrame	Catalog DataFrames, each containing metadata for one chunk of files. Empty chunks are skipped.

Source code in packages/climate-ref/src/climate_ref/datasets/cmip6.py

def iter_local_datasets(
    self, file_or_directory: Path, chunk_size: int = 10_000
) -> Iterator[pd.DataFrame]:
    """
    Stream the data catalog in chunks to bound peak memory.

    Discovery walks the tree once, but parsing and DataFrame construction
    happen ``chunk_size`` files at a time. Chunks flush at directory
    boundaries so files belonging to the same dataset (which share a DRS
    version directory) stay together in a single chunk.

    Parameters
    ----------
    file_or_directory
        Root of the CMIP6 archive (or a single file) to ingest.
    chunk_size
        Soft target for the number of files per chunk. Increasing this
        trades higher peak memory for fewer per-chunk overheads.

    Yields
    ------
    :
        Catalog DataFrames, each containing metadata for one chunk of files.
        Empty chunks are skipped.
    """
    parsing_function = self.get_parsing_function()

    for raw_chunk in iter_built_catalogs(
        paths=[str(file_or_directory)],
        parsing_func=parsing_function,
        include_patterns=["*.nc"],
        depth=10,
        n_jobs=self.n_jobs,
        chunk_size=chunk_size,
    ):
        enriched = self._enrich_parsed_catalog(raw_chunk)
        if enriched.empty:
            continue
        yield enriched

CMIP7DatasetAdapter

Bases: FinaliseableDatasetAdapterMixin, DatasetAdapter

Adapter for CMIP7 datasets

Based on CMIP7 Global Attributes v1.0 (DOI: 10.5281/zenodo.17250297).

Source code in packages/climate-ref/src/climate_ref/datasets/cmip7.py

class CMIP7DatasetAdapter(FinaliseableDatasetAdapterMixin, DatasetAdapter):
    """
    Adapter for CMIP7 datasets

    Based on CMIP7 Global Attributes v1.0 (DOI: 10.5281/zenodo.17250297).
    """

    dataset_cls = CMIP7Dataset
    slug_column = "instance_id"

    columns_requiring_finalisation = frozenset(
        {
            # Optional information
            "realm",
            "nominal_resolution",
            "license_id",
            "external_variables",
            # Parent info
            "branch_time_in_child",
            "branch_time_in_parent",
            "parent_activity_id",
            "parent_experiment_id",
            "parent_mip_era",
            "parent_source_id",
            "parent_time_units",
            "parent_variant_label",
            # Variable metadata
            "standard_name",
            "long_name",
            "units",
            # Time metadata
            "time_units",
            "calendar",
        }
    )

    dataset_specific_metadata = (
        # Core DRS attributes
        "activity_id",
        "institution_id",
        "source_id",
        "experiment_id",
        "variant_label",
        "variable_id",
        "grid_label",
        "frequency",
        "region",
        "branding_suffix",
        "version",
        # Additional mandatory attributes
        "mip_era",
        "realm",
        "nominal_resolution",
        "license_id",
        # Conditionally required attributes
        "external_variables",
        # Parent info
        "branch_time_in_child",
        "branch_time_in_parent",
        "parent_activity_id",
        "parent_experiment_id",
        "parent_mip_era",
        "parent_source_id",
        "parent_time_units",
        "parent_variant_label",
        # Variable metadata
        "standard_name",
        "long_name",
        "units",
        # Time metadata
        "time_units",
        "calendar",
        # Finalisation status
        "finalised",
        # Unique identifier
        slug_column,
    )

    file_specific_metadata = ("start_time", "end_time", "path", "tracking_id")

    # Not stored in the DB; reconstructed by _add_derived_columns on every load.
    derived_metadata = ("branded_variable",)

    version_metadata = "version"

    # CMIP7 DRS directory structure (MIP-DRS7 spec):
    #   <drs_specs>/<mip_era>/<activity_id>/<institution_id>/.../<grid_label>/<version>
    # The leading drs_specs and mip_era are fixed values ("MIP-DRS7" and "CMIP7")
    # and are omitted here. They are added as the "CMIP7." prefix when building instance_id.
    dataset_id_metadata = (
        "activity_id",
        "institution_id",
        "source_id",
        "experiment_id",
        "variant_label",
        "region",
        "frequency",
        "variable_id",
        "branding_suffix",
        "grid_label",
    )

    def __init__(self, n_jobs: int = 1, config: Config | None = None):
        self.n_jobs = n_jobs
        self.config = config or Config.default()

    def get_complete_parser(self) -> DatasetParsingFunction:
        """
        Return the complete parser that opens files to extract full CMIP7 metadata.

        Returns
        -------
        :
            Complete CMIP7 parsing function
        """
        return parse_cmip7_complete

    def _post_finalise_fixes(self, datasets: pd.DataFrame) -> pd.DataFrame:
        """
        Apply CMIP7-specific fixes after finalisation.

        Cleans branch time values that may be stored as strings with units suffixes.

        Parameters
        ----------
        datasets
            DataFrame with finalised metadata

        Returns
        -------
        :
            DataFrame with fixes applied
        """
        if "branch_time_in_child" in datasets.columns:
            datasets["branch_time_in_child"] = clean_branch_time(datasets["branch_time_in_child"])
        if "branch_time_in_parent" in datasets.columns:
            datasets["branch_time_in_parent"] = clean_branch_time(datasets["branch_time_in_parent"])
        return datasets

    def get_parsing_function(self) -> DatasetParsingFunction:
        """
        Get the parsing function for CMIP7 datasets based on configuration

        The parsing function used is determined by the `cmip7_parser` configuration value:
        - "drs": Use the DRS parser (default)
        - "complete": Use the complete parser that extracts all available metadata

        Returns
        -------
        :
            The appropriate parsing function based on configuration
        """
        parser_type = self.config.cmip7_parser
        if parser_type == "complete":
            logger.info("Using complete CMIP7 parser")
            return parse_cmip7_complete
        else:
            logger.info(f"Using DRS CMIP7 parser (config value: {parser_type})")
            return parse_cmip7_drs

    def _enrich_parsed_catalog(self, datasets: pd.DataFrame) -> pd.DataFrame:
        """
        Apply CMIP7-specific post-parse enrichment to a raw catalog.

        Shared between :meth:`find_local_datasets` (whole-tree) and
        :meth:`iter_local_datasets` (streaming) so per-chunk processing is
        identical to a single-pass build.
        """
        # Convert the start_time and end_time columns to cftime objects
        cal = datasets["calendar"] if "calendar" in datasets.columns else "standard"
        if "start_time" in datasets.columns:
            datasets["start_time"] = parse_cftime_dates(datasets["start_time"], cal)
        if "end_time" in datasets.columns:
            datasets["end_time"] = parse_cftime_dates(datasets["end_time"], cal)

        # Clean branch times
        if "branch_time_in_child" in datasets.columns:
            datasets["branch_time_in_child"] = clean_branch_time(datasets["branch_time_in_child"])
        if "branch_time_in_parent" in datasets.columns:
            datasets["branch_time_in_parent"] = clean_branch_time(datasets["branch_time_in_parent"])

        # Build instance_id following CMIP7 DRS format
        # CMIP7.<activity_id>.<institution_id>.<source_id>.<experiment_id>.<variant_label>.
        # <region>.<frequency>.<variable_id>.<branding_suffix>.<grid_label>.<version>
        drs_items = [
            *self.dataset_id_metadata,
            self.version_metadata,
        ]
        datasets = build_instance_id(datasets, drs_items, prefix="CMIP7")

        # Add in any missing metadata columns
        missing_columns = set(self.dataset_specific_metadata + self.file_specific_metadata) - set(
            datasets.columns
        )
        for column in missing_columns:
            datasets[column] = pd.NA

        # Add branded_variable for the raw catalog (before DB ingestion)
        datasets = self._add_derived_columns(datasets)

        return datasets

    def _add_derived_columns(self, catalog: pd.DataFrame) -> pd.DataFrame:
        """
        Add the derived ``branded_variable`` column (``{variable_id}_{branding_suffix}``).

        ``branded_variable`` is not stored in the database as it is derived from
        ``variable_id`` and ``branding_suffix``.
        Both inputs are mandatory CMIP7 DRS facets,
        so a catalog missing the columns or carrying null values is malformed and an exception is raised.
        """
        catalog = super()._add_derived_columns(catalog)

        required = ("variable_id", "branding_suffix")
        missing = [column for column in required if column not in catalog.columns]
        if missing:
            raise ValueError(
                f"Cannot derive 'branded_variable': catalog is missing required column(s) {missing}"
            )

        if catalog.empty:
            catalog["branded_variable"] = pd.Series(dtype="object")
            return catalog

        invalid = catalog["variable_id"].isna() | catalog["branding_suffix"].isna()
        if invalid.any():
            raise ValueError(
                "Cannot derive 'branded_variable': "
                f"'variable_id'/'branding_suffix' is null for {int(invalid.sum())} row(s)"
            )

        catalog["branded_variable"] = (
            catalog["variable_id"].astype(str) + "_" + catalog["branding_suffix"].astype(str)
        )
        return catalog

    def find_local_datasets(self, file_or_directory: Path) -> pd.DataFrame:
        """
        Generate a data catalog from the specified file or directory.

        Each dataset may contain multiple files, which are represented as rows in the data catalog.
        Each dataset has a unique identifier, which is in `slug_column`.

        Parameters
        ----------
        file_or_directory
            File or directory containing the datasets

        Returns
        -------
        :
            Data catalog containing the metadata for the dataset
        """
        parsing_function = self.get_parsing_function()

        datasets = build_catalog(
            paths=[str(file_or_directory)],
            parsing_func=parsing_function,
            include_patterns=["*.nc"],
            depth=10,
            n_jobs=self.n_jobs,
        )

        return self._enrich_parsed_catalog(datasets)

    def iter_local_datasets(
        self, file_or_directory: Path, chunk_size: int = 10_000
    ) -> Iterator[pd.DataFrame]:
        """
        Stream the data catalog in chunks to bound peak memory.

        Discovery walks the tree once, but parsing and DataFrame construction
        happen ``chunk_size`` files at a time. Chunks flush at directory
        boundaries so files belonging to the same dataset (which share a DRS
        version directory) stay together in a single chunk.

        Parameters
        ----------
        file_or_directory
            Root of the CMIP7 archive (or a single file) to ingest.
        chunk_size
            Soft target for the number of files per chunk. Increasing this
            trades higher peak memory for fewer per-chunk overheads.

        Yields
        ------
        :
            Catalog DataFrames, each containing metadata for one chunk of files.
            Empty chunks are skipped.
        """
        parsing_function = self.get_parsing_function()

        for raw_chunk in iter_built_catalogs(
            paths=[str(file_or_directory)],
            parsing_func=parsing_function,
            include_patterns=["*.nc"],
            depth=10,
            n_jobs=self.n_jobs,
            chunk_size=chunk_size,
        ):
            enriched = self._enrich_parsed_catalog(raw_chunk)
            if enriched.empty:
                continue
            yield enriched

find_local_datasets(file_or_directory)

Generate a data catalog from the specified file or directory.

Each dataset may contain multiple files, which are represented as rows in the data catalog. Each dataset has a unique identifier, which is in slug_column.

Parameters:

Name	Type	Description	Default
file_or_directory	Path	File or directory containing the datasets	required

Returns:

Type	Description
DataFrame	Data catalog containing the metadata for the dataset

Source code in packages/climate-ref/src/climate_ref/datasets/cmip7.py

def find_local_datasets(self, file_or_directory: Path) -> pd.DataFrame:
    """
    Generate a data catalog from the specified file or directory.

    Each dataset may contain multiple files, which are represented as rows in the data catalog.
    Each dataset has a unique identifier, which is in `slug_column`.

    Parameters
    ----------
    file_or_directory
        File or directory containing the datasets

    Returns
    -------
    :
        Data catalog containing the metadata for the dataset
    """
    parsing_function = self.get_parsing_function()

    datasets = build_catalog(
        paths=[str(file_or_directory)],
        parsing_func=parsing_function,
        include_patterns=["*.nc"],
        depth=10,
        n_jobs=self.n_jobs,
    )

    return self._enrich_parsed_catalog(datasets)

get_complete_parser()

Return the complete parser that opens files to extract full CMIP7 metadata.

Returns:

Type	Description
DatasetParsingFunction	Complete CMIP7 parsing function

Source code in packages/climate-ref/src/climate_ref/datasets/cmip7.py

def get_complete_parser(self) -> DatasetParsingFunction:
    """
    Return the complete parser that opens files to extract full CMIP7 metadata.

    Returns
    -------
    :
        Complete CMIP7 parsing function
    """
    return parse_cmip7_complete

get_parsing_function()

Get the parsing function for CMIP7 datasets based on configuration

The parsing function used is determined by the cmip7_parser configuration value: - "drs": Use the DRS parser (default) - "complete": Use the complete parser that extracts all available metadata

Returns:

Type	Description
DatasetParsingFunction	The appropriate parsing function based on configuration

Source code in packages/climate-ref/src/climate_ref/datasets/cmip7.py

def get_parsing_function(self) -> DatasetParsingFunction:
    """
    Get the parsing function for CMIP7 datasets based on configuration

    The parsing function used is determined by the `cmip7_parser` configuration value:
    - "drs": Use the DRS parser (default)
    - "complete": Use the complete parser that extracts all available metadata

    Returns
    -------
    :
        The appropriate parsing function based on configuration
    """
    parser_type = self.config.cmip7_parser
    if parser_type == "complete":
        logger.info("Using complete CMIP7 parser")
        return parse_cmip7_complete
    else:
        logger.info(f"Using DRS CMIP7 parser (config value: {parser_type})")
        return parse_cmip7_drs

iter_local_datasets(file_or_directory, chunk_size=10000)

Stream the data catalog in chunks to bound peak memory.

Discovery walks the tree once, but parsing and DataFrame construction happen chunk_size files at a time. Chunks flush at directory boundaries so files belonging to the same dataset (which share a DRS version directory) stay together in a single chunk.

Parameters:

Name	Type	Description	Default
file_or_directory	Path	Root of the CMIP7 archive (or a single file) to ingest.	required
chunk_size	int	Soft target for the number of files per chunk. Increasing this trades higher peak memory for fewer per-chunk overheads.	10000

Yields:

Type	Description
DataFrame	Catalog DataFrames, each containing metadata for one chunk of files. Empty chunks are skipped.

Source code in packages/climate-ref/src/climate_ref/datasets/cmip7.py

def iter_local_datasets(
    self, file_or_directory: Path, chunk_size: int = 10_000
) -> Iterator[pd.DataFrame]:
    """
    Stream the data catalog in chunks to bound peak memory.

    Discovery walks the tree once, but parsing and DataFrame construction
    happen ``chunk_size`` files at a time. Chunks flush at directory
    boundaries so files belonging to the same dataset (which share a DRS
    version directory) stay together in a single chunk.

    Parameters
    ----------
    file_or_directory
        Root of the CMIP7 archive (or a single file) to ingest.
    chunk_size
        Soft target for the number of files per chunk. Increasing this
        trades higher peak memory for fewer per-chunk overheads.

    Yields
    ------
    :
        Catalog DataFrames, each containing metadata for one chunk of files.
        Empty chunks are skipped.
    """
    parsing_function = self.get_parsing_function()

    for raw_chunk in iter_built_catalogs(
        paths=[str(file_or_directory)],
        parsing_func=parsing_function,
        include_patterns=["*.nc"],
        depth=10,
        n_jobs=self.n_jobs,
        chunk_size=chunk_size,
    ):
        enriched = self._enrich_parsed_catalog(raw_chunk)
        if enriched.empty:
            continue
        yield enriched

DatasetAdapter

Bases: Protocol

An adapter to provide a common interface for different dataset types

This allows the same code to work with different dataset types.

Source code in packages/climate-ref/src/climate_ref/datasets/base.py

class DatasetAdapter(Protocol):
    """
    An adapter to provide a common interface for different dataset types

    This allows the same code to work with different dataset types.
    """

    dataset_cls: type[Dataset]
    slug_column: str
    """
    The column in the data catalog that contains the dataset slug.
    The dataset slug is a unique identifier for the dataset that includes the version of the dataset.
    This can be used to group files together that belong to the same dataset.
    """
    dataset_specific_metadata: tuple[str, ...]
    file_specific_metadata: tuple[str, ...] = ()
    columns_requiring_finalisation: frozenset[str] = frozenset()
    """
    Columns that are not available until the dataset has been finalised.

    For adapters that support two-phase parsing (e.g. DRS-only then complete),
    these columns will contain ``pd.NA`` until finalisation opens the files.
    Filtering or grouping on these columns before finalisation will silently
    produce incorrect results.
    """

    version_metadata: str = "version"
    """
    The column in the data catalog that contains the version of the dataset.
    """
    dataset_id_metadata: tuple[str, ...] = ()
    """
    The group of metadata columns that are specific to the dataset excluding the version information.

    Each unique dataset should have the same values for these columns.

    This is generally the columns that describe the `slug` of a dataset,
    excluding the version information.
    """
    derived_metadata: tuple[str, ...] = ()
    """
    Columns that are not stored in the database but are computed from stored metadata.

    These are reconstructed by :meth:`_add_derived_columns` whenever a catalog is loaded,
    so that a DB-loaded catalog has the same columns as a freshly parsed one
    (e.g. CMIP7's ``branded_variable``).
    :meth:`load_catalog` enforces that every column listed here is present after loading.
    """

    def pretty_subset(self, data_catalog: pd.DataFrame) -> pd.DataFrame:
        """
        Get a subset of the data_catalog to pretty print

        Parameters
        ----------
        data_catalog
            Data catalog to subset

        Returns
        -------
        :
            Subset of the data catalog to pretty print

        """
        return data_catalog[
            [
                *self.dataset_id_metadata,
                self.version_metadata,
            ]
        ]

    def find_local_datasets(self, file_or_directory: Path) -> pd.DataFrame:
        """
        Generate a data catalog from the specified file or directory

        This data catalog should contain all the metadata needed by the database.
        The index of the data catalog should be the dataset slug.
        """
        ...

    def validate_data_catalog(self, data_catalog: pd.DataFrame, skip_invalid: bool = False) -> pd.DataFrame:
        """
        Validate a data catalog

        Parameters
        ----------
        data_catalog
            Data catalog to validate
        skip_invalid
            If True, ignore datasets with invalid metadata and remove them from the resulting data catalog.

        Raises
        ------
        ValueError
            If `skip_invalid` is False (default) and the data catalog contains validation errors.

        Returns
        -------
        :
            Validated data catalog
        """
        # Check if the data catalog contains the required columns
        missing_columns = set(self.dataset_specific_metadata + self.file_specific_metadata) - set(
            data_catalog.columns
        )
        if missing_columns:
            raise ValueError(f"Data catalog is missing required columns: {missing_columns}")

        # Verify that the dataset specific columns don't vary by dataset by counting the unique values
        # for each dataset and checking if there are any that have more than one unique value.
        unique_metadata = (
            data_catalog[list(self.dataset_specific_metadata)].groupby(self.slug_column).nunique()
        )
        if unique_metadata.gt(1).any(axis=1).any():
            _log_duplicate_metadata(data_catalog, unique_metadata, self.slug_column)

            if skip_invalid:
                data_catalog = data_catalog[
                    ~data_catalog[self.slug_column].isin(
                        unique_metadata[unique_metadata.gt(1).any(axis=1)].index
                    )
                ]
            else:
                raise ValueError("Dataset specific metadata varies by dataset")

        return data_catalog

    def register_dataset(  # noqa: PLR0912, PLR0915
        self, db: Database, data_catalog_dataset: pd.DataFrame
    ) -> DatasetRegistrationResult:
        """
        Register a dataset in the database using the data catalog

        This assumes that the data catalog has already been validated with `validate_data_catalog`
        to ensure that the dataset-specific metadata is consistent across all files in the dataset.

        Parameters
        ----------
        db
            Database instance
        data_catalog_dataset
            A subset of the data catalog containing the metadata for a single dataset


        Raises
        ------
        RefException
            If the data catalog contains validation errors that should have been caught by
                `validate_data_catalog` (i.e. multiple unique slugs in `slug_column`).


        Returns
        -------
        :
            Registration result with dataset and file change information
        """
        DatasetModel = self.dataset_cls

        unique_slugs = data_catalog_dataset[self.slug_column].unique()
        if len(unique_slugs) != 1:
            raise RefException(f"Found multiple datasets in the same directory: {unique_slugs}")
        slug = unique_slugs[0]

        # Callers are responsible for validating the catalog with  ``validate_data_catalog`` before invoking.
        # This is a strict subset of ``validate_data_catalog`` to catch skipping upstream validation.
        slice_meta = data_catalog_dataset[list(self.dataset_specific_metadata)]
        if (slice_meta.nunique(dropna=False) > 1).any():
            raise RefException(
                f"Dataset {slug} has inconsistent dataset-specific metadata; "
                "callers must pre-validate the catalog with validate_data_catalog."
            )

        # Check if the incoming data is unfinalised (DRS parser) and the dataset
        # already exists as finalised.  In that case, skip the entire update to
        # avoid regressing metadata that was populated during finalisation.
        incoming_finalised = data_catalog_dataset.get("finalised")
        is_unfinalised_ingest = incoming_finalised is not None and not incoming_finalised.iloc[0]

        if is_unfinalised_ingest:
            existing = db.session.query(DatasetModel).filter_by(slug=slug).first()
            if existing and existing.finalised:
                logger.debug(f"Skipping already-finalised dataset: {slug}")
                current_files = db.session.query(DatasetFile).filter_by(dataset_id=existing.id).all()
                existing_paths = {f.path for f in current_files}

                new_file_data = data_catalog_dataset.to_dict(orient="records")
                new_paths = {str(validate_path(r["path"])) for r in new_file_data}
                files_to_add = new_paths - existing_paths

                if files_to_add:
                    new_file_lookup = {}
                    for r in new_file_data:
                        p = str(validate_path(r["path"]))
                        new_file_lookup[p] = {"start_time": r["start_time"], "end_time": r["end_time"]}

                    for file_path in files_to_add:
                        ft = new_file_lookup[file_path]
                        db.session.add(
                            DatasetFile(
                                path=file_path,
                                dataset_id=existing.id,
                                start_time=ft["start_time"],
                                end_time=ft["end_time"],
                            )
                        )

                return DatasetRegistrationResult(
                    dataset=existing,
                    dataset_state=ModelState.UPDATED if files_to_add else None,
                    files_added=list(files_to_add),
                    files_updated=[],
                    files_removed=[],
                    files_unchanged=list(existing_paths & new_paths),
                )

        # Upsert the dataset (create a new dataset or update the metadata)
        dataset_metadata = cast(
            dict[str, Any], data_catalog_dataset[list(self.dataset_specific_metadata)].iloc[0].to_dict()
        )

        # Strip NA/NaN values so we never overwrite existing metadata with empty values.
        # This prevents DRS re-ingestion from regressing columns that were populated
        # during finalisation.
        dataset_metadata = {k: v for k, v in dataset_metadata.items() if not _is_na(v)}

        dataset, dataset_state = db.update_or_create(DatasetModel, defaults=dataset_metadata, slug=slug)
        if dataset_state == ModelState.CREATED:
            logger.info(f"Created new dataset: {dataset}")
        elif dataset_state == ModelState.UPDATED:
            logger.info(f"Updating existing dataset: {dataset}")
        db.session.flush()

        # Initialize result tracking
        files_added: list[str] = []
        files_updated: list[str] = []
        files_removed: list[str] = []
        files_unchanged: list[str] = []

        # Get current files for this dataset
        current_files = db.session.query(DatasetFile).filter_by(dataset_id=dataset.id).all()
        current_file_paths = {f.path: f for f in current_files}

        # Columns to store per file (indexed by path)
        file_meta_cols = [c for c in self.file_specific_metadata if c != "path"]

        # Get new file data from data catalog
        new_file_data = data_catalog_dataset.to_dict(orient="records")
        new_file_lookup = {}
        for dataset_file in new_file_data:
            file_path = str(validate_path(dataset_file["path"]))
            new_file_lookup[file_path] = {c: dataset_file.get(c) for c in file_meta_cols if c in dataset_file}

        new_file_paths = set(new_file_lookup.keys())
        existing_file_paths = set(current_file_paths.keys())

        # Files that exist in the database but are absent from the incoming catalog
        # slice. Removal isn't supported yet (we want to preserve a record of
        # diagnostics that already used the file), but raising here aborts the
        # whole ingest -- including the streaming path, where a dataset can appear
        # in two different chunks (e.g. a CMIP6 mirror that stores the same
        # netCDF file under both an "actual" activity directory and a misfiled
        # parent activity directory). Emit a warning and keep the existing rows
        # in place so subsequent register_dataset calls for the same slug just
        # add their own paths.
        files_to_remove = existing_file_paths - new_file_paths
        if files_to_remove:
            logger.warning(
                f"Dataset {slug}: {len(files_to_remove)} file(s) absent from the current ingest "
                f"are being kept (removal not yet supported): {sorted(files_to_remove)}"
            )

        # Update existing files if any file-specific metadata has changed.
        # Compare via _to_db_str on the incoming value so it matches the on-disk str form
        # (DatasetFile.@validates coerces cftime -> str).
        for file_path, existing_file in current_file_paths.items():
            if file_path in new_file_lookup:
                new_meta = new_file_lookup[file_path]
                changed = any(
                    not _is_na(new_meta.get(c))
                    and hasattr(existing_file, c)
                    and getattr(existing_file, c) != _to_db_str(new_meta[c])
                    for c in file_meta_cols
                    if c in new_meta
                )
                if changed:
                    logger.warning(f"Updating file metadata for {file_path}")
                    for c in file_meta_cols:
                        if c in new_meta and not _is_na(new_meta[c]) and hasattr(existing_file, c):
                            setattr(existing_file, c, new_meta[c])
                    files_updated.append(file_path)
                else:
                    files_unchanged.append(file_path)

        # Add new files (batch operation)
        files_to_add = new_file_paths - existing_file_paths
        if files_to_add:
            files_added = list(files_to_add)
            new_dataset_files = []
            for file_path in files_to_add:
                file_meta = new_file_lookup[file_path]
                # Filter out NA values before passing to DatasetFile constructor
                clean_meta = {c: v for c, v in file_meta.items() if not _is_na(v)}
                new_dataset_files.append(
                    DatasetFile(
                        path=file_path,
                        dataset_id=dataset.id,
                        **clean_meta,
                    )
                )
            db.session.add_all(new_dataset_files)

        # Determine final dataset state
        # If dataset metadata changed, use that state
        # If no metadata changed but files changed, consider it updated
        # If nothing changed, keep the original state (None for existing, CREATED for new)
        final_dataset_state = dataset_state
        if dataset_state is None and (files_added or files_updated or files_removed):
            final_dataset_state = ModelState.UPDATED

        result = DatasetRegistrationResult(
            dataset=dataset,
            dataset_state=final_dataset_state,
            files_added=files_added,
            files_updated=files_updated,
            files_removed=files_removed,
            files_unchanged=files_unchanged,
        )
        change_message = f": ({final_dataset_state.name})" if final_dataset_state else ""
        logger.debug(
            f"Dataset registration complete for {dataset.slug}{change_message} "
            f"{len(files_added)} files added, "
            f"{len(files_updated)} files updated, "
            f"{len(files_removed)} files removed, "
            f"{len(files_unchanged)} files unchanged"
        )

        return result

    def filter_latest_versions(self, catalog: pd.DataFrame) -> pd.DataFrame:
        """
        Filter a data catalog to only include the latest version of each dataset

        Groups by ``dataset_id_metadata`` and keeps only the rows where the version
        matches the maximum version within each group (lexicographic comparison).

        Parameters
        ----------
        catalog
            Data catalog to filter

        Returns
        -------
        :
            Filtered data catalog with only the latest version of each dataset
        """
        if catalog.empty or not self.dataset_id_metadata:
            return catalog

        # Get the latest version for each dataset group
        # Uses transform to compute max version per group, then filters rows matching the max
        # This assumes version can be sorted lexicographically
        max_version_per_group = catalog.groupby(list(self.dataset_id_metadata), sort=False)[
            self.version_metadata
        ].transform("max")

        return catalog[catalog[self.version_metadata] == max_version_per_group]

    def _get_dataset_files(self, db: Database, limit: int | None = None) -> pd.DataFrame:
        dataset_type = self.dataset_cls.__mapper_args__["polymorphic_identity"]

        result = (
            db.session.query(DatasetFile)
            # The join is necessary to be able to order by the dataset columns
            .join(DatasetFile.dataset)
            .where(Dataset.dataset_type == dataset_type)
            # The joinedload is necessary to avoid N+1 queries (one for each dataset)
            # https://docs.sqlalchemy.org/en/14/orm/loading_relationships.html#the-zen-of-joined-eager-loading
            .options(joinedload(DatasetFile.dataset.of_type(self.dataset_cls)))
            .order_by(Dataset.updated_at.desc())
            .limit(limit)
            .all()
        )

        return pd.DataFrame(
            [
                {
                    **{k: getattr(file, k) for k in self.file_specific_metadata},
                    **{k: getattr(file.dataset, k) for k in self.dataset_specific_metadata},
                    "finalised": file.dataset.finalised,
                }
                for file in result
            ],
            index=[file.dataset.id for file in result],
        )

    def _get_datasets(self, db: Database, limit: int | None = None) -> pd.DataFrame:
        result_datasets = (
            db.session.query(self.dataset_cls).order_by(Dataset.updated_at.desc()).limit(limit).all()
        )

        return pd.DataFrame(
            [{k: getattr(dataset, k) for k in self.dataset_specific_metadata} for dataset in result_datasets],
            index=[file.id for file in result_datasets],
        )

    def load_catalog(
        self, db: Database, include_files: bool = True, limit: int | None = None
    ) -> pd.DataFrame:
        """
        Load the data catalog containing the currently tracked datasets/files from the database

        Iterating over different datasets within the data catalog can be done using a `groupby`
        operation for the `instance_id` column.

        Only the latest version of each dataset is returned.

        The index of the data catalog is the primary key of the dataset.
        This should be maintained during any processing.

        Returns
        -------
        :
            Data catalog containing the metadata for the currently ingested datasets
        """
        with db.session.begin():
            # TODO: Paginate this query to avoid loading all the data at once
            if include_files:
                catalog = self._get_dataset_files(db, limit)
            else:
                catalog = self._get_datasets(db, limit)

        # If there are no datasets, return an empty DataFrame
        if catalog.empty:
            empty = pd.DataFrame(columns=self.dataset_specific_metadata + self.file_specific_metadata)
            return self._finalise_loaded_catalog(empty)

        # Convert start_time/end_time strings from DB to cftime objects
        if "start_time" in catalog.columns:
            cal = catalog["calendar"] if "calendar" in catalog.columns else "standard"
            catalog["start_time"] = parse_cftime_dates(catalog["start_time"], cal)
            catalog["end_time"] = parse_cftime_dates(catalog["end_time"], cal)

        return self._finalise_loaded_catalog(self.filter_latest_versions(catalog))

    def _finalise_loaded_catalog(self, catalog: pd.DataFrame) -> pd.DataFrame:
        """
        Add derived columns and enforce the loaded-catalog invariant.

        Every column listed in :attr:`derived_metadata` must be present after loading,
        so downstream code (e.g. the solver applying data requirement filters) can rely on its existence.
        """
        catalog = self._add_derived_columns(catalog)

        missing = set(self.derived_metadata) - set(catalog.columns)
        if missing:
            raise RuntimeError(
                f"{type(self).__name__} did not produce its declared derived column(s): {sorted(missing)}"
            )
        return catalog

    def _add_derived_columns(self, catalog: pd.DataFrame) -> pd.DataFrame:
        """
        Add derived columns to a loaded data catalog.

        Derived columns are computed from stored metadata rather than persisted in
        the database (e.g. CMIP7's ``branded_variable``).
        They must be reconstructed whenever a catalog is loaded from the database.

        Adapters that declare :attr:`derived_metadata` must override this to populate
        every column listed there; the base implementation is a no-op.

        Parameters
        ----------
        catalog
            Data catalog loaded from the database

        Returns
        -------
        :
            Data catalog with any derived columns added
        """
        return catalog

columns_requiring_finalisation = frozenset() class-attribute instance-attribute

Columns that are not available until the dataset has been finalised.

For adapters that support two-phase parsing (e.g. DRS-only then complete), these columns will contain pd.NA until finalisation opens the files. Filtering or grouping on these columns before finalisation will silently produce incorrect results.

dataset_id_metadata = () class-attribute instance-attribute

The group of metadata columns that are specific to the dataset excluding the version information.

Each unique dataset should have the same values for these columns.

This is generally the columns that describe the slug of a dataset, excluding the version information.

derived_metadata = () class-attribute instance-attribute

Columns that are not stored in the database but are computed from stored metadata.

These are reconstructed by :meth:_add_derived_columns whenever a catalog is loaded, so that a DB-loaded catalog has the same columns as a freshly parsed one (e.g. CMIP7's branded_variable). :meth:load_catalog enforces that every column listed here is present after loading.

slug_column instance-attribute

The column in the data catalog that contains the dataset slug. The dataset slug is a unique identifier for the dataset that includes the version of the dataset. This can be used to group files together that belong to the same dataset.

version_metadata = 'version' class-attribute instance-attribute

The column in the data catalog that contains the version of the dataset.

filter_latest_versions(catalog)

Filter a data catalog to only include the latest version of each dataset

Groups by dataset_id_metadata and keeps only the rows where the version matches the maximum version within each group (lexicographic comparison).

Parameters:

Name	Type	Description	Default
catalog	DataFrame	Data catalog to filter	required

Returns:

Type	Description
DataFrame	Filtered data catalog with only the latest version of each dataset

Source code in packages/climate-ref/src/climate_ref/datasets/base.py

def filter_latest_versions(self, catalog: pd.DataFrame) -> pd.DataFrame:
    """
    Filter a data catalog to only include the latest version of each dataset

    Groups by ``dataset_id_metadata`` and keeps only the rows where the version
    matches the maximum version within each group (lexicographic comparison).

    Parameters
    ----------
    catalog
        Data catalog to filter

    Returns
    -------
    :
        Filtered data catalog with only the latest version of each dataset
    """
    if catalog.empty or not self.dataset_id_metadata:
        return catalog

    # Get the latest version for each dataset group
    # Uses transform to compute max version per group, then filters rows matching the max
    # This assumes version can be sorted lexicographically
    max_version_per_group = catalog.groupby(list(self.dataset_id_metadata), sort=False)[
        self.version_metadata
    ].transform("max")

    return catalog[catalog[self.version_metadata] == max_version_per_group]

find_local_datasets(file_or_directory)

Generate a data catalog from the specified file or directory

This data catalog should contain all the metadata needed by the database. The index of the data catalog should be the dataset slug.

Source code in packages/climate-ref/src/climate_ref/datasets/base.py

def find_local_datasets(self, file_or_directory: Path) -> pd.DataFrame:
    """
    Generate a data catalog from the specified file or directory

    This data catalog should contain all the metadata needed by the database.
    The index of the data catalog should be the dataset slug.
    """
    ...

load_catalog(db, include_files=True, limit=None)

Load the data catalog containing the currently tracked datasets/files from the database

Iterating over different datasets within the data catalog can be done using a groupby operation for the instance_id column.

Only the latest version of each dataset is returned.

The index of the data catalog is the primary key of the dataset. This should be maintained during any processing.

Returns:

Type	Description
DataFrame	Data catalog containing the metadata for the currently ingested datasets

Source code in packages/climate-ref/src/climate_ref/datasets/base.py

def load_catalog(
    self, db: Database, include_files: bool = True, limit: int | None = None
) -> pd.DataFrame:
    """
    Load the data catalog containing the currently tracked datasets/files from the database

    Iterating over different datasets within the data catalog can be done using a `groupby`
    operation for the `instance_id` column.

    Only the latest version of each dataset is returned.

    The index of the data catalog is the primary key of the dataset.
    This should be maintained during any processing.

    Returns
    -------
    :
        Data catalog containing the metadata for the currently ingested datasets
    """
    with db.session.begin():
        # TODO: Paginate this query to avoid loading all the data at once
        if include_files:
            catalog = self._get_dataset_files(db, limit)
        else:
            catalog = self._get_datasets(db, limit)

    # If there are no datasets, return an empty DataFrame
    if catalog.empty:
        empty = pd.DataFrame(columns=self.dataset_specific_metadata + self.file_specific_metadata)
        return self._finalise_loaded_catalog(empty)

    # Convert start_time/end_time strings from DB to cftime objects
    if "start_time" in catalog.columns:
        cal = catalog["calendar"] if "calendar" in catalog.columns else "standard"
        catalog["start_time"] = parse_cftime_dates(catalog["start_time"], cal)
        catalog["end_time"] = parse_cftime_dates(catalog["end_time"], cal)

    return self._finalise_loaded_catalog(self.filter_latest_versions(catalog))

pretty_subset(data_catalog)

Get a subset of the data_catalog to pretty print

Parameters:

Name	Type	Description	Default
data_catalog	DataFrame	Data catalog to subset	required

Returns:

Type	Description
DataFrame	Subset of the data catalog to pretty print

Source code in packages/climate-ref/src/climate_ref/datasets/base.py

def pretty_subset(self, data_catalog: pd.DataFrame) -> pd.DataFrame:
    """
    Get a subset of the data_catalog to pretty print

    Parameters
    ----------
    data_catalog
        Data catalog to subset

    Returns
    -------
    :
        Subset of the data catalog to pretty print

    """
    return data_catalog[
        [
            *self.dataset_id_metadata,
            self.version_metadata,
        ]
    ]

register_dataset(db, data_catalog_dataset)

Register a dataset in the database using the data catalog

This assumes that the data catalog has already been validated with validate_data_catalog to ensure that the dataset-specific metadata is consistent across all files in the dataset.

Parameters:

Name	Type	Description	Default
db	Database	Database instance	required
data_catalog_dataset	DataFrame	A subset of the data catalog containing the metadata for a single dataset	required

Raises:

Type	Description
RefException	If the data catalog contains validation errors that should have been caught by validate_data_catalog (i.e. multiple unique slugs in slug_column).

Returns:

Type	Description
DatasetRegistrationResult	Registration result with dataset and file change information

Source code in packages/climate-ref/src/climate_ref/datasets/base.py

def register_dataset(  # noqa: PLR0912, PLR0915
    self, db: Database, data_catalog_dataset: pd.DataFrame
) -> DatasetRegistrationResult:
    """
    Register a dataset in the database using the data catalog

    This assumes that the data catalog has already been validated with `validate_data_catalog`
    to ensure that the dataset-specific metadata is consistent across all files in the dataset.

    Parameters
    ----------
    db
        Database instance
    data_catalog_dataset
        A subset of the data catalog containing the metadata for a single dataset


    Raises
    ------
    RefException
        If the data catalog contains validation errors that should have been caught by
            `validate_data_catalog` (i.e. multiple unique slugs in `slug_column`).


    Returns
    -------
    :
        Registration result with dataset and file change information
    """
    DatasetModel = self.dataset_cls

    unique_slugs = data_catalog_dataset[self.slug_column].unique()
    if len(unique_slugs) != 1:
        raise RefException(f"Found multiple datasets in the same directory: {unique_slugs}")
    slug = unique_slugs[0]

    # Callers are responsible for validating the catalog with  ``validate_data_catalog`` before invoking.
    # This is a strict subset of ``validate_data_catalog`` to catch skipping upstream validation.
    slice_meta = data_catalog_dataset[list(self.dataset_specific_metadata)]
    if (slice_meta.nunique(dropna=False) > 1).any():
        raise RefException(
            f"Dataset {slug} has inconsistent dataset-specific metadata; "
            "callers must pre-validate the catalog with validate_data_catalog."
        )

    # Check if the incoming data is unfinalised (DRS parser) and the dataset
    # already exists as finalised.  In that case, skip the entire update to
    # avoid regressing metadata that was populated during finalisation.
    incoming_finalised = data_catalog_dataset.get("finalised")
    is_unfinalised_ingest = incoming_finalised is not None and not incoming_finalised.iloc[0]

    if is_unfinalised_ingest:
        existing = db.session.query(DatasetModel).filter_by(slug=slug).first()
        if existing and existing.finalised:
            logger.debug(f"Skipping already-finalised dataset: {slug}")
            current_files = db.session.query(DatasetFile).filter_by(dataset_id=existing.id).all()
            existing_paths = {f.path for f in current_files}

            new_file_data = data_catalog_dataset.to_dict(orient="records")
            new_paths = {str(validate_path(r["path"])) for r in new_file_data}
            files_to_add = new_paths - existing_paths

            if files_to_add:
                new_file_lookup = {}
                for r in new_file_data:
                    p = str(validate_path(r["path"]))
                    new_file_lookup[p] = {"start_time": r["start_time"], "end_time": r["end_time"]}

                for file_path in files_to_add:
                    ft = new_file_lookup[file_path]
                    db.session.add(
                        DatasetFile(
                            path=file_path,
                            dataset_id=existing.id,
                            start_time=ft["start_time"],
                            end_time=ft["end_time"],
                        )
                    )

            return DatasetRegistrationResult(
                dataset=existing,
                dataset_state=ModelState.UPDATED if files_to_add else None,
                files_added=list(files_to_add),
                files_updated=[],
                files_removed=[],
                files_unchanged=list(existing_paths & new_paths),
            )

    # Upsert the dataset (create a new dataset or update the metadata)
    dataset_metadata = cast(
        dict[str, Any], data_catalog_dataset[list(self.dataset_specific_metadata)].iloc[0].to_dict()
    )

    # Strip NA/NaN values so we never overwrite existing metadata with empty values.
    # This prevents DRS re-ingestion from regressing columns that were populated
    # during finalisation.
    dataset_metadata = {k: v for k, v in dataset_metadata.items() if not _is_na(v)}

    dataset, dataset_state = db.update_or_create(DatasetModel, defaults=dataset_metadata, slug=slug)
    if dataset_state == ModelState.CREATED:
        logger.info(f"Created new dataset: {dataset}")
    elif dataset_state == ModelState.UPDATED:
        logger.info(f"Updating existing dataset: {dataset}")
    db.session.flush()

    # Initialize result tracking
    files_added: list[str] = []
    files_updated: list[str] = []
    files_removed: list[str] = []
    files_unchanged: list[str] = []

    # Get current files for this dataset
    current_files = db.session.query(DatasetFile).filter_by(dataset_id=dataset.id).all()
    current_file_paths = {f.path: f for f in current_files}

    # Columns to store per file (indexed by path)
    file_meta_cols = [c for c in self.file_specific_metadata if c != "path"]

    # Get new file data from data catalog
    new_file_data = data_catalog_dataset.to_dict(orient="records")
    new_file_lookup = {}
    for dataset_file in new_file_data:
        file_path = str(validate_path(dataset_file["path"]))
        new_file_lookup[file_path] = {c: dataset_file.get(c) for c in file_meta_cols if c in dataset_file}

    new_file_paths = set(new_file_lookup.keys())
    existing_file_paths = set(current_file_paths.keys())

    # Files that exist in the database but are absent from the incoming catalog
    # slice. Removal isn't supported yet (we want to preserve a record of
    # diagnostics that already used the file), but raising here aborts the
    # whole ingest -- including the streaming path, where a dataset can appear
    # in two different chunks (e.g. a CMIP6 mirror that stores the same
    # netCDF file under both an "actual" activity directory and a misfiled
    # parent activity directory). Emit a warning and keep the existing rows
    # in place so subsequent register_dataset calls for the same slug just
    # add their own paths.
    files_to_remove = existing_file_paths - new_file_paths
    if files_to_remove:
        logger.warning(
            f"Dataset {slug}: {len(files_to_remove)} file(s) absent from the current ingest "
            f"are being kept (removal not yet supported): {sorted(files_to_remove)}"
        )

    # Update existing files if any file-specific metadata has changed.
    # Compare via _to_db_str on the incoming value so it matches the on-disk str form
    # (DatasetFile.@validates coerces cftime -> str).
    for file_path, existing_file in current_file_paths.items():
        if file_path in new_file_lookup:
            new_meta = new_file_lookup[file_path]
            changed = any(
                not _is_na(new_meta.get(c))
                and hasattr(existing_file, c)
                and getattr(existing_file, c) != _to_db_str(new_meta[c])
                for c in file_meta_cols
                if c in new_meta
            )
            if changed:
                logger.warning(f"Updating file metadata for {file_path}")
                for c in file_meta_cols:
                    if c in new_meta and not _is_na(new_meta[c]) and hasattr(existing_file, c):
                        setattr(existing_file, c, new_meta[c])
                files_updated.append(file_path)
            else:
                files_unchanged.append(file_path)

    # Add new files (batch operation)
    files_to_add = new_file_paths - existing_file_paths
    if files_to_add:
        files_added = list(files_to_add)
        new_dataset_files = []
        for file_path in files_to_add:
            file_meta = new_file_lookup[file_path]
            # Filter out NA values before passing to DatasetFile constructor
            clean_meta = {c: v for c, v in file_meta.items() if not _is_na(v)}
            new_dataset_files.append(
                DatasetFile(
                    path=file_path,
                    dataset_id=dataset.id,
                    **clean_meta,
                )
            )
        db.session.add_all(new_dataset_files)

    # Determine final dataset state
    # If dataset metadata changed, use that state
    # If no metadata changed but files changed, consider it updated
    # If nothing changed, keep the original state (None for existing, CREATED for new)
    final_dataset_state = dataset_state
    if dataset_state is None and (files_added or files_updated or files_removed):
        final_dataset_state = ModelState.UPDATED

    result = DatasetRegistrationResult(
        dataset=dataset,
        dataset_state=final_dataset_state,
        files_added=files_added,
        files_updated=files_updated,
        files_removed=files_removed,
        files_unchanged=files_unchanged,
    )
    change_message = f": ({final_dataset_state.name})" if final_dataset_state else ""
    logger.debug(
        f"Dataset registration complete for {dataset.slug}{change_message} "
        f"{len(files_added)} files added, "
        f"{len(files_updated)} files updated, "
        f"{len(files_removed)} files removed, "
        f"{len(files_unchanged)} files unchanged"
    )

    return result

validate_data_catalog(data_catalog, skip_invalid=False)

Validate a data catalog

Parameters:

Name	Type	Description	Default
data_catalog	DataFrame	Data catalog to validate	required
skip_invalid	bool	If True, ignore datasets with invalid metadata and remove them from the resulting data catalog.	False

Raises:

Type	Description
ValueError	If skip_invalid is False (default) and the data catalog contains validation errors.

Returns:

Type	Description
DataFrame	Validated data catalog

Source code in packages/climate-ref/src/climate_ref/datasets/base.py

def validate_data_catalog(self, data_catalog: pd.DataFrame, skip_invalid: bool = False) -> pd.DataFrame:
    """
    Validate a data catalog

    Parameters
    ----------
    data_catalog
        Data catalog to validate
    skip_invalid
        If True, ignore datasets with invalid metadata and remove them from the resulting data catalog.

    Raises
    ------
    ValueError
        If `skip_invalid` is False (default) and the data catalog contains validation errors.

    Returns
    -------
    :
        Validated data catalog
    """
    # Check if the data catalog contains the required columns
    missing_columns = set(self.dataset_specific_metadata + self.file_specific_metadata) - set(
        data_catalog.columns
    )
    if missing_columns:
        raise ValueError(f"Data catalog is missing required columns: {missing_columns}")

    # Verify that the dataset specific columns don't vary by dataset by counting the unique values
    # for each dataset and checking if there are any that have more than one unique value.
    unique_metadata = (
        data_catalog[list(self.dataset_specific_metadata)].groupby(self.slug_column).nunique()
    )
    if unique_metadata.gt(1).any(axis=1).any():
        _log_duplicate_metadata(data_catalog, unique_metadata, self.slug_column)

        if skip_invalid:
            data_catalog = data_catalog[
                ~data_catalog[self.slug_column].isin(
                    unique_metadata[unique_metadata.gt(1).any(axis=1)].index
                )
            ]
        else:
            raise ValueError("Dataset specific metadata varies by dataset")

    return data_catalog

IngestionStats

Statistics from ingesting datasets into the database

Source code in packages/climate-ref/src/climate_ref/datasets/__init__.py

@define
class IngestionStats:
    """
    Statistics from ingesting datasets into the database
    """

    datasets_created: int = 0
    datasets_updated: int = 0
    datasets_unchanged: int = 0
    files_added: int = 0
    files_updated: int = 0
    files_removed: int = 0
    files_unchanged: int = 0

    def log_summary(self, prefix: str = "") -> None:
        """Log a summary of the ingestion statistics."""
        prefix_str = f"{prefix} " if prefix else ""
        logger.info(
            f"{prefix_str}Datasets: {self.datasets_created}/{self.datasets_updated}/{self.datasets_unchanged}"
            " (created/updated/unchanged), "
            f"Files: {self.files_added}/{self.files_updated}/{self.files_removed}/{self.files_unchanged}"
            " (created/updated/removed/unchanged)"
        )

    def __iadd__(self, other: "IngestionStats") -> "IngestionStats":
        """Accumulate counts in place from another :class:`IngestionStats`."""
        self.datasets_created += other.datasets_created
        self.datasets_updated += other.datasets_updated
        self.datasets_unchanged += other.datasets_unchanged
        self.files_added += other.files_added
        self.files_updated += other.files_updated
        self.files_removed += other.files_removed
        self.files_unchanged += other.files_unchanged
        return self

__iadd__(other)

Accumulate counts in place from another :class:IngestionStats.

Source code in packages/climate-ref/src/climate_ref/datasets/__init__.py

def __iadd__(self, other: "IngestionStats") -> "IngestionStats":
    """Accumulate counts in place from another :class:`IngestionStats`."""
    self.datasets_created += other.datasets_created
    self.datasets_updated += other.datasets_updated
    self.datasets_unchanged += other.datasets_unchanged
    self.files_added += other.files_added
    self.files_updated += other.files_updated
    self.files_removed += other.files_removed
    self.files_unchanged += other.files_unchanged
    return self

log_summary(prefix='')

Log a summary of the ingestion statistics.

Source code in packages/climate-ref/src/climate_ref/datasets/__init__.py

def log_summary(self, prefix: str = "") -> None:
    """Log a summary of the ingestion statistics."""
    prefix_str = f"{prefix} " if prefix else ""
    logger.info(
        f"{prefix_str}Datasets: {self.datasets_created}/{self.datasets_updated}/{self.datasets_unchanged}"
        " (created/updated/unchanged), "
        f"Files: {self.files_added}/{self.files_updated}/{self.files_removed}/{self.files_unchanged}"
        " (created/updated/removed/unchanged)"
    )

Obs4MIPsDatasetAdapter

Bases: DatasetAdapter

Adapter for obs4MIPs datasets

Source code in packages/climate-ref/src/climate_ref/datasets/obs4mips.py

class Obs4MIPsDatasetAdapter(DatasetAdapter):
    """
    Adapter for obs4MIPs datasets
    """

    dataset_cls: type[Dataset] = Obs4MIPsDataset
    slug_column = "instance_id"

    dataset_specific_metadata = (
        "activity_id",
        "finalised",
        "frequency",
        "grid",
        "grid_label",
        "institution_id",
        "nominal_resolution",
        "product",
        "realm",
        "source_id",
        "source_type",
        "variable_id",
        "variant_label",
        "long_name",
        "units",
        "version",
        "vertical_levels",
        "source_version_number",
        slug_column,
    )

    file_specific_metadata = ("start_time", "end_time", "path")
    version_metadata = "version"
    # See ODS2.5 at https://doi.org/10.5281/zenodo.11500474 under "Directory structure template"
    dataset_id_metadata = (
        "activity_id",
        "institution_id",
        "source_id",
        "frequency",
        "variable_id",
        "nominal_resolution",
        "grid_label",
    )

    def __init__(self, n_jobs: int = 1):
        self.n_jobs = n_jobs

    def find_local_datasets(self, file_or_directory: Path) -> pd.DataFrame:
        """
        Generate a data catalog from the specified file or directory

        Each dataset may contain multiple files, which are represented as rows in the data catalog.
        Each dataset has a unique identifier, which is in `slug_column`.

        Parameters
        ----------
        file_or_directory
            File or directory containing the datasets

        Returns
        -------
        :
            Data catalog containing the metadata for the dataset
        """
        datasets = build_catalog(
            paths=[str(file_or_directory)],
            parsing_func=parse_obs4mips,
            include_patterns=["*.nc"],
            depth=10,
            n_jobs=self.n_jobs,
        )
        if datasets.empty:
            logger.error("No datasets found")
            raise ValueError("No obs4MIPs-compliant datasets found")

        # Convert the start_time and end_time columns to cftime objects
        datasets["start_time"] = parse_cftime_dates(datasets["start_time"])
        datasets["end_time"] = parse_cftime_dates(datasets["end_time"])

        drs_items = [
            *self.dataset_id_metadata,
            self.version_metadata,
        ]

        def _transform(item: str, value: Any) -> str:
            return str(value).replace(" ", "") if item == "nominal_resolution" else str(value)

        datasets = build_instance_id(datasets, drs_items, prefix="obs4MIPs", transform=_transform)
        datasets["finalised"] = True
        return datasets

find_local_datasets(file_or_directory)

Generate a data catalog from the specified file or directory

Each dataset may contain multiple files, which are represented as rows in the data catalog. Each dataset has a unique identifier, which is in slug_column.

Parameters:

Name	Type	Description	Default
file_or_directory	Path	File or directory containing the datasets	required

Returns:

Type	Description
DataFrame	Data catalog containing the metadata for the dataset

Source code in packages/climate-ref/src/climate_ref/datasets/obs4mips.py

def find_local_datasets(self, file_or_directory: Path) -> pd.DataFrame:
    """
    Generate a data catalog from the specified file or directory

    Each dataset may contain multiple files, which are represented as rows in the data catalog.
    Each dataset has a unique identifier, which is in `slug_column`.

    Parameters
    ----------
    file_or_directory
        File or directory containing the datasets

    Returns
    -------
    :
        Data catalog containing the metadata for the dataset
    """
    datasets = build_catalog(
        paths=[str(file_or_directory)],
        parsing_func=parse_obs4mips,
        include_patterns=["*.nc"],
        depth=10,
        n_jobs=self.n_jobs,
    )
    if datasets.empty:
        logger.error("No datasets found")
        raise ValueError("No obs4MIPs-compliant datasets found")

    # Convert the start_time and end_time columns to cftime objects
    datasets["start_time"] = parse_cftime_dates(datasets["start_time"])
    datasets["end_time"] = parse_cftime_dates(datasets["end_time"])

    drs_items = [
        *self.dataset_id_metadata,
        self.version_metadata,
    ]

    def _transform(item: str, value: Any) -> str:
        return str(value).replace(" ", "") if item == "nominal_resolution" else str(value)

    datasets = build_instance_id(datasets, drs_items, prefix="obs4MIPs", transform=_transform)
    datasets["finalised"] = True
    return datasets

PMPClimatologyDatasetAdapter

Bases: Obs4MIPsDatasetAdapter

Adapter for climatology datasets post-processed from obs4MIPs datasets by PMP.

These data look like obs4MIPs datasets and are ingested in the same way, but are treated separately as they may have the same metadata as the obs4MIPs datasets.

Source code in packages/climate-ref/src/climate_ref/datasets/pmp_climatology.py

class PMPClimatologyDatasetAdapter(Obs4MIPsDatasetAdapter):
    """
    Adapter for climatology datasets post-processed from obs4MIPs datasets by PMP.

    These data look like obs4MIPs datasets and are ingested in the same way, but
    are treated separately as they may have the same metadata as the obs4MIPs datasets.
    """

    dataset_cls = PMPClimatologyDataset

get_dataset_adapter(source_type, **kwargs)

Get the appropriate adapter for the specified source type

Parameters:

Name	Type	Description	Default
source_type	str	Type of source dataset	required

Returns:

Type	Description
DatasetAdapter	DatasetAdapter instance

Source code in packages/climate-ref/src/climate_ref/datasets/__init__.py

def get_dataset_adapter(source_type: str, **kwargs: Any) -> DatasetAdapter:
    """
    Get the appropriate adapter for the specified source type

    Parameters
    ----------
    source_type
        Type of source dataset

    Returns
    -------
    :
        DatasetAdapter instance
    """
    if source_type.lower() == SourceDatasetType.CMIP6.value:
        return CMIP6DatasetAdapter(**kwargs)
    elif source_type.lower() == SourceDatasetType.CMIP7.value:
        return CMIP7DatasetAdapter(**kwargs)
    elif source_type.lower() == SourceDatasetType.obs4MIPs.value.lower():
        return Obs4MIPsDatasetAdapter(**kwargs)
    elif source_type.lower() == SourceDatasetType.PMPClimatology.value.lower():
        return PMPClimatologyDatasetAdapter(**kwargs)
    else:
        raise ValueError(f"Unknown source type: {source_type}")

get_slug_column(source_type)

Get the slug column name for a source dataset type.

Parameters:

Name	Type	Description	Default
source_type	SourceDatasetType | str	Source dataset type (enum or string value)	required

Returns:

Type	Description
str	The slug column name for the given source type

Source code in packages/climate-ref/src/climate_ref/datasets/__init__.py

def get_slug_column(source_type: SourceDatasetType | str) -> str:
    """
    Get the slug column name for a source dataset type.

    Parameters
    ----------
    source_type
        Source dataset type (enum or string value)

    Returns
    -------
    :
        The slug column name for the given source type
    """
    if isinstance(source_type, str):
        source_type = SourceDatasetType(source_type)
    return SLUG_COLUMN_BY_SOURCE_TYPE[source_type]

ingest_datasets(adapter, directory, db, *, data_catalog=None, skip_invalid=True, chunk_size=None)

Ingest datasets from a directory into the database.

This is the common ingestion logic shared between the CLI ingest command and provider setup.

Parameters:

Name	Type	Description	Default
adapter	DatasetAdapter	The dataset adapter to use for parsing and registering datasets	required
directory	Path | None	Directory containing the datasets to ingest. Can be None if data_catalog is provided.	required
db	Database	Database instance	required
data_catalog	DataFrame | None	Optional pre-validated data catalog. If provided, directory is ignored and the catalog is used directly. This avoids redundant find/validate operations. When supplied, chunk_size is ignored because the catalog is already fully materialised.	None
skip_invalid	bool	If True, skip datasets that fail validation (default True)	True
chunk_size	int | None	When provided and data_catalog is None, stream the directory in batches of chunk_size files so peak memory is bounded regardless of how many files live under directory. Requires the adapter to implement iter_local_datasets.	None

Returns:

Type	Description
IngestionStats	Statistics about the ingestion (created/updated/unchanged counts)

Raises:

Type	Description
ValueError	If no valid datasets are found in the directory

Source code in packages/climate-ref/src/climate_ref/datasets/__init__.py

def ingest_datasets(  # noqa: PLR0913
    adapter: DatasetAdapter,
    directory: Path | None,
    db: Database,
    *,
    data_catalog: pd.DataFrame | None = None,
    skip_invalid: bool = True,
    chunk_size: int | None = None,
) -> IngestionStats:
    """
    Ingest datasets from a directory into the database.

    This is the common ingestion logic shared between the CLI ingest command
    and provider setup.

    Parameters
    ----------
    adapter
        The dataset adapter to use for parsing and registering datasets
    directory
        Directory containing the datasets to ingest. Can be None if data_catalog is provided.
    db
        Database instance
    data_catalog
        Optional pre-validated data catalog.

        If provided, directory is ignored and the catalog is used directly.
        This avoids redundant find/validate operations.
        When supplied, ``chunk_size`` is ignored because the catalog is already fully materialised.
    skip_invalid
        If True, skip datasets that fail validation (default True)
    chunk_size
        When provided and ``data_catalog`` is None,
        stream the directory in batches of ``chunk_size`` files so peak memory is bounded regardless
        of how many files live under ``directory``.
        Requires the adapter to implement ``iter_local_datasets``.

    Returns
    -------
    :
        Statistics about the ingestion (created/updated/unchanged counts)

    Raises
    ------
    ValueError
        If no valid datasets are found in the directory
    """
    if data_catalog is not None:
        return _ingest_catalog(adapter, db, data_catalog)

    if directory is None:
        raise ValueError("Either directory or data_catalog must be provided")

    if not directory.exists():
        raise ValueError(f"Directory {directory} does not exist")

    # Check for .nc files
    if not any(directory.rglob("*.nc")):
        raise ValueError(f"No .nc files found in {directory}")

    if chunk_size is not None:
        if chunk_size < 1:
            raise ValueError(f"chunk_size must be >= 1, got {chunk_size}")
        iter_fn = getattr(adapter, "iter_local_datasets", None)
        if iter_fn is None:
            raise ValueError(
                f"Adapter {type(adapter).__name__} does not support streaming ingest "
                "(missing iter_local_datasets); omit chunk_size to use whole-catalog mode."
            )

        stats = IngestionStats()
        total_files = 0
        total_datasets = 0
        emitted = False
        for raw_chunk in iter_fn(directory, chunk_size=chunk_size):
            validated_chunk = adapter.validate_data_catalog(raw_chunk, skip_invalid=skip_invalid)
            if validated_chunk.empty:
                continue
            emitted = True
            total_files += len(validated_chunk)
            total_datasets += validated_chunk[adapter.slug_column].nunique()
            stats += _ingest_catalog(adapter, db, validated_chunk)
            # Drop chunk references so the per-chunk pandas memory can be
            # reclaimed before the next chunk is parsed.
            del raw_chunk, validated_chunk

        if not emitted:
            raise ValueError(f"No valid datasets found in {directory}")

        logger.info(f"Ingested {total_files} files across approximately {total_datasets} datasets (streamed)")
        return stats

    data_catalog = adapter.find_local_datasets(directory)
    data_catalog = adapter.validate_data_catalog(data_catalog, skip_invalid=skip_invalid)

    if data_catalog.empty:
        raise ValueError(f"No valid datasets found in {directory}")

    logger.info(
        f"Found {len(data_catalog)} files for {len(data_catalog[adapter.slug_column].unique())} datasets"
    )

    return _ingest_catalog(adapter, db, data_catalog)

sub-packages

Sub-package	Description
base
catalog_builder	Catalog builder for discovering and parsing dataset files into a DataFrame
cmip6
cmip6_parsers	CMIP6 parser functions for extracting metadata from netCDF files
cmip7	CMIP7 Dataset Adapter
cmip7_parsers	CMIP7 parser functions for extracting metadata from netCDF files
mixins	Mixins for dataset adapters that support lazy finalization.
netcdf_utils	Shared utilities for reading NetCDF metadata using netCDF4 directly.
obs4mips
pmp_climatology
utils	Shared utility functions for dataset adapters