Add an archive task to GEFS system to archive files locally

parm/archive/gefs_arcdir.yaml.j2

@@ -0,0 +1,38 @@
+{% 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 = [] %}
+
+{% set tmpl_dict = ({ '${ROTDIR}':ROTDIR,
+ '${RUN}':RUN,
+ '${YMD}':cycle_YMD,
+ '${HH}':cycle_HH,
+ '${GRID}': '1p00',
+ '${MEMDIR}': 'ensstat' }) %}
+
+{% set COMIN_ATMOS_ENSSTAT_1p00 = COM_ATMOS_GRIB_GRID_TMPL | replace_tmpl(tmpl_dict) %}
+
+# Select ensstat files to copy to the arcdir
+{% if RUN == "gefs" %}
+ {% set ensstat_files = [] %}
+ {% if path_exists(COMIN_ATMOS_ENSSTAT_1p00) %}
+ {% 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]) %}
+ {% endfor %}
+ {% endif %}
+{% 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 %}

parm/config/gefs/yaml/defaults.yaml

@@ -14,3 +14,5 @@ base:
  FCST_BREAKPOINTS: "48"
  REPLAY_ICS: "NO"
  USE_OCN_PERTURB_FILES: "false"
+ HPSSARCH: "NO"
+ LOCALARCH: "NO"

scripts/exgdas_enkf_earc.py

@@ -28,11 +28,13 @@ def main():
  'DOHYBVAR', 'DOIAU_ENKF', 'IAU_OFFSET', 'DOIAU',
  'DO_CALC_INCREMENT', 'assim_freq', 'ARCH_CYC',
  'ARCH_WARMICFREQ', 'ARCH_FCSTICFREQ',
- 'IAUFHRS_ENKF']
+ 'IAUFHRS_ENKF', 'NET']
 
  archive_dict = AttrDict()
  for key in keys:
- archive_dict[key] = archive.task_config[key]
+ 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():

scripts/exglobal_archive.py

@@ -31,16 +31,20 @@ def main():
  'restart_interval_gdas', 'restart_interval_gfs',
  'AERO_ANL_RUN', 'AERO_FCST_RUN', 'DOIBP_WAV', 'DO_JEDIOCNVAR',
  'NMEM_ENS', 'DO_JEDIATMVAR', 'DO_VRFY_OCEANDA', 'FHMAX_FITS',
- 'IAUFHRS', 'DO_FIT2OBS']
+ 'IAUFHRS', 'DO_FIT2OBS', 'NET']
 
  archive_dict = AttrDict()
  for key in keys:
- archive_dict[key] = archive.task_config[key]
+ 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* and COMOUT* directory and template variables
  for key in archive.task_config.keys():
- if key.startswith("COMIN_") or key.startswith("COMOUT_"):
- archive_dict[key] = archive.task_config[key]
+ if key.startswith("COM_") or key.startswith("COMIN_") or key.startswith("COMOUT_"):
+ archive_dict[key] = archive.task_config.get(key)
+ if archive_dict[key] is None:
+ print(f"Warning: key ({key}) not found in task_config!")
 
  cwd = os.getcwd()
 

ush/python/pygfs/task/archive.py

@@ -63,9 +63,6 @@ def configure(self, arch_dict: Dict[str, Any]) -> (Dict[str, Any], List[Dict[str
  if not os.path.isdir(arch_dict.ROTDIR):
  raise FileNotFoundError(f"FATAL ERROR: The ROTDIR ({arch_dict.ROTDIR}) does not exist!")
 
- if arch_dict.RUN == "gefs":
- raise NotImplementedError("FATAL ERROR: Archiving is not yet set up for GEFS runs")
-
  if arch_dict.RUN in ["gdas", "gfs"]:
 
  # Copy the cyclone track files and rename the experiments
@@ -75,7 +72,7 @@ def configure(self, arch_dict: Dict[str, Any]) -> (Dict[str, Any], List[Dict[str
  archive_parm = os.path.join(arch_dict.PARMgfs, "archive")
 
  # Collect the dataset to archive locally
- arcdir_j2yaml = os.path.join(archive_parm, "arcdir.yaml.j2")
+ arcdir_j2yaml = os.path.join(archive_parm, f"{arch_dict.NET}_arcdir.yaml.j2")
 
  # Add the glob.glob function for capturing log filenames
  # TODO remove this kludge once log filenames are explicit
@@ -117,6 +114,9 @@ def configure(self, arch_dict: Dict[str, Any]) -> (Dict[str, Any], List[Dict[str
  self.tar_cmd = ""
  return arcdir_set, []
 
+ if arch_dict.NET == "gefs":
+ raise NotImplementedError("GEFS archiving is not yet implemented!")
+
  master_yaml = "master_" + arch_dict.RUN + ".yaml.j2"
 
  parsed_sets = parse_j2yaml(os.path.join(archive_parm, master_yaml),

workflow/rocoto/gefs_tasks.py

@@ -558,7 +558,7 @@ def arch(self):
  'envars': self.envars,
  'cycledef': 'gefs',
  'dependency': dependencies,
- 'command': f'{self.HOMEgfs}/jobs/rocoto/arch_test.sh',
+ 'command': f'{self.HOMEgfs}/jobs/rocoto/arch.sh',
  'job_name': f'{self.pslot}_{task_name}_@H',
  'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log',
  'maxtries': '&MAXTRIES;'