Stat Analysis Implementation

jobs/JGLOBAL_ANALYSIS_STATS

@@ -0,0 +1,36 @@
+#! /usr/bin/env bash
+
+source "${HOMEgfs}/ush/preamble.sh"
+export DATA=${DATA:-${DATAROOT}/${RUN}statanl_${cyc}}
+source "${HOMEgfs}/ush/jjob_header.sh" -e "anlstat" -c "base anlstat"
+
+##############################################
+# Set variables used in the script
+##############################################
+
+
+##############################################
+# Begin JOB SPECIFIC work
+##############################################
+
+
+###############################################################
+# Run relevant script
+
+EXSCRIPT=${ANLSTATSPY:-${SCRgfs}/exglobal_analysis_stats.py}
+${EXSCRIPT}
+status=$?
+[[ ${status} -ne 0 ]] && exit "${status}"
+
+##############################################
+# End JOB SPECIFIC work
+##############################################
+
+##############################################
+# Final processing
+##############################################
+if [[ -e "${pgmout}" ]] ; then
+  cat "${pgmout}"
+fi
+
+exit 0

jobs/rocoto/anlstat.sh

@@ -0,0 +1,18 @@
+#! /usr/bin/env bash
+
+source "${HOMEgfs}/ush/preamble.sh"
+
+###############################################################
+# Source UFSDA workflow modules
+. "${HOMEgfs}/ush/load_ufsda_modules.sh"
+status=$?
+[[ ${status} -ne 0 ]] && exit "${status}"
+
+export job="anlstat"
+export jobid="${job}.$$"
+
+###############################################################
+# Execute the JJOB
+"${HOMEgfs}/jobs/JGLOBAL_ANALYSIS_STATS"
+status=$?
+exit "${status}"

parm/config/gfs/config.analstat

@@ -0,0 +1,11 @@
+#!/bin/bash -x
+
+########## config.anlstat ##########
+# Analysis Stat
+
+echo "BEGIN: config.anlstat"
+
+# Get task specific resources
+source "${EXPDIR}/config.resources" anlstat
+
+echo "END: config.anlstat"

parm/config/gfs/config.anlstat

@@ -0,0 +1,11 @@
+#!/bin/bash -x
+
+########## config.anlstat ##########
+# Analysis Stat
+
+echo "BEGIN: config.anlstat"
+
+# Get task specific resources
+source "${EXPDIR}/config.resources" anlstat
+
+echo "END: config.anlstat"

parm/config/gfs/config.base

@@ -73,9 +73,6 @@ export DO_NPOESS="@DO_NPOESS@"           # NPOESS products
 export DO_TRACKER="@DO_TRACKER@"         # Hurricane track verification
 export DO_GENESIS="@DO_GENESIS@"         # Cyclone genesis verification
 export DO_GENESIS_FSU="@DO_GENESIS_FSU@" # Cyclone genesis verification (FSU)
-export DO_VERFOZN="YES"                  # Ozone data assimilation monitoring
-export DO_VERFRAD="YES"                  # Radiance data assimilation monitoring
-export DO_VMINMON="YES"                  # GSI minimization monitoring
 export DO_MOS="NO"                       # GFS Model Output Statistics - Only supported on WCOSS2
 
 # NO for retrospective parallel; YES for real-time parallel
@@ -474,6 +471,14 @@ 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
+   export DO_ANLSTAT="YES"    # JEDI-based analysis statistics
+else
+   export DO_VERFOZN="YES"    # Ozone data assimilation monitoring
+   export DO_VERFRAD="YES"    # Radiance data assimilation monitoring
+   export DO_VMINMON="YES"    # GSI minimization monitoring
+   if [[ ${DO_AERO} = "YES" || ${DO_JEDIOCNVAR} = "YES" || ${DO_JEDISNOWDA} = "YES " ]]; then
+     export DO_ANLSTAT="YES"  # JEDI-based analysis statistics
+   fi
 fi
 
 # If starting ICs that are not at cycle hour

parm/config/gfs/config.resources

@@ -17,7 +17,7 @@ if (( $# != 1 )); then
     echo "atmensanlinit atmensanlletkf atmensanlfv3inc atmensanlfinal"
     echo "snowanl"
     echo "prepobsaero aeroanlinit aeroanlrun aeroanlfinal"
-    echo "anal sfcanl analcalc analdiag fcst echgres"
+    echo "anal sfcanl analcalc analdiag anlstat fcst echgres"
     echo "upp atmos_products"
     echo "tracker genesis genesis_fsu"
     echo "verfozn verfrad vminmon fit2obs metp arch cleanup"
@@ -626,6 +626,14 @@ case ${step} in
     memory="48GB"
     ;;
 
+  "anlstat")
+    walltime="00:30:00"
+    ntasks=1
+    threads_per_task=1
+    tasks_per_node=$(( max_tasks_per_node / threads_per_task ))
+    memory="24GB"
+    ;;
+
   "sfcanl")
     walltime="00:20:00"
     ntasks=${ntiles:-6}

scripts/exglobal_analysis_stats.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python3
+# exglobal_analysis_stats.py
+# This script will run the OOPS/JEDI code to
+
+import os
+
+from wxflow import Logger, cast_strdict_as_dtypedict
+from pygfs.task.stat_analysis import StatAnalysis
+
+# Initialize root logger
+logger = Logger(level='DEBUG', colored_log=True)
+
+
+if __name__ == '__main__':
+
+    # Take configuration from environment and cast it as python dictionary
+    config = cast_strdict_as_dtypedict(os.environ)
+
+    # Call StatAnalysis
+    anl = StatAnalysis(config)
+    anl.initialize()

sorc/link_workflow.sh

@@ -252,7 +252,7 @@ fi
 #------------------------------
 if [[ -d "${HOMEgfs}/sorc/gdas.cd" ]]; then
   cd "${HOMEgfs}/parm/gdas" || exit 1
-  declare -a gdasapp_comps=("aero" "atm" "io" "ioda" "snow" "soca" "jcb-gdas" "jcb-algorithms")
+  declare -a gdasapp_comps=("aero" "atm" "io" "ioda" "snow" "soca" "stats" "jcb-gdas" "jcb-algorithms")
   for comp in "${gdasapp_comps[@]}"; do
     [[ -d "${comp}" ]] && rm -rf "${comp}"
     ${LINK_OR_COPY} "${HOMEgfs}/sorc/gdas.cd/parm/${comp}" .

ush/python/pygfs/task/stat_analysis.py

@@ -0,0 +1,60 @@
+#!/usr/bin/env python3
+
+import os
+from logging import getLogger
+from typing import Dict, List
+from pprint import pformat
+import numpy as np
+from netCDF4 import Dataset
+
+from wxflow import (AttrDict,
+                    FileHandler,
+                    to_fv3time, to_YMD, to_YMDH, to_timedelta, add_to_datetime,
+                    rm_p,
+                    parse_j2yaml, save_as_yaml,
+                    Jinja,
+                    logit,
+                    Executable,
+                    WorkflowException)
+from pygfs.task.analysis import Analysis
+
+logger = getLogger(__name__.split('.')[-1])
+
+
+class StatAnalysis(Analysis):
+    """
+    Class for global stat analysis tasks
+    """
+
+    @logit(logger, name="StatAnalysis")
+    def __init__(self, config):
+        super().__init__(config)
+
+        _res = int(self.task_config['CASE'][1:])
+        _window_begin = add_to_datetime(self.task_config.current_cycle, -to_timedelta(f"{self.task_config['assim_freq']}H") / 2)
+        _letkfoi_yaml = os.path.join(self.task_config.DATA, f"{self.task_config.RUN}.t{self.task_config['cyc']:02d}z.letkfoi.yaml")
+
+        # Create a local dictionary that is repeatedly used across this class
+        local_dict = AttrDict(
+            {
+                'APREFIX': f"{self.task_config.RUN}.t{self.task_config.cyc:02d}z.",
+                'jedi_yaml': _letkfoi_yaml
+            }
+        )
+
+        # Extend task_config with local_dict
+        self.task_config = AttrDict(**self.task_config, **local_dict)
+
+    @logit(logger)
+    def initialize(self: Analysis) -> None:
+        """
+        Initialize global stat analysis.
+        """
+        super().initialize()
+
+        logger.info(f"Copying files to {self.task_config.DATA}/stats")
+
+        aerostat = os.path.join(self.task_config.COM_CHEM_ANALYSIS, f"{self.task_config['APREFIX']}aerostat")
+        dest = os.path.join(self.task_config.DATA, "stats")
+        statlist = [aerostat, dest]
+        FileHandler({'copy': statlist}).sync()

workflow/applications/applications.py

@@ -63,6 +63,7 @@ def __init__(self, conf: Configuration) -> None:
         self.do_verfozn = _base.get('DO_VERFOZN', True)
         self.do_verfrad = _base.get('DO_VERFRAD', True)
         self.do_vminmon = _base.get('DO_VMINMON', True)
+        self.do_anlstat = _base.get('DO_ANLSTAT', False)
         self.do_tracker = _base.get('DO_TRACKER', True)
         self.do_genesis = _base.get('DO_GENESIS', True)
         self.do_genesis_fsu = _base.get('DO_GENESIS_FSU', False)

workflow/applications/gfs_cycled.py

@@ -74,6 +74,9 @@ def _get_app_configs(self):
         if self.do_vminmon:
             configs += ['vminmon']
 
+        if self.do_anlstat:
+            configs += ['anlstat']
+
         if self.do_tracker:
             configs += ['tracker']
 
@@ -206,6 +209,9 @@ def get_task_names(self):
         if self.do_vminmon:
             gdas_tasks += ['vminmon']
 
+        if self.do_anlstat:
+            gdas_tasks += ['anlstat']
+
         if self.do_gempak:
             gdas_tasks += ['gempak', 'gempakmetancdc']
 
@@ -240,6 +246,9 @@ def get_task_names(self):
         if self.do_vminmon:
             gfs_tasks += ['vminmon']
 
+        if self.do_anlstat:
+            gfs_tasks += ['anlstat']
+
         if self.do_tracker:
             gfs_tasks += ['tracker']
 

workflow/rocoto/gfs_tasks.py

@@ -1731,6 +1731,40 @@ def vminmon(self):
 
         return task
 
+    def anlstat(self):
+        deps = []
+        if self.app_config.do_jediatmvar:
+            dep_dict = {'type': 'task', 'name': f'{self.run}atmanlfinal'}
+            deps.append(rocoto.add_dependency(dep_dict))
+        if self.do_jediocnvar:
+            dep_dict = {'type': 'task', 'name': f'{self.run}ocnanalpost'}
+            deps.append(rocoto.add_dependency(dep_dict))
+        if self.do_jedisnowda:
+            dep_dict = {'type': 'task', 'name': f'{self.run}snowanl'}
+            deps.append(rocoto.add_dependency(dep_dict))
+        if self.do_aero:
+            dep_dict = {'type': 'task', 'name': f'{self.run}aeroanlfinal'}
+            deps.append(rocoto.add_dependency(dep_dict))
+
+        dependencies = rocoto.create_dependency(dep_condition='and', dep=deps)
+
+        resources = self.get_resource('anlstat')
+        task_name = f'{self.run}anlstat'
+        task_dict = {'task_name': task_name,
+                     'resources': resources,
+                     'dependency': dependencies,
+                     'envars': self.envars,
+                     'cycledef': self.run.replace('enkf', ''),
+                     'command': f'{self.HOMEgfs}/jobs/rocoto/anlstat.sh',
+                     'job_name': f'{self.pslot}_{task_name}_@H',
+                     'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log',
+                     'maxtries': '&MAXTRIES;'
+                     }
+
+        task = rocoto.create_task(task_dict)
+
+        return task
+
     def tracker(self):
         deps = []
         dep_dict = {'type': 'metatask', 'name': f'{self.run}atmos_prod'}
@@ -2225,6 +2259,9 @@ def arch(self):
                 if self.app_config.do_vminmon:
                     dep_dict = {'type': 'task', 'name': f'{self.run}vminmon'}
                     deps.append(rocoto.add_dependency(dep_dict))
+                if self.app_config.do_anlstat:
+                    dep_dict = {'type': 'task', 'name': f'{self.run}anlstat'}
+                    deps.append(rocoto.add_dependency(dep_dict))
             elif self.run in ['gdas']:
                 dep_dict = {'type': 'task', 'name': f'{self.run}atmanlprod'}
                 deps.append(rocoto.add_dependency(dep_dict))
@@ -2240,6 +2277,9 @@ def arch(self):
                 if self.app_config.do_vminmon:
                     dep_dict = {'type': 'task', 'name': f'{self.run}vminmon'}
                     deps.append(rocoto.add_dependency(dep_dict))
+                if self.app_config.do_anlstat:
+                    dep_dict = {'type': 'task', 'name': f'{self.run}anlstat'}
+                    deps.append(rocoto.add_dependency(dep_dict))
         if self.run in ['gfs'] and self.app_config.do_tracker:
             dep_dict = {'type': 'task', 'name': f'{self.run}tracker'}
             deps.append(rocoto.add_dependency(dep_dict))