Clean up CMIP7 ScenarioMIP SCM running and post-processing

docs/NAVIGATION.md

@@ -12,6 +12,7 @@ See https://oprypin.github.io/mkdocs-literate-nav/
     - [Understanding the naming conventions](tutorials/understanding-the-naming-conventions.py)
 - [Further background](further-background/index.md)
     - [Dependency pinning and testing](further-background/dependency-pinning-and-testing.md)
-- [Development](development.md)
+- [Development](development/index.md)
+    - [Regression testing strategy](development/regression-testing-strategy.md)
 - [API reference](api/gcages/)
 - [Changelog](changelog.md)

docs/development/regression-testing-strategy.md

@@ -0,0 +1,33 @@
+# Regression testing strategy
+
+We provide workflows that have been used in or replicate key scientific processes.
+To check these, we reproduce the full set of results.
+However, this reproduction is relatively computationally expensive.
+As a result, we have to have a slightly more complicated testing strategy
+that allows us to check reproduction while also offering testing paths
+that facilitate fast iteration during development.
+
+The strategy is as follows.
+For each workflow we support, we put our tests in `tests/regression/[workflow_name]`.
+The workflow tests then go in `tests/regression/[workflow_name]/test_[workflow_name]_workflow.py`.
+These test functions that support running the complete workflows from end-to-end in full.
+As a result, these tests are relatively expensive.
+To avoid making our CI very slow, these tests are marked with `pytest.mark.skip_ci_default`
+and these tests are skipped in the CI, except for commits to the main branch.
+
+We then include tests of individual stages in the workflow
+(e.g. harmonisation, infilling)
+in `tests/regression/[workflow_name]/test_[workflow_name]_[stage].py`
+These tests are deliberately made fast and not marked with `pytest.mark.skip_ci_default`
+so that they are run in the CI on every pull request
+(and give us at least basic coverage of all the different workflow stages).
+If this is possible while still running the stage as it was done in the workflow,
+then we simply compare against the same inputs and outputs as the full workflow (for the given stage).
+However, if this is not possible, we simplify and test against computationally faster
+workflows that capture the key elements as much as possible.
+For example, the AR6 workflow harmonisation is very fast to run,
+so we simply run harmonisation in full when testing the AR6 harmonisation.
+However, the AR6 workflow simple climate model running is slow,
+so we only run the simple climate model with a fraction of its full probabilistic ensemble
+when testing the simple climate model running step
+(and leave the workflow tests to cover the case where the simple climate model is run in full).

src/gcages/ar6/harmonisation.py

@@ -312,7 +312,7 @@ class AR6Harmoniser:
     """
     Number of processes to use for parallel processing.
 
-    Set to 1 to process in serial.
+    Set to `None` to process serially.
     """
 
     @aneris_overrides.validator
@@ -476,7 +476,7 @@ def from_ar6_config(
         n_processes
             Number of processes to use for parallel processing.
 
-            Set to 1 to process in serial.
+            Set to `None` to process serially.
 
         Returns
         -------

src/gcages/ar6/infilling.py

@@ -486,7 +486,7 @@ class AR6Infiller:
     """
     Number of processes to use for parallel processing.
 
-    Set to `None` to process in serial.
+    Set to `None` to process serially.
     """
 
     def __call__(self, in_emissions: pd.DataFrame) -> pd.DataFrame:
@@ -622,7 +622,7 @@ def from_ar6_config(  # noqa: PLR0913
         n_processes
             Number of processes to use for parallel processing.
 
-            Set to `None` to process in serial.
+            Set to `None` to process serially.
 
         Returns
         -------

src/gcages/ar6/post_processing.py

@@ -349,7 +349,14 @@ def __call__(self, in_df: pd.DataFrame) -> PostProcessingResult:
             )
             assert_data_is_all_numeric(in_df)
             assert_has_data_for_times(
-                in_df, name="in_df", times=[2100], allow_nan=False
+                in_df,
+                name="in_df",
+                times=[
+                    *self.gsat_assessment_pre_industrial_period,
+                    *self.gsat_assessment_time_period,
+                    2100,
+                ],
+                allow_nan=False,
             )
 
             if self.raw_gsat_variable_in not in in_df.index.get_level_values(
@@ -557,7 +564,7 @@ def from_ar6_config(  # noqa: PLR0913
         n_processes
             Number of processes to use for parallel processing.
 
-            Set to 1 to process in serial.
+            Set to `None` to process serially.
 
         Returns
         -------

src/gcages/ar6/pre_processing.py

@@ -310,7 +310,9 @@ def run_parallel_pre_processing(  # noqa: PLR0913
         If not supplied, we use a default description.
 
     n_processes
-        Number of parallel processes to use
+        Number of processes to use for parallel processing.
+
+        Set to `None` to process serially.
 
     **kwargs
         Passed to `run_parallel`
@@ -459,7 +461,7 @@ class AR6PreProcessor:
     """
     Number of processes to use for parallel processing.
 
-    Set to `None` to process in serial.
+    Set to `None` to process serially.
     """
 
     def __call__(self, in_emissions: pd.DataFrame) -> pd.DataFrame:
@@ -600,7 +602,7 @@ def from_ar6_config(
         n_processes
             Number of processes to use for parallel processing.
 
-            Set to `None` to process in serial.
+            Set to `None` to process serially.
 
         Returns
         -------

src/gcages/ar6/scm_running.py

@@ -5,7 +5,6 @@
 from __future__ import annotations
 
 import multiprocessing
-import os
 from functools import partial
 from pathlib import Path
 from typing import Any
@@ -23,74 +22,40 @@
     assert_index_is_multiindex,
 )
 from gcages.completeness import assert_all_groups_are_complete
-from gcages.exceptions import MissingOptionalDependencyError
 from gcages.harmonisation import assert_harmonised
 from gcages.hashing import get_file_hash
 from gcages.renaming import SupportedNamingConventions, convert_variable_name
 from gcages.scm_running import (
+    DEFAULT_OUTPUT_VARIABLES,
     convert_openscm_runner_output_names_to_magicc_output_names,
     run_scms,
 )
-from gcages.scm_running.magicc import load_magicc_probabilistic_config
-from gcages.units_helpers import assert_has_no_pint_incompatible_characters
-
-DEFAULT_OUTPUT_VARIABLES: tuple[str, ...] = (
-    # GSAT
-    "Surface Air Temperature Change",
-    # GMST
-    "Surface Air Ocean Blended Temperature Change",
-    # ERFs
-    "Effective Radiative Forcing",
-    "Effective Radiative Forcing|Anthropogenic",
-    "Effective Radiative Forcing|Aerosols",
-    "Effective Radiative Forcing|Aerosols|Direct Effect",
-    "Effective Radiative Forcing|Aerosols|Direct Effect|BC",
-    "Effective Radiative Forcing|Aerosols|Direct Effect|OC",
-    "Effective Radiative Forcing|Aerosols|Direct Effect|SOx",
-    "Effective Radiative Forcing|Aerosols|Indirect Effect",
-    "Effective Radiative Forcing|Greenhouse Gases",
-    "Effective Radiative Forcing|CO2",
-    "Effective Radiative Forcing|CH4",
-    "Effective Radiative Forcing|N2O",
-    "Effective Radiative Forcing|F-Gases",
-    "Effective Radiative Forcing|Montreal Protocol Halogen Gases",
-    "Effective Radiative Forcing|Ozone",
-    # Heat uptake
-    "Heat Uptake",
-    # "Heat Uptake|Ocean",
-    # Atmospheric concentrations
-    "Atmospheric Concentrations|CO2",
-    "Atmospheric Concentrations|CH4",
-    "Atmospheric Concentrations|N2O",
-    # Carbon cycle
-    "Net Atmosphere to Land Flux|CO2",
-    "Net Atmosphere to Ocean Flux|CO2",
-    # Permafrost
-    "Net Land to Atmosphere Flux|CO2|Earth System Feedbacks|Permafrost",
-    "Net Land to Atmosphere Flux|CH4|Earth System Feedbacks|Permafrost",
+from gcages.scm_running.magicc import (
+    check_magicc7_version,
+    combine_probabilistic_and_common_cfg,
+    load_magicc_probabilistic_config,
 )
-"""
-Default output variables
-
-Note that it can be a bit of work
-to get these variables to actually appear in the output,
-depending on which simple climate model you're using.
-"""
+from gcages.units_helpers import assert_has_no_pint_incompatible_characters
 
 
-def check_ar6_magicc7_version() -> None:
+def check_ar6_magicc7_version(magicc_exe_path: Path) -> None:
     """
     Check that the MAGICC7 version is what was used in AR6
-    """
-    try:
-        import openscm_runner.adapters
-    except ImportError as exc:
-        raise MissingOptionalDependencyError(
-            "check_ar6_magicc7_version", requirement="openscm_runner"
-        ) from exc
 
-    if openscm_runner.adapters.MAGICC7.get_version() != "v7.5.3":  # type: ignore
-        raise AssertionError(openscm_runner.adapters.MAGICC7.get_version())  # type: ignore
+    Parameters
+    ----------
+    magicc_exe_path
+        Path to the MAGICC executable to use
+
+    Raises
+    ------
+    AssertionError
+        The MAGICC version is not what we expect
+
+    MissingOptionalDependencyError
+        [openscm-runner](https://github.com/openscm/openscm-runner) is not installed
+    """
+    check_magicc7_version(magicc_exe_path, expected_version="v7.5.3")
 
 
 def load_ar6_magicc_probabilistic_config(filepath: Path) -> list[dict[str, Any]]:
@@ -123,7 +88,12 @@ def load_ar6_magicc_probabilistic_config(filepath: Path) -> list[dict[str, Any]]
 
     cfgs = load_magicc_probabilistic_config(filepath)
 
-    return cfgs
+    # Common config that affect MAGICC behaviour
+    common_cfg = {"startyear": 1750}
+
+    run_config = combine_probabilistic_and_common_cfg(cfgs, common_cfg=common_cfg)
+
+    return run_config
 
 
 @define
@@ -221,7 +191,14 @@ class AR6SCMRunner:
     """
     Number of processes to use for parallel processing.
 
-    Set to `None` to process in serial.
+    Set to `None` to process serially.
+    """
+
+    magicc_exe_path: Path | None = None
+    """
+    Path to the MAGICC executable to use
+
+    Only required if we're running MAGICC
     """
 
     def __call__(
@@ -280,6 +257,20 @@ def __call__(
                 complete_index=self.historical_emissions.index.droplevel("unit"),
             )
 
+        if self.force_interpolate_to_yearly:
+            # TODO: put interpolate to annual steps in pandas-openscm
+            # Interpolate to ensure no nans.
+            for y in range(
+                in_emissions.columns.min(),
+                in_emissions.columns.max() + 1,
+            ):
+                if y not in in_emissions:
+                    in_emissions[y] = np.nan
+
+            in_emissions = (
+                in_emissions.sort_index(axis="columns").T.interpolate("index").T
+            )
+
         openscm_runner_emissions = update_index_levels_func(
             in_emissions,
             {
@@ -290,32 +281,19 @@ def __call__(
                 )
             },
         )
-        if self.force_interpolate_to_yearly:
-            # TODO: put interpolate to annual steps in pandas-openscm
-            # Interpolate to ensure no nans.
-            for y in range(
-                openscm_runner_emissions.columns.min(),
-                openscm_runner_emissions.columns.max() + 1,
-            ):
-                if y not in openscm_runner_emissions:
-                    openscm_runner_emissions[y] = np.nan
-
-            openscm_runner_emissions = (
-                openscm_runner_emissions.sort_index(axis="columns")
-                .T.interpolate("index")
-                .T
-            )
 
         scm_results_maybe = run_scms(
             openscm_runner_emissions,
             climate_models_cfgs=self.climate_models_cfgs,
             output_variables=self.output_variables,
             scenario_group_levels=["model", "scenario"],
+            # TODO: fix value in run_scms
             n_processes=self.n_processes if self.n_processes is not None else 1,
             db=self.db,
             verbose=self.verbose,
             batch_size_scenarios=self.batch_size_scenarios,
             force_rerun=force_rerun,
+            magicc_exe_path=self.magicc_exe_path,
         )
 
         if self.db is not None:
@@ -364,8 +342,9 @@ def from_ar6_config(  # noqa: PLR0913
         output_variables: tuple[str, ...] = DEFAULT_OUTPUT_VARIABLES,
         batch_size_scenarios: int | None = None,
         db: OpenSCMDB | None = None,
+        # TODO: Add helper for loading historical_emissions for MAGICC
         historical_emissions: pd.DataFrame | None = None,
-        harmonisation_year: int | None = None,
+        harmonisation_year: int = 2015,
         verbose: bool = True,
         run_checks: bool = True,
         progress: bool = True,
@@ -426,31 +405,31 @@ def from_ar6_config(  # noqa: PLR0913
         n_processes
             Number of processes to use for parallel processing.
 
-            Set to `None` to process in serial.
+            Set to `None` to process serially.
 
         Returns
         -------
         :
             Initialised SCM runner
         """
-        os.environ["MAGICC_EXECUTABLE_7"] = str(magicc_exe_path)
-        check_ar6_magicc7_version()
+        check_ar6_magicc7_version(magicc_exe_path)
 
         magicc_ar6_prob_cfg = load_ar6_magicc_probabilistic_config(
             magicc_prob_distribution_path
         )
 
-        startyear = 1750
         common_cfg = {
-            "startyear": startyear,
             "out_dynamic_vars": convert_openscm_runner_output_names_to_magicc_output_names(  # noqa: E501
                 output_variables
             ),
             "out_ascii_binary": "BINARY",
             "out_binary_format": 2,
         }
 
-        run_config = [{**common_cfg, **base_cfg} for base_cfg in magicc_ar6_prob_cfg]
+        run_config = combine_probabilistic_and_common_cfg(
+            magicc_ar6_prob_cfg, common_cfg=common_cfg
+        )
+
         magicc_full_distribution_n_config = 600
         if len(run_config) != magicc_full_distribution_n_config:
             raise AssertionError(len(run_config))
@@ -467,4 +446,5 @@ def from_ar6_config(  # noqa: PLR0913
             n_processes=n_processes,
             force_interpolate_to_yearly=True,  # MAGICC safer with annual input
             res_column_type=int,  # annual output by default
+            magicc_exe_path=magicc_exe_path,
         )

src/gcages/cmip7_scenariomip/__init__.py

@@ -11,7 +11,7 @@
     CMIP7ScenarioMIPInfiller,
 )
 from gcages.cmip7_scenariomip.post_processing import (
-    CMIP7ScenarioMIPPostProcessor,
+    create_cmip7_scenariomip_postprocessor,
 )
 from gcages.cmip7_scenariomip.pre_processing import (
     CMIP7ScenarioMIPPreProcessingResult,
@@ -25,11 +25,11 @@
 
 __all__ = [
     "CMIP7ScenarioMIPInfiller",
-    "CMIP7ScenarioMIPPostProcessor",
     "CMIP7ScenarioMIPPreProcessingResult",
     "CMIP7ScenarioMIPPreProcessor",
     "CMIP7ScenarioMIPSCMRunner",
     "ReaggregatorBasic",
     "ReaggregatorLike",
     "create_cmip7_scenariomip_global_harmoniser",
+    "create_cmip7_scenariomip_postprocessor",
 ]