Splits AWIPS jobs into seperate tasks

jobs/rocoto/awips.sh → jobs/rocoto/awips_20sh.sh

@@ -19,14 +19,13 @@ source "${HOMEgfs}/ush/load_fv3gfs_modules.sh"
 status=$?
 (( status != 0 )) && exit "${status}"
 
-export job="awips"
+export job="awips_20sh.sh"
 export jobid="${job}.$$"
 
-# TODO (#1228) - This script is doing more than just calling a j-job
-#   Also, this forces us to call the config files here instead of the j-job
 source "${HOMEgfs}/ush/jjob_header.sh" -e "awips" -c "base awips"
 
-fhrlst=$(echo ${FHRLST} | sed -e 's/_/ /g; s/f/ /g; s/,/ /g')
+# shellcheck disable=SC2153
+fhrlst=$(echo "${FHRLST}" | sed -e 's/_/ /g; s/f/ /g; s/,/ /g')
 
 ###############################################################
 
@@ -45,22 +44,17 @@ for fhr3 in ${fhrlst}; do
     fhmax=84
     if (( fhr >= fhmin && fhr <= fhmax )); then
         if ((fhr % 3 == 0)); then
-            export fcsthrs=${fhr3}
-            ${AWIPS20SH}
-        fi
-
-        if ((fhr % 6 == 0)); then
-            ${AWIPSG2SH}
+            export fcsthrs="${fhr3}"
+            "${AWIPS20SH}"
         fi
     fi
 
     fhmin=90
     fhmax=240
     if (( fhr >= fhmin && fhr <= fhmax )); then
         if ((fhr % 6 == 0)); then
-            export fcsthrs=${fhr3}
-            ${AWIPS20SH}
-            ${AWIPSG2SH}
+            export fcsthrs="${fhr3}"
+            "${AWIPS20SH}"
         fi
     fi
 done

jobs/rocoto/awips_g2sh.sh

@@ -0,0 +1,66 @@
+#! /usr/bin/env bash
+
+source "${HOMEgfs}/ush/preamble.sh"
+
+###############################################################
+## Abstract:
+## Inline awips driver script
+## HOMEgfs   : /full/path/to/workflow
+## EXPDIR : /full/path/to/config/files
+## CDATE  : current analysis date (YYYYMMDDHH)
+## CDUMP  : cycle name (gdas / gfs)
+## PDY    : current date (YYYYMMDD)
+## cyc    : current cycle (HH)
+###############################################################
+
+###############################################################
+# Source FV3GFS workflow modules
+source "${HOMEgfs}/ush/load_fv3gfs_modules.sh"
+status=$?
+(( status != 0 )) && exit "${status}"
+
+export job="awips_g2sh"
+export jobid="${job}.$$"
+
+source "${HOMEgfs}/ush/jjob_header.sh" -e "awips" -c "base awips"
+
+# shellcheck disable=SC2153
+fhrlst=$(echo "${FHRLST}" | sed -e "s/_/ /g; s/f/ /g; s/,/ /g")
+
+###############################################################
+
+################################################################################
+echo
+echo "=============== BEGIN AWIPS ==============="
+
+for fhr3 in ${fhrlst}; do
+    fhr=$(( 10#${fhr3} ))
+    if (( fhr > FHMAX_GFS )); then
+        echo "Nothing to process for FHR = ${fhr3}, cycle"
+        continue
+    fi
+
+    fhmin=0
+    fhmax=84
+    if (( fhr >= fhmin && fhr <= fhmax )); then
+        if ((fhr % 6 == 0)); then
+            "${AWIPSG2SH}"
+        fi
+    fi
+
+    fhmin=90
+    fhmax=240
+    if (( fhr >= fhmin && fhr <= fhmax )); then
+        if ((fhr % 6 == 0)); then
+            export fcsthrs="${fhr3}"
+            "${AWIPSG2SH}"
+        fi
+    fi
+done
+
+
+###############################################################
+# Force Exit out cleanly
+if [[ ${KEEPDATA:-"NO"} == "NO" ]] ; then rm -rf "${DATA}" ; fi
+
+exit 0

workflow/applications/gfs_cycled.py

@@ -234,7 +234,8 @@ def get_task_names(self):
             gfs_tasks += ['gempak']
 
         if self.do_awips:
-            gfs_tasks += ['awips']
+            gfs_tasks += ['awips_20km_1p0']
+            gfs_tasks += ['awips_g2']
             gfs_tasks += ['fbwinds']
 
         if self.do_npoess:

workflow/applications/gfs_forecast_only.py

@@ -119,7 +119,8 @@ def get_task_names(self):
             tasks += ['gempak']
 
         if self.do_awips:
-            tasks += ['awips']
+            tasks += ['awips_20km_1p0']
+            tasks += ['awips_g2']
             tasks += ['fbwinds']
 
         if self.do_wafs:

workflow/rocoto/gfs_tasks.py

@@ -822,7 +822,7 @@ def fbwinds(self):
 
         return task
 
-    def awips(self):
+    def awips_20km_1p0(self):
 
         def _get_awipsgroups(cdump, config):
 
@@ -877,8 +877,68 @@ def _get_awipsgroups(cdump, config):
         vardict = {varname2: varval2, varname3: varval3}
 
         resources = self.get_resource('awips')
-        task = create_wf_task('awips', resources, cdump=self.cdump, envar=awipsenvars, dependency=dependencies,
-                              metatask='awips', varname=varname1, varval=varval1, vardict=vardict)
+        task = create_wf_task('awips_20km_1p0', resources, cdump=self.cdump, envar=awipsenvars, dependency=dependencies,
+                              metatask='awips_20km_1p0', varname=varname1, varval=varval1, vardict=vardict)
+
+        return task
+
+    def awips_g2(self):
+
+        def _get_awipsgroups(cdump, config):
+
+            fhmin = config['FHMIN']
+            fhmax = config['FHMAX']
+            fhout = config['FHOUT']
+
+            # Get a list of all forecast hours
+            fhrs = []
+            if cdump in ['gdas']:
+                fhrs = range(fhmin, fhmax + fhout, fhout)
+            elif cdump in ['gfs']:
+                fhmax = np.max(
+                    [config['FHMAX_GFS_00'], config['FHMAX_GFS_06'], config['FHMAX_GFS_12'], config['FHMAX_GFS_18']])
+                fhout = config['FHOUT_GFS']
+                fhmax_hf = config['FHMAX_HF_GFS']
+                fhout_hf = config['FHOUT_HF_GFS']
+                if fhmax > 240:
+                    fhmax = 240
+                if fhmax_hf > 240:
+                    fhmax_hf = 240
+                fhrs_hf = list(range(fhmin, fhmax_hf + fhout_hf, fhout_hf))
+                fhrs = fhrs_hf + list(range(fhrs_hf[-1] + fhout, fhmax + fhout, fhout))
+
+            nawipsgrp = config['NAWIPSGRP']
+            ngrps = nawipsgrp if len(fhrs) > nawipsgrp else len(fhrs)
+
+            fhrs = [f'f{fhr:03d}' for fhr in fhrs]
+            fhrs = np.array_split(fhrs, ngrps)
+            fhrs = [fhr.tolist() for fhr in fhrs]
+
+            grp = ' '.join([f'_{fhr[0]}-{fhr[-1]}' for fhr in fhrs])
+            dep = ' '.join([fhr[-1] for fhr in fhrs])
+            lst = ' '.join(['_'.join(fhr) for fhr in fhrs])
+
+            return grp, dep, lst
+
+        deps = []
+        dep_dict = {'type': 'metatask', 'name': f'{self.cdump}post'}
+        deps.append(rocoto.add_dependency(dep_dict))
+        dependencies = rocoto.create_dependency(dep=deps)
+
+        awipsenvars = self.envars.copy()
+        awipsenvar_dict = {'FHRGRP': '#grp#',
+                           'FHRLST': '#lst#',
+                           'ROTDIR': self._base.get('ROTDIR')}
+        for key, value in awipsenvar_dict.items():
+            awipsenvars.append(rocoto.create_envar(name=key, value=str(value)))
+
+        varname1, varname2, varname3 = 'grp', 'dep', 'lst'
+        varval1, varval2, varval3 = _get_awipsgroups(self.cdump, self._configs['awips'])
+        vardict = {varname2: varval2, varname3: varval3}
+
+        resources = self.get_resource('awips')
+        task = create_wf_task('awips_g2', resources, cdump=self.cdump, envar=awipsenvars, dependency=dependencies,
+                              metatask='awips_g2', varname=varname1, varval=varval1, vardict=vardict)
 
         return task
 

workflow/rocoto/tasks.py

@@ -22,7 +22,7 @@ class Tasks:
                    'fcst', 'post', 'ocnpost',
                    'verfozn', 'verfrad', 'vminmon', 'metp',
                    'tracker', 'genesis', 'genesis_fsu',
-                   'postsnd', 'awips', 'fbwinds', 'gempak',
+                   'postsnd', 'awips_g2sh', 'awips_20sh', 'fbwinds', 'gempak',
                    'waveawipsbulls', 'waveawipsgridded', 'wavegempak', 'waveinit',
                    'wavepostbndpnt', 'wavepostbndpntbll', 'wavepostpnt', 'wavepostsbs', 'waveprep',
                    'npoess']