climate_ref_pmp.diagnostics

PMP diagnostics.

AnnualCycle

Bases: CommandLineDiagnostic

Calculate the annual cycle for a dataset

Source code in packages/climate-ref-pmp/src/climate_ref_pmp/diagnostics/annual_cycle.py

class AnnualCycle(CommandLineDiagnostic):
    """
    Calculate the annual cycle for a dataset
    """

    reconstruction_inputs = PMP_RECONSTRUCTION_INPUTS

    name = "Annual Cycle"
    slug = "annual-cycle"
    facets = (
        "kind",
        "mip_id",
        "source_id",
        "member_id",
        "experiment_id",
        "variable_id",
        "reference_source_id",
        "level",
        "region",
        "statistic",
        "season",
    )
    version = 5

    _variable_obs_pairs = (
        # ERA-5 as reference dataset, spatial 2-D variables
        ("ts", "ERA-5"),
        ("uas", "ERA-5"),
        ("vas", "ERA-5"),
        ("psl", "ERA-5"),
        # ERA-5 as reference dataset, spatial 3-D variables
        ("ta", "ERA-5"),
        ("ua", "ERA-5"),
        ("va", "ERA-5"),
        ("zg", "ERA-5"),
        # Other reference datasets, spatial 2-D variables
        ("pr", "GPCP-3-3"),
        ("rlds", "CERES-EBAF-4-2"),
        ("rlus", "CERES-EBAF-4-2"),
        ("rlut", "CERES-EBAF-4-2"),
        ("rsds", "CERES-EBAF-4-2"),
        ("rsdt", "CERES-EBAF-4-2"),
        ("rsus", "CERES-EBAF-4-2"),
        ("rsut", "CERES-EBAF-4-2"),
    )

    data_requirements = tuple(
        pair
        for variable_id, obs_source in _variable_obs_pairs
        for pair in make_data_requirement(variable_id, obs_source)
    )

    test_data_spec = TestDataSpecification(
        test_cases=(
            TestCase(
                name="cmip6-ts",
                description="Test with CMIP6 ts data and ERA-5 climatology",
                requests=(
                    RegistryRequest(
                        slug="annual-cycle-era5-ts",
                        registry_name="pmp-climatology",
                        facets={"variable_id": "ts", "source_id": "ERA-5"},
                    ),
                    CMIP6Request(
                        slug="annual-cycle-cmip6-ts",
                        facets={
                            "source_id": "ACCESS-ESM1-5",
                            "experiment_id": "historical",
                            "variable_id": "ts",
                            "member_id": "r1i1p1f1",
                            "table_id": "Amon",
                        },
                        time_span=("2000-01", "2014-12"),
                    ),
                ),
            ),
            TestCase(
                name="cmip6-pr",
                description="Test with CMIP6 pr data and GPCP-3-3 climatology. "
                "Produces double ITCZ pattern in the diagnostics.",
                requests=(
                    RegistryRequest(
                        slug="annual-cycle-gpcp-pr",
                        registry_name="pmp-climatology",
                        facets={"variable_id": "pr", "source_id": "GPCP-3-3"},
                    ),
                    CMIP6Request(
                        slug="annual-cycle-cmip6-pr",
                        facets={
                            "source_id": "ACCESS-ESM1-5",
                            "experiment_id": "historical",
                            "variable_id": "pr",
                            "member_id": "r1i1p1f1",
                            "table_id": "Amon",
                        },
                        time_span=("2000-01", "2014-12"),
                    ),
                ),
            ),
            TestCase(
                name="cmip6-ta",
                description="Test with CMIP6 ta data and ERA-5 climatology. "
                "Exercises the 3D pressure-level path, which captures a level dimension.",
                requests=(
                    RegistryRequest(
                        slug="annual-cycle-era5-ta",
                        registry_name="pmp-climatology",
                        facets={"variable_id": "ta", "source_id": "ERA-5"},
                    ),
                    CMIP6Request(
                        slug="annual-cycle-cmip6-ta",
                        facets={
                            "source_id": "ACCESS-ESM1-5",
                            "experiment_id": "historical",
                            "variable_id": "ta",
                            "member_id": "r1i1p1f1",
                            "table_id": "Amon",
                        },
                        time_span=("2000-01", "2014-12"),
                    ),
                ),
            ),
            TestCase(
                name="cmip7-ts",
                description="CMIP7 test case with converted historical ts from ACCESS-ESM1-5",
                requests=(
                    RegistryRequest(
                        slug="annual-cycle-era5-ts-cmip7",
                        registry_name="pmp-climatology",
                        facets={"variable_id": "ts", "source_id": "ERA-5"},
                    ),
                    CMIP7Request(
                        slug="annual-cycle-cmip7-ts",
                        facets={
                            "source_id": "ACCESS-ESM1-5",
                            "experiment_id": "historical",
                            "variable_id": "ts",
                            "branded_variable": "ts_tavg-u-hxy-u",
                            "variant_label": "r1i1p1f1",
                            "frequency": "mon",
                            "region": "glb",
                        },
                        time_span=("2000-01", "2014-12"),
                    ),
                ),
            ),
        ),
    )

    def __init__(self) -> None:
        self.parameter_file_1 = "pmp_param_annualcycle_1-clims.py"
        self.parameter_file_2 = "pmp_param_annualcycle_2-metrics.py"

    def build_cmds(self, definition: ExecutionDefinition) -> list[list[str]]:  # noqa: PLR0915
        """
        Build the command to run the diagnostic

        Parameters
        ----------
        definition
            Definition of the diagnostic execution

        Returns
        -------
            Command arguments to execute in the PMP environment
        """
        model_source_type = get_model_source_type(definition)
        input_datasets = definition.datasets[model_source_type]
        reference_collection = definition.datasets[SourceDatasetType.PMPClimatology]

        source_id = input_datasets["source_id"].unique()[0]
        experiment_id = input_datasets["experiment_id"].unique()[0]
        variable_id = input_datasets["variable_id"].unique()[0]
        member_id = input_datasets[
            "variant_label" if model_source_type == SourceDatasetType.CMIP7 else "member_id"
        ].unique()[0]

        model_files_raw = input_datasets.path.to_list()
        if len(model_files_raw) == 1:
            model_files = model_files_raw[0]  # If only one file, use it directly
        elif len(model_files_raw) > 1:
            model_files = build_glob_pattern(model_files_raw)  # If multiple files, build a glob pattern
        else:
            raise ValueError("No model files found")

        logger.debug("build_cmd start")

        logger.debug(f"input_datasets: {input_datasets}")
        logger.debug(f"input_datasets.keys(): {input_datasets.keys()}")

        reference_dataset_name = reference_collection["source_id"].unique()[0]
        reference_dataset_path = reference_collection.datasets.iloc[0]["path"]

        logger.debug(f"reference_dataset.datasets: {reference_collection.datasets}")
        logger.debug(f"reference_dataset_name: {reference_dataset_name}")
        logger.debug(f"reference_dataset_path: {reference_dataset_path}")

        output_directory_path = str(definition.output_directory)

        cmds = []

        # ----------------------------------------------
        # PART 1: Build the command to get climatologies
        # ----------------------------------------------
        # Model
        data_name = f"{source_id}_{experiment_id}_{member_id}"
        data_path = model_files

        # PMP stamps this into the climatology filename, which leaks into the bundle provenance.
        # Derived from the diagnostic version rather than the run date so reruns are reproducible.
        clim_version = f"v{self.version}"

        params = {
            "vars": variable_id,
            "infile": data_path,
            "outfile": f"{output_directory_path}/{variable_id}_{data_name}_clims.nc",
            "version": clim_version,
        }

        cmds.append(
            build_pmp_command(
                driver_file="pcmdi_compute_climatologies.py",
                parameter_file=self.parameter_file_1,
                **params,
            )
        )

        # --------------------------------------------------
        # PART 2: Build the command to calculate diagnostics
        # --------------------------------------------------
        # Reference
        obs_dict = {
            variable_id: {
                reference_dataset_name: {
                    "template": reference_dataset_path,
                },
                "default": reference_dataset_name,
            }
        }

        # Generate a JSON file based on the obs_dict
        with open(f"{output_directory_path}/obs_dict.json", "w") as f:
            json.dump(obs_dict, f)

        if variable_id in ["ua", "va", "ta"]:
            levels = ["200", "850"]
        elif variable_id in ["zg"]:
            levels = ["500"]
        else:
            levels = None

        variables = []
        if levels is not None:
            for level in levels:
                variable_id_with_level = f"{variable_id}-{level}"
                variables.append(variable_id_with_level)
        else:
            variables = [variable_id]

        logger.debug(f"variables: {variables}")
        logger.debug(f"levels: {levels}")

        # Build the command for each level
        params = {
            "vars": variables,
            "custom_observations": f"{output_directory_path}/obs_dict.json",
            "test_data_path": output_directory_path,
            "test_data_set": source_id,
            "realization": member_id,
            "filename_template": f"%(variable)_{data_name}_clims.198101-200512.AC.{clim_version}.nc",
            "metrics_output_path": output_directory_path,
            "cmec": "",
        }

        cmds.append(
            build_pmp_command(
                driver_file="mean_climate_driver.py",
                parameter_file=self.parameter_file_2,
                **params,
            )
        )

        logger.debug("build_cmd end")
        logger.debug(f"cmds: {cmds}")

        return cmds

    def build_execution_result(self, definition: ExecutionDefinition) -> ExecutionResult:
        """
        Build a diagnostic result from the output of the PMP driver

        Parameters
        ----------
        definition
            Definition of the diagnostic execution

        Returns
        -------
            Result of the diagnostic execution
        """
        model_source_type = get_model_source_type(definition)
        input_datasets = definition.datasets[model_source_type]
        variable_id = input_datasets["variable_id"].unique()[0]

        if variable_id in ["ua", "va", "ta", "zg"]:
            variable_dir_pattern = f"{variable_id}-???"
        else:
            variable_dir_pattern = variable_id

        results_directory = definition.output_directory

        logger.debug(f"results_directory: {results_directory}")
        logger.debug(f"variable_dir_pattern: {variable_dir_pattern}")

        # Find the CMEC JSON file(s)
        results_files = transform_results_files(list(results_directory.glob("*_cmec.json")))

        if len(results_files) == 1:
            # If only one file, use it directly
            results_file = results_files[0]
            logger.debug(f"results_file: {results_file}")
        elif len(results_files) > 1:
            logger.info(f"More than one cmec file found: {results_files}")
            results_file = combine_results_files(results_files, definition.output_directory)
        else:
            logger.error("Unexpected case: no cmec file found")
            return ExecutionResult.build_from_failure(definition)

        # PMP writes plots and data into a per-variable subdirectory (``ts`` for 2D variables,
        # ``ta-200``/``ta-850`` for 3D pressure-level variables). Glob from the results directory
        # so the ``variable_dir_pattern`` wildcard matches every level subdir.
        # Sort so the committed output.json plot/data key order is deterministic across hosts.
        png_files = [
            definition.as_relative_path(f)
            for f in sorted(results_directory.glob(f"{variable_dir_pattern}/*.png"))
        ]
        data_files = [
            definition.as_relative_path(f)
            for f in sorted(results_directory.glob(f"{variable_dir_pattern}/*.nc"))
        ]

        # Prepare the output bundles
        cmec_output_bundle, cmec_metric_bundle = process_json_result(results_file, png_files, data_files)

        # Add missing dimensions to the output
        member_id_col = "variant_label" if model_source_type == SourceDatasetType.CMIP7 else "member_id"
        reference_collection = definition.datasets[SourceDatasetType.PMPClimatology]
        cmec_metric_bundle = cmec_metric_bundle.prepend_dimensions(
            {
                # PMP scalars are model-performance scores against a reference, not reference
                # (observation) values, so every value's role is ``model``.
                "kind": "model",
                "source_id": input_datasets["source_id"].unique()[0],
                "member_id": input_datasets[member_id_col].unique()[0],
                "experiment_id": input_datasets["experiment_id"].unique()[0],
                "variable_id": input_datasets["variable_id"].unique()[0],
                "reference_source_id": reference_collection["source_id"].unique()[0],
            }
        )

        return ExecutionResult.build_from_output_bundle(
            definition,
            cmec_output_bundle=cmec_output_bundle,
            cmec_metric_bundle=cmec_metric_bundle,
        )

    def execute(self, definition: ExecutionDefinition) -> None:
        """
        Run the diagnostic on the given configuration.

        Parameters
        ----------
        definition : ExecutionDefinition
            The configuration to run the diagnostic on.

        Returns
        -------
        :
            The result of running the diagnostic.
        """
        cmds = self.build_cmds(definition)

        runs = [self.provider.run(cmd) for cmd in cmds]
        logger.debug(f"runs: {runs}")

build_cmds(definition)

Build the command to run the diagnostic

Parameters:

Name	Type	Description	Default
definition	ExecutionDefinition	Definition of the diagnostic execution	required

Returns:

Type	Description
Command arguments to execute in the PMP environment

Source code in packages/climate-ref-pmp/src/climate_ref_pmp/diagnostics/annual_cycle.py

def build_cmds(self, definition: ExecutionDefinition) -> list[list[str]]:  # noqa: PLR0915
    """
    Build the command to run the diagnostic

    Parameters
    ----------
    definition
        Definition of the diagnostic execution

    Returns
    -------
        Command arguments to execute in the PMP environment
    """
    model_source_type = get_model_source_type(definition)
    input_datasets = definition.datasets[model_source_type]
    reference_collection = definition.datasets[SourceDatasetType.PMPClimatology]

    source_id = input_datasets["source_id"].unique()[0]
    experiment_id = input_datasets["experiment_id"].unique()[0]
    variable_id = input_datasets["variable_id"].unique()[0]
    member_id = input_datasets[
        "variant_label" if model_source_type == SourceDatasetType.CMIP7 else "member_id"
    ].unique()[0]

    model_files_raw = input_datasets.path.to_list()
    if len(model_files_raw) == 1:
        model_files = model_files_raw[0]  # If only one file, use it directly
    elif len(model_files_raw) > 1:
        model_files = build_glob_pattern(model_files_raw)  # If multiple files, build a glob pattern
    else:
        raise ValueError("No model files found")

    logger.debug("build_cmd start")

    logger.debug(f"input_datasets: {input_datasets}")
    logger.debug(f"input_datasets.keys(): {input_datasets.keys()}")

    reference_dataset_name = reference_collection["source_id"].unique()[0]
    reference_dataset_path = reference_collection.datasets.iloc[0]["path"]

    logger.debug(f"reference_dataset.datasets: {reference_collection.datasets}")
    logger.debug(f"reference_dataset_name: {reference_dataset_name}")
    logger.debug(f"reference_dataset_path: {reference_dataset_path}")

    output_directory_path = str(definition.output_directory)

    cmds = []

    # ----------------------------------------------
    # PART 1: Build the command to get climatologies
    # ----------------------------------------------
    # Model
    data_name = f"{source_id}_{experiment_id}_{member_id}"
    data_path = model_files

    # PMP stamps this into the climatology filename, which leaks into the bundle provenance.
    # Derived from the diagnostic version rather than the run date so reruns are reproducible.
    clim_version = f"v{self.version}"

    params = {
        "vars": variable_id,
        "infile": data_path,
        "outfile": f"{output_directory_path}/{variable_id}_{data_name}_clims.nc",
        "version": clim_version,
    }

    cmds.append(
        build_pmp_command(
            driver_file="pcmdi_compute_climatologies.py",
            parameter_file=self.parameter_file_1,
            **params,
        )
    )

    # --------------------------------------------------
    # PART 2: Build the command to calculate diagnostics
    # --------------------------------------------------
    # Reference
    obs_dict = {
        variable_id: {
            reference_dataset_name: {
                "template": reference_dataset_path,
            },
            "default": reference_dataset_name,
        }
    }

    # Generate a JSON file based on the obs_dict
    with open(f"{output_directory_path}/obs_dict.json", "w") as f:
        json.dump(obs_dict, f)

    if variable_id in ["ua", "va", "ta"]:
        levels = ["200", "850"]
    elif variable_id in ["zg"]:
        levels = ["500"]
    else:
        levels = None

    variables = []
    if levels is not None:
        for level in levels:
            variable_id_with_level = f"{variable_id}-{level}"
            variables.append(variable_id_with_level)
    else:
        variables = [variable_id]

    logger.debug(f"variables: {variables}")
    logger.debug(f"levels: {levels}")

    # Build the command for each level
    params = {
        "vars": variables,
        "custom_observations": f"{output_directory_path}/obs_dict.json",
        "test_data_path": output_directory_path,
        "test_data_set": source_id,
        "realization": member_id,
        "filename_template": f"%(variable)_{data_name}_clims.198101-200512.AC.{clim_version}.nc",
        "metrics_output_path": output_directory_path,
        "cmec": "",
    }

    cmds.append(
        build_pmp_command(
            driver_file="mean_climate_driver.py",
            parameter_file=self.parameter_file_2,
            **params,
        )
    )

    logger.debug("build_cmd end")
    logger.debug(f"cmds: {cmds}")

    return cmds

build_execution_result(definition)

Build a diagnostic result from the output of the PMP driver

Parameters:

Name	Type	Description	Default
definition	ExecutionDefinition	Definition of the diagnostic execution	required

Returns:

Type	Description
Result of the diagnostic execution

Source code in packages/climate-ref-pmp/src/climate_ref_pmp/diagnostics/annual_cycle.py

def build_execution_result(self, definition: ExecutionDefinition) -> ExecutionResult:
    """
    Build a diagnostic result from the output of the PMP driver

    Parameters
    ----------
    definition
        Definition of the diagnostic execution

    Returns
    -------
        Result of the diagnostic execution
    """
    model_source_type = get_model_source_type(definition)
    input_datasets = definition.datasets[model_source_type]
    variable_id = input_datasets["variable_id"].unique()[0]

    if variable_id in ["ua", "va", "ta", "zg"]:
        variable_dir_pattern = f"{variable_id}-???"
    else:
        variable_dir_pattern = variable_id

    results_directory = definition.output_directory

    logger.debug(f"results_directory: {results_directory}")
    logger.debug(f"variable_dir_pattern: {variable_dir_pattern}")

    # Find the CMEC JSON file(s)
    results_files = transform_results_files(list(results_directory.glob("*_cmec.json")))

    if len(results_files) == 1:
        # If only one file, use it directly
        results_file = results_files[0]
        logger.debug(f"results_file: {results_file}")
    elif len(results_files) > 1:
        logger.info(f"More than one cmec file found: {results_files}")
        results_file = combine_results_files(results_files, definition.output_directory)
    else:
        logger.error("Unexpected case: no cmec file found")
        return ExecutionResult.build_from_failure(definition)

    # PMP writes plots and data into a per-variable subdirectory (``ts`` for 2D variables,
    # ``ta-200``/``ta-850`` for 3D pressure-level variables). Glob from the results directory
    # so the ``variable_dir_pattern`` wildcard matches every level subdir.
    # Sort so the committed output.json plot/data key order is deterministic across hosts.
    png_files = [
        definition.as_relative_path(f)
        for f in sorted(results_directory.glob(f"{variable_dir_pattern}/*.png"))
    ]
    data_files = [
        definition.as_relative_path(f)
        for f in sorted(results_directory.glob(f"{variable_dir_pattern}/*.nc"))
    ]

    # Prepare the output bundles
    cmec_output_bundle, cmec_metric_bundle = process_json_result(results_file, png_files, data_files)

    # Add missing dimensions to the output
    member_id_col = "variant_label" if model_source_type == SourceDatasetType.CMIP7 else "member_id"
    reference_collection = definition.datasets[SourceDatasetType.PMPClimatology]
    cmec_metric_bundle = cmec_metric_bundle.prepend_dimensions(
        {
            # PMP scalars are model-performance scores against a reference, not reference
            # (observation) values, so every value's role is ``model``.
            "kind": "model",
            "source_id": input_datasets["source_id"].unique()[0],
            "member_id": input_datasets[member_id_col].unique()[0],
            "experiment_id": input_datasets["experiment_id"].unique()[0],
            "variable_id": input_datasets["variable_id"].unique()[0],
            "reference_source_id": reference_collection["source_id"].unique()[0],
        }
    )

    return ExecutionResult.build_from_output_bundle(
        definition,
        cmec_output_bundle=cmec_output_bundle,
        cmec_metric_bundle=cmec_metric_bundle,
    )

execute(definition)

Run the diagnostic on the given configuration.

Parameters:

Name	Type	Description	Default
definition	ExecutionDefinition	The configuration to run the diagnostic on.	required

Returns:

Type	Description
None	The result of running the diagnostic.

Source code in packages/climate-ref-pmp/src/climate_ref_pmp/diagnostics/annual_cycle.py

def execute(self, definition: ExecutionDefinition) -> None:
    """
    Run the diagnostic on the given configuration.

    Parameters
    ----------
    definition : ExecutionDefinition
        The configuration to run the diagnostic on.

    Returns
    -------
    :
        The result of running the diagnostic.
    """
    cmds = self.build_cmds(definition)

    runs = [self.provider.run(cmd) for cmd in cmds]
    logger.debug(f"runs: {runs}")

ENSO

Bases: CommandLineDiagnostic

Calculate the ENSO performance metrics for a dataset

Source code in packages/climate-ref-pmp/src/climate_ref_pmp/diagnostics/enso.py

class ENSO(CommandLineDiagnostic):
    """
    Calculate the ENSO performance metrics for a dataset
    """

    reconstruction_inputs = PMP_RECONSTRUCTION_INPUTS

    version = 2

    facets = (
        "kind",
        "mip_id",
        "source_id",
        "member_id",
        "grid_label",
        "experiment_id",
        "metric",
        "reference_source_id",
    )

    def __init__(
        self, metrics_collection: str, experiments: Collection[str] = ("historical", "esm-hist")
    ) -> None:
        self.name = metrics_collection
        self.slug = metrics_collection.lower()
        self.metrics_collection = metrics_collection
        self.parameter_file = "pmp_param_enso.py"
        self.obs_sources: tuple[str, ...]
        self.model_variables: tuple[str, ...]

        if metrics_collection == "ENSO_perf":  # pragma: no cover
            self.model_variables = ("pr", "ts", "tauu")
            self.obs_sources = ("GPCP-Monthly-3-2", "TropFlux-1-0", "HadISST-1-1")
        elif metrics_collection == "ENSO_tel":
            self.model_variables = ("pr", "ts")
            self.obs_sources = ("GPCP-Monthly-3-2", "TropFlux-1-0", "HadISST-1-1")
        elif metrics_collection == "ENSO_proc":
            self.model_variables = ("ts", "tauu", "hfls", "hfss", "rlds", "rlus", "rsds", "rsus")
            self.obs_sources = (
                "GPCP-Monthly-3-2",
                "TropFlux-1-0",
                "HadISST-1-1",
                "CERES-EBAF-4-2",
            )
        else:
            raise ValueError(
                f"Unknown metrics collection: {metrics_collection}. "
                "Valid options are: ENSO_perf, ENSO_tel, ENSO_proc"
            )

        self.data_requirements = self._get_data_requirements(experiments)

        self.test_data_spec = TestDataSpecification(
            test_cases=(
                TestCase(
                    name="cmip6",
                    description=f"Test {metrics_collection} with CMIP6 data",
                    requests=(
                        RegistryRequest(
                            slug=f"enso-{self.slug}-obs",
                            registry_name="obs4ref",
                            source_type="obs4REF",
                            facets={
                                "source_id": self.obs_sources,
                                "variable_id": self.model_variables,
                            },
                        ),
                        CMIP6Request(
                            slug=f"enso-{self.slug}-cmip6",
                            facets={
                                "source_id": "ACCESS-ESM1-5",
                                "experiment_id": "historical",
                                "variable_id": self.model_variables,
                                "member_id": "r1i1p1f1",
                                "table_id": "Amon",
                            },
                            time_span=("2000-01", "2014-12"),
                        ),
                        CMIP6Request(
                            slug=f"enso-{self.slug}-areacella",
                            facets={
                                "source_id": "ACCESS-ESM1-5",
                                "experiment_id": "historical",
                                "variable_id": "areacella",
                                "member_id": "r1i1p1f1",
                                "table_id": "fx",
                            },
                        ),
                        CMIP6Request(
                            slug=f"enso-{self.slug}-sftlf",
                            facets={
                                "source_id": "ACCESS-ESM1-5",
                                "experiment_id": "historical",
                                "variable_id": "sftlf",
                                "member_id": "r1i1p1f1",
                                "table_id": "fx",
                            },
                        ),
                    ),
                ),
                TestCase(
                    name="cmip7",
                    description=f"CMIP7 test case for {metrics_collection}",
                    requests=(
                        RegistryRequest(
                            slug=f"enso-{self.slug}-obs-cmip7",
                            registry_name="obs4ref",
                            source_type="obs4REF",
                            facets={
                                "source_id": self.obs_sources,
                                "variable_id": self.model_variables,
                            },
                        ),
                        CMIP7Request(
                            slug=f"enso-{self.slug}-cmip7",
                            facets={
                                "source_id": "ACCESS-ESM1-5",
                                "experiment_id": "historical",
                                "variable_id": self.model_variables,
                                "branded_variable": tuple(
                                    _BRANDED_VARIABLE_NAMES[v] for v in self.model_variables
                                ),
                                "variant_label": "r1i1p1f1",
                                "frequency": "mon",
                                "region": "glb",
                            },
                            time_span=("2000-01", "2014-12"),
                        ),
                        CMIP7Request(
                            slug=f"enso-{self.slug}-areacella-cmip7",
                            facets={
                                "source_id": "ACCESS-ESM1-5",
                                "experiment_id": "historical",
                                "variable_id": "areacella",
                                "branded_variable": "areacella_ti-u-hxy-u",
                                "variant_label": "r1i1p1f1",
                                "frequency": "fx",
                                "region": "glb",
                            },
                        ),
                        CMIP7Request(
                            slug=f"enso-{self.slug}-sftlf-cmip7",
                            facets={
                                "source_id": "ACCESS-ESM1-5",
                                "experiment_id": "historical",
                                "variable_id": "sftlf",
                                "branded_variable": "sftlf_ti-u-hxy-u",
                                "variant_label": "r1i1p1f1",
                                "frequency": "fx",
                                "region": "glb",
                            },
                        ),
                    ),
                ),
            ),
        )

    def _get_data_requirements(
        self,
        experiments: Collection[str] = ("historical", "esm-hist"),
    ) -> tuple[tuple[DataRequirement, DataRequirement], ...]:
        cmip6_filters = [
            FacetFilter(
                facets={
                    "frequency": "mon",
                    "experiment_id": tuple(experiments),
                    "variable_id": self.model_variables,
                }
            )
        ]

        cmip7_filters = [
            FacetFilter(
                facets={
                    "branded_variable": tuple(_BRANDED_VARIABLE_NAMES[v] for v in self.model_variables),
                    "experiment_id": tuple(experiments),
                    "frequency": "mon",
                    "region": "glb",
                }
            )
        ]

        obs_requirement = DataRequirement(
            source_type=SourceDatasetType.obs4MIPs,
            fallback_source_types=(SourceDatasetType.obs4REF,),
            filters=(
                FacetFilter(facets={"source_id": self.obs_sources, "variable_id": self.model_variables}),
            ),
            group_by=("activity_id",),
        )
        cmip6_requirement = DataRequirement(
            source_type=SourceDatasetType.CMIP6,
            filters=tuple(cmip6_filters),
            group_by=("source_id", "experiment_id", "member_id", "grid_label"),
            constraints=(
                AddSupplementaryDataset.from_defaults("areacella", SourceDatasetType.CMIP6),
                AddSupplementaryDataset.from_defaults("sftlf", SourceDatasetType.CMIP6),
            ),
        )
        cmip7_requirement = DataRequirement(
            source_type=SourceDatasetType.CMIP7,
            filters=tuple(cmip7_filters),
            group_by=("source_id", "experiment_id", "variant_label", "grid_label"),
            constraints=(
                AddSupplementaryDataset.from_defaults("areacella", SourceDatasetType.CMIP7),
                AddSupplementaryDataset.from_defaults("sftlf", SourceDatasetType.CMIP7),
            ),
        )

        return (
            (obs_requirement, cmip6_requirement),
            (obs_requirement, cmip7_requirement),
        )

    def build_cmd(self, definition: ExecutionDefinition) -> Iterable[str]:
        """
        Run the diagnostic on the given configuration.

        Parameters
        ----------
        definition : ExecutionDefinition
            The configuration to run the diagnostic on.

        Returns
        -------
        :
            The result of running the diagnostic.
        """
        mc_name = self.metrics_collection

        # ------------------------------------------------
        # Get the input datasets information for the model
        # ------------------------------------------------
        model_source_type = get_model_source_type(definition)
        input_datasets = definition.datasets[model_source_type]
        source_id = input_datasets["source_id"].unique()[0]
        member_id_col = "variant_label" if model_source_type == SourceDatasetType.CMIP7 else "member_id"
        member_id = input_datasets[member_id_col].unique()[0]
        experiment_id = input_datasets["experiment_id"].unique()[0]
        variable_ids = set(input_datasets["variable_id"].unique()) - {"areacella", "sftlf"}
        mod_run = f"{source_id}_{member_id}"

        # We only need one entry for the model run
        dict_mod: dict[str, dict[str, Any]] = {mod_run: {}}

        def extract_variable(dc: DatasetCollection, variable: str) -> list[str]:
            return dc.datasets[input_datasets["variable_id"] == variable]["path"].to_list()  # type: ignore

        # TO DO: Get the path to the files per variable
        for variable in variable_ids:
            list_files = extract_variable(input_datasets, variable)
            list_areacella = extract_variable(input_datasets, "areacella")
            list_sftlf = extract_variable(input_datasets, "sftlf")

            if len(list_files) > 0:
                dict_mod[mod_run][variable] = {
                    "path + filename": list_files,
                    "varname": variable,
                    "path + filename_area": list_areacella,
                    "areaname": "areacella",
                    "path + filename_landmask": list_sftlf,
                    "landmaskname": "sftlf",
                }

        # -------------------------------------------------------
        # Get the input datasets information for the observations
        # -------------------------------------------------------
        reference_dataset = definition.datasets[SourceDatasetType.obs4MIPs]
        reference_dataset_names = reference_dataset["source_id"].unique()

        dict_obs: dict[str, dict[str, Any]] = {}

        # TO DO: Get the path to the files per variable and per source
        for obs_name in reference_dataset_names:
            dict_obs[obs_name] = {}
            for variable in variable_ids:
                # Get the list of files for the current variable and observation source
                list_files = reference_dataset.datasets[
                    (reference_dataset["variable_id"] == variable)
                    & (reference_dataset["source_id"] == obs_name)
                ]["path"].to_list()
                # If the list is not empty, add it to the dictionary
                if len(list_files) > 0:
                    dict_obs[obs_name][variable] = {
                        "path + filename": list_files,
                        "varname": variable,
                    }

        # Create input directory
        dict_datasets = {
            "model": dict_mod,
            "observations": dict_obs,
            "metricsCollection": mc_name,
            "experiment_id": experiment_id,
        }

        # Create JSON file for dictDatasets
        json_file = os.path.join(
            definition.output_directory, f"input_{mc_name}_{source_id}_{experiment_id}_{member_id}.json"
        )
        with open(json_file, "w") as f:
            json.dump(dict_datasets, f, indent=4)
        logger.debug(f"JSON file created: {json_file}")

        driver_file = _get_resource("climate_ref_pmp.drivers", "enso_driver.py", use_resources=True)
        return [
            "python",
            driver_file,
            "--metrics_collection",
            mc_name,
            "--experiment_id",
            experiment_id,
            "--input_json_path",
            json_file,
            "--output_directory",
            str(definition.output_directory),
        ]

    def build_execution_result(self, definition: ExecutionDefinition) -> ExecutionResult:
        """
        Build a diagnostic result from the output of the PMP driver

        Parameters
        ----------
        definition
            Definition of the diagnostic execution

        Returns
        -------
            Result of the diagnostic execution
        """
        model_source_type = get_model_source_type(definition)
        input_datasets = definition.datasets[model_source_type]
        source_id = input_datasets["source_id"].unique()[0]
        experiment_id = input_datasets["experiment_id"].unique()[0]
        member_id_col = "variant_label" if model_source_type == SourceDatasetType.CMIP7 else "member_id"
        member_id = input_datasets[member_id_col].unique()[0]
        mc_name = self.metrics_collection
        pattern = f"{mc_name}_{source_id}_{experiment_id}_{member_id}"

        # Find the results files
        results_files = list(definition.output_directory.glob(f"{pattern}_cmec.json"))
        logger.debug(f"Results files: {results_files}")

        if len(results_files) != 1:  # pragma: no cover
            logger.warning(f"A single cmec output file not found: {results_files}")
            return ExecutionResult.build_from_failure(definition)

        # Sort so the committed output.json plot/data key order is deterministic across hosts.
        output_dir = definition.output_directory
        png_files = [definition.as_relative_path(f) for f in sorted(output_dir.glob("*.png"))]
        data_files = [definition.as_relative_path(f) for f in sorted(output_dir.glob("*.nc"))]

        cmec_output, cmec_metric = process_json_result(results_files[0], png_files, data_files)

        cmec_metric_bundle = cmec_metric.remove_dimensions(
            [
                "model",
                "realization",
            ],
        ).prepend_dimensions(
            {
                # PMP scalars are model-performance scores against a reference, not reference
                # (observation) values, so every value's role is ``model``.
                "kind": "model",
                "mip_id": model_source_type.value,
                "source_id": source_id,
                "member_id": member_id,
                "grid_label": input_datasets["grid_label"].unique()[0],
                "experiment_id": experiment_id,
            }
        )

        return ExecutionResult.build_from_output_bundle(
            definition,
            cmec_output_bundle=cmec_output,
            cmec_metric_bundle=cmec_metric_bundle,
        )

build_cmd(definition)

Run the diagnostic on the given configuration.

Parameters:

Name	Type	Description	Default
definition	ExecutionDefinition	The configuration to run the diagnostic on.	required

Returns:

Type	Description
Iterable[str]	The result of running the diagnostic.

Source code in packages/climate-ref-pmp/src/climate_ref_pmp/diagnostics/enso.py

def build_cmd(self, definition: ExecutionDefinition) -> Iterable[str]:
    """
    Run the diagnostic on the given configuration.

    Parameters
    ----------
    definition : ExecutionDefinition
        The configuration to run the diagnostic on.

    Returns
    -------
    :
        The result of running the diagnostic.
    """
    mc_name = self.metrics_collection

    # ------------------------------------------------
    # Get the input datasets information for the model
    # ------------------------------------------------
    model_source_type = get_model_source_type(definition)
    input_datasets = definition.datasets[model_source_type]
    source_id = input_datasets["source_id"].unique()[0]
    member_id_col = "variant_label" if model_source_type == SourceDatasetType.CMIP7 else "member_id"
    member_id = input_datasets[member_id_col].unique()[0]
    experiment_id = input_datasets["experiment_id"].unique()[0]
    variable_ids = set(input_datasets["variable_id"].unique()) - {"areacella", "sftlf"}
    mod_run = f"{source_id}_{member_id}"

    # We only need one entry for the model run
    dict_mod: dict[str, dict[str, Any]] = {mod_run: {}}

    def extract_variable(dc: DatasetCollection, variable: str) -> list[str]:
        return dc.datasets[input_datasets["variable_id"] == variable]["path"].to_list()  # type: ignore

    # TO DO: Get the path to the files per variable
    for variable in variable_ids:
        list_files = extract_variable(input_datasets, variable)
        list_areacella = extract_variable(input_datasets, "areacella")
        list_sftlf = extract_variable(input_datasets, "sftlf")

        if len(list_files) > 0:
            dict_mod[mod_run][variable] = {
                "path + filename": list_files,
                "varname": variable,
                "path + filename_area": list_areacella,
                "areaname": "areacella",
                "path + filename_landmask": list_sftlf,
                "landmaskname": "sftlf",
            }

    # -------------------------------------------------------
    # Get the input datasets information for the observations
    # -------------------------------------------------------
    reference_dataset = definition.datasets[SourceDatasetType.obs4MIPs]
    reference_dataset_names = reference_dataset["source_id"].unique()

    dict_obs: dict[str, dict[str, Any]] = {}

    # TO DO: Get the path to the files per variable and per source
    for obs_name in reference_dataset_names:
        dict_obs[obs_name] = {}
        for variable in variable_ids:
            # Get the list of files for the current variable and observation source
            list_files = reference_dataset.datasets[
                (reference_dataset["variable_id"] == variable)
                & (reference_dataset["source_id"] == obs_name)
            ]["path"].to_list()
            # If the list is not empty, add it to the dictionary
            if len(list_files) > 0:
                dict_obs[obs_name][variable] = {
                    "path + filename": list_files,
                    "varname": variable,
                }

    # Create input directory
    dict_datasets = {
        "model": dict_mod,
        "observations": dict_obs,
        "metricsCollection": mc_name,
        "experiment_id": experiment_id,
    }

    # Create JSON file for dictDatasets
    json_file = os.path.join(
        definition.output_directory, f"input_{mc_name}_{source_id}_{experiment_id}_{member_id}.json"
    )
    with open(json_file, "w") as f:
        json.dump(dict_datasets, f, indent=4)
    logger.debug(f"JSON file created: {json_file}")

    driver_file = _get_resource("climate_ref_pmp.drivers", "enso_driver.py", use_resources=True)
    return [
        "python",
        driver_file,
        "--metrics_collection",
        mc_name,
        "--experiment_id",
        experiment_id,
        "--input_json_path",
        json_file,
        "--output_directory",
        str(definition.output_directory),
    ]

build_execution_result(definition)

Build a diagnostic result from the output of the PMP driver

Parameters:

Name	Type	Description	Default
definition	ExecutionDefinition	Definition of the diagnostic execution	required

Returns:

Type	Description
Result of the diagnostic execution

Source code in packages/climate-ref-pmp/src/climate_ref_pmp/diagnostics/enso.py

def build_execution_result(self, definition: ExecutionDefinition) -> ExecutionResult:
    """
    Build a diagnostic result from the output of the PMP driver

    Parameters
    ----------
    definition
        Definition of the diagnostic execution

    Returns
    -------
        Result of the diagnostic execution
    """
    model_source_type = get_model_source_type(definition)
    input_datasets = definition.datasets[model_source_type]
    source_id = input_datasets["source_id"].unique()[0]
    experiment_id = input_datasets["experiment_id"].unique()[0]
    member_id_col = "variant_label" if model_source_type == SourceDatasetType.CMIP7 else "member_id"
    member_id = input_datasets[member_id_col].unique()[0]
    mc_name = self.metrics_collection
    pattern = f"{mc_name}_{source_id}_{experiment_id}_{member_id}"

    # Find the results files
    results_files = list(definition.output_directory.glob(f"{pattern}_cmec.json"))
    logger.debug(f"Results files: {results_files}")

    if len(results_files) != 1:  # pragma: no cover
        logger.warning(f"A single cmec output file not found: {results_files}")
        return ExecutionResult.build_from_failure(definition)

    # Sort so the committed output.json plot/data key order is deterministic across hosts.
    output_dir = definition.output_directory
    png_files = [definition.as_relative_path(f) for f in sorted(output_dir.glob("*.png"))]
    data_files = [definition.as_relative_path(f) for f in sorted(output_dir.glob("*.nc"))]

    cmec_output, cmec_metric = process_json_result(results_files[0], png_files, data_files)

    cmec_metric_bundle = cmec_metric.remove_dimensions(
        [
            "model",
            "realization",
        ],
    ).prepend_dimensions(
        {
            # PMP scalars are model-performance scores against a reference, not reference
            # (observation) values, so every value's role is ``model``.
            "kind": "model",
            "mip_id": model_source_type.value,
            "source_id": source_id,
            "member_id": member_id,
            "grid_label": input_datasets["grid_label"].unique()[0],
            "experiment_id": experiment_id,
        }
    )

    return ExecutionResult.build_from_output_bundle(
        definition,
        cmec_output_bundle=cmec_output,
        cmec_metric_bundle=cmec_metric_bundle,
    )

ExtratropicalModesOfVariability

Bases: CommandLineDiagnostic

Calculate the extratropical modes of variability for a given area

Source code in packages/climate-ref-pmp/src/climate_ref_pmp/diagnostics/variability_modes.py

class ExtratropicalModesOfVariability(CommandLineDiagnostic):
    """
    Calculate the extratropical modes of variability for a given area
    """

    reconstruction_inputs = PMP_RECONSTRUCTION_INPUTS

    ts_modes = ("PDO", "NPGO", "AMO")
    psl_modes = ("NAO", "NAM", "PNA", "NPO", "SAM")

    version = 3

    facets = (
        "kind",
        "mip_id",
        "source_id",
        "member_id",
        "experiment_id",
        "reference_source_id",
        "mode",
        "season",
        "method",
        "statistic",
    )

    def __init__(self, mode_id: str):
        super().__init__()
        self.mode_id = mode_id.upper()
        self.name = f"Extratropical modes of variability: {mode_id}"
        self.slug = f"extratropical-modes-of-variability-{mode_id.lower()}"

        def _get_data_requirements(
            obs_source: str,
            obs_variable: str,
            model_variable: str,
            extra_experiments: str | tuple[str, ...] | list[str] = (),
        ) -> tuple[tuple[DataRequirement, DataRequirement], ...]:
            cmip6_filters = [
                FacetFilter(
                    facets={
                        "frequency": "mon",
                        "table_id": "Amon",
                        "experiment_id": ("historical", "esm-hist", "hist-GHG", *extra_experiments),
                        "variable_id": model_variable,
                    }
                )
            ]

            cmip7_filters = [
                FacetFilter(
                    facets={
                        "branded_variable": (_BRANDED_VARIABLE_NAMES[model_variable],),
                        "experiment_id": ("historical", "esm-hist", "hist-GHG", *extra_experiments),
                        "frequency": "mon",
                        "region": "glb",
                    }
                )
            ]

            obs_requirement = DataRequirement(
                source_type=SourceDatasetType.obs4MIPs,
                fallback_source_types=(SourceDatasetType.obs4REF,),
                filters=(FacetFilter(facets={"source_id": (obs_source,), "variable_id": (obs_variable,)}),),
                group_by=("source_id", "variable_id"),
            )
            cmip6_requirement = DataRequirement(
                source_type=SourceDatasetType.CMIP6,
                filters=tuple(cmip6_filters),
                group_by=("source_id", "experiment_id", "member_id", "grid_label"),
            )
            cmip7_requirement = DataRequirement(
                source_type=SourceDatasetType.CMIP7,
                filters=tuple(cmip7_filters),
                group_by=("source_id", "experiment_id", "variant_label", "grid_label"),
            )

            return (
                (obs_requirement, cmip6_requirement),
                (obs_requirement, cmip7_requirement),
            )

        if self.mode_id in self.ts_modes:
            self.parameter_file = "pmp_param_MoV-ts.py"
            self.data_requirements = _get_data_requirements("HadISST-1-1", "ts", "ts")
            self.test_data_spec = TestDataSpecification(
                test_cases=(
                    TestCase(
                        name="cmip6",
                        description=f"Test {self.mode_id} with CMIP6 ts data and HadISST obs",
                        requests=(
                            RegistryRequest(
                                slug=f"mov-{self.mode_id.lower()}-obs",
                                registry_name="obs4ref",
                                source_type="obs4REF",
                                facets={"source_id": "HadISST-1-1", "variable_id": "ts"},
                            ),
                            CMIP6Request(
                                slug=f"mov-{self.mode_id.lower()}-cmip6",
                                facets={
                                    "source_id": "ACCESS-ESM1-5",
                                    "experiment_id": "historical",
                                    "variable_id": "ts",
                                    "member_id": "r1i1p1f1",
                                    "frequency": "mon",
                                    "table_id": "Amon",
                                },
                                time_span=("2000-01", "2014-12"),
                            ),
                        ),
                    ),
                    TestCase(
                        name="cmip7",
                        description=f"CMIP7 test case for {self.mode_id}",
                        requests=(
                            RegistryRequest(
                                slug=f"mov-{self.mode_id.lower()}-obs-cmip7",
                                registry_name="obs4ref",
                                source_type="obs4REF",
                                facets={"source_id": "HadISST-1-1", "variable_id": "ts"},
                            ),
                            CMIP7Request(
                                slug=f"mov-{self.mode_id.lower()}-cmip7",
                                facets={
                                    "source_id": "ACCESS-ESM1-5",
                                    "experiment_id": "historical",
                                    "variable_id": "ts",
                                    "branded_variable": "ts_tavg-u-hxy-u",
                                    "variant_label": "r1i1p1f1",
                                    "frequency": "mon",
                                    "region": "glb",
                                },
                                time_span=("2000-01", "2014-12"),
                            ),
                        ),
                    ),
                ),
            )
        elif self.mode_id in self.psl_modes:
            self.parameter_file = "pmp_param_MoV-psl.py"
            self.data_requirements = _get_data_requirements(
                "20CR-V2", "psl", "psl", extra_experiments=("amip",)
            )
            self.test_data_spec = TestDataSpecification(
                test_cases=(
                    TestCase(
                        name="cmip6",
                        description=f"Test {self.mode_id} with CMIP6 psl data and 20CR-V2 obs",
                        requests=(
                            Obs4MIPsRequest(
                                slug=f"mov-{self.mode_id.lower()}-obs",
                                facets={"source_id": "20CR-V2", "variable_id": "psl"},
                            ),
                            CMIP6Request(
                                slug=f"mov-{self.mode_id.lower()}-cmip6",
                                facets={
                                    "source_id": "ACCESS-ESM1-5",
                                    "experiment_id": "historical",
                                    "variable_id": "psl",
                                    "member_id": "r1i1p1f1",
                                    "frequency": "mon",
                                    "table_id": "Amon",
                                },
                                time_span=("2000-01", "2014-12"),
                            ),
                        ),
                    ),
                    TestCase(
                        name="cmip7",
                        description=f"CMIP7 test case for {self.mode_id}",
                        requests=(
                            Obs4MIPsRequest(
                                slug=f"mov-{self.mode_id.lower()}-obs-cmip7",
                                facets={"source_id": "20CR-V2", "variable_id": "psl"},
                            ),
                            CMIP7Request(
                                slug=f"mov-{self.mode_id.lower()}-cmip7",
                                facets={
                                    "source_id": "ACCESS-ESM1-5",
                                    "experiment_id": "historical",
                                    "variable_id": "psl",
                                    "branded_variable": "psl_tavg-u-hxy-u",
                                    "variant_label": "r1i1p1f1",
                                    "frequency": "mon",
                                    "region": "glb",
                                },
                                time_span=("2000-01", "2014-12"),
                            ),
                        ),
                    ),
                ),
            )
        else:
            raise ValueError(
                f"Unknown mode_id '{self.mode_id}'. Must be one of {self.ts_modes + self.psl_modes}"
            )

    def build_cmd(self, definition: ExecutionDefinition) -> Iterable[str]:
        """
        Build the command to run the diagnostic

        Parameters
        ----------
        definition
            Definition of the diagnostic execution

        Returns
        -------
            Command arguments to execute in the PMP environment
        """
        model_source_type = get_model_source_type(definition)
        input_datasets = definition.datasets[model_source_type]
        source_id = input_datasets["source_id"].unique()[0]
        experiment_id = input_datasets["experiment_id"].unique()[0]
        member_id_col = "variant_label" if model_source_type == SourceDatasetType.CMIP7 else "member_id"
        member_id = input_datasets[member_id_col].unique()[0]

        logger.debug(f"input_datasets: {input_datasets}")
        logger.debug(f"source_id: {source_id}")
        logger.debug(f"experiment_id: {experiment_id}")
        logger.debug(f"member_id: {member_id}")

        reference_dataset = definition.datasets[SourceDatasetType.obs4MIPs]
        reference_dataset_name = reference_dataset["source_id"].unique()[0]
        reference_dataset_path = reference_dataset.datasets.iloc[0]["path"]

        logger.debug(f"reference_dataset: {reference_dataset}")
        logger.debug(f"reference_dataset_name: {reference_dataset_name}")
        logger.debug(f"reference_dataset_path: {reference_dataset_path}")

        model_files = input_datasets.path.to_list()

        if isinstance(model_files, list):
            modpath = get_wildcard_pattern(model_files)
            logger.debug(f"model_files: {model_files}")
            logger.debug(f"modpath: {modpath}")
        else:
            modpath = model_files

        if isinstance(reference_dataset_path, list):
            reference_data_path = " ".join([str(p) for p in reference_dataset_path])
        else:
            reference_data_path = reference_dataset_path

        # Build the command to run the PMP driver script
        params: dict[str, str | int | None] = {
            "variability_mode": self.mode_id,
            "modpath": modpath,
            "modpath_lf": "none",
            "mip": model_source_type.value,
            "exp": experiment_id,
            "realization": member_id,
            "modnames": source_id,
            "reference_data_name": reference_dataset_name,
            "reference_data_path": reference_data_path,
            "results_dir": str(definition.output_directory),
            "cmec": None,
            "no_provenance": None,
        }

        # Add conditional parameters
        if self.mode_id in ["SAM"]:  # pragma: no cover
            params["osyear"] = 1950
            params["oeyear"] = 2005

        if self.mode_id in ["NPO", "NPGO"]:
            params["eofn_obs"] = 2
            params["eofn_mod"] = 2
            params["eofn_mod_max"] = 2

        # Pass the parameters using **kwargs
        return build_pmp_command(
            driver_file="variability_modes_driver.py",
            parameter_file=self.parameter_file,
            **params,
        )

    def build_execution_result(self, definition: ExecutionDefinition) -> ExecutionResult:
        """
        Build a diagnostic result from the output of the PMP driver

        Parameters
        ----------
        definition
            Definition of the diagnostic execution

        Returns
        -------
            Result of the diagnostic execution
        """
        model_source_type = get_model_source_type(definition)
        mip = model_source_type.value

        # Use mip-scoped glob to avoid matching files from other MIP runs
        results_files = list(definition.output_directory.glob(f"*_{mip}_*_cmec.json"))
        if len(results_files) != 1:  # pragma: no cover
            logger.warning(f"A single cmec output file not found: {results_files}")
            return ExecutionResult.build_from_failure(definition)

        clean_up_json(results_files[0])

        # Sort so the committed output.json plot/data key order is deterministic across hosts.
        output_dir = definition.output_directory
        png_files = [definition.as_relative_path(f) for f in sorted(output_dir.glob("*.png"))]
        data_files = [definition.as_relative_path(f) for f in sorted(output_dir.glob("*.nc"))]

        cmec_output_bundle, cmec_metric_bundle = process_json_result(results_files[0], png_files, data_files)
        input_datasets = definition.datasets[model_source_type]
        reference_collection = definition.datasets[SourceDatasetType.obs4MIPs]
        member_id_col = "variant_label" if model_source_type == SourceDatasetType.CMIP7 else "member_id"
        cmec_metric_bundle = cmec_metric_bundle.remove_dimensions(
            [
                "model",
                "realization",
                "reference",
            ],
        ).prepend_dimensions(
            {
                # PMP scalars are model-performance scores against a reference, not reference
                # (observation) values, so every value's role is ``model``.
                "kind": "model",
                "mip_id": model_source_type.value,
                "source_id": input_datasets["source_id"].unique()[0],
                "member_id": input_datasets[member_id_col].unique()[0],
                "experiment_id": input_datasets["experiment_id"].unique()[0],
                "reference_source_id": reference_collection["source_id"].unique()[0],
            }
        )

        return ExecutionResult.build_from_output_bundle(
            definition,
            cmec_output_bundle=cmec_output_bundle,
            cmec_metric_bundle=cmec_metric_bundle,
        )

build_cmd(definition)

Build the command to run the diagnostic

Parameters:

Name	Type	Description	Default
definition	ExecutionDefinition	Definition of the diagnostic execution	required

Returns:

Type	Description
Command arguments to execute in the PMP environment

Source code in packages/climate-ref-pmp/src/climate_ref_pmp/diagnostics/variability_modes.py

def build_cmd(self, definition: ExecutionDefinition) -> Iterable[str]:
    """
    Build the command to run the diagnostic

    Parameters
    ----------
    definition
        Definition of the diagnostic execution

    Returns
    -------
        Command arguments to execute in the PMP environment
    """
    model_source_type = get_model_source_type(definition)
    input_datasets = definition.datasets[model_source_type]
    source_id = input_datasets["source_id"].unique()[0]
    experiment_id = input_datasets["experiment_id"].unique()[0]
    member_id_col = "variant_label" if model_source_type == SourceDatasetType.CMIP7 else "member_id"
    member_id = input_datasets[member_id_col].unique()[0]

    logger.debug(f"input_datasets: {input_datasets}")
    logger.debug(f"source_id: {source_id}")
    logger.debug(f"experiment_id: {experiment_id}")
    logger.debug(f"member_id: {member_id}")

    reference_dataset = definition.datasets[SourceDatasetType.obs4MIPs]
    reference_dataset_name = reference_dataset["source_id"].unique()[0]
    reference_dataset_path = reference_dataset.datasets.iloc[0]["path"]

    logger.debug(f"reference_dataset: {reference_dataset}")
    logger.debug(f"reference_dataset_name: {reference_dataset_name}")
    logger.debug(f"reference_dataset_path: {reference_dataset_path}")

    model_files = input_datasets.path.to_list()

    if isinstance(model_files, list):
        modpath = get_wildcard_pattern(model_files)
        logger.debug(f"model_files: {model_files}")
        logger.debug(f"modpath: {modpath}")
    else:
        modpath = model_files

    if isinstance(reference_dataset_path, list):
        reference_data_path = " ".join([str(p) for p in reference_dataset_path])
    else:
        reference_data_path = reference_dataset_path

    # Build the command to run the PMP driver script
    params: dict[str, str | int | None] = {
        "variability_mode": self.mode_id,
        "modpath": modpath,
        "modpath_lf": "none",
        "mip": model_source_type.value,
        "exp": experiment_id,
        "realization": member_id,
        "modnames": source_id,
        "reference_data_name": reference_dataset_name,
        "reference_data_path": reference_data_path,
        "results_dir": str(definition.output_directory),
        "cmec": None,
        "no_provenance": None,
    }

    # Add conditional parameters
    if self.mode_id in ["SAM"]:  # pragma: no cover
        params["osyear"] = 1950
        params["oeyear"] = 2005

    if self.mode_id in ["NPO", "NPGO"]:
        params["eofn_obs"] = 2
        params["eofn_mod"] = 2
        params["eofn_mod_max"] = 2

    # Pass the parameters using **kwargs
    return build_pmp_command(
        driver_file="variability_modes_driver.py",
        parameter_file=self.parameter_file,
        **params,
    )

build_execution_result(definition)

Build a diagnostic result from the output of the PMP driver

Parameters:

Name	Type	Description	Default
definition	ExecutionDefinition	Definition of the diagnostic execution	required

Returns:

Type	Description
Result of the diagnostic execution

Source code in packages/climate-ref-pmp/src/climate_ref_pmp/diagnostics/variability_modes.py

def build_execution_result(self, definition: ExecutionDefinition) -> ExecutionResult:
    """
    Build a diagnostic result from the output of the PMP driver

    Parameters
    ----------
    definition
        Definition of the diagnostic execution

    Returns
    -------
        Result of the diagnostic execution
    """
    model_source_type = get_model_source_type(definition)
    mip = model_source_type.value

    # Use mip-scoped glob to avoid matching files from other MIP runs
    results_files = list(definition.output_directory.glob(f"*_{mip}_*_cmec.json"))
    if len(results_files) != 1:  # pragma: no cover
        logger.warning(f"A single cmec output file not found: {results_files}")
        return ExecutionResult.build_from_failure(definition)

    clean_up_json(results_files[0])

    # Sort so the committed output.json plot/data key order is deterministic across hosts.
    output_dir = definition.output_directory
    png_files = [definition.as_relative_path(f) for f in sorted(output_dir.glob("*.png"))]
    data_files = [definition.as_relative_path(f) for f in sorted(output_dir.glob("*.nc"))]

    cmec_output_bundle, cmec_metric_bundle = process_json_result(results_files[0], png_files, data_files)
    input_datasets = definition.datasets[model_source_type]
    reference_collection = definition.datasets[SourceDatasetType.obs4MIPs]
    member_id_col = "variant_label" if model_source_type == SourceDatasetType.CMIP7 else "member_id"
    cmec_metric_bundle = cmec_metric_bundle.remove_dimensions(
        [
            "model",
            "realization",
            "reference",
        ],
    ).prepend_dimensions(
        {
            # PMP scalars are model-performance scores against a reference, not reference
            # (observation) values, so every value's role is ``model``.
            "kind": "model",
            "mip_id": model_source_type.value,
            "source_id": input_datasets["source_id"].unique()[0],
            "member_id": input_datasets[member_id_col].unique()[0],
            "experiment_id": input_datasets["experiment_id"].unique()[0],
            "reference_source_id": reference_collection["source_id"].unique()[0],
        }
    )

    return ExecutionResult.build_from_output_bundle(
        definition,
        cmec_output_bundle=cmec_output_bundle,
        cmec_metric_bundle=cmec_metric_bundle,
    )

sub-packages

Sub-package	Description
annual_cycle
enso
variability_modes