Prepare observations for the updates to the ensemble members.

env/HERA.env

@@ -4,9 +4,9 @@ if [[ $# -ne 1 ]]; then
 
     echo "Must specify an input argument to set runtime environment variables!"
     echo "argument can be any one of the following:"
-    echo "atmanlrun atmensanlrun aeroanlrun snowanl"
+    echo "atmanlrun atmensanlrun aeroanlrun prepsnowobs snowanl"
     echo "anal sfcanl fcst post metp"
-    echo "eobs eupd ecen efcs epos"
+    echo "eobs eupd ecen efcs epos epsn"
     echo "postsnd awips gempak"
     exit 1
 
@@ -42,7 +42,7 @@ if [[ "${step}" = "prep" ]] || [[ "${step}" = "prepbufr" ]]; then
     export sys_tp="HERA"
     export launcher_PREP="srun"
 
-elif [[ "${step}" = "prepsnowobs" ]]; then
+elif [[ "${step}" = "prepsnowobs" ]] || [[ "${step}" = "epsn" ]]; then
 
     export APRUN_CALCFIMS="${launcher} -n 1"
 

jobs/JGLOBAL_PREP_SNOW_OBS

@@ -1,32 +1,53 @@
 #! /usr/bin/env bash
 
 source "${HOMEgfs}/ush/preamble.sh"
-export DATA=${DATA:-${DATAROOT}/${RUN}snowanl_${cyc}}
-source "${HOMEgfs}/ush/jjob_header.sh" -e "prepsnowobs" -c "base prepsnowobs"
+if (( 10#${ENSMEM:-0} > 0 )); then
+  source "${HOMEgfs}/ush/jjob_header.sh" -e "epsn" -c "base prepsnowobs epsn"
+  export CASE=${CASE_ENS}
+else
+  source "${HOMEgfs}/ush/jjob_header.sh" -e "prepsnowobs" -c "base prepsnowobs"
+fi
+
+##############################################
+# Begin JOB SPECIFIC work
+##############################################
+
+# Restart conditions for GFS cycle come from GDAS
+rCDUMP=${RUN}
+export rCDUMP="${RUN/gfs/gdas}"
+
+# Get OBS for ENKFGDAS run from GDAS
+oCDUMP=${RUN}
+export oCDUMP="${RUN/enkfgdas/gdas}"
 
 ##############################################
 # Set variables used in the script
 ##############################################
 # Ignore possible spelling error (nothing is misspelled)
 # shellcheck disable=SC2153
-GDATE=$(date --utc +%Y%m%d%H -d "${PDY} ${cyc} - ${assim_freq} hours")
-gPDY=${GDATE:0:8}
-gcyc=${GDATE:8:2}
-GDUMP="gdas"
+GDATE=$(date --utc -d "${PDY} ${cyc} - ${assim_freq} hours" +%Y%m%d%H)
+declare -rx GDATE
+# shellcheck disable=
+declare -rx gPDY="${GDATE:0:8}"
+declare -rx gcyc="${GDATE:8:2}"
+
 
 ##############################################
 # Begin JOB SPECIFIC work
 ##############################################
 # Generate COM variables from templates
-YMD=${PDY} HH=${cyc} declare_from_tmpl -rx COM_OBS
+RUN=${oCDUMP} MEMDIR="" YMD=${PDY} HH=${cyc} declare_from_tmpl -rx COMIN_OBS:COM_OBS_TMPL
+RUN=${RUN} YMD=${PDY} HH=${cyc} declare_from_tmpl -rx COMOUT_OBS:COM_OBS_TMPL
 
-RUN=${GDUMP} YMD=${gPDY} HH=${gcyc} declare_from_tmpl -rx \
+RUN=${rCDUMP} YMD=${gPDY} HH=${gcyc} declare_from_tmpl -rx \
     COM_ATMOS_RESTART_PREV:COM_ATMOS_RESTART_TMPL
 
+mkdir -m 775 -p "${COMOUT_OBS}"
+
 ###############################################################
 # Run relevant script
-EXSCRIPT=${GDASSNOWPREPPY:-${SCRgfs}/exglobal_prep_snow_obs.py}
-${EXSCRIPT}
+###############################################################
+${GDASSNOWPREPPY:-${SCRgfs}/exglobal_prep_snow_obs.py}
 status=$?
 [[ ${status} -ne 0 ]] && (echo "FATAL ERROR: Error executing ${EXSCRIPT}, ABORT!"; exit "${status}")
 
@@ -41,4 +62,10 @@ if [[ -e "${pgmout}" ]] ; then
   cat "${pgmout}"
 fi
 
+##########################################
+# Remove the Temporary working directory
+##########################################
+cd "${DATAROOT}" || true
+[[ ${KEEPDATA} = "NO" ]] && rm -rf "${DATA}"
+
 exit 0

parm/config/gfs/config.com

@@ -38,7 +38,7 @@ if [[ "${RUN_ENVIR:-emc}" == "nco" ]]; then
     COM_OBS_TMPL=$(compath.py "${envir}/obsproc/${obsproc_ver}")'/${RUN}.${YMD}/${HH}/atmos'
     COM_RTOFS_TMPL=$(compath.py "${envir}/${WAVECUR_DID}/${rtofs_ver}")
 else
-    COM_OBS_TMPL='${ROTDIR}/${RUN}.${YMD}/${HH}/obs'
+    COM_OBS_TMPL='${ROTDIR}/${RUN}.${YMD}/${HH}/obs/${MEMDIR}'
     COM_RTOFS_TMPL='${DMPDIR}'
 fi
 declare -rx COM_OBS_TMPL COM_RTOFS_TMPL

parm/config/gfs/config.epsn

@@ -0,0 +1,13 @@
+#! /usr/bin/env bash
+
+########## config.epsn ##########
+# Snow Obs Prep specific
+
+echo "BEGIN: config.epsn"
+
+source "${EXPDIR}/config.prepsnowobs"
+
+# Get task specific resources
+. "${EXPDIR}/config.resources" epsn
+
+echo "END: config.epsn"

parm/config/gfs/config.resources

@@ -18,7 +18,7 @@ if (( $# != 1 )); then
     echo "upp atmos_products"
     echo "tracker genesis genesis_fsu"
     echo "verfozn verfrad vminmon fit2obs metp arch cleanup"
-    echo "eobs ediag eomg eupd ecen esfc efcs epos earc"
+    echo "eobs ediag eomg eupd ecen esfc efcs epos earc epsn"
     echo "init_chem mom6ic oceanice_products"
     echo "waveinit waveprep wavepostsbs wavepostbndpnt wavepostbndpntbll wavepostpnt"
     echo "wavegempak waveawipsbulls waveawipsgridded"
@@ -83,11 +83,11 @@ case ${step} in
     fi
     ;;
 
-  "prepsnowobs")
-    export wtime_prepsnowobs="00:05:00"
-    export npe_prepsnowobs=1
-    export nth_prepsnowobs=1
-    export npe_node_prepsnowobs=1
+  "prepsnowobs" | "epsn")
+    declare -x "wtime_${step}"="00:05:00"
+    declare -x "npe_${step}"=1
+    declare -x "nth_${step}"=1
+    declare -x "npe_node_${step}"=1
     ;;
 
   "prepatmiodaobs")

ush/python/pygfs/task/snow_analysis.py

@@ -45,7 +45,8 @@ def __init__(self, config):
                 'npz': self.config.LEVS - 1,
                 'SNOW_WINDOW_BEGIN': _window_begin,
                 'SNOW_WINDOW_LENGTH': f"PT{self.config['assim_freq']}H",
-                'OPREFIX': f"{self.runtime_config.RUN}.t{self.runtime_config.cyc:02d}z.",
+                'OPREFIX_IN': f"{self.runtime_config.RUN.replace('enkf','')}.t{self.runtime_config.cyc:02d}z.",
+                'OPREFIX_OUT': f"{self.runtime_config.RUN}.t{self.runtime_config.cyc:02d}z.",
                 'APREFIX': f"{self.runtime_config.RUN}.t{self.runtime_config.cyc:02d}z.",
                 'jedi_yaml': _letkfoi_yaml
             }
@@ -73,8 +74,8 @@ def prepare_GTS(self) -> None:
 
         # create a temporary dict of all keys needed in this method
         localconf = AttrDict()
-        keys = ['HOMEgfs', 'DATA', 'current_cycle', 'COM_OBS', 'COM_ATMOS_RESTART_PREV',
-                'OPREFIX', 'CASE', 'OCNRES', 'ntiles']
+        keys = ['HOMEgfs', 'DATA', 'current_cycle', 'COMIN_OBS', 'COMOUT_OBS', 'COM_ATMOS_RESTART_PREV',
+                'OPREFIX_IN', 'OPREFIX_OUT', 'CASE', 'OCNRES', 'ntiles']
         for key in keys:
             localconf[key] = self.task_config[key]
 
@@ -126,7 +127,7 @@ def _gtsbufr2iodax(exe, yaml_file):
         # Ensure the IODA snow depth GTS file is produced by the IODA converter
         # If so, copy to COM_OBS/
         try:
-            FileHandler(prep_gts_config.gtsioda).sync()
+            FileHandler(prep_gts_config.gtsioda[self.runtime_config.RUN]).sync()
         except OSError as err:
             logger.exception(f"{self.task_config.BUFR2IODAX} failed to produce GTS ioda files")
             raise OSError(err)
@@ -152,8 +153,8 @@ def prepare_IMS(self) -> None:
 
         # create a temporary dict of all keys needed in this method
         localconf = AttrDict()
-        keys = ['DATA', 'current_cycle', 'COM_OBS', 'COM_ATMOS_RESTART_PREV',
-                'OPREFIX', 'CASE', 'OCNRES', 'ntiles', 'FIXgfs']
+        keys = ['DATA', 'current_cycle', 'COMIN_OBS', 'COMOUT_OBS', 'COM_ATMOS_RESTART_PREV',
+                'OPREFIX_IN', 'OPREFIX_OUT', 'CASE', 'OCNRES', 'ntiles', 'FIXgfs']
         for key in keys:
             localconf[key] = self.task_config[key]
 
@@ -227,7 +228,7 @@ def prepare_IMS(self) -> None:
             logger.exception(f"{self.task_config.IMS2IODACONV} failed to produce {output_file}")
             raise FileNotFoundError(f"{os.path.join(localconf.DATA, output_file)}")
         else:
-            logger.info(f"Copy {output_file} to {self.task_config.COM_OBS}")
+            logger.info(f"Copy {output_file} to {self.task_config.COMOUT_OBS}")
             FileHandler(prep_ims_config.ims2ioda).sync()
 
     @logit(logger)
@@ -249,7 +250,7 @@ def initialize(self) -> None:
         # create a temporary dict of all keys needed in this method
         localconf = AttrDict()
         keys = ['DATA', 'current_cycle', 'COM_OBS', 'COM_ATMOS_RESTART_PREV',
-                'OPREFIX', 'CASE', 'OCNRES', 'ntiles']
+                'OPREFIX_OUT', 'CASE', 'OCNRES', 'ntiles']
         for key in keys:
             localconf[key] = self.task_config[key]
 

workflow/applications/gfs_cycled.py

@@ -60,6 +60,8 @@ def _get_app_configs(self):
                 configs += ['atmensanlinit', 'atmensanlrun', 'atmensanlfinal']
             else:
                 configs += ['eobs', 'eomg', 'ediag', 'eupd']
+            if self.do_jedisnowda:
+                configs += ['epsn']
             configs += ['ecen', 'esfc', 'efcs', 'echgres', 'epos', 'earc']
 
         if self.do_fit2obs:
@@ -168,6 +170,8 @@ def get_task_names(self):
             else:
                 hybrid_tasks += ['eobs', 'eupd', 'echgres']
                 hybrid_tasks += ['ediag'] if self.lobsdiag_forenkf else ['eomg']
+            if self.do_jedisnowda:
+                hybrid_tasks += ['epsn']
             hybrid_after_eupd_tasks += ['ecen', 'esfc', 'efcs', 'epos', 'earc', 'cleanup']
 
         # Collect all "gdas" cycle tasks

workflow/rocoto/gfs_tasks.py

@@ -2457,6 +2457,45 @@ def atmensanlfinal(self):
 
         return task
 
+    def epsn(self):
+
+        deps = []
+        dep_dict = {'type': 'task', 'name': f'{self.cdump.replace("enkf","")}prep'}
+        deps.append(rocoto.add_dependency(dep_dict))
+        dep_dict = {'type': 'metatask', 'name': 'enkfgdasepmn', 'offset': '-06:00:00'}
+        deps.append(rocoto.add_dependency(dep_dict))
+        dependencies = rocoto.create_dependency(dep_condition='and', dep=deps)
+
+        epsnenvars = self.envars.copy()
+        epsnenvars_dict = {'ENSMEM': '#member#',
+                           'MEMDIR': 'mem#member#'
+                           }
+        for key, value in epsnenvars_dict.items():
+            epsnenvars.append(rocoto.create_envar(name=key, value=str(value)))
+
+        resources = self.get_resource('epsn')
+        task_name = f'{self.cdump}prepsnowobs_mem#member#'
+        task_dict = {'task_name': task_name,
+                     'resources': resources,
+                     'dependency': dependencies,
+                     'envars': epsnenvars,
+                     'cycledef': self.cdump.replace('enkf', ''),
+                     'command': f'{self.HOMEgfs}/jobs/rocoto/prepsnowobs.sh',
+                     'job_name': f'{self.pslot}_{task_name}_@H',
+                     'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log',
+                     'maxtries': '&MAXTRIES;'
+                     }
+
+        member_var_dict = {'member': ' '.join([str(mem).zfill(3) for mem in range(1, self.nmem + 1)])}
+        metatask_dict = {'task_name': f'{self.cdump}prepsnowobs',
+                         'var_dict': member_var_dict,
+                         'task_dict': task_dict
+                         }
+
+        task = rocoto.create_task(metatask_dict)
+
+        return task
+
     def ecen(self):
 
         def _get_ecengroups():