Source code for polaris.suite

import argparse
import importlib.resources as imp_res
import sys
from typing import List

from polaris.setup import setup_tasks


[docs] def setup_suite(component, suite_name, work_dir, config_file=None, machine=None, baseline_dir=None, component_path=None, copy_executable=False, clean=False, model=None): """ Set up a suite of tasks Parameters ---------- component : str The component ('ocean', 'landice', etc.) of the suite suite_name : str The name of the suite. A file ``<suite_name>.txt`` must exist within the core's ``suites`` package that lists the paths of the tasks in the suite config_file : str, optional Configuration file with custom options for setting up and running tasks machine : str, optional The name of one of the machines with defined config options, which can be listed with ``polaris list --machines`` work_dir : str, optional A directory that will serve as the base for creating task directories baseline_dir : str, optional Location of baselines that can be compared to component_path : str, optional The relative or absolute path to the location where the model and default namelists have been built copy_executable : bool, optional Whether to copy the MPAS executable to the work directory clean : bool, optional Whether to delete the contents of the base work directory before setting up the suite model : str, optional The model to run """ text = imp_res.files(f'polaris.{component}.suites').joinpath( f'{suite_name}.txt').read_text() tasks, cached = _parse_suite(text) setup_tasks(work_dir, tasks, config_file=config_file, machine=machine, baseline_dir=baseline_dir, component_path=component_path, suite_name=suite_name, cached=cached, copy_executable=copy_executable, clean=clean, model=model)
def main(): parser = argparse.ArgumentParser( description='Set up a regression suite', prog='polaris suite') parser.add_argument("-c", "--component", dest="component", help="The component for the suite.", metavar="COMPONENT", required=True) parser.add_argument("-t", "--task_suite", dest="task_suite", help="Path to file containing a suite to setup.", metavar="SUITE", required=True) parser.add_argument("-f", "--config_file", dest="config_file", help="Configuration file for task setup.", metavar="FILE") parser.add_argument("-m", "--machine", dest="machine", help="The name of the machine for loading machine-" "related config options.", metavar="MACH") parser.add_argument("-b", "--baseline_dir", dest="baseline_dir", help="Location of baselines that can be compared to.", metavar="PATH") parser.add_argument("-w", "--work_dir", dest="work_dir", required=True, help="If set, script will setup the suite in " "work_dir rather in this script's location.", metavar="PATH") parser.add_argument("-p", "--component_path", dest="component_path", help="The path where the component executable and " "default namelists have been built.", metavar="PATH") parser.add_argument("--copy_executable", dest="copy_executable", action="store_true", help="If the model executable should be copied to the " "work directory.") parser.add_argument("--clean", dest="clean", action="store_true", help="If the base work directory should be deleted " "before setting up the suite.") parser.add_argument("--model", dest="model", help="The model to run (one of 'mpas-ocean', 'omega', " "or 'mpas-seaice')") args = parser.parse_args(sys.argv[2:]) setup_suite(component=args.component, suite_name=args.task_suite, work_dir=args.work_dir, config_file=args.config_file, machine=args.machine, baseline_dir=args.baseline_dir, component_path=args.component_path, copy_executable=args.copy_executable, clean=args.clean, model=args.model) def _parse_suite(text): """ Parse the text of a file defining a suite """ tasks: List[str] = list() cached: List[List[str]] = list() for task in text.split('\n'): task = task.strip() if len(task) == 0 or task.startswith('#'): # a blank line or comment continue if task == 'cached': cached[-1] = ['_all'] elif task.startswith('cached:'): steps = task[len('cached:'):].strip().split(' ') cached[-1].extend(steps) else: tasks.append(task) cached.append(list()) return tasks, cached