Prepare GTS snow depth observations for JEDI-based Land DA (#1761)

This PR:

- adds a job to prepare GTS snow depth observations as a task in the workflow.
This task depends on the `prep.sh` job to bring GTS adpsfc bufr data. To test this
type of data DMPDIR in `config.base` needed to be pointed to "/scratch1/NCEPDEV/global/Jiarui.Dong/JEDI/GlobalWorkflow/para_gfs/glopara_dump". 
- land_analysis.py introduces a method `prepare_GTS` for this type of data.
- Updates are necessary in the GDASApp repo. See companion PR NOAA-EMC/GDASApp#541

The `preplandobs` job runs at the all four cycles in the workflow. The `prepare_IMS` job runs
only at 18z cycle, and this is controlled in the script.

Externals.cfg

@@ -57,7 +57,7 @@ protocol = git
 required = False
 
 [GDASApp]
-hash = fd8aa2f
+hash = 7966501
 local_path = sorc/gdas.cd
 repo_url = https://github.com/NOAA-EMC/GDASApp.git
 protocol = git

parm/config/gfs/config.preplandobs

@@ -8,6 +8,9 @@ echo "BEGIN: config.preplandobs"
 # Get task specific resources
 . "${EXPDIR}/config.resources" preplandobs
 
+export GTS_OBS_LIST="${HOMEgfs}/sorc/gdas.cd/parm/land/prep/prep_gts.yaml"
+export BUFR2IODAX="${HOMEgfs}/exec/bufr2ioda.x"
+export BUFR2IODAYAML="${HOMEgfs}/sorc/gdas.cd/test/testinput/bufr_adpsfc_snow.yaml"
 export FIMS_NML_TMPL="${HOMEgfs}/sorc/gdas.cd/parm/land/prep/fims.nml.j2"
 export IMS_OBS_LIST="${HOMEgfs}/sorc/gdas.cd/parm/land/prep/prep_ims.yaml"
 export CALCFIMSEXE="${HOMEgfs}/exec/calcfIMS.exe"

scripts/exglobal_prep_land_obs.py

@@ -1,8 +1,8 @@
 #!/usr/bin/env python3
 # exglobal_land_analysis_prepare.py
 # This script creates a LandAnalysis object
-# and runs the prepare_IMS method
-# which perform the pre-processing for IMS data
+# and runs the prepare_GTS and prepare_IMS method
+# which perform the pre-processing for GTS and IMS data
 import os
 
 from wxflow import Logger, cast_strdict_as_dtypedict
@@ -20,4 +20,6 @@
 
     # Instantiate the land prepare task
     LandAnl = LandAnalysis(config)
-    LandAnl.prepare_IMS()
+    LandAnl.prepare_GTS()
+    if f"{ LandAnl.runtime_config.cyc }" == '18':
+        LandAnl.prepare_IMS()

sorc/checkout.sh

@@ -160,7 +160,7 @@ if [[ ${checkout_gsi} == "YES" ]]; then
 fi
 
 if [[ ${checkout_gdas} == "YES" ]]; then
-  checkout "gdas.cd" "https://github.com/NOAA-EMC/GDASApp.git" "fd8aa2f"; errs=$((errs + $?))
+  checkout "gdas.cd" "https://github.com/NOAA-EMC/GDASApp.git" "7966501"; errs=$((errs + $?))
 fi
 
 if [[ ${checkout_gsi} == "YES" || ${checkout_gdas} == "YES" ]]; then

sorc/link_workflow.sh

@@ -308,6 +308,7 @@ if [[ -d "${HOMEgfs}/sorc/gdas.cd" ]]; then
                        "soca_setcorscales.x" \
                        "soca_gridgen.x" \
                        "soca_var.x" \
+                       "bufr2ioda.x" \
                        "calcfIMS.exe" \
                        "apply_incr.exe" )
   for gdasexe in "${JEDI_EXE[@]}"; do

ush/python/pygfs/task/land_analysis.py

@@ -54,6 +54,81 @@ def __init__(self, config):
         # task_config is everything that this task should need
         self.task_config = AttrDict(**self.config, **self.runtime_config, **local_dict)
 
+    @logit(logger)
+    def prepare_GTS(self) -> None:
+        """Prepare the GTS data for a global land analysis
+
+        This method will prepare GTS data for a global land analysis using JEDI.
+        This includes:
+        - processing GTS bufr snow depth observation data to IODA format
+
+        Parameters
+        ----------
+        Analysis: parent class for GDAS task
+
+        Returns
+        ----------
+        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', 'ntiles']
+        for key in keys:
+            localconf[key] = self.task_config[key]
+
+        # Read and render the GTS_OBS_LIST yaml
+        logger.info(f"Reading {self.task_config.GTS_OBS_LIST}")
+        prep_gts_config = parse_j2yaml(self.task_config.GTS_OBS_LIST, localconf)
+        logger.debug(f"{self.task_config.GTS_OBS_LIST}:\n{pformat(prep_gts_config)}")
+
+        # copy the GTS obs files from COM_OBS to DATA/obs
+        logger.info("Copying GTS obs for bufr2ioda.x")
+        FileHandler(prep_gts_config.gtsbufr).sync()
+
+        # generate bufr2ioda YAML file
+        bufr2ioda_yaml = os.path.join(self.runtime_config.DATA, "bufr_adpsfc_snow.yaml")
+        logger.info(f"Generate BUFR2IODA YAML file: {bufr2ioda_yaml}")
+        temp_yaml = parse_j2yaml(self.task_config.BUFR2IODAYAML, self.task_config)
+        save_as_yaml(temp_yaml, bufr2ioda_yaml)
+        logger.info(f"Wrote bufr2ioda YAML to: {bufr2ioda_yaml}")
+
+        logger.info("Link BUFR2IODAX into DATA/")
+        exe_src = self.task_config.BUFR2IODAX
+        exe_dest = os.path.join(localconf.DATA, os.path.basename(exe_src))
+        if os.path.exists(exe_dest):
+            rm_p(exe_dest)
+        os.symlink(exe_src, exe_dest)
+
+        output_file = f"{localconf.OPREFIX}adpsfc_snow.nc4"
+        if os.path.isfile(f"{os.path.join(localconf.DATA, output_file)}"):
+            rm_p(output_file)
+
+        # execute BUFR2IODAX to convert GTS bufr data into IODA format
+        yaml_file = f"bufr_adpsfc_snow.yaml"
+        if not os.path.isfile(f"{os.path.join(localconf.DATA, yaml_file)}"):
+            logger.exception(f"{yaml_file} not found")
+            raise FileNotFoundError(f"{os.path.join(localconf.DATA, yaml_file)}")
+        exe = Executable(self.task_config.BUFR2IODAX)
+        exe.add_default_arg(os.path.join(localconf.DATA, f"{yaml_file}"))
+
+        logger.info(f"Executing {exe}")
+        try:
+            exe()
+        except OSError:
+            raise OSError(f"Failed to execute {exe}")
+        except Exception:
+            raise WorkflowException(f"An error occured during execution of {exe}")
+
+        # 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()
+        except OSError as err:
+            logger.exception(f"{self.task_config.BUFR2IODAX} failed to produce {output_file}")
+            raise OSError(err)
+
     @logit(logger)
     def prepare_IMS(self) -> None:
         """Prepare the IMS data for a global land analysis

workflow/rocoto/gfs_cycled_xml.py

@@ -20,12 +20,6 @@ def get_cycledefs(self):
         sdate = sdate + to_timedelta(interval)
         strings.append(f'\t<cycledef group="gdas">{sdate.strftime("%Y%m%d%H%M")} {edate.strftime("%Y%m%d%H%M")} {interval}</cycledef>')
 
-        if self._app_config.do_jedilandda:
-            sdate_land_str = sdate.replace(hour=18, minute=0, second=0).strftime("%Y%m%d%H%M")
-            edate_land_str = edate.strftime("%Y%m%d%H%M")
-            if edate >= sdate:
-                strings.append(f'\t<cycledef group="gdas_land_prep">{sdate_land_str} {edate_land_str} 24:00:00</cycledef>')
-
         if self._app_config.gfs_cyc != 0:
             sdate_gfs = self._base['SDATE_GFS']
             edate_gfs = self._base['EDATE_GFS']

workflow/rocoto/gfs_tasks.py

@@ -337,26 +337,16 @@ def preplandobs(self):
         dependencies = rocoto.create_dependency(dep=deps)
 
         resources = self.get_resource('preplandobs')
-        task = create_wf_task('preplandobs', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies,
-                              cycledef=f'{self.cdump}_land_prep')
+        task = create_wf_task('preplandobs', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies)
 
         return task
 
     def landanl(self):
 
         deps = []
-        dep_dict = {'type': 'task', 'name': f'{self.cdump}prep'}
-        deps.append(rocoto.add_dependency(dep_dict))
-
-        # Either gdaspreplandobs (runs in 18z cycle) or not 18z cycle
-        sub_deps = []
         dep_dict = {'type': 'task', 'name': f'{self.cdump}preplandobs'}
-        sub_deps.append(rocoto.add_dependency(dep_dict))
-        dep_dict = {'type': 'strneq', 'left': '@H', 'right': 18}
-        sub_deps.append(rocoto.add_dependency(dep_dict))
-        deps.append(rocoto.create_dependency(dep_condition='xor', dep=sub_deps))
-
-        dependencies = rocoto.create_dependency(dep_condition='and', dep=deps)
+        deps.append(rocoto.add_dependency(dep_dict))
+        dependencies = rocoto.create_dependency(dep=deps)
 
         resources = self.get_resource('landanl')
         task = create_wf_task('landanl', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies)