diff --git a/changelog/+esmvaltool-reference-datasets.feature.md b/changelog/+esmvaltool-reference-datasets.feature.md new file mode 100644 index 000000000..1c6088e32 --- /dev/null +++ b/changelog/+esmvaltool-reference-datasets.feature.md @@ -0,0 +1,5 @@ +ESMValTool reference (observational/reanalysis) datasets are now ingested into the database under a dedicated `esmvaltool-reference` dataset type during `ref providers setup`, so the data each diagnostic uses can be recorded for provenance and surfaced in the frontend. +Because this data is not CMOR/obs4MIPs compliant, its metadata is parsed from the ESMValTool `OBS`/`OBS6`, `native6` and `obs4MIPs` layout conventions rather than from file global attributes. +ESMValTool diagnostics now declare the reference datasets they use via a `reference_datasets` specification instead of hardcoding them inside recipe construction, giving a single source of truth for provenance. +Each execution now records the reference datasets the diagnostic compared against, so they appear alongside the model datasets in the execution's dataset list. +The near-identical obs4MIPs and PMP climatology dataset models were also consolidated onto a shared mixin while remaining distinct dataset types. diff --git a/packages/climate-ref-core/src/climate_ref_core/diagnostics.py b/packages/climate-ref-core/src/climate_ref_core/diagnostics.py index 05aa8e331..5fc512173 100644 --- a/packages/climate-ref-core/src/climate_ref_core/diagnostics.py +++ b/packages/climate-ref-core/src/climate_ref_core/diagnostics.py @@ -5,7 +5,7 @@ from __future__ import annotations import pathlib -from collections.abc import Iterable, Sequence +from collections.abc import Iterable, Mapping, Sequence from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable from attrs import field, frozen @@ -414,6 +414,29 @@ def apply_filters(self, data_catalog: pd.DataFrame) -> pd.DataFrame: return data_catalog[select] +@frozen +class ReferenceDatasetSelector: + """ + Identifies ingested reference datasets a diagnostic uses, for provenance recording. + + Unlike :class:`DataRequirement`, this does not drive the solver's input selection. It is a + provider-agnostic description of reference (observational/reanalysis) datasets that should be + recorded against each execution of the diagnostic so their use can be tracked and surfaced, + even though the diagnostic references them internally rather than receiving them as inputs. + """ + + source_type: SourceDatasetType + """The dataset type the reference datasets are ingested as (e.g. ``esmvaltool-reference``).""" + + facets: Mapping[str, str] + """ + Metadata facets that must all match (AND) an ingested dataset for it to be linked. + + Keys are columns on the corresponding dataset model (e.g. ``project``, ``source_id``, + ``table_id``). All datasets matching the facets are linked, across their variables/versions. + """ + + @runtime_checkable class AbstractDiagnostic(Protocol): """ @@ -636,6 +659,23 @@ def run(self, definition: ExecutionDefinition, *, capture_regression: bool = Fal # Build the result from the output bundle return self.build_execution_result(definition) + def reference_dataset_selectors(self) -> Sequence[ReferenceDatasetSelector]: + """ + Return the reference datasets to record as provenance for each execution. + + These are datasets the diagnostic compares against but does not receive as solver inputs + (e.g. observational data baked into a recipe). The solver links every ingested dataset that + matches a selector to the execution, so the reference data used can be tracked and surfaced. + + The default returns no selectors; diagnostics with reference data override this. + + Returns + ------- + : + The reference dataset selectors for this diagnostic. + """ + return () + def prepare_regression_output(self, definition: ExecutionDefinition) -> None: """ Normalise native output for deterministic regression capture. diff --git a/packages/climate-ref-core/src/climate_ref_core/source_types.py b/packages/climate-ref-core/src/climate_ref_core/source_types.py index fa3741177..0425fe6ef 100644 --- a/packages/climate-ref-core/src/climate_ref_core/source_types.py +++ b/packages/climate-ref-core/src/climate_ref_core/source_types.py @@ -30,6 +30,7 @@ class SourceDatasetType(enum.Enum): CMIP7 = "cmip7" obs4MIPs = "obs4mips" PMPClimatology = "pmp-climatology" + ESMValToolReference = "esmvaltool-reference" @classmethod @functools.lru_cache(maxsize=1) diff --git a/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/__init__.py b/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/__init__.py index 0a74f033c..9f443316d 100644 --- a/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/__init__.py +++ b/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/__init__.py @@ -5,7 +5,7 @@ from __future__ import annotations from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any import pooch from loguru import logger @@ -65,6 +65,46 @@ def get_data_path(self) -> Path | None: """Get the path where ESMValTool data is cached.""" return resolve_cache_dir("esmvaltool") + def ingest_data(self, config: Config, db: Any) -> None: + """ + Ingest fetched ESMValTool reference data into the database. + + This records the observational/reanalysis datasets bundled with ESMValTool so + their use can be tracked for provenance and surfaced in the frontend. The data is + not CMOR compliant, so it is registered under its own dataset type. + + Note: requires the ``climate-ref`` package. When ``climate-ref-esmvaltool`` is used + standalone the ingestion is skipped with a warning, mirroring the PMP provider. + """ + try: + from climate_ref.datasets import ingest_datasets # noqa: PLC0415 + from climate_ref.datasets.esmvaltool_reference import ( # noqa: PLC0415 + ESMValToolReferenceDatasetAdapter, + ) + except ImportError: + logger.info( + f"Skipping {self.slug} data ingestion: climate-ref package not installed. " + "Run `ref datasets ingest --source-type esmvaltool-reference` manually if needed." + ) + return + + # Reference data is fetched under ``/ESMValTool``; the same + # location build_cmd() configures as the ESMValCore rootpath. + registry = dataset_registry_manager[_DATASETS_REGISTRY_NAME] + data_path = registry.abspath / "ESMValTool" # type: ignore[attr-defined] + if not data_path.exists(): + logger.warning( + f"ESMValTool reference data not found at {data_path}. " + f"Run `ref providers setup --provider {self.slug}` first." + ) + return + + try: + stats = ingest_datasets(ESMValToolReferenceDatasetAdapter(), data_path, db, skip_invalid=True) + stats.log_summary("ESMValTool reference ingestion complete:") + except ValueError as e: + logger.warning(f"No valid ESMValTool reference datasets found: {e}") + # Initialise the diagnostics manager. provider = ESMValToolProvider( diff --git a/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/diagnostics/base.py b/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/diagnostics/base.py index 34cc29706..2b6481de0 100644 --- a/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/diagnostics/base.py +++ b/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/diagnostics/base.py @@ -18,10 +18,12 @@ CommandLineDiagnostic, ExecutionDefinition, ExecutionResult, + ReferenceDatasetSelector, ) from climate_ref_core.metric_values.typing import MetricValueKind, SeriesMetricValue from climate_ref_core.pycmec.metric import CMECMetric, MetricCV from climate_ref_core.pycmec.output import CMECOutput, OutputCV +from climate_ref_esmvaltool.diagnostics.reference import ESMValToolReferenceSpec from climate_ref_esmvaltool.recipe import ( fix_annual_statistics_keep_year, load_recipe, @@ -73,6 +75,16 @@ class ESMValToolDiagnostic(CommandLineDiagnostic): base_recipe: ClassVar[str] + reference_datasets: ClassVar[tuple[ESMValToolReferenceSpec, ...]] = () + """ + Reference (observational/reanalysis) datasets this diagnostic compares against. + + These are the non-CMIP datasets the diagnostic uses; they are declared here as the single + source of truth so they can be recorded for provenance and surfaced in the frontend. + ``update_recipe`` builds the recipe entries from these specs rather than hardcoding them. + Diagnostics whose references have not yet been declared leave this empty. + """ + reconstruction_inputs = (f"executions/{_STABLE_SESSION_NAME}/{_PROVENANCE_GLOB}",) """Raw provenance YAML that :meth:`build_execution_result` re-scans to discover the run's outputs. @@ -83,6 +95,26 @@ class ESMValToolDiagnostic(CommandLineDiagnostic): the native sanitiser rewrites them to ```` and the captured blobs stay portable. """ + def reference_dataset_selectors(self) -> tuple[ReferenceDatasetSelector, ...]: + """ + Map the declared :attr:`reference_datasets` to provenance selectors. + + Each spec resolves to the ingested ESMValTool reference datasets matching its project and + dataset name (and MIP table where declared), across their variables. + """ + selectors = [] + for spec in self.reference_datasets: + facets = {"project": spec.project, "source_id": spec.dataset} + if spec.mip is not None: + facets["table_id"] = spec.mip + selectors.append( + ReferenceDatasetSelector( + source_type=SourceDatasetType.ESMValToolReference, + facets=facets, + ) + ) + return tuple(selectors) + @staticmethod @abstractmethod def update_recipe( diff --git a/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/diagnostics/enso.py b/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/diagnostics/enso.py index 4e775e2e3..93cccc877 100644 --- a/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/diagnostics/enso.py +++ b/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/diagnostics/enso.py @@ -17,9 +17,20 @@ from climate_ref_core.pycmec.output import CMECOutput from climate_ref_core.testing import TestCase, TestDataSpecification from climate_ref_esmvaltool.diagnostics.base import ESMValToolDiagnostic, get_cmip_source_type +from climate_ref_esmvaltool.diagnostics.reference import ESMValToolReferenceSpec from climate_ref_esmvaltool.recipe import dataframe_to_recipe from climate_ref_esmvaltool.types import MetricBundleArgs, OutputBundleArgs, Recipe +# Reference reanalysis for ENSO characteristics; one observational dataset per run. +_TROPFLUX = ESMValToolReferenceSpec( + project="OBS6", + dataset="TROPFLUX", + mip="Omon", + tier=2, + obs_type="reanaly", + version="v1", +) + class ENSOBasicClimatology(ESMValToolDiagnostic): """ @@ -302,6 +313,8 @@ class ENSOCharacteristics(ESMValToolDiagnostic): slug = "enso-characteristics" base_recipe = "ref/recipe_enso_characteristics.yml" + reference_datasets = (_TROPFLUX,) + data_requirements = ( ( DataRequirement( @@ -421,21 +434,11 @@ def update_recipe( # TODO: update the observational data requirement once available on ESGF. # Observations - use only one per run recipe["datasets"].append( - # { - # "dataset": "NOAA-ERSSTv5", - # "version": "v5", - # "project": "OBS6", - # "type": "reanaly", - # "tier": 2, - # } - { - "dataset": "TROPFLUX", - "version": "v1", - "project": "OBS6", - "type": "reanaly", - "tier": 2, - "mip": "Omon", - } + # Alternative observational dataset: + # ESMValToolReferenceSpec( + # project="OBS6", dataset="NOAA-ERSSTv5", tier=2, obs_type="reanaly", version="v5" + # ) + _TROPFLUX.to_recipe_dataset() ) @staticmethod diff --git a/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/diagnostics/reference.py b/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/diagnostics/reference.py new file mode 100644 index 000000000..871a94f7f --- /dev/null +++ b/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/diagnostics/reference.py @@ -0,0 +1,69 @@ +""" +Declarative reference-dataset specifications for ESMValTool diagnostics. + +ESMValTool diagnostics compare model output against reference (observational/reanalysis) +datasets that are not published as CMIP data. Historically each diagnostic hardcoded these +references as raw dictionaries inside ``update_recipe``. :class:`ESMValToolReferenceSpec` +makes them a declared, inspectable property of the diagnostic instead: a single source of +truth that feeds both recipe construction (:meth:`ESMValToolReferenceSpec.to_recipe_dataset`) +and dataset provenance. +""" + +from __future__ import annotations + +from typing import Any + +from attrs import frozen + + +@frozen +class ESMValToolReferenceSpec: + """ + A single reference dataset that an ESMValTool diagnostic depends on. + + The fields mirror the ESMValTool/ESMValCore dataset keys used in a recipe. Optional fields + are omitted from the generated recipe entry when unset, matching how the references were + previously written by hand. + """ + + project: str + """ESMValCore project, e.g. ``OBS``, ``OBS6``, ``native6`` or ``obs4MIPs``.""" + + dataset: str + """Reference dataset name, e.g. ``OSI-450-nh``, ``HadCRUT5``, ``ERA5``.""" + + mip: str | None = None + """MIP table where applicable, e.g. ``OImon``, ``Amon``.""" + + tier: int | None = None + """ESMValTool data tier (accessibility).""" + + obs_type: str | None = None + """ESMValTool observation ``type`` where applicable, e.g. ``reanaly``, ``sat``, ``ground``.""" + + version: str | None = None + """Dataset version, e.g. ``v3``, ``5.0.1.0-analysis``.""" + + supplementary_variables: tuple[dict[str, str], ...] | None = None + """Optional supplementary variables (e.g. cell measures) attached to the recipe entry.""" + + def to_recipe_dataset(self) -> dict[str, Any]: + """ + Build the ESMValTool recipe dataset entry for this reference. + + Keys are emitted in alphabetical order to match the historical hardcoded dictionaries, + so the generated recipe YAML (dumped with ``sort_keys=False``) is byte-for-byte unchanged. + """ + entry: dict[str, Any] = {"dataset": self.dataset} + if self.mip is not None: + entry["mip"] = self.mip + entry["project"] = self.project + if self.supplementary_variables is not None: + entry["supplementary_variables"] = [dict(sv) for sv in self.supplementary_variables] + if self.tier is not None: + entry["tier"] = self.tier + if self.obs_type is not None: + entry["type"] = self.obs_type + if self.version is not None: + entry["version"] = self.version + return entry diff --git a/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/diagnostics/regional_historical_changes.py b/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/diagnostics/regional_historical_changes.py index 68d6941ad..89240fe24 100644 --- a/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/diagnostics/regional_historical_changes.py +++ b/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/diagnostics/regional_historical_changes.py @@ -18,9 +18,28 @@ from climate_ref_core.pycmec.output import CMECOutput from climate_ref_core.testing import TestCase, TestDataSpecification from climate_ref_esmvaltool.diagnostics.base import ESMValToolDiagnostic, fillvalues_to_nan +from climate_ref_esmvaltool.diagnostics.reference import ESMValToolReferenceSpec from climate_ref_esmvaltool.recipe import dataframe_to_recipe from climate_ref_esmvaltool.types import MetricBundleArgs, OutputBundleArgs, Recipe +# Reference datasets not published as CMIP-style obs4MIPs, used by the regional-historical +# diagnostics. Declared once and shared across the diagnostics that use them. +_HADCRUT5 = ESMValToolReferenceSpec( + project="OBS", + dataset="HadCRUT5", + tier=2, + obs_type="ground", + version="5.0.1.0-analysis", +) +_GPCP_V2_3 = ESMValToolReferenceSpec(project="obs4MIPs", dataset="GPCP-V2.3") +_ERA5_NATIVE6 = ESMValToolReferenceSpec( + project="native6", + dataset="ERA5", + tier=3, + obs_type="reanaly", + version="v1", +) + ANNUAL_CYCLE_DIAGNOSTICS = ( ("hus", "specific_humidity_annual_cycle"), ("pr", "precipitation_annual_cycle"), @@ -114,6 +133,8 @@ class RegionalHistoricalAnnualCycle(ESMValToolDiagnostic): slug = "regional-historical-annual-cycle" base_recipe = "ref/recipe_ref_annual_cycle_region.yml" + reference_datasets = (_HADCRUT5, _GPCP_V2_3, _ERA5_NATIVE6) + variables = ( "hus", "pr", @@ -316,30 +337,9 @@ def update_recipe( """Update the recipe.""" # Extra datasets that are not published as obs4MIPs. extra_datasets: dict[str, list[dict[str, str | int]]] = { - "tas": [ - { - "dataset": "HadCRUT5", - "project": "OBS", - "tier": 2, - "type": "ground", - "version": "5.0.1.0-analysis", - }, - ], - "pr": [ - { - "dataset": "GPCP-V2.3", - "project": "obs4MIPs", - }, - ], - "hus": [ - { - "dataset": "ERA5", - "project": "native6", - "tier": 3, - "type": "reanaly", - "version": "v1", - }, - ], + "tas": [_HADCRUT5.to_recipe_dataset()], + "pr": [_GPCP_V2_3.to_recipe_dataset()], + "hus": [_ERA5_NATIVE6.to_recipe_dataset()], } # Remove the unused regions alias. @@ -655,6 +655,8 @@ class RegionalHistoricalTrend(ESMValToolDiagnostic): slug = "regional-historical-trend" base_recipe = "ref/recipe_ref_trend_regions.yml" + reference_datasets = (_HADCRUT5, _GPCP_V2_3, _ERA5_NATIVE6) + variables = ( "hus", "pr", @@ -850,37 +852,9 @@ def update_recipe( """Update the recipe.""" # Extra datasets that are not published as CMIP6-style obs4MIPs. extra_datasets: dict[str, list[dict[str, str | int]]] = { - "tas": [ - { - "dataset": "HadCRUT5", - "project": "OBS", - "tier": 2, - "type": "ground", - "version": "5.0.1.0-analysis", - }, - ], - "pr": [ - { - "dataset": "GPCP-V2.3", - "project": "obs4MIPs", - }, - { - "dataset": "ERA5", - "project": "native6", - "type": "reanaly", - "tier": 3, - "version": "v1", - }, - ], - "hus": [ - { - "dataset": "ERA5", - "project": "native6", - "tier": 3, - "type": "reanaly", - "version": "v1", - }, - ], + "tas": [_HADCRUT5.to_recipe_dataset()], + "pr": [_GPCP_V2_3.to_recipe_dataset(), _ERA5_NATIVE6.to_recipe_dataset()], + "hus": [_ERA5_NATIVE6.to_recipe_dataset()], } # Update the datasets. diff --git a/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/diagnostics/sea_ice_area_basic.py b/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/diagnostics/sea_ice_area_basic.py index dda493309..a0659a991 100644 --- a/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/diagnostics/sea_ice_area_basic.py +++ b/packages/climate-ref-esmvaltool/src/climate_ref_esmvaltool/diagnostics/sea_ice_area_basic.py @@ -1,4 +1,5 @@ import pandas +from attrs import evolve from climate_ref_core.constraints import ( AddSupplementaryDataset, @@ -12,6 +13,7 @@ from climate_ref_core.metric_values.typing import FileDefinition, SeriesDefinition from climate_ref_core.testing import TestCase, TestDataSpecification from climate_ref_esmvaltool.diagnostics.base import ESMValToolDiagnostic, get_cmip_source_type +from climate_ref_esmvaltool.diagnostics.reference import ESMValToolReferenceSpec from climate_ref_esmvaltool.recipe import dataframe_to_recipe from climate_ref_esmvaltool.types import Recipe @@ -25,6 +27,18 @@ "sh": "February", } +# Reference sea-ice concentration from EUMETSAT OSI SAF, one dataset per hemisphere. +_OSI_450_NH = ESMValToolReferenceSpec( + project="OBS", + dataset="OSI-450-nh", + mip="OImon", + tier=2, + obs_type="reanaly", + version="v3", + supplementary_variables=({"short_name": "areacello", "mip": "fx"},), +) +_OSI_450_SH = evolve(_OSI_450_NH, dataset="OSI-450-sh") + class SeaIceAreaBasic(ESMValToolDiagnostic): """ @@ -35,6 +49,8 @@ class SeaIceAreaBasic(ESMValToolDiagnostic): slug = "sea-ice-area-basic" base_recipe = "ref/recipe_ref_sea_ice_area_basic.yml" + reference_datasets = (_OSI_450_NH, _OSI_450_SH) + data_requirements = ( ( DataRequirement( @@ -202,23 +218,11 @@ def update_recipe( for dataset in recipe["datasets"]: dataset.pop("timerange") - # Update observational datasets - nh_obs = { - "dataset": "OSI-450-nh", - "mip": "OImon", - "project": "OBS", - "supplementary_variables": [ - { - "short_name": "areacello", - "mip": "fx", - }, - ], - "tier": 2, - "type": "reanaly", - "version": "v3", - } + # Update observational datasets. sh shares nh's supplementary_variables list (shallow + # copy) so the two hemispheres differ only by dataset name, exactly as before. + nh_obs = _OSI_450_NH.to_recipe_dataset() sh_obs = nh_obs.copy() - sh_obs["dataset"] = "OSI-450-sh" + sh_obs["dataset"] = _OSI_450_SH.dataset diagnostics = recipe["diagnostics"] diagnostics["siarea_min"]["variables"]["sea_ice_area_nh_sep"]["additional_datasets"] = [nh_obs] diagnostics["siarea_min"]["variables"]["sea_ice_area_sh_feb"]["additional_datasets"] = [sh_obs] diff --git a/packages/climate-ref-esmvaltool/tests/unit/diagnostics/recipes/recipe_enso_characteristics_cmip6.yml b/packages/climate-ref-esmvaltool/tests/unit/diagnostics/recipes/recipe_enso_characteristics_cmip6.yml index ce2e2a519..52e9b8d3e 100644 --- a/packages/climate-ref-esmvaltool/tests/unit/diagnostics/recipes/recipe_enso_characteristics_cmip6.yml +++ b/packages/climate-ref-esmvaltool/tests/unit/diagnostics/recipes/recipe_enso_characteristics_cmip6.yml @@ -20,11 +20,11 @@ datasets: mip: Omon timerange: 18500116T120000/20141216T120000 - dataset: TROPFLUX - version: v1 + mip: Omon project: OBS6 - type: reanaly tier: 2 - mip: Omon + type: reanaly + version: v1 preprocessors: ssta_enso: custom_order: true diff --git a/packages/climate-ref-esmvaltool/tests/unit/diagnostics/recipes/recipe_enso_characteristics_cmip7.yml b/packages/climate-ref-esmvaltool/tests/unit/diagnostics/recipes/recipe_enso_characteristics_cmip7.yml index a2d08b387..9303d3206 100644 --- a/packages/climate-ref-esmvaltool/tests/unit/diagnostics/recipes/recipe_enso_characteristics_cmip7.yml +++ b/packages/climate-ref-esmvaltool/tests/unit/diagnostics/recipes/recipe_enso_characteristics_cmip7.yml @@ -23,11 +23,11 @@ datasets: region: glb timerange: 18500116T120000/20211216T120000 - dataset: TROPFLUX - version: v1 + mip: Omon project: OBS6 - type: reanaly tier: 2 - mip: Omon + type: reanaly + version: v1 preprocessors: ssta_enso: custom_order: true diff --git a/packages/climate-ref-esmvaltool/tests/unit/diagnostics/recipes/recipe_regional_historical_trend_obs4mips.yml b/packages/climate-ref-esmvaltool/tests/unit/diagnostics/recipes/recipe_regional_historical_trend_obs4mips.yml index 9474f492b..7d4bc01b7 100644 --- a/packages/climate-ref-esmvaltool/tests/unit/diagnostics/recipes/recipe_regional_historical_trend_obs4mips.yml +++ b/packages/climate-ref-esmvaltool/tests/unit/diagnostics/recipes/recipe_regional_historical_trend_obs4mips.yml @@ -360,8 +360,8 @@ diagnostics: project: obs4MIPs - dataset: ERA5 project: native6 - type: reanaly tier: 3 + type: reanaly version: v1 scripts: plot: diff --git a/packages/climate-ref-esmvaltool/tests/unit/diagnostics/test_reference.py b/packages/climate-ref-esmvaltool/tests/unit/diagnostics/test_reference.py new file mode 100644 index 000000000..77488fab2 --- /dev/null +++ b/packages/climate-ref-esmvaltool/tests/unit/diagnostics/test_reference.py @@ -0,0 +1,74 @@ +"""Tests for the declarative reference-dataset spec.""" + +from climate_ref_esmvaltool.diagnostics.reference import ESMValToolReferenceSpec +from climate_ref_esmvaltool.diagnostics.sea_ice_area_basic import SeaIceAreaBasic + +from climate_ref_core.datasets import SourceDatasetType + + +def test_to_recipe_dataset_full_alphabetical_order(): + spec = ESMValToolReferenceSpec( + project="OBS", + dataset="OSI-450-nh", + mip="OImon", + tier=2, + obs_type="reanaly", + version="v3", + supplementary_variables=({"short_name": "areacello", "mip": "fx"},), + ) + entry = spec.to_recipe_dataset() + assert entry == { + "dataset": "OSI-450-nh", + "mip": "OImon", + "project": "OBS", + "supplementary_variables": [{"short_name": "areacello", "mip": "fx"}], + "tier": 2, + "type": "reanaly", + "version": "v3", + } + # Key order is significant: recipes are dumped with sort_keys=False. + assert list(entry.keys()) == [ + "dataset", + "mip", + "project", + "supplementary_variables", + "tier", + "type", + "version", + ] + + +def test_to_recipe_dataset_omits_unset_optionals(): + spec = ESMValToolReferenceSpec(project="obs4MIPs", dataset="GPCP-V2.3") + assert spec.to_recipe_dataset() == {"dataset": "GPCP-V2.3", "project": "obs4MIPs"} + + +def test_to_recipe_dataset_builds_independent_mutable_entries(): + spec = ESMValToolReferenceSpec( + project="native6", dataset="ERA5", tier=3, obs_type="reanaly", version="v1" + ) + first = spec.to_recipe_dataset() + second = spec.to_recipe_dataset() + # Distinct objects so callers can mutate one without affecting the other. + assert first == second + assert first is not second + first["dataset"] = "OTHER" + assert second["dataset"] == "ERA5" + + +def test_diagnostic_maps_specs_to_reference_selectors(): + selectors = SeaIceAreaBasic().reference_dataset_selectors() + assert [s.source_type for s in selectors] == [SourceDatasetType.ESMValToolReference] * 2 + assert [dict(s.facets) for s in selectors] == [ + {"project": "OBS", "source_id": "OSI-450-nh", "table_id": "OImon"}, + {"project": "OBS", "source_id": "OSI-450-sh", "table_id": "OImon"}, + ] + + +def test_reference_selectors_omit_table_id_when_mip_unset(): + # A spec without a mip (e.g. an obs4MIPs reference) matches by project + dataset only. + class _Diag(SeaIceAreaBasic): + reference_datasets = (ESMValToolReferenceSpec(project="obs4MIPs", dataset="GPCP-V2.3"),) + + (selector,) = _Diag().reference_dataset_selectors() + assert dict(selector.facets) == {"project": "obs4MIPs", "source_id": "GPCP-V2.3"} diff --git a/packages/climate-ref-esmvaltool/tests/unit/test_provider.py b/packages/climate-ref-esmvaltool/tests/unit/test_provider.py index 09b0f5576..0a3fbb323 100644 --- a/packages/climate-ref-esmvaltool/tests/unit/test_provider.py +++ b/packages/climate-ref-esmvaltool/tests/unit/test_provider.py @@ -1,3 +1,4 @@ +import builtins import importlib.metadata from pathlib import Path @@ -14,6 +15,7 @@ def test_provider(): diagnostics_per_module = { "__init__.py": 0, "base.py": 0, + "reference.py": 0, "cloud_scatterplots.py": 5, "enso.py": 2, "regional_historical_changes.py": 3, @@ -94,3 +96,78 @@ def test_validate_setup_all_valid(self, mocker): result = provider.validate_setup(mock_config) assert result is True + + def _mock_registry(self, mocker, abspath): + """Point the ESMValTool data registry at ``abspath`` for ingest_data tests.""" + mock_registry = mocker.Mock() + mock_registry.abspath = abspath + mocker.patch( + "climate_ref_esmvaltool.dataset_registry_manager", + {_DATASETS_REGISTRY_NAME: mock_registry}, + ) + + def test_ingest_data_skips_when_climate_ref_not_installed(self, mocker, caplog): + """Test ingest_data gracefully skips when climate-ref package is not installed.""" + mock_config = mocker.Mock() + mock_db = mocker.Mock() + + original_import = builtins.__import__ + + def mock_import(name, *args, **kwargs): + if name.startswith("climate_ref.datasets"): + raise ImportError("No module named 'climate_ref'") + return original_import(name, *args, **kwargs) + + mocker.patch.object(builtins, "__import__", side_effect=mock_import) + + provider.ingest_data(mock_config, mock_db) + + assert "climate-ref package not installed" in caplog.text + + def test_ingest_data_path_not_exists(self, mocker, tmp_path, caplog): + """Test ingest_data logs a warning when the ESMValTool data directory is absent.""" + mock_config = mocker.Mock() + mock_db = mocker.Mock() + + # ESMValTool subdirectory does not exist under the registry cache. + self._mock_registry(mocker, tmp_path) + + provider.ingest_data(mock_config, mock_db) + + assert "ESMValTool reference data not found" in caplog.text + + def test_ingest_data_no_valid_datasets(self, mocker, tmp_path, caplog): + """Test ingest_data handles the case when no valid datasets are found.""" + mock_config = mocker.Mock() + mock_db = mocker.Mock() + + (tmp_path / "ESMValTool").mkdir() + self._mock_registry(mocker, tmp_path) + mocker.patch( + "climate_ref.datasets.ingest_datasets", + side_effect=ValueError("No valid datasets found"), + ) + + provider.ingest_data(mock_config, mock_db) + + assert "No valid ESMValTool reference datasets found" in caplog.text + + def test_ingest_data_success(self, mocker, tmp_path): + """Test ingest_data calls ingest_datasets with the expected parameters.""" + mock_config = mocker.Mock() + mock_db = mocker.Mock() + + (tmp_path / "ESMValTool").mkdir() + self._mock_registry(mocker, tmp_path) + + mock_stats = mocker.Mock() + mock_ingest = mocker.patch( + "climate_ref.datasets.ingest_datasets", + return_value=mock_stats, + ) + + provider.ingest_data(mock_config, mock_db) + + mock_ingest.assert_called_once() + assert mock_ingest.call_args.kwargs["skip_invalid"] is True + mock_stats.log_summary.assert_called_once() diff --git a/packages/climate-ref/src/climate_ref/datasets/__init__.py b/packages/climate-ref/src/climate_ref/datasets/__init__.py index 6bb9691bd..f9205ff09 100644 --- a/packages/climate-ref/src/climate_ref/datasets/__init__.py +++ b/packages/climate-ref/src/climate_ref/datasets/__init__.py @@ -13,6 +13,7 @@ from climate_ref.datasets.base import DatasetAdapter from climate_ref.datasets.cmip6 import CMIP6DatasetAdapter from climate_ref.datasets.cmip7 import CMIP7DatasetAdapter +from climate_ref.datasets.esmvaltool_reference import ESMValToolReferenceDatasetAdapter from climate_ref.datasets.obs4mips import Obs4MIPsDatasetAdapter from climate_ref.datasets.pmp_climatology import PMPClimatologyDatasetAdapter from climate_ref_core.datasets import SourceDatasetType @@ -24,6 +25,7 @@ SourceDatasetType.CMIP7: "instance_id", SourceDatasetType.obs4MIPs: "instance_id", SourceDatasetType.PMPClimatology: "instance_id", + SourceDatasetType.ESMValToolReference: "instance_id", } @@ -247,6 +249,8 @@ def get_dataset_adapter(source_type: str, **kwargs: Any) -> DatasetAdapter: return Obs4MIPsDatasetAdapter(**kwargs) elif source_type.lower() == SourceDatasetType.PMPClimatology.value.lower(): return PMPClimatologyDatasetAdapter(**kwargs) + elif source_type.lower() == SourceDatasetType.ESMValToolReference.value.lower(): + return ESMValToolReferenceDatasetAdapter(**kwargs) else: raise ValueError(f"Unknown source type: {source_type}") @@ -256,6 +260,7 @@ def get_dataset_adapter(source_type: str, **kwargs: Any) -> DatasetAdapter: "CMIP6DatasetAdapter", "CMIP7DatasetAdapter", "DatasetAdapter", + "ESMValToolReferenceDatasetAdapter", "IngestionStats", "Obs4MIPsDatasetAdapter", "PMPClimatologyDatasetAdapter", diff --git a/packages/climate-ref/src/climate_ref/datasets/esmvaltool_reference.py b/packages/climate-ref/src/climate_ref/datasets/esmvaltool_reference.py new file mode 100644 index 000000000..bf93e2bc9 --- /dev/null +++ b/packages/climate-ref/src/climate_ref/datasets/esmvaltool_reference.py @@ -0,0 +1,216 @@ +""" +Adapter for ESMValTool reference (observational/reanalysis) datasets. + +ESMValTool reference data is *not* CMOR/obs4MIPs compliant, so metadata cannot be read +from global attributes the way :mod:`climate_ref.datasets.obs4mips` does. Instead it is +parsed from the ESMValCore DRS path and filename templates that ESMValTool itself uses to +locate the data at run time (see ``climate_ref_esmvaltool.diagnostics.base``): + +* ``OBS`` / ``OBS6`` (metadata from the filename): + ``OBS/Tier{tier}/{dataset}/{project}_{dataset}_{type}_{version}_{mip}_{short_name}_{timerange}.nc`` +* ``native6`` (metadata from the directory; raw non-CMOR filename): + ``native6/Tier{tier}/{dataset}/{version}/{frequency}/{short_name}/*.nc`` +* ``obs4MIPs``: ``obs4MIPs/{dataset}/{version}/{short_name}_*.nc`` +""" + +from __future__ import annotations + +import traceback +from pathlib import Path +from typing import Any + +import pandas as pd +from loguru import logger + +from climate_ref.datasets.base import DatasetAdapter +from climate_ref.datasets.catalog_builder import build_catalog +from climate_ref.datasets.utils import build_instance_id, parse_cftime_dates, parse_drs_daterange +from climate_ref.models.dataset import Dataset, ESMValToolReferenceDataset + +# Top-level ESMValTool reference project directories (relative to the ``ESMValTool`` data root). +# ``OBS6`` data lives under the ``OBS`` directory; the project is recovered from the filename. +_PROJECT_ANCHORS = ("OBS", "native6", "obs4MIPs") + +_SLUG_PREFIX = "esmvaltool-reference" + +# Metadata columns (in order) that make up the dataset ``instance_id`` slug. +_INSTANCE_ID_FACETS = ("project", "source_id", "table_id", "variable_id", "version") + + +def _tier_from_segment(segment: str) -> int | None: + """Parse ``Tier2`` -> ``2``; return ``None`` if the segment is not a tier.""" + if segment.startswith("Tier") and segment[4:].isdigit(): + return int(segment[4:]) + return None + + +def _parse_obs(rel: tuple[str, ...], filename: str) -> dict[str, Any]: + # rel == ("OBS", "Tier{n}", "{dataset}", ..., filename) + tier = _tier_from_segment(rel[1]) + dataset = rel[2] + stem = filename[:-3] if filename.endswith(".nc") else filename + tokens = stem.split("_") + # {project}_{dataset}_{type}_{version}_{mip}_{short_name}[_{timerange}] + if len(tokens) < 6: # noqa: PLR2004 + raise ValueError(f"unexpected OBS filename structure: {filename}") + project, _, data_type, version, mip, short_name = tokens[:6] + # The timerange is the trailing token; use ``tokens[-1]`` (matching ``_parse_obs4mips``) + # so an unexpected extra segment does not silently drop the date range. + timerange = tokens[-1] if len(tokens) > 6 else None # noqa: PLR2004 + start_time, end_time = parse_drs_daterange(timerange) if timerange else (None, None) + return { + "project": project, + "source_id": dataset, + "variable_id": short_name, + "table_id": mip, + "version": version, + "data_type": data_type, + "tier": tier, + "start_time": start_time, + "end_time": end_time, + } + + +def _parse_native6(rel: tuple[str, ...]) -> dict[str, Any]: + # rel == ("native6", "Tier{n}", "{dataset}", "{version}", "{frequency}", "{short_name}", filename) + if len(rel) < 7: # noqa: PLR2004 + raise ValueError(f"unexpected native6 path structure: {'/'.join(rel)}") + tier = _tier_from_segment(rel[1]) + dataset, version, frequency, short_name = rel[2], rel[3], rel[4], rel[5] + return { + "project": "native6", + "source_id": dataset, + "variable_id": short_name, + "table_id": frequency, + "version": version, + "data_type": None, + "tier": tier, + # native6 filenames are raw (non-CMOR) and carry no reliable DRS date range. + "start_time": None, + "end_time": None, + } + + +def _parse_obs4mips(rel: tuple[str, ...], filename: str) -> dict[str, Any]: + # rel == ("obs4MIPs", "{dataset}", "{version}", filename) + if len(rel) < 4: # noqa: PLR2004 + raise ValueError(f"unexpected obs4MIPs path structure: {'/'.join(rel)}") + dataset, version = rel[1], rel[2] + stem = filename[:-3] if filename.endswith(".nc") else filename + tokens = stem.split("_") + if not tokens[0]: + raise ValueError(f"unexpected obs4MIPs filename structure: {filename}") + short_name = tokens[0] + timerange = tokens[-1] if len(tokens) > 1 else None + start_time, end_time = parse_drs_daterange(timerange) if timerange else (None, None) + return { + "project": "obs4MIPs", + "source_id": dataset, + "variable_id": short_name, + # obs4MIPs reference files here are monthly; use "mon" as the non-null grouping key. + "table_id": "mon", + "version": version, + "data_type": None, + "tier": None, + "start_time": start_time, + "end_time": end_time, + } + + +def parse_esmvaltool_reference(file: str, **kwargs: Any) -> dict[str, Any]: + """ + Parse a single ESMValTool reference file into a metadata record. + + Dispatches on the top-level project directory (``OBS``/``native6``/``obs4MIPs``) and + reads metadata from the path/filename rather than the file contents, because the data + is not CMOR compliant. + """ + try: + path = Path(file) + parts = path.parts + + anchor_idx = next((i for i, part in enumerate(parts) if part in _PROJECT_ANCHORS), None) + if anchor_idx is None: + raise ValueError( + f"{file} is not under a known ESMValTool reference project ({', '.join(_PROJECT_ANCHORS)})" + ) + rel = parts[anchor_idx:] + anchor = rel[0] + + if anchor == "OBS": + info = _parse_obs(rel, path.name) + elif anchor == "native6": + info = _parse_native6(rel) + else: + info = _parse_obs4mips(rel, path.name) + + info["path"] = str(file) + info["long_name"] = None + info["units"] = None + return info + except (ValueError, IndexError) as err: + logger.warning(str(err)) + return {"INVALID_ASSET": file, "TRACEBACK": str(err)} + except Exception: + logger.warning(traceback.format_exc()) + return {"INVALID_ASSET": file, "TRACEBACK": traceback.format_exc()} + + +class ESMValToolReferenceDatasetAdapter(DatasetAdapter): + """ + Adapter for ESMValTool reference datasets. + + See the module docstring for the layout conventions this adapter understands. + """ + + dataset_cls: type[Dataset] = ESMValToolReferenceDataset + slug_column = "instance_id" + + dataset_specific_metadata = ( + "project", + "source_id", + "variable_id", + "table_id", + "version", + "data_type", + "tier", + "long_name", + "units", + "finalised", + slug_column, + ) + + file_specific_metadata = ("start_time", "end_time", "path") + version_metadata = "version" + dataset_id_metadata = ( + "project", + "source_id", + "table_id", + "variable_id", + ) + + 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 (rows). The unique dataset identifier is + the ``instance_id`` slug in :attr:`slug_column`. + """ + datasets = build_catalog( + paths=[str(file_or_directory)], + parsing_func=parse_esmvaltool_reference, + include_patterns=["*.nc"], + depth=10, + n_jobs=self.n_jobs, + ) + if datasets.empty: + logger.error("No datasets found") + raise ValueError("No ESMValTool reference datasets found") + + datasets["start_time"] = parse_cftime_dates(datasets["start_time"]) + datasets["end_time"] = parse_cftime_dates(datasets["end_time"]) + datasets["finalised"] = True + return build_instance_id(datasets, list(_INSTANCE_ID_FACETS), prefix=_SLUG_PREFIX) diff --git a/packages/climate-ref/src/climate_ref/migrations/versions/2026-07-03T0000_a7b8c9d0e1f2_add_esmvaltool_reference_dataset.py b/packages/climate-ref/src/climate_ref/migrations/versions/2026-07-03T0000_a7b8c9d0e1f2_add_esmvaltool_reference_dataset.py new file mode 100644 index 000000000..9b81730d7 --- /dev/null +++ b/packages/climate-ref/src/climate_ref/migrations/versions/2026-07-03T0000_a7b8c9d0e1f2_add_esmvaltool_reference_dataset.py @@ -0,0 +1,106 @@ +"""add esmvaltool reference dataset + +Revision ID: a7b8c9d0e1f2 +Revises: f6a7b8c9d0e1 +Create Date: 2026-07-03 00:00:00.000000 + +""" + +from collections.abc import Sequence +from typing import Union + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "a7b8c9d0e1f2" +down_revision: Union[str, None] = "f6a7b8c9d0e1" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # ``dataset_type`` is a ``SourceDatasetType`` enum. On PostgreSQL it is backed by a native + # ENUM, so the new value must be added to the type explicitly. On SQLite the enum is stored + # as a VARCHAR sized to the longest member; ``ESMValToolReference`` (19 chars) is longer than + # the previous maximum ``PMPClimatology`` (14), so the column type must be widened to match. + bind = op.get_bind() + if bind.dialect.name == "postgresql": + op.execute("ALTER TYPE sourcedatasettype ADD VALUE IF NOT EXISTS 'ESMValToolReference'") + else: + with op.batch_alter_table("dataset", schema=None) as batch_op: + batch_op.alter_column( + "dataset_type", + existing_type=sa.String(length=14), + type_=sa.Enum( + "CMIP6", + "CMIP7", + "obs4MIPs", + "PMPClimatology", + "ESMValToolReference", + name="sourcedatasettype", + ), + existing_nullable=False, + ) + + op.create_table( + "esmvaltool_reference_dataset", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("project", sa.String(), nullable=False), + sa.Column("source_id", sa.String(), nullable=False), + sa.Column("variable_id", sa.String(), nullable=False), + sa.Column("table_id", sa.String(), nullable=False), + sa.Column("version", sa.String(), nullable=False), + sa.Column("data_type", sa.String(), nullable=True), + sa.Column("tier", sa.Integer(), nullable=True), + sa.Column("long_name", sa.String(), nullable=True), + sa.Column("units", sa.String(), nullable=True), + sa.Column("instance_id", sa.String(), nullable=False), + sa.ForeignKeyConstraint( + ["id"], ["dataset.id"], name=op.f("fk_esmvaltool_reference_dataset_id_dataset") + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_esmvaltool_reference_dataset")), + ) + with op.batch_alter_table("esmvaltool_reference_dataset", schema=None) as batch_op: + batch_op.create_index( + batch_op.f("ix_esmvaltool_reference_dataset_instance_id"), ["instance_id"], unique=False + ) + batch_op.create_index( + batch_op.f("ix_esmvaltool_reference_dataset_project"), ["project"], unique=False + ) + batch_op.create_index( + batch_op.f("ix_esmvaltool_reference_dataset_source_id"), ["source_id"], unique=False + ) + batch_op.create_index( + batch_op.f("ix_esmvaltool_reference_dataset_variable_id"), ["variable_id"], unique=False + ) + + +def downgrade() -> None: + with op.batch_alter_table("esmvaltool_reference_dataset", schema=None) as batch_op: + batch_op.drop_index(batch_op.f("ix_esmvaltool_reference_dataset_variable_id")) + batch_op.drop_index(batch_op.f("ix_esmvaltool_reference_dataset_source_id")) + batch_op.drop_index(batch_op.f("ix_esmvaltool_reference_dataset_project")) + batch_op.drop_index(batch_op.f("ix_esmvaltool_reference_dataset_instance_id")) + + op.drop_table("esmvaltool_reference_dataset") + + # On SQLite, narrow ``dataset_type`` back to the previous enum member set. On PostgreSQL the + # ``ESMValToolReference`` value is intentionally left in the enum type: PostgreSQL does not + # support removing a value from an enum without recreating the type, and leaving it is harmless. + bind = op.get_bind() + if bind.dialect.name != "postgresql": + with op.batch_alter_table("dataset", schema=None) as batch_op: + batch_op.alter_column( + "dataset_type", + existing_type=sa.Enum( + "CMIP6", + "CMIP7", + "obs4MIPs", + "PMPClimatology", + "ESMValToolReference", + name="sourcedatasettype", + ), + type_=sa.String(length=14), + existing_nullable=False, + ) diff --git a/packages/climate-ref/src/climate_ref/models/dataset.py b/packages/climate-ref/src/climate_ref/models/dataset.py index 230bd82d9..687c55978 100644 --- a/packages/climate-ref/src/climate_ref/models/dataset.py +++ b/packages/climate-ref/src/climate_ref/models/dataset.py @@ -207,15 +207,19 @@ class CMIP6Dataset(Dataset): __mapper_args__: ClassVar[Any] = {"polymorphic_identity": SourceDatasetType.CMIP6} # type: ignore -class Obs4MIPsDataset(Dataset): +class ReferenceDatasetMixin: """ - Represents a obs4mips dataset + Shared metadata columns for reference (observational) dataset types. - TODO: Should the metadata fields be part of the file or dataset? - """ + obs4MIPs, PMP climatology and (in future) other reference dataset types each live in + their own table with their own version lineage. They are deliberately kept as separate + dataset types because they describe data drawn from different sources and may carry + different versions of what looks like the same data. - __tablename__ = "obs4mips_dataset" - id: Mapped[int] = mapped_column(ForeignKey("dataset.id"), primary_key=True) + This mixin only removes the duplication in their column definitions; it does not merge + the types. Each concrete model still declares its own ``__tablename__``, primary-key + foreign key, and ``polymorphic_identity``. + """ activity_id: Mapped[str] = mapped_column() frequency: Mapped[str] = mapped_column() @@ -245,10 +249,22 @@ class Obs4MIPsDataset(Dataset): """ Unique identifier for the dataset. """ + + +class Obs4MIPsDataset(ReferenceDatasetMixin, Dataset): + """ + Represents a obs4mips dataset + + TODO: Should the metadata fields be part of the file or dataset? + """ + + __tablename__ = "obs4mips_dataset" + id: Mapped[int] = mapped_column(ForeignKey("dataset.id"), primary_key=True) + __mapper_args__: ClassVar[Any] = {"polymorphic_identity": SourceDatasetType.obs4MIPs} # type: ignore -class PMPClimatologyDataset(Dataset): +class PMPClimatologyDataset(ReferenceDatasetMixin, Dataset): """ Represents a climatology dataset from PMP @@ -258,35 +274,59 @@ class PMPClimatologyDataset(Dataset): __tablename__ = "pmp_climatology_dataset" id: Mapped[int] = mapped_column(ForeignKey("dataset.id"), primary_key=True) - activity_id: Mapped[str] = mapped_column() - frequency: Mapped[str] = mapped_column() - grid: Mapped[str] = mapped_column() - grid_label: Mapped[str] = mapped_column() - institution_id: Mapped[str] = mapped_column() - long_name: Mapped[str] = mapped_column() - nominal_resolution: Mapped[str] = mapped_column() - realm: Mapped[str] = mapped_column() - product: Mapped[str] = mapped_column() - source_id: Mapped[str] = mapped_column() - source_type: Mapped[str] = mapped_column() - units: Mapped[str] = mapped_column() - variable_id: Mapped[str] = mapped_column() - variant_label: Mapped[str] = mapped_column() - version: Mapped[str] = mapped_column() + __mapper_args__: ClassVar[Any] = {"polymorphic_identity": SourceDatasetType.PMPClimatology} # type: ignore + + +class ESMValToolReferenceDataset(Dataset): """ - Dataset version string. + Represents a reference (observational/reanalysis) dataset used by ESMValTool. - Only write this through an ORM instance: the base-table ``version_key`` ordering key is - synced by the ``_sync_version_key`` mapper event, which Core-level updates bypass. + Unlike obs4MIPs and PMP climatology datasets, ESMValTool reference data is *not* + CMOR/obs4MIPs compliant. It arrives in ESMValTool's own layout (``OBS``/``OBS6``, + ``native6`` and non-compliant ``obs4MIPs`` subtrees), so the available metadata is + limited to what the ESMValCore DRS path and filename encode. This model therefore + has its own, smaller column set rather than reusing :class:`ReferenceDatasetMixin`. + + It is a deliberately separate dataset type: the data describes different sources and + may carry different versions to the obs4MIPs data of the same name. """ - vertical_levels: Mapped[int] = mapped_column() - source_version_number: Mapped[str] = mapped_column() - instance_id: Mapped[str] = mapped_column() + __tablename__ = "esmvaltool_reference_dataset" + id: Mapped[int] = mapped_column(ForeignKey("dataset.id"), primary_key=True) + + project: Mapped[str] = mapped_column(index=True) + """ESMValCore project the data is loaded as: ``OBS``, ``OBS6``, ``native6`` or ``obs4MIPs``.""" + + source_id: Mapped[str] = mapped_column(index=True) + """Reference dataset name, e.g. ``CERES-EBAF``, ``ERA5``, ``OSI-450-nh``.""" + + variable_id: Mapped[str] = mapped_column(index=True) + """ESMValTool short name of the variable, e.g. ``tas``, ``sic``, ``rlut``.""" + + table_id: Mapped[str] = mapped_column() + """ + MIP table (``OBS``/``OBS6``, e.g. ``Amon``) or the frequency (``native6``, e.g. ``mon``). + + Always populated so it can act as a non-null part of the dataset identity/grouping key. + """ + + version: Mapped[str] = mapped_column() + """Dataset version as encoded in the ESMValTool layout, e.g. ``v3``, ``Ed4.2``.""" + + data_type: Mapped[str] = mapped_column(nullable=True) + """ESMValTool observation type where available: ``reanaly``, ``sat``, ``ground``.""" + + tier: Mapped[int] = mapped_column(nullable=True) + """ESMValTool data tier (accessibility), where the layout encodes it.""" + + long_name: Mapped[str] = mapped_column(nullable=True) + units: Mapped[str] = mapped_column(nullable=True) + + instance_id: Mapped[str] = mapped_column(index=True) """ Unique identifier for the dataset. """ - __mapper_args__: ClassVar[Any] = {"polymorphic_identity": SourceDatasetType.PMPClimatology} # type: ignore + __mapper_args__: ClassVar[Any] = {"polymorphic_identity": SourceDatasetType.ESMValToolReference} # type: ignore class CMIP7Dataset(Dataset): diff --git a/packages/climate-ref/src/climate_ref/reference_provenance.py b/packages/climate-ref/src/climate_ref/reference_provenance.py new file mode 100644 index 000000000..f53dcfdad --- /dev/null +++ b/packages/climate-ref/src/climate_ref/reference_provenance.py @@ -0,0 +1,103 @@ +""" +Record reference-dataset provenance for diagnostic executions. + +Diagnostics compare their inputs against reference (observational/reanalysis) datasets that are +not part of the solver's input selection (see +:meth:`climate_ref_core.diagnostics.Diagnostic.reference_dataset_selectors`). This module resolves +those selectors to ingested dataset rows and links them to an execution, so the reference data an +execution used can be tracked and surfaced without affecting the execution's dataset hash or the +data the diagnostic actually receives. +""" + +from __future__ import annotations + +from collections.abc import Sequence + +from loguru import logger + +from climate_ref.database import Database +from climate_ref.datasets import get_dataset_adapter +from climate_ref.models.execution import Execution, execution_datasets +from climate_ref_core.diagnostics import Diagnostic, ReferenceDatasetSelector + + +def resolve_reference_dataset_ids(db: Database, selectors: Sequence[ReferenceDatasetSelector]) -> list[int]: + """ + Resolve reference dataset selectors to ingested dataset primary keys. + + Parameters + ---------- + db + Database instance. + selectors + The reference dataset selectors to resolve. + + Returns + ------- + : + Deduplicated dataset ids matching the selectors, preserving first-seen order. + A warning is logged for any selector that matches nothing (e.g. reference data not ingested). + """ + ids: list[int] = [] + seen: set[int] = set() + for selector in selectors: + model = get_dataset_adapter(selector.source_type.value).dataset_cls + query = db.session.query(model.id) + for column, value in selector.facets.items(): + query = query.filter(getattr(model, column) == value) + + rows = query.all() + if not rows: + logger.warning( + f"No ingested {selector.source_type.value} datasets match reference selector " + f"{dict(selector.facets)}; run `ref providers setup` to ingest reference data." + ) + for (dataset_id,) in rows: + if dataset_id not in seen: + seen.add(dataset_id) + ids.append(dataset_id) + return ids + + +def link_reference_datasets(db: Database, execution: Execution, diagnostic: Diagnostic) -> int: + """ + Link a diagnostic's reference datasets to an execution for provenance. + + Idempotent: dataset ids already linked to the execution (e.g. from a previous call or as model + inputs) are skipped. + + Parameters + ---------- + db + Database instance. + execution + The execution to record provenance against. Must already be persisted (``execution.id`` set). + diagnostic + The diagnostic whose reference datasets should be linked. + + Returns + ------- + : + The number of new reference-dataset links created. + """ + selectors = diagnostic.reference_dataset_selectors() + if not selectors: + return 0 + + dataset_ids = resolve_reference_dataset_ids(db, selectors) + if not dataset_ids: + return 0 + + existing = { + row[0] + for row in db.session.query(execution_datasets.c.dataset_id).filter( + execution_datasets.c.execution_id == execution.id + ) + } + new_ids = [dataset_id for dataset_id in dataset_ids if dataset_id not in existing] + if new_ids: + db.session.execute( + execution_datasets.insert(), + [{"execution_id": execution.id, "dataset_id": dataset_id} for dataset_id in new_ids], + ) + return len(new_ids) diff --git a/packages/climate-ref/src/climate_ref/solver.py b/packages/climate-ref/src/climate_ref/solver.py index bddcbf344..e37297186 100644 --- a/packages/climate-ref/src/climate_ref/solver.py +++ b/packages/climate-ref/src/climate_ref/solver.py @@ -32,6 +32,7 @@ from climate_ref.models.diagnostic import recompute_promoted_version from climate_ref.models.execution import Execution from climate_ref.provider_registry import ProviderRegistry +from climate_ref.reference_provenance import link_reference_datasets from climate_ref_core.constraints import apply_constraint from climate_ref_core.datasets import ( DatasetCollection, @@ -782,6 +783,11 @@ def solve_required_executions( # noqa: PLR0912, PLR0913, PLR0915 # Add links to the datasets used in the execution execution.register_datasets(db, definition.datasets) + # Record reference (observational) datasets the diagnostic compares against. + # These are provenance only: they are not solver inputs and do not affect the + # dataset hash or the data passed to the diagnostic. + link_reference_datasets(db, execution, potential_execution.diagnostic) + if execute: # Detach the row before the surrounding ``with begin()`` commits. # Otherwise expire-on-commit marks ``execution.id`` stale, diff --git a/packages/climate-ref/tests/unit/datasets/test_datasets.py b/packages/climate-ref/tests/unit/datasets/test_datasets.py index 6af8e1ade..b1f01467a 100644 --- a/packages/climate-ref/tests/unit/datasets/test_datasets.py +++ b/packages/climate-ref/tests/unit/datasets/test_datasets.py @@ -100,6 +100,10 @@ def test_validate_data_catalog_metadata_variance(caplog): (SourceDatasetType.CMIP6.value, "climate_ref.datasets.cmip6.CMIP6DatasetAdapter"), (SourceDatasetType.CMIP7.value, "climate_ref.datasets.cmip7.CMIP7DatasetAdapter"), (SourceDatasetType.obs4MIPs.value, "climate_ref.datasets.obs4mips.Obs4MIPsDatasetAdapter"), + ( + SourceDatasetType.ESMValToolReference.value, + "climate_ref.datasets.esmvaltool_reference.ESMValToolReferenceDatasetAdapter", + ), ], ) def test_get_dataset_adapter_valid(source_type, expected_adapter): diff --git a/packages/climate-ref/tests/unit/datasets/test_esmvaltool_reference.py b/packages/climate-ref/tests/unit/datasets/test_esmvaltool_reference.py new file mode 100644 index 000000000..8fbbc7fe0 --- /dev/null +++ b/packages/climate-ref/tests/unit/datasets/test_esmvaltool_reference.py @@ -0,0 +1,165 @@ +"""Tests for the ESMValTool reference dataset adapter and its custom-format parser.""" + +from pathlib import Path + +import pytest + +from climate_ref.config import Config +from climate_ref.database import Database +from climate_ref.datasets import ingest_datasets +from climate_ref.datasets.esmvaltool_reference import ( + ESMValToolReferenceDatasetAdapter, + parse_esmvaltool_reference, +) + + +@pytest.mark.parametrize( + "path, expected", + [ + pytest.param( + "/root/ESMValTool/OBS/Tier2/CERES-EBAF/OBS_CERES-EBAF_sat_Ed4.2_Amon_rlut_200003-202311.nc", + { + "project": "OBS", + "source_id": "CERES-EBAF", + "variable_id": "rlut", + "table_id": "Amon", + "version": "Ed4.2", + "data_type": "sat", + "tier": 2, + "start_time": "2000-03-01", + "end_time": "2023-11-30", + }, + id="obs", + ), + pytest.param( + "/root/ESMValTool/OBS/Tier2/TROPFLUX/OBS6_TROPFLUX_reanaly_v1_Omon_tos_197901-201812.nc", + { + "project": "OBS6", + "source_id": "TROPFLUX", + "variable_id": "tos", + "table_id": "Omon", + "version": "v1", + "data_type": "reanaly", + "tier": 2, + }, + id="obs6", + ), + pytest.param( + "/root/ESMValTool/native6/Tier3/ERA5/v1/mon/hus/era5_specific_humidity_1980_monthly.nc", + { + "project": "native6", + "source_id": "ERA5", + "variable_id": "hus", + "table_id": "mon", + "version": "v1", + "data_type": None, + "tier": 3, + "start_time": None, + "end_time": None, + }, + id="native6", + ), + pytest.param( + "/root/ESMValTool/obs4MIPs/GPCP-V2.3/v20180519/pr_GPCP-SG_L3_v2.3_197901-201710.nc", + { + "project": "obs4MIPs", + "source_id": "GPCP-V2.3", + "variable_id": "pr", + "table_id": "mon", + "version": "v20180519", + "tier": None, + }, + id="obs4mips", + ), + ], +) +def test_parse_layouts(path, expected): + result = parse_esmvaltool_reference(path) + assert "INVALID_ASSET" not in result + for key, value in expected.items(): + assert result[key] == value, key + assert result["path"] == path + + +def test_parse_rejects_unknown_layout(): + result = parse_esmvaltool_reference("/root/somewhere/random.nc") + assert result["INVALID_ASSET"] == "/root/somewhere/random.nc" + + +def test_parse_rejects_malformed_obs_filename(): + result = parse_esmvaltool_reference("/root/ESMValTool/OBS/Tier2/FOO/OBS_FOO_sat.nc") + assert "INVALID_ASSET" in result + + +def test_parse_rejects_empty_obs4mips_filename(): + # An empty filename stem would otherwise yield an empty variable_id rather than being flagged. + result = parse_esmvaltool_reference("/root/ESMValTool/obs4MIPs/FOO/v1/.nc") + assert "INVALID_ASSET" in result + + +def _touch(path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(b"") + + +@pytest.fixture +def reference_tree(tmp_path) -> Path: + """A minimal ESMValTool reference tree of empty .nc files (parser is path-based).""" + root = tmp_path / "ESMValTool" + files = [ + # single-file OBS dataset + "OBS/Tier2/CERES-EBAF/OBS_CERES-EBAF_sat_Ed4.2_Amon_rlut_200003-202311.nc", + # two-file OBS dataset (same slug -> one dataset, two files) + "OBS/Tier2/OSI-450-nh/OBS_OSI-450-nh_reanaly_v3_OImon_sic_197901-197912.nc", + "OBS/Tier2/OSI-450-nh/OBS_OSI-450-nh_reanaly_v3_OImon_sic_198001-198012.nc", + # two-file native6 dataset + "native6/Tier3/ERA5/v1/mon/hus/era5_specific_humidity_1980_monthly.nc", + "native6/Tier3/ERA5/v1/mon/hus/era5_specific_humidity_1981_monthly.nc", + # single-file obs4MIPs dataset + "obs4MIPs/GPCP-V2.3/v20180519/pr_GPCP-SG_L3_v2.3_197901-201710.nc", + ] + for rel in files: + _touch(root / rel) + return root + + +@pytest.fixture +def db() -> Database: + config = Config.default() + database = Database("sqlite:///:memory:") + database.migrate(config) + yield database + database.close() + + +def test_ingest_end_to_end(reference_tree, db): + adapter = ESMValToolReferenceDatasetAdapter() + + stats = ingest_datasets(adapter, reference_tree, db, skip_invalid=True) + + # 4 distinct datasets across the three layouts, 6 files total + assert stats.datasets_created == 4 + assert stats.files_added == 6 + + catalog = adapter.load_catalog(db) + assert set(catalog["instance_id"]) == { + "esmvaltool-reference.OBS.CERES-EBAF.Amon.rlut.Ed4.2", + "esmvaltool-reference.OBS.OSI-450-nh.OImon.sic.v3", + "esmvaltool-reference.native6.ERA5.mon.hus.v1", + "esmvaltool-reference.obs4MIPs.GPCP-V2.3.mon.pr.v20180519", + } + # the OSI-450 and ERA5 datasets each own two files + counts = catalog["instance_id"].value_counts() + assert counts["esmvaltool-reference.OBS.OSI-450-nh.OImon.sic.v3"] == 2 + assert counts["esmvaltool-reference.native6.ERA5.mon.hus.v1"] == 2 + + +def test_ingest_is_idempotent(reference_tree, db): + adapter = ESMValToolReferenceDatasetAdapter() + + ingest_datasets(adapter, reference_tree, db, skip_invalid=True) + stats = ingest_datasets(adapter, reference_tree, db, skip_invalid=True) + + assert stats.datasets_created == 0 + assert stats.datasets_unchanged == 4 + assert stats.files_added == 0 diff --git a/packages/climate-ref/tests/unit/test_reference_provenance.py b/packages/climate-ref/tests/unit/test_reference_provenance.py new file mode 100644 index 000000000..5b9a56283 --- /dev/null +++ b/packages/climate-ref/tests/unit/test_reference_provenance.py @@ -0,0 +1,118 @@ +"""Tests for reference-dataset provenance linkage.""" + +import pytest + +from climate_ref.config import Config +from climate_ref.database import Database +from climate_ref.models.dataset import ESMValToolReferenceDataset +from climate_ref.models.execution import Execution +from climate_ref.reference_provenance import link_reference_datasets, resolve_reference_dataset_ids +from climate_ref_core.datasets import SourceDatasetType +from climate_ref_core.diagnostics import ReferenceDatasetSelector + + +@pytest.fixture +def db() -> Database: + config = Config.default() + database = Database("sqlite:///:memory:") + database.migrate(config) + yield database + database.close() + + +def _add_reference_dataset(db, *, source_id, variable_id, table_id, version="v1", project="OBS") -> int: + slug = f"esmvaltool-reference.{project}.{source_id}.{table_id}.{variable_id}.{version}" + dataset = ESMValToolReferenceDataset( + slug=slug, + project=project, + source_id=source_id, + variable_id=variable_id, + table_id=table_id, + version=version, + instance_id=slug, + finalised=True, + ) + db.session.add(dataset) + db.session.flush() + return dataset.id + + +class _StubDiagnostic: + """Minimal stand-in exposing reference selectors.""" + + def __init__(self, selectors): + self._selectors = selectors + + def reference_dataset_selectors(self): + return self._selectors + + +_OSI = ReferenceDatasetSelector( + source_type=SourceDatasetType.ESMValToolReference, + facets={"project": "OBS", "source_id": "OSI-450-nh", "table_id": "OImon"}, +) + + +def test_resolve_matches_only_selected_facets(db): + with db.session.begin(): + osi_id = _add_reference_dataset(db, source_id="OSI-450-nh", variable_id="sic", table_id="OImon") + _add_reference_dataset(db, source_id="CERES-EBAF", variable_id="rlut", table_id="Amon") + + assert resolve_reference_dataset_ids(db, [_OSI]) == [osi_id] + + +def test_resolve_matches_all_variables_of_a_dataset(db): + # A selector without a variable facet links every variable of the reference dataset. + with db.session.begin(): + ids = { + _add_reference_dataset( + db, source_id="ERA5", project="native6", variable_id="hus", table_id="mon" + ), + _add_reference_dataset(db, source_id="ERA5", project="native6", variable_id="pr", table_id="mon"), + } + selector = ReferenceDatasetSelector( + source_type=SourceDatasetType.ESMValToolReference, + facets={"project": "native6", "source_id": "ERA5"}, + ) + assert set(resolve_reference_dataset_ids(db, [selector])) == ids + + +def test_resolve_warns_and_returns_empty_when_nothing_ingested(db, caplog): + with caplog.at_level("WARNING"): + result = resolve_reference_dataset_ids(db, [_OSI]) + assert result == [] + assert "No ingested esmvaltool-reference datasets match reference selector" in caplog.text + + +def test_link_records_reference_datasets_idempotently(db): + with db.session.begin(): + osi_id = _add_reference_dataset(db, source_id="OSI-450-nh", variable_id="sic", table_id="OImon") + execution = Execution(execution_group_id=1, output_fragment="frag", dataset_hash="hash") + db.session.add(execution) + db.session.flush() + execution_id = execution.id + + diagnostic = _StubDiagnostic([_OSI]) + + with db.session.begin(): + execution = db.session.get(Execution, execution_id) + assert link_reference_datasets(db, execution, diagnostic) == 1 + assert {d.id for d in execution.datasets} == {osi_id} + + # Second call links nothing new. + with db.session.begin(): + execution = db.session.get(Execution, execution_id) + assert link_reference_datasets(db, execution, diagnostic) == 0 + assert {d.id for d in execution.datasets} == {osi_id} + + +def test_link_noop_for_diagnostic_without_reference_datasets(db): + with db.session.begin(): + execution = Execution(execution_group_id=1, output_fragment="frag", dataset_hash="hash") + db.session.add(execution) + db.session.flush() + execution_id = execution.id + + with db.session.begin(): + execution = db.session.get(Execution, execution_id) + assert link_reference_datasets(db, execution, _StubDiagnostic([])) == 0