Stat Analysis Implementation

jobs/JGLOBAL_ANALYSIS_STATS

@@ -6,12 +6,18 @@
 ##############################################
 # Set variables used in the script
 ##############################################
-
+# 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"
 
 ##############################################
 # Begin JOB SPECIFIC work
 ##############################################
 
+# Generate COM variables from templates
+YMD=${PDY} HH=${cyc} declare_from_tmpl -rx COM_OBS COM_CHEM_ANALYSIS
 
 ###############################################################
 # Run relevant script

parm/config/gfs/config.anlstat

@@ -0,0 +1,13 @@
+#!/bin/bash -x
+
+########## config.anlstat ##########
+# Analysis Stat
+
+echo "BEGIN: config.anlstat"
+
+# Get task specific resources
+source "${EXPDIR}/config.resources" anlstat
+
+export JEDIEXE=${HOMEgfs}/sorc/gdas.cd/build/bin/ioda-stats.x
+
+echo "END: config.anlstat"

parm/config/gfs/config.resources

@@ -17,7 +17,7 @@ if (( $# != 1 )); then
     echo "atmensanlinit atmensanlobs atmensanlsol atmensanlletkf atmensanlfv3inc atmensanlfinal"
     echo "snowanl esnowrecen"
     echo "prepobsaero aeroanlinit aeroanlvar aeroanlfinal aeroanlgenb"
-    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"
@@ -707,6 +707,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

@@ -1,11 +1,14 @@
 #!/usr/bin/env python3
-# exglobal_analysis_stats.py
-# This script will run the OOPS/JEDI code to
-
+# exglobal_stat_analysis.py
+# This script creates an StatAnalysis class
+# and runs the initialize method
+# which create and stage the runtime directory
+# and create the YAML configuration
+# for a global stat analysis
 import os
 
 from wxflow import Logger, cast_strdict_as_dtypedict
-from pygfs.task.atm_analysis import AtmAnalysis
+from pygfs.task.stat_analysis import StatAnalysis
 
 # Initialize root logger
 logger = Logger(level='DEBUG', colored_log=True)
@@ -16,5 +19,9 @@
     # Take configuration from environment and cast it as python dictionary
     config = cast_strdict_as_dtypedict(os.environ)
 
-    # placeholder
-    print('Do nothing right now!')
+    # Instantiate the atm analysis task
+    StatAnl = StatAnalysis(config, 'statanlvar')
+
+    # Initialize JEDI variational analysis
+    StatAnl.initialize_jedi()
+    StatAnl.initialize_analysis()

sorc/link_workflow.sh

@@ -374,6 +374,7 @@ if [[ -d "${HOMEgfs}/sorc/gdas.cd/build" ]]; then
                        "gdas_socahybridweights.x" \
                        "gdasapp_land_ensrecenter.x" \
                        "bufr2ioda.x" \
+                       "ioda-stats.x" \
                        "calcfIMS.exe" \
                        "apply_incr.exe" )
   for gdasexe in "${JEDI_EXE[@]}"; do

ush/python/pygfs/task/stat_analysis.py

@@ -0,0 +1,133 @@
+#!/usr/bin/env python3
+
+import os
+import glob
+import gzip
+import tarfile
+from logging import getLogger
+from pprint import pformat
+from typing import Optional, Dict, Any
+
+from wxflow import (AttrDict,
+                    FileHandler,
+                    add_to_datetime, to_fv3time, to_timedelta, to_YMDH,
+                    Task,
+                    parse_j2yaml, save_as_yaml,
+                    logit)
+from pygfs.jedi import Jedi
+
+logger = getLogger(__name__.split('.')[-1])
+
+
+class StatAnalysis(Task):
+    """
+    Class for JEDI-based global stat analysis tasks
+    """
+    @logit(logger, name="StatAnalysis")
+    def __init__(self, config: Dict[str, Any], yaml_name: Optional[str] = None):
+        """
+        Constructor global stat analysis task
+
+        This method will construct a global stat analysis task.
+        This includes:
+        - extending the task_config attribute AttrDict to include parameters required for this task
+        - instantiate the Jedi attribute object
+
+        Parameters
+        ----------
+        config: Dict
+            dictionary object containing task configuration
+        yaml_name: str, optional
+            name of YAML file for JEDI configuration
+
+        Returns
+        ----------
+        None
+        """
+        super().__init__(config)
+
+        _res = int(self.task_config.CASE[1:])
+        # _res_anl = int(self.task_config.CASE_ANL[1:])
+        _window_begin = add_to_datetime(self.task_config.current_cycle, -to_timedelta(f"{self.task_config.assim_freq}H") / 2)
+        print(_window_begin)
+
+        # Create a local dictionary that is repeatedly used across this class
+        local_dict = AttrDict(
+            {
+                'npx_ges': _res + 1,
+                'npy_ges': _res + 1,
+                'npz_ges': self.task_config.LEVS - 1,
+                'npz': self.task_config.LEVS - 1,
+                #'npx_anl': _res_anl + 1,
+                #'npy_anl': _res_anl + 1,
+                'npz_anl': self.task_config.LEVS - 1,
+                'ATM_WINDOW_BEGIN': _window_begin,
+                'ATM_WINDOW_LENGTH': f"PT{self.task_config.assim_freq}H",
+                'OPREFIX': f"{self.task_config.RUN}.t{self.task_config.cyc:02d}z.",
+                'APREFIX': f"{self.task_config.RUN}.t{self.task_config.cyc:02d}z.",
+                'GPREFIX': f"gdas.t{self.task_config.previous_cycle.hour:02d}z."
+            }
+        )
+
+        # Extend task_config with local_dict
+        self.task_config = AttrDict(**self.task_config, **local_dict)
+
+        # Create JEDI object
+        self.jedi = Jedi(self.task_config, yaml_name)
+
+    @logit(logger)
+    def initialize_jedi(self):
+        """Initialize JEDI application
+
+        This method will initialize a JEDI application used in the global stat analysis.
+        This includes:
+        - generating and saving JEDI YAML config
+        - linking the JEDI executable
+
+        Parameters
+        ----------
+        None
+
+        Returns
+        ----------
+        None
+        """
+
+        # get JEDI-to-FV3 increment converter config and save to YAML file
+        logger.info(f"Generating JEDI YAML config: {self.jedi.yaml}")
+        self.jedi.set_config(self.task_config)
+        logger.debug(f"JEDI config:\n{pformat(self.jedi.config)}")
+
+        # save JEDI config to YAML file
+        logger.debug(f"Writing JEDI YAML config to: {self.jedi.yaml}")
+        save_as_yaml(self.jedi.config, self.jedi.yaml)
+
+        # link JEDI executable
+        logger.info(f"Linking JEDI executable {self.task_config.JEDIEXE} to {self.jedi.exe}")
+        self.jedi.link_exe(self.task_config)
+
+    @logit(logger)
+    def initialize_analysis(self) -> None:
+        """
+        Initialize a global stat analysis
+
+        This method will initialize a global stat analysis.
+        This includes:
+        - copying stat files
+
+        Parameters
+        ----------
+        None
+
+        Returns
+        ----------
+        None
+        """
+        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

@@ -55,6 +55,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', True)
         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)