climate_ref.datasets.obs4mips

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"

    activity_id = "obs4MIPs"
    """
    The collection this adapter ingests into, whatever the file itself claims.

    The obs4REF collection republishes obs4MIPs files unchanged,
    so the file attribute cannot tell the two collections apart.
    """

    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"
    # The facets ESGF publishes obs4MIPs datasets under, so that a dataset's ``instance_id``
    # is the same one the ESGF index knows it by and can be searched for directly.
    # The ODS2.5 directory template (https://doi.org/10.5281/zenodo.11500474) additionally
    # nests data under ``nominal_resolution``, which is descriptive rather than identifying.
    dataset_id_metadata = (
        "activity_id",
        "institution_id",
        "source_id",
        "frequency",
        "variable_id",
        "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"],
            n_jobs=self.n_jobs,
        )
        if datasets.empty:
            logger.error("No datasets found")
            raise ValueError("No obs4MIPs-compliant datasets found")

        self._warn_if_misfiled(datasets)
        datasets["activity_id"] = self.activity_id

        # 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,
        ]

        datasets = build_instance_id(datasets, drs_items)
        datasets["finalised"] = True
        return datasets

    def _warn_if_misfiled(self, datasets: pd.DataFrame) -> None:
        """
        Warn when the files look like they belong to the other collection.

        The registry republishes obs4MIPs files unchanged, so an ``obs4REF`` directory
        or activity id is only a hint. The files are ingested either way.
        """
        other = "obs4REF" if self.activity_id == "obs4MIPs" else "obs4MIPs"
        misfiled = in_collection_directory(datasets["path"], other)
        if other == "obs4REF":
            # An obs4MIPs attribute on an obs4REF file is expected, the reverse is not.
            misfiled |= datasets["activity_id"] == other
        count = int(misfiled.sum())
        if count:
            logger.warning(
                f"{count} of {len(datasets)} files look like {other} data but are being ingested as "
                f"{self.activity_id}. Use `--source-type {other.lower()}` if that is not intended."
            )

activity_id = 'obs4MIPs' class-attribute instance-attribute

The collection this adapter ingests into, whatever the file itself claims.

The obs4REF collection republishes obs4MIPs files unchanged, so the file attribute cannot tell the two collections apart.

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"],
        n_jobs=self.n_jobs,
    )
    if datasets.empty:
        logger.error("No datasets found")
        raise ValueError("No obs4MIPs-compliant datasets found")

    self._warn_if_misfiled(datasets)
    datasets["activity_id"] = self.activity_id

    # 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,
    ]

    datasets = build_instance_id(datasets, drs_items)
    datasets["finalised"] = True
    return datasets

Obs4REFDatasetAdapter

Bases: Obs4MIPsDatasetAdapter

Adapter for obs4REF datasets

obs4REF is the REF-curated collection of observational data. It shares the obs4MIPs metadata conventions and parser, but is ingested as its own dataset type so it is never mistaken for published obs4MIPs data.

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

class Obs4REFDatasetAdapter(Obs4MIPsDatasetAdapter):
    """
    Adapter for obs4REF datasets

    obs4REF is the REF-curated collection of observational data.
    It shares the obs4MIPs metadata conventions and parser,
    but is ingested as its own dataset type so it is never mistaken for published obs4MIPs data.
    """

    dataset_cls: type[Dataset] = Obs4REFDataset
    activity_id = "obs4REF"

in_collection_directory(paths, activity_id)

Whether each file sits under the named collection's directory.

Parameters:

Name	Type	Description	Default
paths	Series	File paths from a catalog.	required
activity_id	str	The collection to look for, obs4MIPs or obs4REF.	required

Returns:

Type	Description
Series	A boolean mask over paths.

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

def in_collection_directory(paths: pd.Series, activity_id: str) -> pd.Series:
    """
    Whether each file sits under the named collection's directory.

    Parameters
    ----------
    paths
        File paths from a catalog.
    activity_id
        The collection to look for, ``obs4MIPs`` or ``obs4REF``.

    Returns
    -------
    :
        A boolean mask over ``paths``.
    """
    return paths.astype(str).str.replace("\\", "/", regex=False).str.contains(f"/{activity_id}/", regex=False)

parse_obs4mips(file, **kwargs)

Parser for obs4MIPs and obs4REF files

obs4REF files follow the obs4MIPs metadata conventions, so the same parser reads both. The adapter that called the parser decides which collection the file belongs to.

Parameters:

Name	Type	Description	Default
file	str	File to parse	required
kwargs	Any	Additional keyword arguments (not used, but required for protocol compatibility)	{}

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

def parse_obs4mips(file: str, **kwargs: Any) -> dict[str, Any]:
    """
    Parser for obs4MIPs and obs4REF files

    obs4REF files follow the obs4MIPs metadata conventions, so the same parser reads both.
    The adapter that called the parser decides which collection the file belongs to.

    Parameters
    ----------
    file
        File to parse
    kwargs
        Additional keyword arguments (not used, but required for protocol compatibility)
    """
    keys = sorted(
        list(
            {
                "activity_id",
                "frequency",
                "grid",
                "grid_label",
                "institution_id",
                "nominal_resolution",
                "realm",
                "product",
                "source_id",
                "source_type",
                "variable_id",
                "variant_label",
                "source_version_number",
            }
        )
    )

    try:
        with netCDF4.Dataset(file, "r") as ds:
            activity_id = getattr(ds, "activity_id", "")
            if activity_id not in ("obs4MIPs", "obs4REF"):
                traceback_message = f"{file} is not an obs4MIPs or obs4REF dataset"
                raise TypeError(traceback_message)

            global_attrs = read_global_attrs(ds, keys)
            missing_fields = [key for key in keys if global_attrs.get(key) is None]

            if missing_fields:
                traceback_message = str(missing_fields) + " are missing from the file metadata"
                raise AttributeError(traceback_message)
            info = {**global_attrs}

            variable_id = global_attrs["variable_id"]

            if variable_id:
                var_attrs = read_variable_attrs(ds, variable_id, ["long_name", "units"])
                info.update(var_attrs)

            vertical_levels = read_vertical_levels(ds)
            start_time, end_time = read_time_bounds(ds)

            info["vertical_levels"] = vertical_levels
            info["start_time"] = start_time
            info["end_time"] = end_time
            if not (start_time and end_time):
                info["time_range"] = None
            else:
                info["time_range"] = f"{start_time}-{end_time}"
        info["path"] = str(file)
        # Parsing the version like for CMIP6 fails because some obs4REF paths
        # do not include "v" in the version directory name.
        # TODO: fix obs4REF paths
        info["version"] = Path(file).parent.name
        if not info["version"].startswith("v"):
            info["version"] = "v{version}".format(**info)
        return info

    except (TypeError, AttributeError) as err:
        if (len(err.args)) == 1:
            logger.warning(str(err.args[0]))
        else:
            logger.warning(str(err.args))
        return {"INVALID_ASSET": file, "TRACEBACK": str(err)}
    except Exception:
        logger.warning(traceback.format_exc())
        return {"INVALID_ASSET": file, "TRACEBACK": traceback.format_exc()}