Add run and finalize methods to marine LETKF task

ci/cases/pr/C48mx500_hybAOWCDA.yaml

@@ -0,0 +1,24 @@
+experiment:
+  system: gfs
+  mode: cycled
+
+arguments:
+  pslot: {{ 'pslot' | getenv }}
+  app: S2S
+  resdetatmos: 48
+  resdetocean: 5.0
+  resensatmos: 48
+  comroot: {{ 'RUNTESTS' | getenv }}/COMROOT
+  expdir: {{ 'RUNTESTS' | getenv }}/EXPDIR
+  icsdir: {{ 'ICSDIR_ROOT' | getenv }}/C48mx500/20240610
+  idate: 2021032412
+  edate: 2021032418
+  nens: 3
+  gfs_cyc: 0
+  start: warm
+  yaml: {{ HOMEgfs }}/ci/cases/yamls/soca_gfs_defaults_ci.yaml
+
+skip_ci_on_hosts:
+  - wcoss2
+  - orion
+  - hercules

env/HERA.env

@@ -155,10 +155,10 @@ elif [[ "${step}" = "ocnanalecen" ]]; then
     export NTHREADS_OCNANALECEN=${NTHREADSmax}
     export APRUN_OCNANALECEN="${APRUN_default} --cpus-per-task=${NTHREADS_OCNANALECEN}"
 
-elif [[ "${step}" = "marineanalletkf" ]]; then
+elif [[ "${step}" = "marineanlletkf" ]]; then
 
-    export NTHREADS_MARINEANALLETKF=${NTHREADSmax}
-    export APRUN_MARINEANALLETKF="${APRUN_default} --cpus-per-task=${NTHREADS_MARINEANALLETKF}"
+    export NTHREADS_MARINEANLLETKF=${NTHREADSmax}
+    export APRUN_MARINEANLLETKF="${APRUN_default} --cpus-per-task=${NTHREADS_MARINEANLLETKF}"
 
 elif [[ "${step}" = "anal" ]] || [[ "${step}" = "analcalc" ]]; then
 

env/ORION.env

@@ -148,10 +148,10 @@ elif [[ "${step}" = "ocnanalecen" ]]; then
     export NTHREADS_OCNANALECEN=${NTHREADSmax}
     export APRUN_OCNANALECEN="${APRUN_default} --cpus-per-task=${NTHREADS_OCNANALECEN}"
 
-elif [[ "${step}" = "marineanalletkf" ]]; then
+elif [[ "${step}" = "marineanlletkf" ]]; then
 
-    export NTHREADS_MARINEANALLETKF=${NTHREADSmax}
-    export APRUN_MARINEANALLETKF="${APRUN_default} --cpus-per-task=${NTHREADS_MARINEANALLETKF}"
+    export NTHREADS_MARINEANLLETKF=${NTHREADSmax}
+    export APRUN_MARINEANLLETKF="${APRUN_default} --cpus-per-task=${NTHREADS_MARINEANLLETKF}"
 
 elif [[ "${step}" = "anal" ]] || [[ "${step}" = "analcalc" ]]; then
 

env/WCOSS2.env

@@ -126,10 +126,10 @@ elif [[ "${step}" = "ocnanalecen" ]]; then
     export NTHREADS_OCNANALECEN=${NTHREADSmax}
     export APRUN_OCNANALECEN="${APRUN_default} --cpus-per-task=${NTHREADS_OCNANALECEN}"
 
-elif [[ "${step}" = "marineanalletkf" ]]; then
+elif [[ "${step}" = "marineanlletkf" ]]; then
 
-    export NTHREADS_MARINEANALLETKF=${NTHREADSmax}
-    export APRUN_MARINEANALLETKF="${APRUN_default} --cpus-per-task=${NTHREADS_MARINEANALLETKF}"
+    export NTHREADS_MARINEANLLETKF=${NTHREADSmax}
+    export APRUN_MARINEANLLETKF="${APRUN_default} --cpus-per-task=${NTHREADS_MARINEANLLETKF}"
 
 elif [[ "${step}" = "atmanlfv3inc" ]]; then
 

jobs/JGLOBAL_MARINE_ANALYSIS_LETKF

@@ -1,6 +1,6 @@
 #!/bin/bash
 source "${HOMEgfs}/ush/preamble.sh"
-source "${HOMEgfs}/ush/jjob_header.sh" -e "marineanalletkf" -c "base ocnanal marineanalletkf"
+source "${HOMEgfs}/ush/jjob_header.sh" -e "marineanlletkf" -c "base ocnanal marineanlletkf"
 
 ##############################################
 # Set variables used in the script
@@ -13,10 +13,13 @@ gPDY=${GDATE:0:8}
 gcyc=${GDATE:8:2}
 
 YMD=${gPDY} HH=${gcyc} declare_from_tmpl -rx \
-   COMIN_OCEAN_HISTORY_PREV:COM_OCEAN_HISTORY_TMPL \
-   COMIN_ICE_HISTORY_PREV:COM_ICE_HISTORY_TMPL
+  COMIN_OCEAN_HISTORY_PREV:COM_OCEAN_HISTORY_TMPL \
+  COMIN_ICE_HISTORY_PREV:COM_ICE_HISTORY_TMPL
 
-YMD=${PDY} HH=${cyc} declare_from_tmpl -rx COMIN_OBS:COM_OBS_TMPL 
+YMD=${PDY} HH=${cyc} declare_from_tmpl -rx \
+  COMIN_OBS:COM_OBS_TMPL \
+  COMOUT_OCEAN_LETKF:COM_OCEAN_LETKF_TMPL \
+  COMOUT_ICE_LETKF:COM_ICE_LETKF_TMPL
 
 ##############################################
 # Begin JOB SPECIFIC work
@@ -25,7 +28,7 @@ YMD=${PDY} HH=${cyc} declare_from_tmpl -rx COMIN_OBS:COM_OBS_TMPL
 ###############################################################
 # Run relevant script
 
-EXSCRIPT=${GDASOCNLETKFPY:-${HOMEgfs}/scripts/exgdas_global_marine_analysis_letkf.py}
+EXSCRIPT=${GDASOCNLETKFPY:-${HOMEgfs}/scripts/exglobal_marine_analysis_letkf.py}
 ${EXSCRIPT}
 status=$?
 [[ ${status} -ne 0 ]] && exit "${status}"

jobs/JGLOBAL_MARINE_BMAT

@@ -2,17 +2,19 @@
 
 source "${HOMEgfs}/ush/preamble.sh"
 
-if (( 10#${ENSMEM:-0} > 0 )); then
+# source config.base, config.ocnanal and config.marinebmat
+# and pass marinebmat to ${machine}.env
+source "${HOMEgfs}/ush/jjob_header.sh" -e "marinebmat" -c "base ocnanal marinebmat"
+
+
+if (( 10#${NMEM_ENS:-0} > 0 )); then
   export DATAjob="${DATAROOT}/${RUN}marinebmat.${PDY:-}${cyc}"
-  export DATA="${DATAjob}/${jobid}"
+  export DATA=${DATA:-${DATAROOT}/${jobid:?}}
   # Create the directory to hold ensemble perturbations
   export DATAenspert="${DATAjob}/enspert"
   if [[ ! -d "${DATAenspert}" ]]; then mkdir -p "${DATAenspert}"; fi
 fi
 
-# source config.base, config.ocnanal and config.marinebmat
-# and pass marinebmat to ${machine}.env
-source "${HOMEgfs}/ush/jjob_header.sh" -e "marinebmat" -c "base ocnanal marinebmat"
 
 ##############################################
 # Set variables used in the script

jobs/rocoto/marineanalletkf.sh → jobs/rocoto/marineanlletkf.sh

@@ -8,7 +8,7 @@ source "${HOMEgfs}/ush/preamble.sh"
 status=$?
 [[ ${status} -ne 0 ]] && exit "${status}"
 
-export job="marineanalletkf"
+export job="marineanlletkf"
 export jobid="${job}.$$"
 
 ###############################################################

parm/config/gfs/config.com

@@ -82,12 +82,14 @@ declare -rx COM_OCEAN_HISTORY_TMPL=${COM_BASE}'/model/ocean/history'
 declare -rx COM_OCEAN_RESTART_TMPL=${COM_BASE}'/model/ocean/restart'
 declare -rx COM_OCEAN_INPUT_TMPL=${COM_BASE}'/model/ocean/input'
 declare -rx COM_OCEAN_ANALYSIS_TMPL=${COM_BASE}'/analysis/ocean'
+declare -rx COM_OCEAN_LETKF_TMPL=${COM_BASE}'/analysis/ocean/letkf'
 declare -rx COM_OCEAN_BMATRIX_TMPL=${COM_BASE}'/bmatrix/ocean'
 declare -rx COM_OCEAN_NETCDF_TMPL=${COM_BASE}'/products/ocean/netcdf'
 declare -rx COM_OCEAN_GRIB_TMPL=${COM_BASE}'/products/ocean/grib2'
 declare -rx COM_OCEAN_GRIB_GRID_TMPL=${COM_OCEAN_GRIB_TMPL}'/${GRID}'
 
 declare -rx COM_ICE_ANALYSIS_TMPL=${COM_BASE}'/analysis/ice'
+declare -rx COM_ICE_LETKF_TMPL=${COM_BASE}'/analysis/ice/letkf'
 declare -rx COM_ICE_BMATRIX_TMPL=${COM_BASE}'/bmatrix/ice'
 declare -rx COM_ICE_INPUT_TMPL=${COM_BASE}'/model/ice/input'
 declare -rx COM_ICE_HISTORY_TMPL=${COM_BASE}'/model/ice/history'

parm/config/gfs/config.marineanalletkf → parm/config/gfs/config.marineanlletkf

@@ -1,18 +1,19 @@
 #!/bin/bash
 
-########## config.marineanalletkf ##########
+########## config.marineanlletkf ##########
 # Ocn Analysis specific
 
-echo "BEGIN: config.marineanalletkf"
+echo "BEGIN: config.marineanlletkf"
 
 # Get task specific resources
-. "${EXPDIR}/config.resources" marineanalletkf
+. "${EXPDIR}/config.resources" marineanlletkf
 
 export MARINE_LETKF_EXEC="${JEDI_BIN}/gdas.x"
 export MARINE_LETKF_YAML_TMPL="${PARMgfs}/gdas/soca/letkf/letkf.yaml.j2"
 export MARINE_LETKF_STAGE_YAML_TMPL="${PARMgfs}/gdas/soca/letkf/letkf_stage.yaml.j2"
+export MARINE_LETKF_SAVE_YAML_TMPL="${PARMgfs}/gdas/soca/letkf/letkf_save.yaml.j2"
 
 export GRIDGEN_EXEC="${JEDI_BIN}/gdas_soca_gridgen.x"
 export GRIDGEN_YAML="${PARMgfs}/gdas/soca/gridgen/gridgen.yaml"
 
-echo "END: config.marineanalletkf"
+echo "END: config.marineanlletkf"

parm/config/gfs/config.resources

@@ -26,7 +26,7 @@ if (( $# != 1 )); then
     echo "waveinit waveprep wavepostsbs wavepostbndpnt wavepostbndpntbll wavepostpnt"
     echo "wavegempak waveawipsbulls waveawipsgridded"
     echo "postsnd awips gempak npoess"
-    echo "ocnanalprep prepoceanobs marinebmat ocnanalrun ocnanalecen marineanalletkf ocnanalchkpt ocnanalpost ocnanalvrfy"
+    echo "ocnanalprep prepoceanobs marinebmat ocnanalrun ocnanalecen marineanlletkf ocnanalchkpt ocnanalpost ocnanalvrfy"
     exit 1
 
 fi
@@ -600,7 +600,7 @@ case ${step} in
     tasks_per_node=$(( max_tasks_per_node / threads_per_task ))
     ;;
 
-  "marineanalletkf")
+  "marineanlletkf")
     ntasks=16
     case ${OCNRES} in
       "025")

ush/forecast_postdet.sh

@@ -460,12 +460,13 @@ MOM6_postdet() {
     fi
 
     # GEFS perturbations
-    # TODO if [[ $RUN} == "gefs" ]] block maybe be needed
+    if [[ "${RUN}" == "gefs" ]]; then 
     #     to ensure it does not interfere with the GFS when ensemble is updated in the GFS
-    if (( MEMBER > 0 )) && [[ "${ODA_INCUPD:-False}" == "True" ]]; then
-      ${NCP} "${COMIN_OCEAN_ANALYSIS}/mom6_increment.nc" "${DATA}/INPUT/mom6_increment.nc" \
-      || ( echo "FATAL ERROR: Unable to copy ensemble MOM6 increment, ABORT!"; exit 1 )
-    fi
+      if (( MEMBER > 0 )) && [[ "${ODA_INCUPD:-False}" == "True" ]]; then
+        ${NCP} "${COMIN_OCEAN_ANALYSIS}/mom6_increment.nc" "${DATA}/INPUT/mom6_increment.nc" \
+        || ( echo "FATAL ERROR: Unable to copy ensemble MOM6 increment, ABORT!"; exit 1 )
+      fi
+    fi # if [[ "${RUN}" == "gefs" ]]; then 
   fi  # if [[ "${RERUN}" == "NO" ]]; then
 
   # Link output files

ush/python/pygfs/task/marine_bmat.py

@@ -322,7 +322,7 @@ def finalize(self: Task) -> None:
         FileHandler({'copy': diagb_list}).sync()
 
         # Copy the ensemble perturbation diagnostics to the ROTDIR
-        if self.task_config.DOHYBVAR == "YES" or self.task_config.NMEM_ENS > 3:
+        if self.task_config.DOHYBVAR == "YES" or self.task_config.NMEM_ENS > 2:
             window_middle_iso = self.task_config.MARINE_WINDOW_MIDDLE.strftime('%Y-%m-%dT%H:%M:%SZ')
             weight_list = []
             src = os.path.join(self.task_config.DATA, f"ocn.ens_weights.incr.{window_middle_iso}.nc")

ush/python/pygfs/task/marine_letkf.py

@@ -6,11 +6,13 @@
 from pygfs.task.analysis import Analysis
 from typing import Dict
 from wxflow import (AttrDict,
+                    Executable,
                     FileHandler,
                     logit,
                     parse_j2yaml,
                     to_timedelta,
-                    to_YMDH)
+                    to_YMDH,
+                    WorkflowException)
 
 logger = getLogger(__name__.split('.')[-1])
 
@@ -64,17 +66,35 @@ def initialize(self):
         logger.info("initialize")
 
         # make directories and stage ensemble background files
-        ensbkgconf = AttrDict()
-        keys = ['previous_cycle', 'current_cycle', 'DATA', 'NMEM_ENS',
-                'PARMgfs', 'ROTDIR', 'COM_OCEAN_HISTORY_TMPL', 'COM_ICE_HISTORY_TMPL']
-        for key in keys:
-            ensbkgconf[key] = self.task_config[key]
-        ensbkgconf.RUN = 'enkfgdas'
-        soca_ens_bkg_stage_list = parse_j2yaml(self.task_config.SOCA_ENS_BKG_STAGE_YAML_TMPL, ensbkgconf)
-        FileHandler(soca_ens_bkg_stage_list).sync()
         soca_fix_stage_list = parse_j2yaml(self.task_config.SOCA_FIX_YAML_TMPL, self.task_config)
         FileHandler(soca_fix_stage_list).sync()
-        letkf_stage_list = parse_j2yaml(self.task_config.MARINE_LETKF_STAGE_YAML_TMPL, self.task_config)
+        stageconf = AttrDict()
+        keys = ['current_cycle',
+                'previous_cycle',
+                'COM_ICE_LETKF_TMPL',
+                'COM_OCEAN_LETKF_TMPL',
+                'COM_ICE_HISTORY_TMPL',
+                'COM_OCEAN_HISTORY_TMPL',
+                'COMIN_OCEAN_HISTORY_PREV',
+                'COMIN_ICE_HISTORY_PREV',
+                'COMOUT_ICE_LETKF',
+                'COMOUT_OCEAN_LETKF',
+                'DATA',
+                'NMEM_ENS',
+                'PARMgfs',
+                'ROTDIR',
+                'WINDOW_BEGIN',
+                'WINDOW_MIDDLE']
+        for key in keys:
+            stageconf[key] = self.task_config[key]
+        stageconf.RUN = 'enkfgdas'
+
+        # stage ensemble background files
+        soca_ens_bkg_stage_list = parse_j2yaml(self.task_config.SOCA_ENS_BKG_STAGE_YAML_TMPL, stageconf)
+        FileHandler(soca_ens_bkg_stage_list).sync()
+
+        # stage letkf-specific files
+        letkf_stage_list = parse_j2yaml(self.task_config.MARINE_LETKF_STAGE_YAML_TMPL, stageconf)
         FileHandler(letkf_stage_list).sync()
 
         obs_list = parse_j2yaml(self.task_config.OBS_YAML, self.task_config)
@@ -83,7 +103,7 @@ def initialize(self):
         obs_files = []
         for ob in obs_list['observers']:
             obs_name = ob['obs space']['name'].lower()
-            obs_filename = f"{self.task_config.RUN}.t{self.task_config.cyc}z.{obs_name}.{to_YMDH(self.task_config.current_cycle)}.nc"
+            obs_filename = f"{self.task_config.RUN}.t{self.task_config.cyc}z.{obs_name}.{to_YMDH(self.task_config.current_cycle)}.nc4"
             obs_files.append((obs_filename, ob))
 
         obs_files_to_copy = []
@@ -102,12 +122,7 @@ def initialize(self):
         FileHandler({'copy': obs_files_to_copy}).sync()
 
         # make the letkf.yaml
-        letkfconf = AttrDict()
-        keys = ['WINDOW_BEGIN', 'WINDOW_MIDDLE', 'RUN', 'gcyc', 'NMEM_ENS']
-        for key in keys:
-            letkfconf[key] = self.task_config[key]
-        letkfconf.RUN = 'enkfgdas'
-        letkf_yaml = parse_j2yaml(self.task_config.MARINE_LETKF_YAML_TMPL, letkfconf)
+        letkf_yaml = parse_j2yaml(self.task_config.MARINE_LETKF_YAML_TMPL, stageconf)
         letkf_yaml.observations.observers = obs_to_use
         letkf_yaml.save(self.task_config.letkf_yaml_file)
 
@@ -133,6 +148,32 @@ def run(self):
 
         logger.info("run")
 
+        exec_cmd_gridgen = Executable(self.task_config.APRUN_MARINEANLLETKF)
+        exec_cmd_gridgen.add_default_arg(self.task_config.GRIDGEN_EXEC)
+        exec_cmd_gridgen.add_default_arg(self.task_config.GRIDGEN_YAML)
+
+        try:
+            logger.debug(f"Executing {exec_cmd_gridgen}")
+            exec_cmd_gridgen()
+        except OSError:
+            raise OSError(f"Failed to execute {exec_cmd_gridgen}")
+        except Exception:
+            raise WorkflowException(f"An error occured during execution of {exec_cmd_gridgen}")
+        pass
+
+        exec_cmd_letkf = Executable(self.task_config.APRUN_MARINEANLLETKF)
+        for letkf_exec_arg in self.task_config.letkf_exec_args:
+            exec_cmd_letkf.add_default_arg(letkf_exec_arg)
+
+        try:
+            logger.debug(f"Executing {exec_cmd_letkf}")
+            exec_cmd_letkf()
+        except OSError:
+            raise OSError(f"Failed to execute {exec_cmd_letkf}")
+        except Exception:
+            raise WorkflowException(f"An error occured during execution of {exec_cmd_letkf}")
+        pass
+
     @logit(logger)
     def finalize(self):
         """Method finalize for ocean and sea ice LETKF task
@@ -145,3 +186,12 @@ def finalize(self):
         """
 
         logger.info("finalize")
+
+        letkfsaveconf = AttrDict()
+        keys = ['current_cycle', 'DATA', 'NMEM_ENS', 'WINDOW_BEGIN',
+                'PARMgfs', 'ROTDIR', 'COM_OCEAN_LETKF_TMPL', 'COM_ICE_LETKF_TMPL']
+        for key in keys:
+            letkfsaveconf[key] = self.task_config[key]
+        letkfsaveconf.RUN = 'enkfgdas'
+        letkf_save_list = parse_j2yaml(self.task_config.MARINE_LETKF_SAVE_YAML_TMPL, letkfsaveconf)
+        FileHandler(letkf_save_list).sync()

ush/python/pygfs/utils/marine_da_utils.py

@@ -1,6 +1,7 @@
 import f90nml
-import os
 from logging import getLogger
+import os
+import re
 import xarray as xr
 
 from wxflow import (FileHandler,

workflow/applications/gfs_cycled.py

@@ -46,7 +46,7 @@ def _get_app_configs(self):
         if self.do_jediocnvar:
             configs += ['prepoceanobs', 'ocnanalprep', 'marinebmat', 'ocnanalrun']
             if self.do_hybvar:
-                configs += ['ocnanalecen']
+                configs += ['marineanlletkf', 'ocnanalecen']
             configs += ['ocnanalchkpt', 'ocnanalpost']
             if self.do_vrfy_oceanda:
                 configs += ['ocnanalvrfy']
@@ -148,7 +148,7 @@ def get_task_names(self):
         if self.do_jediocnvar:
             gdas_gfs_common_tasks_before_fcst += ['prepoceanobs', 'ocnanalprep', 'marinebmat', 'ocnanalrun']
             if self.do_hybvar:
-                gdas_gfs_common_tasks_before_fcst += ['ocnanalecen']
+                gdas_gfs_common_tasks_before_fcst += ['marineanlletkf', 'ocnanalecen']
             gdas_gfs_common_tasks_before_fcst += ['ocnanalchkpt', 'ocnanalpost']
             if self.do_vrfy_oceanda:
                 gdas_gfs_common_tasks_before_fcst += ['ocnanalvrfy']