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
guess_magicc_exe_path	Guess the path to the MAGICC executable

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

guess_magicc_exe_path

guess_magicc_exe_path() -> Path

Guess the path to the MAGICC executable

Uses the MAGICC_EXECUTABLE_7 environment variable. If that isn't set, it guesses.

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_path() -> Path:
    """
    Guess the path to the MAGICC executable

    Uses the `MAGICC_EXECUTABLE_7` environment variable.
    If that isn't set, it guesses.

    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:
        return Path(env_var)

    guess = None
    guess_path = (
        Path(__file__).parents[2]
        / "tests"
        / "regression"
        / "ar6"
        / "ar6-workflow-inputs"
        / "magicc-v7.5.3"
        / "bin"
    )
    if platform.system() == "Darwin":
        if platform.processor() == "arm":
            guess = guess_path / "magicc-darwin-arm64"

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

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

    if guess is not None:
        if guess.exists():
            return guess

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

    msg = "No guess about where the MAGICC executable is for your system"
    raise FileNotFoundError(msg)