Refactor Local Archiving: Move YAML Logic to archive_vars.py and Handle Ensemble/Deterministic Archiving

dev/jobs/JGLOBAL_ARCHIVE_VRFY

@@ -20,6 +20,12 @@ for grid in "0p25" "0p50" "1p00"; do
         "COMIN_ATMOS_GRIB_${grid}:COM_ATMOS_GRIB_GRID_TMPL"
 done
 
+# GEFS-specific: Ensemble statistics path
+if [[ "${RUN}" == "gefs" ]]; then
+    MEMDIR="ensstat" YMD=${PDY} HH=${cyc} GRID="1p00" declare_from_tmpl -rx \
+        COMIN_ATMOS_ENSSTAT_1p00:COM_ATMOS_GRIB_GRID_TMPL
+fi
+
 ###############################################################
 # Run archive script
 ###############################################################

dev/scripts/exglobal_archive_vrfy.py

@@ -3,6 +3,7 @@
 import os
 
 from pygfs.task.archive import Archive
+from pygfs.utils.archive_vars import ArchiveVrfyVars
 from wxflow import AttrDict, Logger, cast_strdict_as_dtypedict, logit, chdir
 
 # initialize root logger
@@ -14,38 +15,17 @@ def main():
 
     config = cast_strdict_as_dtypedict(os.environ)
 
-    # Instantiate the Archive object
+    # Instantiate the Archive task object
     archive = Archive(config)
 
-    # update these keys to be 3 digits if they are part of archive.task_config.keys
-    for key in ['OCNRES', 'ICERES']:
-        try:
-            archive.task_config[key] = f"{archive.task_config[key]:03d}"
-        except KeyError as ee:
-            logger.info(f"key ({key}) not found in archive.task_config!")
-
-    # Pull out all the configuration keys needed to run the rest of archive steps
-    keys = ['current_cycle', 'RUN', 'PSLOT', 'ROTDIR', 'PARMgfs',
-            'ARCDIR', 'MODE', 'DO_JEDIATMENS', 'DO_FIT2OBS', 'DO_JEDIATMVAR', 'FHMIN_GFS',
-            'DO_JEDISNOWDA', 'DO_AERO_ANL', 'DO_PREP_OBS_AERO', 'NET', 'MODE', 'FHOUT_GFS',
-            'FHMAX_HF_GFS', 'FHOUT_GFS', 'FHMAX_FITS', 'FHMAX', 'FHOUT', 'FHMAX_GFS', 'DO_GSISOILDA', 'DO_LAND_IAU']
-
-    archive_dict = AttrDict()
-    for key in keys:
-        try:
-            archive_dict[key] = archive.task_config[key]
-        except KeyError as ee:
-            logger.warning(f"WARNING: key ({key}) not found in archive.task_config!")
-
-    # Also import all COMIN* and COMOUT* directory and template variables
-    for key in archive.task_config.keys():
-        if key.startswith(("COM_", "COMIN_", "COMOUT_")):
-            archive_dict[key] = archive.task_config.get(key)
+    # Collect all archive variables in complete arch_dict for YAML templates
+    # Use static utility methods from ArchiveVrfyVars
+    arch_dict = ArchiveVrfyVars.get_all_yaml_vars(archive.task_config)
 
-    with chdir(config.ROTDIR):
+    # Pass arch_dict to configure_vrfy which will render the Jinja2 YAML
+    arcdir_set = archive.configure_vrfy(arch_dict)
 
-        # Determine which archives to create
-        arcdir_set = archive.configure_vrfy(archive_dict)
+    with chdir(config.ROTDIR):
 
         # Populate the product archive (ARCDIR)
         archive.execute_store_products(arcdir_set)

dev/scripts/exglobal_enkf_earc_vrfy.py

@@ -3,6 +3,7 @@
 import os
 
 from pygfs.task.archive import Archive
+from pygfs.utils.archive_vars import ArchiveVrfyVars
 from wxflow import AttrDict, Logger, cast_strdict_as_dtypedict, chdir, logit
 
 # initialize root logger
@@ -14,37 +15,20 @@ def main():
 
     config = cast_strdict_as_dtypedict(os.environ)
 
-    # Instantiate the Archive object
+    # Instantiate the Archive task object
     archive = Archive(config)
 
-    # Pull out all the configuration keys needed to run the rest of archive steps
-    keys = ['current_cycle', 'RUN', 'PSLOT', 'ROTDIR', 'PARMgfs',
-            'ARCDIR', 'MODE', 'DO_JEDIATMENS', 'DO_FIT2OBS', 'DO_JEDIATMVAR',
-            'DO_JEDISNOWDA', 'DO_AERO_ANL', 'DO_PREP_OBS_AERO', 'NET', 'MODE', 'FHOUT_GFS',
-            'FHMAX_HF_GFS', 'FHOUT_GFS', 'FHMAX_FITS', 'FHMAX', 'FHOUT', 'FHMAX_GFS', 'DO_GSISOILDA', 'DO_LAND_IAU']
+    # Collect all archive variables in complete arch_dict for YAML templates
+    # Use static utility methods from ArchiveVrfyVars
+    arch_dict = ArchiveVrfyVars.get_all_yaml_vars(archive.task_config)
 
-    archive_dict = AttrDict()
-    for key in keys:
-        archive_dict[key] = archive.task_config.get(key)
-        if archive_dict[key] is None:
-            print(f"Warning: key ({key}) not found in task_config!")
+    # Pass arch_dict to configure_vrfy which will render the Jinja2 YAML
+    arcdir_set = archive.configure_vrfy(arch_dict)
 
-    # Also import all COMIN* directory and template variables
-    for key in archive.task_config.keys():
-        if key.startswith("COMIN"):
-            archive_dict[key] = archive.task_config[key]
+    with chdir(config.ROTDIR):
 
-    cwd = os.getcwd()
-
-    os.chdir(config.ROTDIR)
-
-    # Determine which archives to create
-    arcdir_set = archive.configure_vrfy(archive_dict)
-
-    # Populate the product archive (ARCDIR)
-    archive.execute_store_products(arcdir_set)
-
-    os.chdir(cwd)
+        # Populate the product archive (ARCDIR)
+        archive.execute_store_products(arcdir_set)
 
 
 if __name__ == '__main__':

parm/archive/enkf_arcdir.yaml.j2

@@ -0,0 +1,25 @@
+# Variables provided by archive_vars.py:
+# - cycle_HH, cycle_YMDH, cycle_YMD, head
+# - ARCDIR
+# - All COMIN_* paths
+
+# Ensemble (EnKF) archiving template
+# Used for: enkfgdas, enkfgfs, enkfgcafs, enkfgcdas
+
+# Build head string (e.g., 'gfs.t00z.')
+{% set head = RUN + ".t" + cycle_HH + "z." %}
+
+mkdir:
+    - "{{ ARCDIR }}"
+
+copy_opt:
+    # Ensemble analysis statistics
+    {% if DO_JEDIATMENS == True %}
+    - ["{{ COMIN_ATMOS_ANALYSIS_ENSSTAT }}/{{ head }}stat.atm.tar",
+       "{{ ARCDIR }}/atmensstat.{{ RUN }}.{{ cycle_YMDH }}"]
+    {% else %}
+    - ["{{ COMIN_ATMOS_ANALYSIS_ENSSTAT }}/{{ head }}enkfstat.txt",
+       "{{ ARCDIR }}/enkfstat.{{ RUN }}.{{ cycle_YMDH }}"]
+    - ["{{ COMIN_ATMOS_ANALYSIS_ENSSTAT }}/{{ head }}gsistat_ensmean.txt",
+       "{{ ARCDIR }}/gsistat.{{ RUN }}.{{ cycle_YMDH }}.ensmean"]
+    {% endif %}

parm/archive/gcafs_arcdir.yaml.j2

@@ -1,107 +1,58 @@
-{% set cycle_HH = current_cycle | strftime("%H") %}
-{% set cycle_YMDH = current_cycle | to_YMDH %}
-{% set cycle_YMD = current_cycle | to_YMD %}
-{% set head = RUN + ".t" + cycle_HH + "z." %}
-
-# Select data to store in the ARCDIR and VFYARC from deterministic runs
-# This file set will contain all source-destination pairs to send to the FileHandler for copying
-{% set file_set = [] %}
-
-# Declare the VFYARC where Fit2Obs data will be sent
-{% set VFYARC = ROTDIR ~ "/vrfyarch" %}
+# Variables provided by archive_vars.py:
+# - cycle_HH, cycle_YMDH, cycle_YMD, head
+# - VFYARC
+# - All COMIN_* paths
 
-# Deterministic files
-{% if "enkf" not in RUN %}
-    # Common files to be added to both the gcafs and gcdas keys below
-    {% set det_files = [] %}
+# Deterministic GCAFS/GCDAS archiving template
+# Used for: gcafs, gcdas
 
-    # Deterministic analysis files (generated for cycled experiments)
-    {% set det_anl_files = [] %}
-
-    {% if DO_AERO_ANL %}
-        {% do det_anl_files.append([COMIN_CHEM_ANALYSIS ~ "/" ~ head ~ "aerostat.tgz",
-                                    ARCDIR ~ "/aerostat." ~ RUN ~ "." ~ cycle_YMDH ~ ".tgz"]) %}
-    {% endif %}
+# Build head string (e.g., 'gfs.t00z.')
+{% set head = RUN + ".t" + cycle_HH + "z." %}
 
-    {% if DO_PREP_OBS_AERO == True %}
-        {% do det_anl_files.append([COMIN_OBS ~ "/" ~ head ~ "aeroobs",
-                                    ARCDIR ~ "/aeroobs." ~ RUN ~ "." ~ cycle_YMDH]) %}
-        {% do det_anl_files.append([COMIN_OBS ~ "/" ~ head ~ "aeroawobs",
-                                    ARCDIR ~ "/aeroawobs." ~ RUN ~ "." ~ cycle_YMDH]) %}
-    {% endif %}
+mkdir:
+    - "{{ ARCDIR }}"
+{% if DO_FIT2OBS == True %}
+    - "{{ VFYARC }}/{{ RUN }}.{{ cycle_YMD }}/{{ cycle_HH }}"
+{% endif %}
 
-    # GCAFS-specific files
-    {% set gfs_files = [] %}
+copy_req:
+{% if RUN == "gcafs" %}
+    # GCAFS forecast files - REQUIRED
     {% for fhr in range(0, FHMAX_GFS + 1, FHOUT_GFS) %}
-        {% do gfs_files.append([COMIN_ATMOS_GRIB_1p00 ~ "/" ~ head ~ "pres_a.1p00.f" ~ '%03d'|format(fhr) ~ ".grib2",
-                                ARCDIR ~ "/pgbf" ~ '%02d'|format(fhr) ~ "." ~ RUN ~ "." ~ cycle_YMDH ~ ".grib2"]) %}
+    - ["{{ COMIN_ATMOS_GRIB_1p00 }}/{{ head }}pres_a.1p00.f{{ '%03d'|format(fhr) }}.grib2",
+       "{{ ARCDIR }}/pgbf{{ '%02d'|format(fhr) }}.{{ RUN }}.{{ cycle_YMDH }}.grib2"]
     {% endfor %}
 
-    # GCAFS Fit2Obs data
-    {% set fit2obs_files = [] %}
-    {% for fhr in range(0, FHMAX_FITS + 1, 6) %}
-        {% set sfcfile = "/" + head + "sfc.f" + '%03d'|format(fhr) + ".nc" %}
-        {% set sigfile = "/" + head + "atm.f" + '%03d'|format(fhr) + ".nc" %}
-        {% do fit2obs_files.append([COMIN_ATMOS_HISTORY ~ "/" ~ sfcfile,
-                                    VFYARC ~ "/" ~ RUN ~ "." ~ cycle_YMD ~ "/" ~ cycle_HH ~ "/" ~ sfcfile ]) %}
-        {% do fit2obs_files.append([COMIN_ATMOS_HISTORY ~ "/" ~ sigfile,
-                                    VFYARC ~ "/" ~ RUN ~ "." ~ cycle_YMD ~ "/" ~ cycle_HH ~ "/" ~ sigfile ]) %}
-    {% endfor %}
+    {% if DO_FIT2OBS == True %}
+    # GCAFS Fit2Obs data - REQUIRED if DO_FIT2OBS is enabled
+        {% for fhr in range(0, FHMAX_FITS + 1, 6) %}
+    - ["{{ COMIN_ATMOS_HISTORY }}/{{ head }}sfc.f{{ '%03d'|format(fhr) }}.nc",
+       "{{ VFYARC }}/{{ RUN }}.{{ cycle_YMD }}/{{ cycle_HH }}/{{ head }}sfc.f{{ '%03d'|format(fhr) }}.nc"]
+    - ["{{ COMIN_ATMOS_HISTORY }}/{{ head }}atm.f{{ '%03d'|format(fhr) }}.nc",
+       "{{ VFYARC }}/{{ RUN }}.{{ cycle_YMD }}/{{ cycle_HH }}/{{ head }}atm.f{{ '%03d'|format(fhr) }}.nc"]
+        {% endfor %}
+    {% endif %}
+{% endif %}
 
-    # GCDAS-specific files
-    {% set gdas_files = [] %}
+{% if RUN == "gcdas" %}
+    # GCDAS forecast files - REQUIRED
     {% for fhr in range(0, FHMAX + 1, FHOUT) %}
-        {% do gdas_files.append([COMIN_ATMOS_GRIB_1p00 ~ "/" ~ head ~ "pres_a.1p00.f" ~ '%03d'|format(fhr) ~ ".grib2",
-                                 ARCDIR ~ "/pgbf" ~ '%02d'|format(fhr) ~ "." ~ RUN ~ "." ~ cycle_YMDH ~ ".grib2"]) %}
+    - ["{{ COMIN_ATMOS_GRIB_1p00 }}/{{ head }}pres_a.1p00.f{{ '%03d'|format(fhr) }}.grib2",
+       "{{ ARCDIR }}/pgbf{{ '%02d'|format(fhr) }}.{{ RUN }}.{{ cycle_YMDH }}.grib2"]
     {% endfor %}
+{% endif %}
 
-    # Now append the necessary file pairs to file_set
-    # Common deterministic files
-    {% set file_set = file_set + det_files %}
-    {% if MODE == "cycled" %}
-        {% set file_set = file_set + det_anl_files %}
-    {% endif %}
-
-    # Run-specific deterministic files
-    {% if RUN == "gcafs" %}
-        {% set file_set = file_set + gfs_files %}
-        # Fit2Obs files
-        {% if DO_FIT2OBS == True %}
-            {% set file_set = file_set + fit2obs_files %}
-        {% endif %}
-    {% elif RUN == "gcdas" %}
-        {% set file_set = file_set + gdas_files %}
+{% if MODE == "cycled" %}
+    # Deterministic analysis files (cycled mode only) - REQUIRED
+    {% if DO_AERO_ANL %}
+    - ["{{ COMIN_CHEM_ANALYSIS }}/{{ head }}aerostat.tgz",
+       "{{ ARCDIR }}/aerostat.{{ RUN }}.{{ cycle_YMDH }}.tgz"]
     {% endif %}
 
-{% else %}  # End of deterministic files
-
-    # Ensemble analysis files
-    {% set enkf_files = [] %}
-    {% if DO_JEDIATMENS == True %}
-        {% do enkf_files.append([COMIN_ATMOS_ANALYSIS_ENSSTAT ~ "/" ~ head ~ "atmensstat",
-                                 ARCDIR ~ "/atmensstat." ~ RUN ~ "." ~ cycle_YMDH ]) %}
-    {% else %}
-        {% do enkf_files.append([COMIN_ATMOS_ANALYSIS_ENSSTAT ~ "/" ~ head ~ "enkfstat.txt",
-                                 ARCDIR ~ "/enkfstat." ~ RUN ~ "." ~ cycle_YMDH ]) %}
-        {% do enkf_files.append([COMIN_ATMOS_ANALYSIS_ENSSTAT ~ "/" ~ head ~ "gsistat_ensmean.txt",
-                                 ARCDIR ~ "/gsistat." ~ RUN ~ "." ~ cycle_YMDH ~ ".ensmean"]) %}
+    {% if DO_PREP_OBS_AERO == True %}
+    - ["{{ COMIN_OBS }}/{{ head }}aeroobs",
+       "{{ ARCDIR }}/aeroobs.{{ RUN }}.{{ cycle_YMDH }}"]
+    - ["{{ COMIN_OBS }}/{{ head }}aeroawobs",
+       "{{ ARCDIR }}/aeroawobs.{{ RUN }}.{{ cycle_YMDH }}"]
     {% endif %}
-
-    # Construct the final file set
-    {% set file_set = file_set + enkf_files %}
-
 {% endif %}
-
-
-# Actually write the yaml
-mkdir:
-    - "{{ ARCDIR }}"
-
-    {% if DO_FIT2OBS == True %}
-    - "{{ VFYARC }}/{{ RUN }}.{{ cycle_YMD }}/{{ cycle_HH }}"
-    {% endif %}
-
-copy:
-   {% for source_dest_pair in file_set %}
-   - {{ source_dest_pair }}
-   {% endfor %}

parm/archive/gefs_arcdir.yaml.j2

@@ -1,36 +1,20 @@
-{% set cycle_HH = current_cycle | strftime("%H") %}
-{% set cycle_YMDH = current_cycle | to_YMDH %}
-{% set cycle_YMD = current_cycle | to_YMD %}
-{% set head = RUN + ".t" + cycle_HH + "z." %}
-
-# Declare the GEFS_ARCH where atmos data will be sent
-{% set GEFS_ARCH = ROTDIR ~ "/gefsarch" %}
-
-{% set file_set = [] %}
+# Variables provided by archive_vars.py:
+# - cycle_HH, cycle_YMDH, cycle_YMD, head
+# - VFYARC (archive directory)
+# - COMIN_ATMOS_ENSSTAT_1p00 (calculated in Python with MEMDIR='ensstat')
 
-{% set tmpl_dict = ({ '${ROTDIR}':ROTDIR,
-                      '${RUN}':RUN,
-                      '${YMD}':cycle_YMD,
-                      '${HH}':cycle_HH,
-                      '${GRID}': '1p00',
-                      '${MEMDIR}': 'ensstat' }) %}
+# Build head string (e.g., 'gfs.t00z.')
+{% set head = RUN + ".t" + cycle_HH + "z." %}
 
-{% set COMIN_ATMOS_ENSSTAT_1p00 = COM_ATMOS_GRIB_GRID_TMPL | replace_tmpl(tmpl_dict) %}
+# Create directories first
+mkdir:
+    - "{{ VFYARC }}"
 
-# Select ensstat files to copy to the arcdir
-{% set ensstat_files = [] %}
-{% if path_exists(COMIN_ATMOS_ENSSTAT_1p00) %}
+# Define all source-destination pairs for archiving
+# Use copy_req for files that MUST exist (raise error if missing)
+copy_req:
+    # GEFS ensemble mean forecast files - REQUIRED
     {% for fhr in range(FHMIN_GFS, FHMAX_GFS + FHOUT_GFS, FHOUT_GFS) %}
-        {% do ensstat_files.append([COMIN_ATMOS_ENSSTAT_1p00 ~ "/" ~ head ~ "mean.pres_." ~
-                                    "1p00" ~ ".f" ~ '%03d'|format(fhr) ~ ".grib2",
-                                    GEFS_ARCH]) %}
+    - ["{{ COMIN_ATMOS_ENSSTAT_1p00 }}/{{ head }}mean.pres_.1p00.f{{ '%03d'|format(fhr) }}.grib2",
+       "{{ VFYARC }}/{{ head }}mean.pres_.1p00.f{{ '%03d'|format(fhr) }}.grib2"]
     {% endfor %}
-{% endif %}
-{% set file_set = ensstat_files %}
-# Actually write the yaml
-mkdir:
-    - "{{ GEFS_ARCH }}"
-copy:
-   {% for source_dest_pair in file_set %}
-   - {{ source_dest_pair }}
-   {% endfor %}