Add support for cached output files

When constructing a step, if ``cached=True``, the outputs for
this step will be downloaded to the appropriate local database and
symlinked instead of being computed.  Inputs (other than the cached
outputs) and the run method will be skipped.

Each MPAS core can optionally have a database (a python dictionary
in a json file called cached_files.json) that keeps track of which
files are available in the cache and what date stamp is in the
filename.

When setting up test cases, a user can supply test-case numbers
with a "c" suffix to indicate that they should be cached.  When
setting up test cases individually with a path, a user can supply
a list of steps in the test case that should use cached outputs.
A test suite can supply a line with "cached" or
"cached: <step> <step>" to indicate either that all steps in the
test case or the listed steps should use cached outputs.

Author: xylar

compass/mpas_core.py

@@ -1,3 +1,7 @@
+from importlib import resources
+import json
+
+
 class MpasCore:
     """
     The base class for housing all the tests for a given MPAS core, such as
@@ -10,6 +14,11 @@ class MpasCore:
 
     test_groups : dict
         A dictionary of test groups for the MPAS core with their names as keys
+
+    cached_files : dict
+        A dictionary that maps from output file names in test cases to cached
+        files in the ``compass_cache`` database for the MPAS core.  These
+        file mappings are read in from ``cached_files.json`` in the MPAS core.
     """
 
     def __init__(self, name):
@@ -26,6 +35,9 @@ def __init__(self, name):
         # test groups are added with add_test_groups()
         self.test_groups = dict()
 
+        self.cached_files = dict()
+        self._read_cached_files()
+
     def add_test_group(self, test_group):
         """
         Add a test group to the MPAS core
@@ -36,3 +48,16 @@ def add_test_group(self, test_group):
             the test group to add
         """
         self.test_groups[test_group.name] = test_group
+
+    def _read_cached_files(self):
+        """ Read in the dictionary of cached files from cached_files.json """
+
+        package = f'compass.{self.name}'
+        filename = 'cached_files.json'
+        try:
+            with resources.path(package, filename) as path:
+                with open(path) as data_file:
+                    self.cached_files = json.load(data_file)
+        except FileNotFoundError:
+            # no cached files for this core
+            pass

compass/setup.py

@@ -3,6 +3,7 @@
 import configparser
 import os
 import pickle
+import warnings
 
 from compass.mpas_cores import get_mpas_cores
 from compass.config import add_config, ensure_absolute_paths
@@ -11,7 +12,8 @@
 
 
 def setup_cases(tests=None, numbers=None, config_file=None, machine=None,
-                work_dir=None, baseline_dir=None, mpas_model_path=None):
+                work_dir=None, baseline_dir=None, mpas_model_path=None,
+                cached=None):
     """
     Set up one or more test cases
 
@@ -20,8 +22,10 @@ def setup_cases(tests=None, numbers=None, config_file=None, machine=None,
     tests : list of str, optional
         Relative paths for a test cases to set up
 
-    numbers : list of int, optional
-        Case numbers to setup, as listed from ``compass list``
+    numbers : list of str, optional
+        Case numbers to setup, as listed from ``compass list``, optionally with
+        a suffix ``c`` to indicate that all steps in that test case should be
+        cached
 
     config_file : str, optional
         Configuration file with custom options for setting up and running test
@@ -41,6 +45,10 @@ def setup_cases(tests=None, numbers=None, config_file=None, machine=None,
         The relative or absolute path to the root of a branch where the MPAS
         model has been built
 
+    cached : list of list of str, optional
+        For each test in ``tests``, which steps (if any) should be cached,
+        or "_all" if all steps should be cached
+
     Returns
     -------
     test_cases : dict of compass.TestCase
@@ -56,6 +64,14 @@ def setup_cases(tests=None, numbers=None, config_file=None, machine=None,
     if tests is None and numbers is None:
         raise ValueError('At least one of tests or numbers is needed.')
 
+    if cached is not None:
+        if tests is None:
+            warnings.warn('Ignoring "cached" argument becasue "tests" was '
+                          'not provided')
+        elif len(cached) != len(tests):
+            raise ValueError('A list of cached steps must be provided for '
+                             'each test in "tests"')
+
     if work_dir is None:
         work_dir = os.getcwd()
 
@@ -71,18 +87,36 @@ def setup_cases(tests=None, numbers=None, config_file=None, machine=None,
     if numbers is not None:
         keys = list(all_test_cases)
         for number in numbers:
+            cache_all = False
+            if number.endswith('c'):
+                cache_all = True
+                number = int(number[:-1])
+            else:
+                number = int(number)
+
             if number >= len(keys):
                 raise ValueError('test number {} is out of range.  There are '
                                  'only {} tests.'.format(number, len(keys)))
             path = keys[number]
-            test_cases[path] = all_test_cases[path]
+            test_case = all_test_cases[path]
+            if cache_all:
+                for step in test_case.steps.values():
+                    step.cached = True
+            test_cases[path] = test_case
 
     if tests is not None:
-        for path in tests:
+        for index, path in enumerate(tests):
             if path not in all_test_cases:
                 raise ValueError('Test case with path {} is not in '
                                  'test_cases'.format(path))
-            test_cases[path] = all_test_cases[path]
+            test_case = all_test_cases[path]
+            if cached is not None:
+                step_names = cached[index]
+                if len(step_names) > 0 and step_names[0] == '_all':
+                    step_names = list(test_case.steps.keys())
+                for step_name in step_names:
+                    test_case.steps[step_name].cached = True
+            test_cases[path] = test_case
 
     # get the MPAS core of the first test case.  We'll assume all tests are
     # for this core
@@ -133,6 +167,10 @@ def setup_case(path, test_case, config_file, machine, work_dir, baseline_dir,
     """
 
     print('  {}'.format(path))
+    cached_steps = [step.name for step in test_case.steps.values() if step.cached]
+    if len(cached_steps) > 0:
+        cached_steps = ' '.join(cached_steps)
+        print(f'    steps with cached outputs: {cached_steps}')
 
     config = configparser.ConfigParser(
         interpolation=configparser.ExtendedInterpolation())
@@ -253,10 +291,12 @@ def main():
                         help="Relative path for a test case to set up",
                         metavar="PATH")
     parser.add_argument("-n", "--case_number", nargs='+', dest="case_num",
-                        type=int,
+                        type=str,
                         help="Case number(s) to setup, as listed from "
                              "'compass list'. Can be a space-separated"
-                             "list of case numbers.", metavar="NUM")
+                             "list of case numbers.  A suffix 'c' indicates"
+                             "that all steps in the test should use cached"
+                             "outputs.", metavar="NUM")
     parser.add_argument("-f", "--config_file", dest="config_file",
                         help="Configuration file for test case setup",
                         metavar="FILE")
@@ -274,13 +314,21 @@ def main():
                         help="The path to the build of the MPAS model for the "
                              "core.",
                         metavar="PATH")
+    parser.add_argument("--cached", dest="cached", nargs='+',
+                        help="A list of steps in the test case supplied with"
+                             "--test that should use cached outputs, or "
+                             "'_all' if all steps should be cached",
+                        metavar="STEP")
 
     args = parser.parse_args(sys.argv[2:])
+    cached = None
     if args.test is None:
         tests = None
     else:
         tests = [args.test]
+        if args.cached is not None:
+            cached = [args.cached]
     setup_cases(tests=tests, numbers=args.case_num,
                 config_file=args.config_file, machine=args.machine,
                 work_dir=args.work_dir, baseline_dir=args.baseline_dir,
-                mpas_model_path=args.mpas_model)
+                mpas_model_path=args.mpas_model, cached=cached)

compass/step.py

@@ -76,9 +76,9 @@ class Step:
         time or the step will raise an exception
 
     outputs : list of str
-        a list of absolute paths of output files produced by this step and
-        available as inputs to other test cases and steps.  These files must
-        exist after the test has run or an exception will be raised
+        a list of absolute paths of output files produced by this step (or
+        cached) and available as inputs to other test cases and steps.  These
+        files must exist after the test has run or an exception will be raised
 
     namelist_data : dict
         a dictionary used internally to keep track of updates to the default
@@ -111,10 +111,14 @@ class Step:
     log_filename : str
         At run time, the name of a log file where output/errors from the step
         are being logged, or ``None`` if output is to stdout/stderr
+
+    cached : bool
+        Whether to get all of the outputs for the step from the database of
+        cached outputs for this MPAS core
     """
 
     def __init__(self, test_case, name, subdir=None, cores=1, min_cores=1,
-                 threads=1, max_memory=1000, max_disk=1000):
+                 threads=1, max_memory=1000, max_disk=1000, cached=False):
         """
         Create a new test case
 
@@ -150,6 +154,10 @@ def __init__(self, test_case, name, subdir=None, cores=1, min_cores=1,
             the amount of disk space that the step is allowed to use in MB.
             This is currently just a placeholder for later use with task
             parallelism
+
+        cached : bool, optional
+            Whether to get all of the outputs for the step from the database of
+            cached outputs for this MPAS core
         """
         self.name = name
         self.test_case = test_case
@@ -186,6 +194,9 @@ def __init__(self, test_case, name, subdir=None, cores=1, min_cores=1,
         self.logger = None
         self.log_filename = None
 
+        # output caching
+        self.cached = cached
+
     def setup(self):
         """
         Set up the test case in the work directory, including downloading any
@@ -454,6 +465,22 @@ def process_inputs_and_outputs(self):
         step_dir = self.work_dir
         config = self.config
 
+        # process the outputs first because cached outputs will add more inputs
+        if self.cached:
+            # forget about the inputs -- we won't used them, but we will add
+            # the cached outputs as inputs
+            self.input_data = list()
+            for output in self.outputs:
+                filename = os.path.join(self.path, output)
+                if filename not in self.mpas_core.cached_files:
+                    raise ValueError(f'The file {filename} has not been added '
+                                     f'to the cache database')
+                target = self.mpas_core.cached_files[filename]
+                self.add_input_file(
+                    filename=output,
+                    target=target,
+                    database='compass_cache')
+
         inputs = []
         for entry in self.input_data:
             filename = entry['filename']
@@ -534,6 +561,10 @@ def _generate_namelists(self):
         by parsing the files and dictionaries in the step's ``namelist_data``.
         """
 
+        if self.cached:
+            # no need for namelists
+            return
+
         step_work_dir = self.work_dir
         config = self.config
 
@@ -570,6 +601,9 @@ def _generate_streams(self):
         Writes out a streams file in the work directory with new values given
         by parsing the files and dictionaries in the step's ``streams_data``.
         """
+        if self.cached:
+            # no need for streams
+            return
 
         step_work_dir = self.work_dir
         config = self.config

compass/suite.py

@@ -37,7 +37,7 @@ def setup_suite(mpas_core, suite_name, config_file=None, machine=None,
         directories
 
     baseline_dir : str, optional
-        Location of baseslines that can be compared to
+        Location of baselines that can be compared to
 
     mpas_model_path : str, optional
         The relative or absolute path to the root of a branch where the MPAS
@@ -51,20 +51,16 @@ def setup_suite(mpas_core, suite_name, config_file=None, machine=None,
 
     text = resources.read_text('compass.{}.suites'.format(mpas_core),
                                '{}.txt'.format(suite_name))
-    tests = list()
-    for test in text.split('\n'):
-        test = test.strip()
-        if (len(test) > 0 and test not in tests
-                and not test.startswith('#')):
-            tests.append(test)
+
+    tests, cached = _parse_suite(text)
 
     if work_dir is None:
         work_dir = os.getcwd()
     work_dir = os.path.abspath(work_dir)
 
     test_cases = setup_cases(tests, config_file=config_file, machine=machine,
                              work_dir=work_dir, baseline_dir=baseline_dir,
-                             mpas_model_path=mpas_model_path)
+                             mpas_model_path=mpas_model_path, cached=cached)
 
     test_suite = {'name': suite_name,
                   'test_cases': test_cases,
@@ -108,8 +104,8 @@ def clean_suite(mpas_core, suite_name, work_dir=None):
 
     text = resources.read_text('compass.{}.suites'.format(mpas_core),
                                '{}.txt'.format(suite_name))
-    tests = [test.strip() for test in text.split('\n') if
-             len(test.strip()) > 0 and not test.startswith('#')]
+
+    tests, _ = _parse_suite(text)
 
     if work_dir is None:
         work_dir = os.getcwd()
@@ -186,3 +182,26 @@ def _get_required_cores(test_cases):
             max_of_min_cores = max(max_of_min_cores, step.min_cores)
 
     return max_cores, max_of_min_cores
+
+
+def _parse_suite(text):
+    """ Parse the text of a file defining a test suite """
+
+    tests = list()
+    cached = list()
+    for test in text.split('\n'):
+        test = test.strip()
+        if len(test) == 0 or test.startswith('#'):
+            # a blank line or comment
+            continue
+
+        if test == 'cached':
+            cached[-1] = ['_all']
+        elif test.startswith('cached:'):
+            steps = test[len('cached:'):].strip().split(' ')
+            cached[-1] = steps
+        else:
+            tests.append(test)
+            cached.append(list())
+
+    return tests, cached

compass/testcase.py

@@ -145,6 +145,8 @@ def run(self):
         cwd = os.getcwd()
         for step_name in self.steps_to_run:
             step = self.steps[step_name]
+            if step.cached:
+                continue
             step.config = self.config
             new_log_file = self.new_step_log_file
             if self.log_filename is not None: