extend stage_ic to gdas for cycled parallels (#2294)

ci/cases/pr/C48C48_ufs_hybatmDA.yaml

@@ -9,7 +9,6 @@ arguments:
   resensatmos: 48
   comroot: {{ 'RUNTESTS' | getenv }}/COMROOT
   expdir: {{ 'RUNTESTS' | getenv }}/EXPDIR
-  icsdir: {{ 'ICSDIR_ROOT' | getenv }}/C48C48
   idate: 2021032312
   edate: 2021032400
   nens: 2

parm/config/gfs/config.base.emc.dyn

@@ -397,7 +397,7 @@ export FHMAX_FITS=132
 [[ "${FHMAX_FITS}" -gt "${FHMAX_GFS}" ]] && export FHMAX_FITS=${FHMAX_GFS}
 
 # The monitor jobs are not yet supported for JEDIATMVAR
-if [[ ${DO_JEDIATMVAR} = "NO" ]]; then
+if [[ ${DO_JEDIATMVAR} = "YES" ]]; then
    export DO_VERFOZN="NO"     # Ozone data assimilation monitoring
    export DO_VERFRAD="NO"     # Radiance data assimilation monitoring
    export DO_VMINMON="NO"     # GSI minimization monitoring

parm/config/gfs/config.stage_ic

@@ -7,6 +7,8 @@
 # Get task specific resources
 source "${EXPDIR}/config.resources" stage_ic
 
+# Stage ICs for coupled system
+if [[ "${DO_COUPLED}" = "YES" ]]; then
 case "${CASE}" in
   "C48" | "C96")
     export CPL_ATMIC="workflow_${CASE}_refactored"
@@ -37,5 +39,12 @@
     exit 1
     ;;
 esac
+fi # End block for coupled ICs
+
+# Stage ICs for cycled runs
+if [[ "$MODE" = "cycled" ]]; then
+  export CPL_ATMIC="C48C48"
+  export CPL_ATMDAIC="C48C48"
+fi
 
 echo "END: config.stage_ic"

scripts/exglobal_stage_ic.sh

@@ -25,6 +25,8 @@
   echo "FATAL ERROR: Unable to copy ${1} to ${2} (Error code ${3})"
 }
 
+if [[ "${MODE}" == "forecast-only" ]]; then
+
 ###############################################################
 for MEMDIR in "${MEMDIR_ARRAY[@]}"; do
 
@@ -152,6 +154,59 @@
 
 done # for MEMDIR in "${MEMDIR_ARRAY[@]}"; do
 
+# Stage files for MODE="cycled"
+else
+  # Stage deterministic ICs from previous cycle
+  RUN=${rCDUMP} YMD=${gPDY} HH=${gcyc} generate_com COM_TOP:COM_TOP_TMPL
+  COM_TOP_BASE=$(dirname $COM_TOP)
+  [[ ! -d "${COM_TOP_BASE}" ]] && mkdir -p "${COM_TOP_BASE}"
+  [[ -d "${COM_TOP_BASE}/${gcyc}" ]] && rm -rf "${COM_TOP_BASE}/${gcyc}"
+  src="${BASE_CPLIC}/${CPL_ATMIC:-}/${RUN}.${gPDY}/${gcyc}"
+  tgt="${COM_TOP_BASE}/"
+  ${NLN} "${src}" "${tgt}"
+  rc=$?
+  ((rc != 0)) && error_message "${src}" "${tgt}" "${rc}"
+  err=$((err + rc))
+
+  # Stage ensemble ICs from previous cycle
+  if [[ "${DOHYBVAR}" == "YES" ]]; then
+    RUN=enkf${rCDUMP} YMD=${gPDY} HH=${gcyc} generate_com COM_TOP:COM_TOP_TMPL
+    COM_TOP_BASE=$(dirname $COM_TOP)
+    [[ ! -d "${COM_TOP_BASE}" ]] && mkdir -p "${COM_TOP_BASE}"
+    [[ -d "${COM_TOP_BASE}/${gcyc}" ]] && rm -rf "${COM_TOP_BASE}/${gcyc}"
+    src="${BASE_CPLIC}/${CPL_ATMIC:-}/enkf${RUN}.${gPDY}/${gcyc}"
+    tgt="${COM_TOP_BASE}/"
+    ${NLN} "${src}" "${tgt}"
+    rc=$?
+    ((rc != 0)) && error_message "${src}" "${tgt}" "${rc}"
+    err=$((err + rc))
+  fi
+
+  # Stage bias correction files
+  RUN=${rCDUMP} YMD=${PDY} HH=${cyc} generate_com COM_ATMOS_ANALYSIS:COM_ATMOS_ANALYSIS_TMPL
+  [[ ! -d "${COM_ATMOS_ANALYSIS}" ]] && mkdir -p "${COM_ATMOS_ANALYSIS}"
+  src="${BASE_CPLIC}/${CPL_ATMIC:-}/${RUN}.${PDY}/${cyc}/analysis/atmos"
+  tgt="${COM_ATMOS_ANALYSIS}/"
+  if [[ "${DO_JEDIATMVAR}" == "YES" || "${DO_JEDIATMENS}" == "YES" ]]; then
+    flist="radbcor"
+  else
+    flist="abias abias_air abias_pc radstat"
+  fi
+  for ftype in $flist; do
+    file=${rCDUMP}.t${cyc}z.$ftype
+    ${NCP} "${src}/$file" "${tgt}"
+    rc=$?
+    ((rc != 0)) && error_message "${src}" "${tgt}" "${rc}"
+    err=$((err + rc))
+  done
+  if [[ "${DO_JEDIATMVAR}" == "YES" || "${DO_JEDIATMENS}" == "YES" ]]; then
+    cd ${tgt}
+    tar -xvf ${rCDUMP}.t${cyc}z.radbcor
+    cd ${DATA}
+  fi
+
+fi
+
 ###############################################################
 # Check for errors and exit if any of the above failed
 if [[ "${err}" -ne 0 ]]; then

ush/python/pygfs/task/atm_analysis.py

@@ -94,7 +94,7 @@ def initialize(self: Analysis) -> None:
                     'NMEM_ENS', 'DATA', 'current_cycle', 'ntiles']
             for key in keys:
                 localconf[key] = self.task_config[key]
-            localconf.RUN = 'enkf' + self.task_config.RUN
+            localconf.RUN = 'enkfgdas'
             localconf.dirname = 'ens'
             FileHandler(self.get_fv3ens_dict(localconf)).sync()
 

workflow/applications/gfs_cycled.py

@@ -35,7 +35,7 @@ def _get_app_configs(self):
         Returns the config_files that are involved in the cycled app
         """
 
-        configs = ['prep']
+        configs = ['stage_ic', 'prep']
 
         if self.do_jediatmvar:
             configs += ['prepatmiodaobs', 'atmanlinit', 'atmanlrun', 'atmanlfinal']
@@ -166,7 +166,8 @@ def get_task_names(self):
             hybrid_after_eupd_tasks += ['ecen', 'esfc', 'efcs', 'epos', 'earc', 'cleanup']
 
         # Collect all "gdas" cycle tasks
-        gdas_tasks = gdas_gfs_common_tasks_before_fcst.copy()
+        gdas_tasks = ['stage_ic']
+        gdas_tasks += gdas_gfs_common_tasks_before_fcst.copy()
 
         if not self.do_jediatmvar:
             gdas_tasks += ['analdiag']

workflow/rocoto/gfs_tasks.py

@@ -24,14 +24,22 @@ def stage_ic(self):
 
         # Atm ICs
         if self.app_config.do_atm:
-            prefix = f"{cpl_ic['BASE_CPLIC']}/{cpl_ic['CPL_ATMIC']}/@Y@m@d@H/atmos"
-            for file in ['gfs_ctrl.nc'] + \
-                        [f'{datatype}_data.tile{tile}.nc'
-                         for datatype in ['gfs', 'sfc']
-                         for tile in range(1, self.n_tiles + 1)]:
+            if self.cdump in 'gdas':
+                prefix = f"{cpl_ic['BASE_CPLIC']}/{cpl_ic['CPL_ATMIC']}/gdas.@Y@m@d/@H/model_data/atmos/restart"
+                PDY = self._base.get('SDATE').strftime("%Y%m%d%H")[0:8]
+                cyc = self._base.get('SDATE').strftime("%Y%m%d%H")[8:10]
+                file = PDY + '.' + cyc + '0000.fv_core.res.nc'
                 data = f"{prefix}/{file}"
-                dep_dict = {'type': 'data', 'data': data}
-                deps.append(rocoto.add_dependency(dep_dict))
+                dep_dict = {'type': 'data', 'data': data, 'offset': f"-{timedelta_to_HMS(self._base['cycle_interval'])}"}
+            else:
+                prefix = f"{cpl_ic['BASE_CPLIC']}/{cpl_ic['CPL_ATMIC']}/@Y@m@d@H/atmos"
+                for file in ['gfs_ctrl.nc'] + \
+                            [f'{datatype}_data.tile{tile}.nc'
+                             for datatype in ['gfs', 'sfc']
+                             for tile in range(1, self.n_tiles + 1)]:
+                    data = f"{prefix}/{file}"
+                    dep_dict = {'type': 'data', 'data': data}
+            deps.append(rocoto.add_dependency(dep_dict))
         else:  # data-atmosphere
             # TODO - need more information about how these forcings are stored
             prefix = f"{cpl_ic['BASE_CPLIC']}/{cpl_ic['CPL_DATM']}/@Y@m@d@H"
@@ -70,13 +78,14 @@ def stage_ic(self):
 
         dependencies = rocoto.create_dependency(dep_condition='and', dep=deps)
 
+        cycledef = 'gdas_half' if self.cdump in ['gdas'] else self.cdump
         resources = self.get_resource('stage_ic')
         task_name = f'{self.cdump}stage_ic'
         task_dict = {'task_name': task_name,
                      'resources': resources,
                      'dependency': dependencies,
                      'envars': self.envars,
-                     'cycledef': self.cdump,
+                     'cycledef': cycledef,
                      'command': f'{self.HOMEgfs}/jobs/rocoto/stage_ic.sh',
                      'job_name': f'{self.pslot}_{task_name}_@H',
                      'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log',
@@ -833,7 +842,7 @@ def _fcst_cycled(self):
         dependencies = rocoto.create_dependency(dep_condition='and', dep=dependencies)
 
         if self.cdump in ['gdas']:
-            dep_dict = {'type': 'cycleexist', 'condition': 'not', 'offset': f"-{timedelta_to_HMS(self._base['cycle_interval'])}"}
+            dep_dict = {'type': 'task', 'name': f'{self.cdump}stage_ic'}
             dependencies.append(rocoto.add_dependency(dep_dict))
             dependencies = rocoto.create_dependency(dep_condition='or', dep=dependencies)
 
@@ -2535,7 +2544,7 @@ def efcs(self):
         dep_dict = {'type': 'task', 'name': f'{self.cdump}esfc'}
         deps.append(rocoto.add_dependency(dep_dict))
         dependencies = rocoto.create_dependency(dep_condition='and', dep=deps)
-        dep_dict = {'type': 'cycleexist', 'condition': 'not', 'offset': f"-{timedelta_to_HMS(self._base['cycle_interval'])}"}
+        dep_dict = {'type': 'task', 'name': f'gdasstage_ic'}
         dependencies.append(rocoto.add_dependency(dep_dict))
         dependencies = rocoto.create_dependency(dep_condition='or', dep=dependencies)