gcages.testing

Code to support our tests

This is here, rather than in our tests directory because of the issues that come when you turn your tests into a package using __init__.py files (for details, see https://docs.pytest.org/en/7.1.x/explanation/goodpractices.html#choosing-an-import-mode).

Functions:

Name	Description
assert_frame_equal	Assert two pd.DataFrame's are equal.
get_ar6_all_emissions	Get all emissions from AR6 for a given model-scenario
get_ar6_harmonised_emissions	Get all harmonised emissions from AR6 for a given model-scenario
get_ar6_infilled_emissions	Get all infilled emissions from AR6 for a given model-scenario
get_ar6_metadata_outputs	Get metadata from AR6 for a given model-scenario
get_ar6_raw_emissions	Get all raw emissions from AR6 for a given model-scenario
get_ar6_temperature_outputs	Get temperature outputs we've downloaded from AR6 for a given model-scenario
get_cmip7_scenariomip_complete_emissions	Get complete emissions from CMIP7 ScenarioMIP outputs
get_cmip7_scenariomip_harmonised_emissions	Get harmonised emissions from CMIP7 ScenarioMIP outputs
get_cmip7_scenariomip_pre_processed_emissions	Get pre-processed emissions from CMIP7 ScenarioMIP outputs
get_key_testing_model_scenario_parameters	Create a pytest parameterization decorator for model-scenario pairs.
guess_magicc_exe	Guess the MAGICC executable based on the operating system

assert_frame_equal

assert_frame_equal(
    res: DataFrame,
    exp: DataFrame,
    rtol: float = 1e-08,
    **kwargs: Any,
) -> None

Assert two pd.DataFrame's are equal.

This is a very thin wrapper around pd.testing.assert_frame_equal that makes some use of pandas_indexing to give slightly nicer and clearer errors.

Parameters:

Name	Type	Description	Default
res	DataFrame	Result	required
exp	DataFrame	Expected value	required
rtol	float	Relative tolerance	1e-08
**kwargs	Any	Passed to pd.testing.assert_frame_equal	{}

Raises:

Type	Description
AssertionError	The frames aren't equal

Source code in src/gcages/testing.py

def assert_frame_equal(
    res: pd.DataFrame, exp: pd.DataFrame, rtol: float = 1e-8, **kwargs: Any
) -> None:
    """
    Assert two [pd.DataFrame][pandas.DataFrame]'s are equal.

    This is a very thin wrapper around
    [pd.testing.assert_frame_equal][pandas.testing.assert_frame_equal]
    that makes some use of [pandas_indexing][]
    to give slightly nicer and clearer errors.

    Parameters
    ----------
    res
        Result

    exp
        Expected value

    rtol
        Relative tolerance

    **kwargs
        Passed to [pd.testing.assert_frame_equal][pandas.testing.assert_frame_equal]

    Raises
    ------
    AssertionError
        The frames aren't equal
    """
    try:
        from pandas_indexing.core import uniquelevel
    except ImportError as exc:
        raise MissingOptionalDependencyError(
            "assert_frame_equal", requirement="pandas_indexing"
        ) from exc

    for idx_name in res.index.names:
        idx_diffs = uniquelevel(res, idx_name).symmetric_difference(
            uniquelevel(exp, idx_name)
        )
        if not idx_diffs.empty:
            msg = f"Differences in the {idx_name} (res on the left): {idx_diffs=}"
            raise AssertionError(msg)

    pd.testing.assert_frame_equal(
        res.reorder_levels(exp.index.names).T,
        exp.T,
        check_like=True,
        check_exact=False,
        rtol=rtol,
        **kwargs,
    )

get_ar6_all_emissions cached

get_ar6_all_emissions(
    model: str,
    scenario: str,
    processed_ar6_output_data_dir: Path,
) -> DataFrame

Get all emissions from AR6 for a given model-scenario

Parameters:

Name	Type	Description	Default
model	str	Model	required
scenario	str	Scenario	required
processed_ar6_output_data_dir	Path	Directory in which the AR6 was processed into individual model-scenario files (In the repo, see tests/regression/ar6/convert_ar6_res_to_checking_csvs.py.)	required

Returns:

Type	Description
DataFrame	All emissions from AR6 for model-scenario

Source code in src/gcages/testing.py

@functools.cache
def get_ar6_all_emissions(
    model: str, scenario: str, processed_ar6_output_data_dir: Path
) -> pd.DataFrame:
    """
    Get all emissions from AR6 for a given model-scenario

    Parameters
    ----------
    model
        Model

    scenario
        Scenario

    processed_ar6_output_data_dir
        Directory in which the AR6 was processed into individual model-scenario files

        (In the repo, see `tests/regression/ar6/convert_ar6_res_to_checking_csvs.py`.)

    Returns
    -------
    :
        All emissions from AR6 for `model`-`scenario`
    """
    filename_emissions = f"ar6_scenarios__{model}__{scenario}__emissions.csv"
    filename_emissions = filename_emissions.replace("/", "_").replace(" ", "_")
    emissions_file = processed_ar6_output_data_dir / filename_emissions

    res = load_timeseries_csv(
        emissions_file,
        index_columns=["model", "scenario", "variable", "region", "unit"],
        out_columns_type=int,
    )

    return res

get_ar6_harmonised_emissions cached

get_ar6_harmonised_emissions(
    model: str,
    scenario: str,
    processed_ar6_output_data_dir: Path,
) -> DataFrame

Get all harmonised emissions from AR6 for a given model-scenario

Parameters:

Name	Type	Description	Default
model	str	Model	required
scenario	str	Scenario	required
processed_ar6_output_data_dir	Path	Directory in which the AR6 was processed into individual model-scenario files (In the repo, see tests/regression/ar6/convert_ar6_res_to_checking_csvs.py.)	required

Returns:

Type	Description
DataFrame	All harmonised emissions from AR6 for model-scenario

Source code in src/gcages/testing.py

@functools.cache
def get_ar6_harmonised_emissions(
    model: str, scenario: str, processed_ar6_output_data_dir: Path
) -> pd.DataFrame:
    """
    Get all harmonised emissions from AR6 for a given model-scenario

    Parameters
    ----------
    model
        Model

    scenario
        Scenario

    processed_ar6_output_data_dir
        Directory in which the AR6 was processed into individual model-scenario files

        (In the repo, see `tests/regression/ar6/convert_ar6_res_to_checking_csvs.py`.)

    Returns
    -------
    :
        All harmonised emissions from AR6 for `model`-`scenario`
    """
    try:
        from pandas_indexing.selectors import ismatch
    except ImportError as exc:
        raise MissingOptionalDependencyError(
            "get_ar6_harmonised_emissions", requirement="pandas_indexing"
        ) from exc

    all_emissions = get_ar6_all_emissions(
        model=model,
        scenario=scenario,
        processed_ar6_output_data_dir=processed_ar6_output_data_dir,
    )
    res: pd.DataFrame = all_emissions.loc[ismatch(variable="**Harmonized**")].dropna(
        how="all", axis="columns"
    )

    return res

get_ar6_infilled_emissions cached

get_ar6_infilled_emissions(
    model: str,
    scenario: str,
    processed_ar6_output_data_dir: Path,
) -> DataFrame

Get all infilled emissions from AR6 for a given model-scenario

Parameters:

Name	Type	Description	Default
model	str	Model	required
scenario	str	Scenario	required
processed_ar6_output_data_dir	Path	Directory in which the AR6 output was processed into model-scenario files (In the repo, see tests/regression/ar6/convert_ar6_res_to_checking_csvs.py.)	required

Returns:

Type	Description
DataFrame	All infilled emissions from AR6 for model-scenario

Source code in src/gcages/testing.py

@functools.cache
def get_ar6_infilled_emissions(
    model: str, scenario: str, processed_ar6_output_data_dir: Path
) -> pd.DataFrame:
    """
    Get all infilled emissions from AR6 for a given model-scenario

    Parameters
    ----------
    model
        Model

    scenario
        Scenario

    processed_ar6_output_data_dir
        Directory in which the AR6 output was processed into model-scenario files

        (In the repo, see `tests/regression/ar6/convert_ar6_res_to_checking_csvs.py`.)

    Returns
    -------
    :
        All infilled emissions from AR6 for `model`-`scenario`
    """
    try:
        from pandas_indexing.selectors import ismatch
    except ImportError as exc:
        raise MissingOptionalDependencyError(
            "get_ar6_infilled_emissions", requirement="pandas_indexing"
        ) from exc

    all_emissions = get_ar6_all_emissions(
        model=model,
        scenario=scenario,
        processed_ar6_output_data_dir=processed_ar6_output_data_dir,
    )
    res: pd.DataFrame = all_emissions.loc[ismatch(variable="**Infilled**")].dropna(
        how="all", axis="columns"
    )

    return res

get_ar6_metadata_outputs cached

get_ar6_metadata_outputs(
    model: str,
    scenario: str,
    ar6_output_data_dir: Path,
    filename: str = "AR6_Scenarios_Database_metadata_indicators_v1.1_meta.csv",
) -> DataFrame

Get metadata from AR6 for a given model-scenario

Parameters:

Name	Type	Description	Default
model	str	Model	required
scenario	str	Scenario	required
ar6_output_data_dir	Path	Directory in which the AR6 output was saved	required

Returns:

Type	Description
DataFrame	Metadata from AR6 for model-scenario

Source code in src/gcages/testing.py

@functools.cache
def get_ar6_metadata_outputs(
    model: str,
    scenario: str,
    ar6_output_data_dir: Path,
    filename: str = "AR6_Scenarios_Database_metadata_indicators_v1.1_meta.csv",
) -> pd.DataFrame:
    """
    Get metadata from AR6 for a given model-scenario

    Parameters
    ----------
    model
        Model

    scenario
        Scenario

    ar6_output_data_dir
        Directory in which the AR6 output was saved

    Returns
    -------
    :
        Metadata from AR6 for `model`-`scenario`
    """
    res = load_timeseries_csv(
        ar6_output_data_dir / filename,
        lower_column_names=False,
        index_columns=["Model", "Scenario"],
    ).loc[[(model, scenario)]]

    res.index = res.index.rename({"Model": "model", "Scenario": "scenario"})

    return res

get_ar6_raw_emissions cached

get_ar6_raw_emissions(
    model: str,
    scenario: str,
    processed_ar6_output_data_dir: Path,
) -> DataFrame

Get all raw emissions from AR6 for a given model-scenario

Parameters:

Name	Type	Description	Default
model	str	Model	required
scenario	str	Scenario	required
processed_ar6_output_data_dir	Path	Directory in which the AR6 was processed into individual model-scenario files (In the repo, see tests/regression/ar6/convert_ar6_res_to_checking_csvs.py.)	required

Returns:

Type	Description
DataFrame	All raw emissions from AR6 for model-scenario

Source code in src/gcages/testing.py

@functools.cache
def get_ar6_raw_emissions(
    model: str, scenario: str, processed_ar6_output_data_dir: Path
) -> pd.DataFrame:
    """
    Get all raw emissions from AR6 for a given model-scenario

    Parameters
    ----------
    model
        Model

    scenario
        Scenario

    processed_ar6_output_data_dir
        Directory in which the AR6 was processed into individual model-scenario files

        (In the repo, see `tests/regression/ar6/convert_ar6_res_to_checking_csvs.py`.)

    Returns
    -------
    :
        All raw emissions from AR6 for `model`-`scenario`
    """
    try:
        from pandas_indexing.selectors import ismatch
    except ImportError as exc:
        raise MissingOptionalDependencyError(
            "get_ar6_raw_emissions", requirement="pandas_indexing"
        ) from exc

    all_emissions = get_ar6_all_emissions(
        model=model,
        scenario=scenario,
        processed_ar6_output_data_dir=processed_ar6_output_data_dir,
    )
    res: pd.DataFrame = all_emissions.loc[ismatch(variable="Emissions**")].dropna(
        how="all", axis="columns"
    )

    return res

get_ar6_temperature_outputs cached

get_ar6_temperature_outputs(
    model: str,
    scenario: str,
    processed_ar6_output_data_dir: Path,
    dropna: bool = True,
) -> DataFrame

Get temperature outputs we've downloaded from AR6 for a given model-scenario

Parameters:

Name	Type	Description	Default
model	str	Model	required
scenario	str	Scenario	required
processed_ar6_output_data_dir	Path	Directory in which the AR6 output was processed into model-scenario files (In the repo, see tests/regression/ar6/convert_ar6_res_to_checking_csvs.py.)	required
dropna	bool	Drop time columns that only contain NaN	True

Returns:

Type	Description
DataFrame	All temperature outputs we've downloaded from AR6 for model-scenario

Source code in src/gcages/testing.py

@functools.cache
def get_ar6_temperature_outputs(
    model: str, scenario: str, processed_ar6_output_data_dir: Path, dropna: bool = True
) -> pd.DataFrame:
    """
    Get temperature outputs we've downloaded from AR6 for a given model-scenario

    Parameters
    ----------
    model
        Model

    scenario
        Scenario

    processed_ar6_output_data_dir
        Directory in which the AR6 output was processed into model-scenario files

        (In the repo, see `tests/regression/ar6/convert_ar6_res_to_checking_csvs.py`.)

    dropna
        Drop time columns that only contain NaN

    Returns
    -------
    :
        All temperature outputs we've downloaded from AR6 for `model`-`scenario`
    """
    filename_temperatures = f"ar6_scenarios__{model}__{scenario}__temperatures.csv"
    filename_temperatures = filename_temperatures.replace("/", "_").replace(" ", "_")
    temperatures_file = processed_ar6_output_data_dir / filename_temperatures

    res = load_timeseries_csv(
        temperatures_file,
        index_columns=["model", "scenario", "variable", "region", "unit"],
        out_columns_type=int,
    )
    if dropna:
        res = res.dropna(axis="columns", how="all")

    return res

get_cmip7_scenariomip_complete_emissions cached

get_cmip7_scenariomip_complete_emissions(
    model: str,
    scenario: str,
    processed_cmip7_scenariomip_output_data_dir: Path,
) -> DataFrame

Get complete emissions from CMIP7 ScenarioMIP outputs

Parameters:

Name	Type	Description	Default
model	str	Model for which to retrieve outputs	required
scenario	str	Scenario for which to retrieve outputs	required
processed_cmip7_scenariomip_output_data_dir	Path	Directory in which the CMIP7 ScenarioMIP output was saved	required

Returns:

Type	Description
DataFrame	All complete emissions from CMIP7 ScenarioMIP for model-scenario

Source code in src/gcages/testing.py

@functools.cache
def get_cmip7_scenariomip_complete_emissions(
    model: str, scenario: str, processed_cmip7_scenariomip_output_data_dir: Path
) -> pd.DataFrame:
    """
    Get complete emissions from CMIP7 ScenarioMIP outputs

    Parameters
    ----------
    model
        Model for which to retrieve outputs

    scenario
        Scenario for which to retrieve outputs

    processed_cmip7_scenariomip_output_data_dir
        Directory in which the CMIP7 ScenarioMIP output was saved

    Returns
    -------
    :
        All complete emissions from CMIP7 ScenarioMIP for `model`-`scenario`
    """
    try:
        from pandas_indexing.selectors import ismatch as pix_ismatch
    except ImportError as exc:
        raise MissingOptionalDependencyError(
            "get_cmip7_scenariomip_complete_emissions", requirement="pandas_indexing"
        ) from exc

    res = load_timeseries_csv(
        processed_cmip7_scenariomip_output_data_dir
        / f"{model}_{scenario}_complete.csv",
        index_columns=["model", "scenario", "variable", "region", "unit"],
        out_columns_type=int,
        out_columns_name="year",
    )
    res = res.loc[:, 2023:2100]
    # Select scenario and drop aggregated/cumulative rows
    res = res.loc[
        pix_ismatch(scenario=scenario)
        & ~pix_ismatch(variable=["**Kyoto**", "Cumulative**", "**CO2", "**GHG**"])
    ]

    return res

get_cmip7_scenariomip_harmonised_emissions cached

get_cmip7_scenariomip_harmonised_emissions(
    model: str,
    scenario: str,
    processed_cmip7_scenariomip_output_data_dir: Path,
) -> DataFrame

Get harmonised emissions from CMIP7 ScenarioMIP outputs

Parameters:

Name	Type	Description	Default
model	str	Model for which to retrieve outputs	required
scenario	str	Scenario for which to retrieve outputs	required
processed_cmip7_scenariomip_output_data_dir	Path	Directory in which the CMIP7 ScenarioMIP output was saved	required

Returns:

Type	Description
DataFrame	All harmonised emissions from CMIP7 ScenarioMIP for model-scenario

Source code in src/gcages/testing.py

@functools.cache
def get_cmip7_scenariomip_harmonised_emissions(
    model: str, scenario: str, processed_cmip7_scenariomip_output_data_dir: Path
) -> pd.DataFrame:
    """
    Get harmonised emissions from CMIP7 ScenarioMIP outputs

    Parameters
    ----------
    model
        Model for which to retrieve outputs

    scenario
        Scenario for which to retrieve outputs

    processed_cmip7_scenariomip_output_data_dir
        Directory in which the CMIP7 ScenarioMIP output was saved

    Returns
    -------
    :
        All harmonised emissions from CMIP7 ScenarioMIP for `model`-`scenario`
    """
    res = load_timeseries_csv(
        processed_cmip7_scenariomip_output_data_dir
        / f"{model}_{scenario}_harmonised.csv",
        index_columns=["model", "scenario", "variable", "region", "unit", "workflow"],
        out_columns_type=int,
    )
    return res

get_cmip7_scenariomip_pre_processed_emissions cached

get_cmip7_scenariomip_pre_processed_emissions(
    model: str,
    scenario: str,
    processed_cmip7_scenariomip_output_data_dir: Path,
) -> DataFrame

Get pre-processed emissions from CMIP7 ScenarioMIP outputs

Parameters:

Name	Type	Description	Default
model	str	Model for which to retrieve outputs	required
scenario	str	Scenario for which to retrieve outputs	required
processed_cmip7_scenariomip_output_data_dir	Path	Directory in which the CMIP7 ScenarioMIP output was saved	required

Returns:

Type	Description
DataFrame	All pre-processed emissions from CMIP7 ScenarioMIP for model-scenario

Source code in src/gcages/testing.py

@functools.cache
def get_cmip7_scenariomip_pre_processed_emissions(
    model: str, scenario: str, processed_cmip7_scenariomip_output_data_dir: Path
) -> pd.DataFrame:
    """
    Get pre-processed emissions from CMIP7 ScenarioMIP outputs

    Parameters
    ----------
    model
        Model for which to retrieve outputs

    scenario
        Scenario for which to retrieve outputs

    processed_cmip7_scenariomip_output_data_dir
        Directory in which the CMIP7 ScenarioMIP output was saved

    Returns
    -------
    :
        All pre-processed emissions from CMIP7 ScenarioMIP for `model`-`scenario`
    """
    res = load_timeseries_csv(
        processed_cmip7_scenariomip_output_data_dir
        / f"{model}_{scenario}_pre-processed.csv",
        index_columns=["model", "scenario", "variable", "region", "unit", "stage"],
        out_columns_type=int,
    )

    return res

get_key_testing_model_scenario_parameters

get_key_testing_model_scenario_parameters(
    model_scenarios: tuple[tuple[str, str]],
) -> MarkDecorator

Create a pytest parameterization decorator for model-scenario pairs.

Parameters:

Name	Type	Description	Default
model_scenarios	tuple[tuple[str, str]]	Tuples of (model, scenario) pairs to parameterize.	required

Returns:

Type	Description
MarkDecorator	A pytest decorator that runs a test for each (model, scenario) pair.

Raises:

Type	Description
MissingOptionalDependencyError	If pytest is not installed.

Examples:

>>> @get_key_testing_model_scenario_parameters((("m1", "s1"), ("m2", "s2")))
... def test_func(model, scenario): ...

Source code in src/gcages/testing.py

def get_key_testing_model_scenario_parameters(
    model_scenarios: tuple[tuple[str, str]],
) -> pytest.MarkDecorator:
    """
    Create a pytest parameterization decorator for model-scenario pairs.

    Parameters
    ----------
    model_scenarios
        Tuples of (model, scenario) pairs to parameterize.

    Returns
    -------
    :
        A pytest decorator that runs a test for each (model, scenario) pair.

    Raises
    ------
    MissingOptionalDependencyError
        If pytest is not installed.

    Examples
    --------
    >>> @get_key_testing_model_scenario_parameters((("m1", "s1"), ("m2", "s2")))
    ... def test_func(model, scenario): ...
    """
    try:
        import pytest
    except ImportError as exc:
        raise MissingOptionalDependencyError(
            "get_key_testing_model_scenario_parameters", requirement="pytest"
        ) from exc

    return pytest.mark.parametrize(
        "model, scenario",
        [(model, scenario) for model, scenario in model_scenarios],
    )

guess_magicc_exe

guess_magicc_exe(magicc_executables_dir: Path) -> Path

Guess the MAGICC executable based on the operating system

If the MAGICC_EXECUTABLE_7 environment variable is set, that is simply used and this function becomes almost a no-op.

Parameters:

Name	Type	Description	Default
magicc_executables_dir	Path	Directory in which MAGICC executables are stored	required

Returns:

Type	Description
Path	Path to the MAGICC executable

Raises:

Type	Description
FileNotFoundError	The guessed path to the MAGICC executable does not exist

Source code in src/gcages/testing.py

def guess_magicc_exe(magicc_executables_dir: Path) -> Path:
    """
    Guess the MAGICC executable based on the operating system

    If the `MAGICC_EXECUTABLE_7` environment variable is set,
    that is simply used and this function becomes almost a no-op.

    Parameters
    ----------
    magicc_executables_dir
        Directory in which MAGICC executables are stored

    Returns
    -------
    :
        Path to the MAGICC executable

    Raises
    ------
    FileNotFoundError
        The guessed path to the MAGICC executable does not exist
    """
    env_var = os.environ.get("MAGICC_EXECUTABLE_7", None)
    if env_var is not None:
        res = Path(env_var)
        if not res.exists():
            msg = (
                f"Path specified by envionment variable `MAGICC_EXECUTABLE_7`, {res}, "
                "does not exist"
            )
            raise FileNotFoundError(msg)

        return res

    guess = None
    if platform.system() == "Darwin":
        if platform.processor() == "arm":
            guess = magicc_executables_dir / "magicc-darwin-arm64"

    elif platform.system() == "Linux":
        guess = magicc_executables_dir / "magicc"

    elif platform.system() == "Windows":
        guess = magicc_executables_dir / "magicc.exe"

    if guess is None:
        msg = (
            f"No guess about where the MAGICC executable is "
            "for your system and procesor, "
            f"{platform.system()=} {platform.processor()=}"
        )

        raise NotImplementedError(msg)

    if not guess.exists():
        msg = f"Guessed that the MAGICC executable was in: {guess}"
        raise FileNotFoundError(msg)

    return guess