Refactor Tar Archiving Logic to Python-Based Implementation

dev/scripts/exglobal_archive_tars.py

@@ -3,6 +3,7 @@
 import os
 
 from pygfs.task.archive import Archive
+from pygfs.utils.archive_tar_vars import ArchiveTarVars
 from wxflow import AttrDict, Logger, cast_strdict_as_dtypedict, logit, chdir
 
 # initialize root logger
@@ -24,35 +25,9 @@ def main():
         except KeyError:
             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 = ['ATARDIR', 'current_cycle', 'FHMIN', 'FHMAX', 'FHOUT', 'RUN', 'PDY',
-            'DO_VERFRAD', 'DO_VMINMON', 'DO_VERFOZN', 'DO_ICE', 'DO_PREP_OBS_AERO',
-            'PARMgfs', 'DO_OCN', 'DO_WAVE', 'WRITE_DOPOST', 'PSLOT',
-            'DO_JEDISNOWDA', 'DO_ARCHCOM', 'ARCHCOM_TO', 'ROTDIR', 'ARCH_WARMICFREQ',
-            'ARCH_FCSTICFREQ', 'ARCH_CYC', 'assim_freq', 'ARCDIR', 'SDATE',
-            'FHMIN_GFS', 'FHMAX_GFS', 'FHOUT_GFS', 'ARCH_GAUSSIAN', 'MODE',
-            'FHOUT_OCN', 'FHOUT_ICE', 'FHOUT_OCN_GFS', 'FHOUT_ICE_GFS', 'DO_BUFRSND', 'DOHYBVAR',
-            'ARCH_GAUSSIAN_FHMAX', 'ARCH_GAUSSIAN_FHINC', 'ARCH_GAUSSIAN_FHINC',
-            'DOIAU', 'OCNRES', 'ICERES', 'NUM_SND_COLLECTIVES', 'FHOUT_WAV', 'FHOUT_WAV_GFS',
-            'FHOUT_HF_WAV', 'FHMAX_WAV', 'FHMAX_HF_WAV', 'FHMAX_WAV_GFS',
-            'restart_interval_gdas', 'restart_interval_gfs', 'DO_ARCHCOM',
-            'DO_AERO_ANL', 'DO_AERO_FCST', 'DO_CA', 'DOBNDPNT_WAVE', 'DO_JEDIOCNVAR', 'DOHYBVAR_OCN',
-            'DOLETKF_OCN', 'NMEM_ENS', 'DO_JEDIATMVAR', 'FHMAX_FITS', 'waveGRD',
-            'IAUFHRS', 'DO_FIT2OBS', 'NET', 'FHOUT_HF_GFS', 'FHMAX_HF_GFS',
-            'OFFSET_START_HOUR', 'ARCH_EXPDIR', 'EXPDIR', 'ARCH_EXPDIR_FREQ', 'ARCH_HASHES',
-            'ARCH_DIFFS', 'SDATE', 'EDATE', 'HOMEgfs', 'DO_GEMPAK', 'DATASETS_YAML',
-            'WAVE_OUT_GRIDS', 'DO_GSISOILDA', 'DO_LAND_IAU', 'TARBALL_TYPE', 'ATMINC_GRID']
-
-    archive_dict = AttrDict()
-    for key in keys:
-        archive_dict[key] = archive.task_config.get(key)
-        if archive_dict[key] is None:
-            logger.warning(f"WARNING: key ({key}) not found in 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 variables for YAML templates (config keys, COM paths, cycle vars, member paths)
+    # Note: COM paths are relative to ROTDIR for portability in tar archives
+    archive_dict = ArchiveTarVars.get_all_yaml_vars(archive.task_config)
 
     with chdir(config.ROTDIR):
         logger.debug(f"Changed working directory to {config.ROTDIR}")

dev/scripts/exglobal_archive_vrfy.py

@@ -3,7 +3,7 @@
 import os
 
 from pygfs.task.archive import Archive
-from pygfs.utils.archive_vars import ArchiveVrfyVars
+from pygfs.utils.archive_vrfy_vars import ArchiveVrfyVars
 from wxflow import AttrDict, Logger, cast_strdict_as_dtypedict, logit, chdir
 
 # initialize root logger

dev/scripts/exglobal_enkf_earc_tars.py

@@ -3,6 +3,7 @@
 import os
 
 from pygfs.task.archive import Archive
+from pygfs.utils.archive_tar_vars import ArchiveTarVars
 from wxflow import AttrDict, Logger, cast_strdict_as_dtypedict, chdir, logit
 
 # initialize root logger
@@ -17,30 +18,13 @@ def main():
     # Instantiate the Archive object
     archive = Archive(config)
 
-    # Pull out all the configuration keys needed to run the rest of archive steps
-    keys = ['ATARDIR', 'current_cycle', 'IAUFHRS', 'RUN', 'PDY',
-            'PSLOT', 'DO_ARCHCOM', 'ARCHCOM_TO', 'ROTDIR', 'PARMgfs',
-            'ARCDIR', 'SDATE', 'MODE', 'ENSGRP', 'NMEM_EARCGRP',
-            'NMEM_ENS', 'DO_CALC_INCREMENT_ENKF_GFS', 'DO_JEDIATMENS',
-            'lobsdiag_forenkf', 'FHMIN_ENKF', 'FHMAX_ENKF_GFS',
-            'FHOUT_ENKF_GFS', 'FHMAX_ENKF', 'FHOUT_ENKF', 'ENKF_SPREAD',
-            'restart_interval_enkfgdas', 'restart_interval_enkfgfs',
-            'DOHYBVAR', 'DOIAU_ENKF', 'IAU_OFFSET', 'DOIAU', 'DO_CA',
-            'DO_CALC_INCREMENT', 'assim_freq', 'ARCH_CYC', 'DO_JEDISNOWDA',
-            'ARCH_WARMICFREQ', 'ARCH_FCSTICFREQ', 'DOHYBVAR_OCN',
-            'DOLETKF_OCN', 'IAUFHRS_ENKF', 'NET', 'NMEM_ENS_GFS', 'DO_GSISOILDA', 'DO_LAND_IAU']
-
-    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!")
-
-    # Also import all COMIN* directory and template variables
-    for key in archive.task_config.keys():
-        if key.startswith(("COM_", "COMIN_")):
-            archive_dict[key] = archive.task_config.get(key)
+    # Collect all variables for YAML templates (config keys, COM paths, cycle vars, member paths)
+    # Note: COM paths are relative to ROTDIR for portability in tar archives
+    archive_dict = ArchiveTarVars.get_all_yaml_vars(archive.task_config)
 
+    # Change to ROTDIR for the entire archiving workflow so that relative paths
+    # in YAML templates (e.g., logs/..., enkfgdas.20211221/...) are resolved
+    # correctly during both file existence checks (glob) and tar creation
     with chdir(config.ROTDIR):
 
         # Determine which archives to create

dev/scripts/exglobal_enkf_earc_vrfy.py

@@ -3,7 +3,7 @@
 import os
 
 from pygfs.task.archive import Archive
-from pygfs.utils.archive_vars import ArchiveVrfyVars
+from pygfs.utils.archive_vrfy_vars import ArchiveVrfyVars
 from wxflow import AttrDict, Logger, cast_strdict_as_dtypedict, chdir, logit
 
 # initialize root logger

parm/archive/chem.yaml.j2

@@ -4,4 +4,4 @@ chem:
     target: "{{ ATARDIR }}/{{ cycle_YMDH }}/chem.tar"
     required:
         # TODO explicitize this set
-        - "{{ COMIN_CHEM_HISTORY | relpath(ROTDIR) }}/{{ head }}*"
+        - "{{ COMIN_CHEM_HISTORY }}/{{ head }}*"

parm/archive/enkf.yaml.j2

@@ -7,9 +7,7 @@ enkf:
             {% for mem in range(1, nmem_ens + 1) %}
         - "logs/{{ cycle_YMDH }}/{{ RUN }}_fcst_mem{{ '%03d' % mem }}.log"
             {% endfor %}
-            {% set fhrs = range(fhmin, fhmax + fhout, fhout) %}
-            {% set ngrps = fhrs | length %}
-            {% for grp in range(0, ngrps) %}
+            {% for grp in range(0, enkf_epos_ngrps) %}
         - "logs/{{ cycle_YMDH }}/{{ RUN }}_epos{{ '%03d' % grp }}.log"
             {% endfor %}
             {% if not DO_JEDIATMENS %}
@@ -18,7 +16,7 @@ enkf:
         {% endif %}
         - "logs/{{ cycle_YMDH }}/{{ RUN }}_esfc.log"
         {% if not DO_JEDIATMENS %}
-            {% for grp in range(IAUFHRS | length) %}
+            {% for grp in range(iaufhrs_str | length) %}
         - "logs/{{ cycle_YMDH }}/{{ RUN }}_ecen{{ '%03d' % grp }}.log"
             {% endfor %}
         {% endif %}
@@ -49,7 +47,7 @@ enkf:
             {% if DOHYBVAR %}
                 {% if DOIAU %}
                     {% for fhr in range(3, fhmax + 1, 3) %}
-        - "{{ COMIN_ATMOS_HISTORY | relpath(ROTDIR) }}/{{ head |replace('enkf','')}}ensres.atm.f{{ '%03d' % fhr }}.nc"
+        - "{{ COMIN_ATMOS_HISTORY }}/{{ head |replace('enkf','')}}ensres.atm.f{{ '%03d' % fhr }}.nc"
                     {% endfor %}
                 {% endif %}
             {% endif %}
@@ -58,10 +56,10 @@ enkf:
         # Ensemble mean and spread
         {% if is_gdas %}
             {% for fhr in range(3, fhmax + 1, 3) %}
-        - "{{ COMIN_ATMOS_HISTORY_ENSSTAT | relpath(ROTDIR) }}/{{ head }}ensmean.atm.f{{ '%03d' % fhr }}.nc"
-        - "{{ COMIN_ATMOS_HISTORY_ENSSTAT | relpath(ROTDIR) }}/{{ head }}ensmean.sfc.f{{ '%03d' % fhr }}.nc"
+        - "{{ COMIN_ATMOS_HISTORY_ENSSTAT }}/{{ head }}ensmean.atm.f{{ '%03d' % fhr }}.nc"
+        - "{{ COMIN_ATMOS_HISTORY_ENSSTAT }}/{{ head }}ensmean.sfc.f{{ '%03d' % fhr }}.nc"
                 {% if ENKF_SPREAD %}
-        - "{{ COMIN_ATMOS_HISTORY_ENSSTAT | relpath(ROTDIR) }}/{{ head }}ensspread.atm.f{{ '%03d' % fhr }}.nc"
+        - "{{ COMIN_ATMOS_HISTORY_ENSSTAT }}/{{ head }}ensspread.atm.f{{ '%03d' % fhr }}.nc"
                 {% endif %}
             {% endfor %}
         {% endif %}
@@ -91,47 +89,47 @@ enkf:
             {% endif %}
         {% endif %}
         {% for file in da_stat_files %}
-        - "{{ COMIN_ATMOS_ANALYSIS_ENSSTAT | relpath(ROTDIR) }}/{{ head }}{{ file }}"
+        - "{{ COMIN_ATMOS_ANALYSIS_ENSSTAT }}/{{ head }}{{ file }}"
         {% endfor %}
         {% for file in da_conf_files %}
-        - "{{ COMIN_CONF | relpath(ROTDIR) }}/{{ head }}{{ file }}"
+        - "{{ COMIN_CONF }}/{{ head }}{{ file }}"
         {% endfor %}
 
         {% if DO_JEDISNOWDA %}
-        - "{{ COMIN_SNOW_ANALYSIS_ENSSTAT | relpath(ROTDIR) }}/{{ head }}snow_analysis.ioda_hofx.ensmean.tar.gz"
+        - "{{ COMIN_SNOW_ANALYSIS_ENSSTAT }}/{{ head }}snow_analysis.ioda_hofx.ensmean.tar.gz"
             {% for itile in range(1,7) %}
         # Snow analysis is 3dvar
-        - "{{ COMIN_SNOW_ANALYSIS_ENSSTAT | relpath(ROTDIR) }}/{{ cycle_YMD }}.{{ cycle_HH }}0000.snow_increment.sfc_data.tile{{ itile }}.nc"
+        - "{{ COMIN_SNOW_ANALYSIS_ENSSTAT }}/{{ cycle_YMD }}.{{ cycle_HH }}0000.snow_increment.sfc_data.tile{{ itile }}.nc"
             {% endfor %}
         {% endif %}
 
         # Ensemble mean analyses/increments
         # soil DA increments
         {% if DO_GSISOILDA %}
-            {% for fhr in IAUFHRS %}
-        - "{{ COMIN_ATMOS_ANALYSIS_ENSSTAT | relpath(ROTDIR) }}/{{ head }}ensmean_increment.sfc.i{{ '%03d' % fhr }}.nc"
+            {% for fhr in iaufhrs_str %}
+        - "{{ COMIN_ATMOS_ANALYSIS_ENSSTAT }}/{{ head }}ensmean_increment.sfc.i{{ fhr }}.nc"
             {% endfor %}
         {% endif %}
 
         # Atmospheric analyses/increment
-        {% for fhr in IAUFHRS %}
+        {% for fhr in iaufhrs_str %}
             {% if not DO_JEDIATMENS %}
                 {% if do_calc_increment %}
         # Store analyses instead of increments
-        - "{{ COMIN_ATMOS_ANALYSIS_ENSSTAT | relpath(ROTDIR) }}/{{ head }}ensmean_analysis.atm.a{{ '%03d' % fhr }}.nc"
+        - "{{ COMIN_ATMOS_ANALYSIS_ENSSTAT }}/{{ head }}ensmean_analysis.atm.a{{ fhr }}.nc"
                 {% else %}
         # Store increments
-        - "{{ COMIN_ATMOS_ANALYSIS_ENSSTAT | relpath(ROTDIR) }}/{{ head }}ensmean_increment.atm.i{{ '%03d' % fhr }}.nc"
+        - "{{ COMIN_ATMOS_ANALYSIS_ENSSTAT }}/{{ head }}ensmean_increment.atm.i{{ fhr }}.nc"
                 {% endif %}
             {% else %}
-        - "{{ COMIN_ATMOS_ANALYSIS_ENSSTAT | relpath(ROTDIR) }}/{{ head }}csg_ensmean_jedi_analysis.atm.a{{ '%03d' % fhr }}.nc"
+        - "{{ COMIN_ATMOS_ANALYSIS_ENSSTAT }}/{{ head }}csg_ensmean_jedi_analysis.atm.a{{ fhr }}.nc"
             {% endif %}
 
         {% endfor %}
 
         # End of analysis mean increments/analyses
 
         {% if DOHYBVAR_OCN %}
-        - "{{ COMIN_OCEAN_ANALYSIS_ENSSTAT | relpath(ROTDIR) }}/{{ head }}bg_ensvar.nc"
-        - "{{ COMIN_ICE_ANALYSIS_ENSSTAT | relpath(ROTDIR) }}/{{ head }}bg_ensvar.nc"
+        - "{{ COMIN_OCEAN_ANALYSIS_ENSSTAT }}/{{ head }}bg_ensvar.nc"
+        - "{{ COMIN_ICE_ANALYSIS_ENSSTAT }}/{{ head }}bg_ensvar.nc"
         {% endif %}

parm/archive/enkf_arcdir.yaml.j2

@@ -1,4 +1,4 @@
-# Variables provided by archive_vars.py:
+# Variables provided by archive_vrfy_vars.py:
 # - cycle_HH, cycle_YMDH, cycle_YMD, head
 # - ARCDIR
 # - All COMIN_* paths

parm/archive/enkf_grp.yaml.j2

@@ -2,49 +2,40 @@ enkf_grp:
     name: "ENKF_GRP"
     target: "{{ ATARDIR }}/{{ cycle_YMDH }}/{{ RUN }}_grp{{ ENSGRP }}.tar"
     required:
-        {% for mem in range(first_group_mem, last_group_mem + 1) %}
-            {% set imem = mem - first_group_mem %}
-        # Construct member COM directories
-            {% set COMIN_ATMOS_ANALYSIS_MEM = COMIN_ATMOS_ANALYSIS_MEM_list[imem] %}
-            {% set COMIN_ATMOS_HISTORY_MEM = COMIN_ATMOS_HISTORY_MEM_list[imem] %}
-            {% set COMIN_ATMOS_RESTART_MEM = COMIN_ATMOS_RESTART_MEM_list[imem] %}
-
         # Forecast data
-            {% if is_gdas %}
-                {% for fhr in range(3, 10, 3) %}
-        - "{{ COMIN_ATMOS_HISTORY_MEM | relpath(ROTDIR) }}/{{ head }}atm.f{{ "%03d" % fhr }}.nc"
-        - "{{ COMIN_ATMOS_HISTORY_MEM | relpath(ROTDIR) }}/{{ head }}sfc.f{{ "%03d" % fhr }}.nc"
-                {% endfor %}
-            {% endif %}
+        {% if is_gdas %}
+            {% for fhr in range(3, 10, 3) %}
+        - "{{ COMIN_ATMOS_HISTORY_MEM }}/{{ head }}atm.f{{ "%03d" % fhr }}.nc"
+        - "{{ COMIN_ATMOS_HISTORY_MEM }}/{{ head }}sfc.f{{ "%03d" % fhr }}.nc"
+            {% endfor %}
+        {% endif %}
 
         # Store the individual member analysis data
-            {% if not lobsdiag_forenkf %}
-        - "{{ COMIN_ATMOS_RESTART_MEM | relpath(ROTDIR) }}/{{ head }}gsistat.txt"
-            {% endif %}
-            {% if DO_JEDIATMENS %}
-                {% for iaufhr in IAUFHRS %}
-                    {% for itile in range(6) %}
-        - "{{ COMIN_ATMOS_ANALYSIS_MEM | relpath(ROTDIR) }}/{{ head }}recentered_jedi_increment.atm.i{{ '%03d' % iaufhr }}.tile{{ itile+1 }}.nc"
-                    {% endfor %}
+        {% if not lobsdiag_forenkf %}
+        - "{{ COMIN_ATMOS_RESTART_MEM }}/{{ head }}gsistat.txt"
+        {% endif %}
+        {% if DO_JEDIATMENS %}
+            {% for iaufhr in iaufhrs_str %}
+                {% for itile in range(6) %}
+        - "{{ COMIN_ATMOS_ANALYSIS_MEM }}/{{ head }}recentered_jedi_increment.atm.i{{ iaufhr }}.tile{{ itile+1 }}.nc"
                 {% endfor %}
+            {% endfor %}
+        {% else %}
+            {% if do_calc_increment %}
+        - "{{ COMIN_ATMOS_ANALYSIS_MEM }}/{{ head }}analysis.atm.a006.nc"
             {% else %}
-                {% if do_calc_increment %}
-        - "{{ COMIN_ATMOS_ANALYSIS_MEM | relpath(ROTDIR) }}/{{ head }}analysis.atm.a006.nc"
-                {% else %}
-                    {% for iaufhr in IAUFHRS %}
-                        {% set iaufhr = iaufhr %}
-        - "{{ COMIN_ATMOS_ANALYSIS_MEM | relpath(ROTDIR) }}/{{ head }}recentered_increment.atm.i{{ '%03d' % iaufhr }}.nc"
-                    {% endfor %}  # iaufhr in IAUFHRS
-                {% endif %}
+                {% for iaufhr in iaufhrs_str %}
+                    {% set iaufhr = iaufhr %}
+        - "{{ COMIN_ATMOS_ANALYSIS_MEM }}/{{ head }}recentered_increment.atm.i{{ iaufhr }}.nc"
+                {% endfor %}  # iaufhr in iaufhrs_str
             {% endif %}
+        {% endif %}
 
         # soil DA increments
-            {% if is_gdas %}
-                {% if DO_GSISOILDA %}
-                    {% for iaufhr in IAUFHRS %}
-        - "{{ COMIN_ATMOS_ANALYSIS_MEM | relpath(ROTDIR) }}/{{ head }}increment.sfc.i{{ '%03d' % iaufhr }}.nc"
-                    {% endfor %}
-                {% endif %}
+        {% if is_gdas %}
+            {% if DO_GSISOILDA %}
+                {% for iaufhr in iaufhrs_str %}
+        - "{{ COMIN_ATMOS_ANALYSIS_MEM }}/{{ head }}increment.sfc.i{{ iaufhr }}.nc"
+                {% endfor %}
             {% endif %}
-
-        {% endfor %}  # first_group_mem to last_group_mem
+        {% endif %}

parm/archive/enkf_marine_grp.yaml.j2

@@ -2,35 +2,18 @@ enkf_marine_grp:
     name: "ENKF_MARINE_GRP"
     target: "{{ ATARDIR }}/{{ cycle_YMDH }}/{{ RUN }}_marine_grp{{ ENSGRP }}.tar"
     required:
-        {% for mem in range(first_group_mem, last_group_mem + 1) %}
-        {% set imem = mem - first_group_mem %}
-        # Construct member COM directories
-        {% set COMIN_OCEAN_ANALYSIS_MEM = COMIN_OCEAN_ANALYSIS_MEM_list[imem] %}
-        {% set COMIN_OCEAN_LETKF_MEM = COMIN_OCEAN_LETKF_MEM_list[imem] %}
-        {% set COMIN_OCEAN_HISTORY_MEM = COMIN_OCEAN_HISTORY_MEM_list[imem] %}
-        {% set COMIN_ICE_ANALYSIS_MEM = COMIN_ICE_ANALYSIS_MEM_list[imem] %}
-        {% set COMIN_ICE_LETKF_MEM = COMIN_ICE_LETKF_MEM_list[imem] %}
-        {% set COMIN_ICE_HISTORY_MEM = COMIN_ICE_HISTORY_MEM_list[imem] %}
-
         # Forecast data
         {% if is_gdas %}
-        - "{{ COMIN_OCEAN_HISTORY_MEM | relpath(ROTDIR) }}/{{ head }}inst.f009.nc"
+        - "{{ COMIN_OCEAN_HISTORY_MEM }}/{{ head }}inst.f009.nc"
 
         {% for body in ["ic", "inst.f009"] %}
-        - "{{ COMIN_ICE_HISTORY_MEM | relpath(ROTDIR) }}/{{ head }}{{ body }}.nc"
+        - "{{ COMIN_ICE_HISTORY_MEM }}/{{ head }}{{ body }}.nc"
         {% endfor %}
         {% endif %}
 
         # Store the individual member analysis data
-        {% if DOIAU_ENKF %}
-        {% set anl_delta = "-3H" | to_timedelta %}
-        {% else %}
-        {% set anl_delta = "0H" | to_timedelta %}
-        {% endif %}
-        {% set anl_time = current_cycle | add_to_datetime(anl_delta) %}
         {% if DOLETKF_OCN %}
-        - "{{ COMIN_OCEAN_LETKF_MEM | relpath(ROTDIR) }}/{{ head }}jedi_analysis.a006.nc"
-        - "{{ COMIN_ICE_LETKF_MEM | relpath(ROTDIR) }}/{{ head }}jedi_analysis.a006.nc"
-        - "{{ COMIN_CONF | relpath(ROTDIR) }}/soca_letkf.yaml"
+        - "{{ COMIN_OCEAN_LETKF_MEM }}/{{ head }}jedi_analysis.a006.nc"
+        - "{{ COMIN_ICE_LETKF_MEM }}/{{ head }}jedi_analysis.a006.nc"
+        - "{{ COMIN_CONF }}/soca_letkf.yaml"
         {% endif %}
-        {% endfor %}  # first_group_mem to last_group_mem