Merge pull request #172 from E3SM-Project/2.2.5

2.2.5

Author: sterlingbaldwin

.gitignore

@@ -152,7 +152,7 @@ tags
 
 #### visualstudiocode ####
 .vscode/*
-!.vscode/settings.json
+.vscode/settings.json
 !.vscode/tasks.json
 !.vscode/launch.json
 !.vscode/extensions.json

conda/build.sh

@@ -0,0 +1 @@
+$PYTHON setup.py install

conda/conda_build.sh

@@ -1,5 +1,5 @@
 #!/bin/bash
-export VERSION="2.2.4"
+export VERSION="2.2.5"
 export BUILD_NAME="0"
 export CONDA_BLD_PATH=~/conda-bld
 USER="e3sm"

processflow/__main__.py

@@ -8,10 +8,9 @@
 
 from time import sleep
 
-from processflow.lib.events import EventList
 from processflow.lib.finalize import finalize
 from processflow.lib.initialize import initialize
-from processflow.lib.util import print_debug, print_line, print_message
+from processflow.lib.util import print_debug, print_line
 
 os.environ['UVCDAT_ANONYMOUS_LOG'] = 'no'
 os.environ['NCO_PATH_OVERRIDE'] = 'no'
@@ -27,8 +26,6 @@ def main(cl_args=None):
         kwargs (dict): when running in test mode, arguments are passed directly through the kwargs
             which bypasses the argument parsing.
     """
-    # create global EventList
-    event_list = EventList()
 
     # The master configuration object
     config = {}
@@ -43,12 +40,10 @@ def main(cl_args=None):
                       "    command line: {}".format(cl_args,
                                                     ' '.join(sys.argv[:])))
 
-    config, runmanager = initialize(
-        argv=cl_args,
-        event_list=event_list)
+    config, runmanager = initialize(argv=cl_args)
 
     if isinstance(config, int):
-        print("Error in setup, exiting")
+        print_line("Error in setup, exiting", status='error')
         return -1
     logging.info('Config setup complete')
     debug = True if config['global'].get('debug') else False
@@ -67,43 +62,43 @@ def main(cl_args=None):
         while True:
 
             if debug:
-                print_line(' -- checking data -- ', event_list)
+                print_line(' -- checking data --')
             runmanager.check_data_ready()
 
             if debug:
-                print_line(' -- starting ready jobs --', event_list)
+                print_line(' -- starting ready jobs --')
             runmanager.start_ready_jobs()
 
             if debug:
-                print_line(' -- monitoring running jobs --', event_list)
+                print_line(' -- monitoring running jobs --')
             runmanager.monitor_running_jobs(debug=debug)
 
             if debug:
-                print_line(' -- writing out state -- ', event_list)
+                print_line(' -- writing out state --')
             runmanager.write_job_sets(state_path)
 
             status = runmanager.is_all_done()
             if status >= 0:
                 msg = "Finishing up run"
-                print_line(msg, event_list)
+                print_line(msg)
                 finalize(
                     config=config,
-                    event_list=event_list,
                     status=status,
                     runmanager=runmanager)
                 # SUCCESS EXIT
                 return 0
 
             if debug:
-                print_line(' -- sleeping', event_list)
+                print_line(' -- sleeping')
             sleep(loop_delay)
     except KeyboardInterrupt as e:
-        print_message('\n----- KEYBOARD INTERRUPT -----')
-        runmanager.write_job_sets(state_path)
-        print_message('-----  cleanup complete  -----', 'ok')
+        print_line('----- KEYBOARD INTERRUPT -----', status='err')
+        if debug:
+            import ipdb; ipdb.set_trace()
     except Exception as e:
-        print_message('----- AN UNEXPECTED EXCEPTION OCCURED -----')
+        print_line('----- AN UNEXPECTED EXCEPTION OCCURED -----', status='err')
         print_debug(e)
+    finally:
         runmanager.write_job_sets(state_path)
 # -----------------------------------------------
 

processflow/jobs/amwg.py

@@ -24,8 +24,13 @@ def __init__(self, *args, **kwargs):
                 to pass to the resource manager
         """
         super(AMWG, self).__init__(*args, **kwargs)
+        if not os.environ.get('NCARG_ROOT'):
+            msg = 'ERROR: NCL doesnt appear to be installed in your environment, unable to run AMWG'
+            print_line(msg, status='err')
+            self._status = JobStatus.FAILED
+
         self._job_type = 'amwg'
-        self._requires = 'climo'
+        self._requires = ['climo']
         self._data_required = ['climo_regrid']
 
         config = kwargs.get('config')
@@ -77,6 +82,7 @@ def __init__(self, *args, **kwargs):
             self._host_path = ''
             self._output_path = ''
 
+        self.setup_job_args(config)
     # -----------------------------------------------
 
     def setup_data(self, config, filemanager, case):
@@ -92,7 +98,7 @@ def _dep_filter(self, job):
         find the climo job we're waiting for, assuming there's only
         one climo job in this case with the same start and end years
         """
-        if job.job_type != self._requires:
+        if job.job_type not in self._requires:
             return False
         if job.start_year != self.start_year:
             return False
@@ -131,14 +137,13 @@ def setup_dependencies(self, *args, **kwargs):
             self.depends_on.append(self_climo.id)
     # -----------------------------------------------
 
-    def execute(self, config, event_list, custom_args=None, dryrun=False):
+    def execute(self, config, *args, custom_args=None, dryrun=False, **kwargs):
         """
         Execute the AMWG job
 
         Parameters
         ----------
             config (dict): the global config object
-            event_list (EventList): an EventList to push notifications into
             dryrun (bool): if true this job will generate all scripts,
                 setup data, and exit without submitting the job
         Returns
@@ -210,12 +215,11 @@ def execute(self, config, event_list, custom_args=None, dryrun=False):
             output_path=csh_template_out)
 
         # create the run command and submit it
-        self._has_been_executed = True
         cmd = ['csh', csh_template_out]
-        return self._submit_cmd_to_manager(config, cmd, event_list)
+        return self._submit_cmd_to_manager(config, cmd)
     # -----------------------------------------------
 
-    def postvalidate(self, config, event_list, *args, **kwargs):
+    def postvalidate(self, config, *args, **kwargs):
         """
         Validates that the job ran correctly
 
@@ -261,7 +265,6 @@ def postvalidate(self, config, event_list, *args, **kwargs):
                 return self._check_tar(
                     img_source_tar, 
                     img_source, 
-                    event_list, 
                     config, 
                     expected_files) == 0
             else:
@@ -307,45 +310,43 @@ def postvalidate(self, config, event_list, *args, **kwargs):
             number_missing = 0
             if os.path.exists(img_source_tar):
                 number_missing = self._check_tar(
-                    img_source_tar, img_source, event_list, config, expected_files)
+                    img_source_tar, img_source, config, expected_files)
 
         if number_missing == 0:
             msg = '{prefix}: all expected output images found'.format(
                 prefix=self.msg_prefix())
-            print_line(msg, event_list)
+            print_line(msg)
             logging.info(msg)
             self._check_links(config, img_source)
             return True
         elif number_missing > 0 and number_missing <= 2:
             msg = '{prefix}: this job was found to be missing plots, please check the console output for additional information'.format(
                 prefix=self.msg_prefix())
-            print_line(msg, event_list)
+            print_line(msg)
             self._check_links(config, img_source)
             return True
         else:
             return False
     # -----------------------------------------------
 
-    def handle_completion(self, filemanager, event_list, config, *args, **kwargs):
+    def handle_completion(self, filemanager, config, *args, **kwargs):
         """
         Sets up variables needed to web hosting
 
         Parameters
         ----------
-            event_list (EventList): an EventList to push user notifications into
+            filemanager: the global filemanager instance
             config (dict): the global config object
         """
-        if self.status == JobStatus.COMPLETED:
-            msg = '{prefix}: Job complete'.format(
-                prefix=self.msg_prefix())
-        else:
-            msg = '{prefix}: Job failed'.format(
-                prefix=self.msg_prefix())
-        print_line(msg, event_list)
-        logging.info(msg)
+        if self.status != JobStatus.COMPLETED:
+            msg = f'{self.msg_prefix()}: Job failed, not running completion handler'
+            print_line(msg, status='error')
+            return
 
         # if hosting is turned off, simply return
         if not config['global'].get('host'):
+            msg = f'{self.msg_prefix()}: Job completion handler done\n'
+            print_line(msg)
             return
 
         img_source = os.path.join(
@@ -361,16 +362,18 @@ def handle_completion(self, filemanager, event_list, config, *args, **kwargs):
             else:
                 msg = '{prefix}: Unable to find output directory or tar archive'.format(
                     prefix=self.msg_prefix())
-                print_line(msg, event_list)
+                print_line(msg)
                 self.status = JobStatus.FAILED
                 logging.info(msg)
                 return
 
         self.setup_hosting(
             always_copy=config['global']['always_copy'],
             img_source=img_source,
-            host_path=self._host_path,
-            event_list=event_list)
+            host_path=self._host_path)
+        
+        msg = f'{self.msg_prefix()}: Job completion handler done\n'
+        print_line(msg)
 
     # -----------------------------------------------
 
@@ -475,11 +478,11 @@ def _change_input_file_names(self):
             os.rename(input_file, new_name)
     # -----------------------------------------------
 
-    def _check_tar(self, img_source_tar, img_source, event_list, config, expected_files):
+    def _check_tar(self, img_source_tar, img_source, config, expected_files):
         number_missing = 0
         msg = '{prefix}: extracting images from tar archive'.format(
             prefix=self.msg_prefix())
-        print_line(msg, event_list)
+        print_line(msg)
         call(['tar', '-xf', img_source_tar,
               '--directory', self._output_path])
         passed = True
@@ -497,7 +500,7 @@ def _check_tar(self, img_source_tar, img_source, event_list, config, expected_fi
                         prefix=self.msg_prefix(),
                         dir=setpath)
                     logging.error(msg)
-                    print_line(msg, event_list)
+                    print_line(msg)
             else:
                 count = len(os.listdir(setpath))
                 if count < expected_files[setname]:
@@ -507,6 +510,71 @@ def _check_tar(self, img_source_tar, img_source, event_list, config, expected_fi
                         numProduced=count,
                         numExpected=expected_files[setname])
                     logging.error(msg)
-                    print_line(msg, event_list)
+                    print_line(msg)
                     number_missing += 1
         return number_missing
+    
+    def validate(self, config):
+        messages = []
+        if 'job_options' in config.keys():
+            amwg_global_config = config['job_options'].get('amwg')
+            if amwg_global_config:
+                if 'diag_home' not in amwg_global_config.keys():
+                    msg = "Global job_options does not contain the amwg code path"
+                    messages.append(msg)
+                if not amwg_global_config.get('frequency'):
+                    config['job_options']['amwg']['frequency'] = set()
+                if not isinstance(amwg_global_config['frequency'], set):
+                    amwg_global_config['frequency']  = set(amwg_global_config['frequency'])
+                amwg_sets = amwg_global_config.get('plot_sets')
+                if amwg_sets:
+                    if not isinstance(amwg_sets, set):
+                        amwg_sets = set(amwg_sets)
+                    allowed_sets = set([str(x) for x in range(1, 17)] + ['all', '4a'])
+                    for s in amwg_sets:
+                        if s not in allowed_sets:
+                            msg = "{} is not an allowed AMWG set".format(s)
+                            messages.append(msg)
+
+        for case in config['simulations']:
+            if "amwg" not in case['jobs']:
+                continue
+            case_options = case['jobs']['amwg']
+
+            freqs = case_options.get('frequency')
+            if freqs:
+                if not isinstance(freqs, set):
+                    freqs = set(freqs)
+                freqs.update(amwg_global_config['frequency'])
+
+            case_sets = case_options.get('plot_sets')
+            if case_sets:
+                if not isinstance(case_sets, set):
+                    case_sets = set(case_sets)
+                allowed_sets = set([str(x) for x in range(1, 17)] + ['all', '4a'])
+                for s in case_sets:
+                    if s not in allowed_sets:
+                        msg = "case {case} contains invalid AMWG set {set}".format(
+                            case=case_options['shortname'], set=s)
+                        messages.append(msg)
+                case_sets.update(amwg_sets)
+            
+            if not amwg_global_config.get('output_grid_name') and not case_options.get('output_grid_name'):
+                msg = 'Please specify a name for the climo output_grid_name in either the global job_options under amwg, or under the job entry in each simulation\'s config'
+                messages.append(msg)
+            else:
+                if not case_options.get('output_grid_name'):
+                    case_options['output_grid_name'] = amwg_global_config.get('output_grid_name')
+            if not amwg_global_config.get('atm_map_path') and not case_options.get('atm_map_path'):
+                msg = 'Please specify a atm_map_path in either the global job_options under amwg, or under the job entry in each simulation\'s config'
+                messages.append(msg)
+            else:
+                if not case_options.get('atm_map_path'):
+                    case_options['atm_map_path'] = amwg_global_config.get('atm_map_path')
+        
+        if messages:
+            return messages
+        return None
+
+            
+