Use assim_freq to configure "offset" in dependencies (#2039)

Use the assim_freq from config.base to configure "offset" in dependencies
instead of hard-coding them.

Resolves #2038

workflow/applications/applications.py

@@ -1,8 +1,9 @@
 #!/usr/bin/env python3
 
 from typing import Dict, List, Any
+from datetime import timedelta
 from hosts import Host
-from wxflow import Configuration
+from wxflow import Configuration, to_timedelta
 from abc import ABC, ABCMeta, abstractmethod
 
 __all__ = ['AppConfig']
@@ -170,14 +171,14 @@ def get_task_names(self) -> Dict[str, List[str]]:
         pass
 
     @staticmethod
-    def get_gfs_interval(gfs_cyc: int) -> str:
+    def get_gfs_interval(gfs_cyc: int) -> timedelta:
         """
         return interval in hours based on gfs_cyc
         """
 
-        gfs_internal_map = {'0': None, '1': '24:00:00', '2': '12:00:00', '4': '06:00:00'}
+        gfs_internal_map = {'1': '24H', '2': '12H', '4': '6H'}
 
         try:
-            return gfs_internal_map[str(gfs_cyc)]
+            return to_timedelta(gfs_internal_map[str(gfs_cyc)])
         except KeyError:
             raise KeyError(f'Invalid gfs_cyc = {gfs_cyc}')

workflow/applications/gfs_cycled.py

@@ -1,6 +1,6 @@
 from typing import Dict, Any
 from applications.applications import AppConfig
-from wxflow import Configuration
+from wxflow import Configuration, to_timedelta
 from datetime import timedelta
 
 
@@ -265,47 +265,45 @@ def get_gfs_cyc_dates(base: Dict[str, Any]) -> Dict[str, Any]:
 
         base_out = base.copy()
 
-        gfs_cyc = base['gfs_cyc']
         sdate = base['SDATE']
         edate = base['EDATE']
-        base_out['INTERVAL'] = '06:00:00'  # Cycled interval is 6 hours
-
-        interval_gfs = AppConfig.get_gfs_interval(gfs_cyc)
+        base_out['INTERVAL'] = to_timedelta(f"{base['assim_freq']}H")
 
         # Set GFS cycling dates
-        hrinc = 0
-        hrdet = 0
-        if gfs_cyc == 0:
-            return base_out
-        elif gfs_cyc == 1:
-            hrinc = 24 - sdate.hour
-            hrdet = edate.hour
-        elif gfs_cyc == 2:
-            if sdate.hour in [0, 12]:
-                hrinc = 12
-            elif sdate.hour in [6, 18]:
+        gfs_cyc = base['gfs_cyc']
+        if gfs_cyc != 0:
+            interval_gfs = AppConfig.get_gfs_interval(gfs_cyc)
+            hrinc = 0
+            hrdet = 0
+            if gfs_cyc == 1:
+                hrinc = 24 - sdate.hour
+                hrdet = edate.hour
+            elif gfs_cyc == 2:
+                if sdate.hour in [0, 12]:
+                    hrinc = 12
+                elif sdate.hour in [6, 18]:
+                    hrinc = 6
+                if edate.hour in [6, 18]:
+                    hrdet = 6
+            elif gfs_cyc == 4:
                 hrinc = 6
-            if edate.hour in [6, 18]:
-                hrdet = 6
-        elif gfs_cyc == 4:
-            hrinc = 6
-        sdate_gfs = sdate + timedelta(hours=hrinc)
-        edate_gfs = edate - timedelta(hours=hrdet)
-        if sdate_gfs > edate:
-            print('W A R N I N G!')
-            print('Starting date for GFS cycles is after Ending date of experiment')
-            print(f'SDATE = {sdate.strftime("%Y%m%d%H")},     EDATE = {edate.strftime("%Y%m%d%H")}')
-            print(f'SDATE_GFS = {sdate_gfs.strftime("%Y%m%d%H")}, EDATE_GFS = {edate_gfs.strftime("%Y%m%d%H")}')
-            gfs_cyc = 0
-
-        base_out['gfs_cyc'] = gfs_cyc
-        base_out['SDATE_GFS'] = sdate_gfs
-        base_out['EDATE_GFS'] = edate_gfs
-        base_out['INTERVAL_GFS'] = interval_gfs
-
-        fhmax_gfs = {}
-        for hh in ['00', '06', '12', '18']:
-            fhmax_gfs[hh] = base.get(f'FHMAX_GFS_{hh}', base.get('FHMAX_GFS_00', 120))
-        base_out['FHMAX_GFS'] = fhmax_gfs
+            sdate_gfs = sdate + timedelta(hours=hrinc)
+            edate_gfs = edate - timedelta(hours=hrdet)
+            if sdate_gfs > edate:
+                print('W A R N I N G!')
+                print('Starting date for GFS cycles is after Ending date of experiment')
+                print(f'SDATE = {sdate.strftime("%Y%m%d%H")},     EDATE = {edate.strftime("%Y%m%d%H")}')
+                print(f'SDATE_GFS = {sdate_gfs.strftime("%Y%m%d%H")}, EDATE_GFS = {edate_gfs.strftime("%Y%m%d%H")}')
+                gfs_cyc = 0
+
+            base_out['gfs_cyc'] = gfs_cyc
+            base_out['SDATE_GFS'] = sdate_gfs
+            base_out['EDATE_GFS'] = edate_gfs
+            base_out['INTERVAL_GFS'] = interval_gfs
+
+            fhmax_gfs = {}
+            for hh in ['00', '06', '12', '18']:
+                fhmax_gfs[hh] = base.get(f'FHMAX_GFS_{hh}', base.get('FHMAX_GFS_00', 120))
+            base_out['FHMAX_GFS'] = fhmax_gfs
 
         return base_out

workflow/rocoto/gefs_xml.py

@@ -2,7 +2,7 @@
 
 from rocoto.workflow_xml import RocotoXML
 from applications.applications import AppConfig
-from wxflow import to_timedelta
+from wxflow import to_timedelta, timedelta_to_HMS
 from typing import Dict
 
 
@@ -16,13 +16,17 @@ def __init__(self, app_config: AppConfig, rocoto_config: Dict) -> None:
     def get_cycledefs(self):
         sdate = self._base['SDATE']
         edate = self._base['EDATE']
-        interval = self._base.get('INTERVAL_GFS', '24:00:00')
+        interval = self._base.get('INTERVAL_GFS', to_timedelta('24H'))
+        sdate_str = sdate.strftime("%Y%m%d%H%M")
+        edate_str = edate.strftime("%Y%m%d%H%M")
+        interval_str = timedelta_to_HMS(interval)
         strings = []
-        strings.append(f'\t<cycledef group="gefs">{sdate.strftime("%Y%m%d%H%M")} {edate.strftime("%Y%m%d%H%M")} {interval}</cycledef>')
+        strings.append(f'\t<cycledef group="gefs">{sdate_str} {edate_str} {interval_str}</cycledef>')
 
-        sdate = sdate + to_timedelta(interval)
+        sdate = sdate + interval
         if sdate <= edate:
-            strings.append(f'\t<cycledef group="gefs_seq">{sdate.strftime("%Y%m%d%H%M")} {edate.strftime("%Y%m%d%H%M")} {interval}</cycledef>')
+            sdate_str = sdate.strftime("%Y%m%d%H%M")
+            strings.append(f'\t<cycledef group="gefs_seq">{sdate_str} {edate_str} {interval_str}</cycledef>')
 
         strings.append('')
         strings.append('')

workflow/rocoto/gfs_cycled_xml.py

@@ -2,7 +2,7 @@
 
 from rocoto.workflow_xml import RocotoXML
 from applications.applications import AppConfig
-from wxflow import to_timedelta
+from wxflow import to_timedelta, timedelta_to_HMS
 from typing import Dict
 
 
@@ -14,21 +14,29 @@ def __init__(self, app_config: AppConfig, rocoto_config: Dict) -> None:
     def get_cycledefs(self):
         sdate = self._base['SDATE']
         edate = self._base['EDATE']
-        interval = self._base.get('INTERVAL', '06:00:00')
+        interval = to_timedelta(f"{self._base['assim_freq']}H")
+        sdate_str = sdate.strftime("%Y%m%d%H%M")
+        edate_str = edate.strftime("%Y%m%d%H%M")
+        interval_str = timedelta_to_HMS(interval)
         strings = []
-        strings.append(f'\t<cycledef group="gdas_half">{sdate.strftime("%Y%m%d%H%M")} {sdate.strftime("%Y%m%d%H%M")} {interval}</cycledef>')
-        sdate = sdate + to_timedelta(interval)
-        strings.append(f'\t<cycledef group="gdas">{sdate.strftime("%Y%m%d%H%M")} {edate.strftime("%Y%m%d%H%M")} {interval}</cycledef>')
+        strings.append(f'\t<cycledef group="gdas_half">{sdate_str} {sdate_str} {interval_str}</cycledef>')
+        sdate = sdate + interval
+        sdate_str = sdate.strftime("%Y%m%d%H%M")
+        strings.append(f'\t<cycledef group="gdas">{sdate_str} {edate_str} {interval_str}</cycledef>')
 
         if self._app_config.gfs_cyc != 0:
             sdate_gfs = self._base['SDATE_GFS']
             edate_gfs = self._base['EDATE_GFS']
             interval_gfs = self._base['INTERVAL_GFS']
-            strings.append(f'\t<cycledef group="gfs">{sdate_gfs.strftime("%Y%m%d%H%M")} {edate_gfs.strftime("%Y%m%d%H%M")} {interval_gfs}</cycledef>')
+            sdate_gfs_str = sdate_gfs.strftime("%Y%m%d%H%M")
+            edate_gfs_str = edate_gfs.strftime("%Y%m%d%H%M")
+            interval_gfs_str = timedelta_to_HMS(interval_gfs)
+            strings.append(f'\t<cycledef group="gfs">{sdate_gfs_str} {edate_gfs_str} {interval_gfs_str}</cycledef>')
 
-            sdate_gfs = sdate_gfs + to_timedelta(interval_gfs)
+            sdate_gfs = sdate_gfs + interval_gfs
+            sdate_gfs_str = sdate_gfs.strftime("%Y%m%d%H%M")
             if sdate_gfs <= edate_gfs:
-                strings.append(f'\t<cycledef group="gfs_seq">{sdate_gfs.strftime("%Y%m%d%H%M")} {edate_gfs.strftime("%Y%m%d%H%M")} {interval_gfs}</cycledef>')
+                strings.append(f'\t<cycledef group="gfs_seq">{sdate_gfs_str} {edate_gfs_str} {interval_gfs_str}</cycledef>')
 
         strings.append('')
         strings.append('')

workflow/rocoto/gfs_forecast_only_xml.py

@@ -2,7 +2,7 @@
 
 from rocoto.workflow_xml import RocotoXML
 from applications.applications import AppConfig
-from wxflow import to_timedelta
+from wxflow import to_timedelta, timedelta_to_HMS
 from typing import Dict
 
 
@@ -14,13 +14,13 @@ def __init__(self, app_config: AppConfig, rocoto_config: Dict) -> None:
     def get_cycledefs(self):
         sdate = self._base['SDATE']
         edate = self._base['EDATE']
-        interval = self._base.get('INTERVAL_GFS', '24:00:00')
+        interval = self._base.get('INTERVAL_GFS', to_timedelta('24H'))
         strings = []
-        strings.append(f'\t<cycledef group="gfs">{sdate.strftime("%Y%m%d%H%M")} {edate.strftime("%Y%m%d%H%M")} {interval}</cycledef>')
+        strings.append(f'\t<cycledef group="gfs">{sdate.strftime("%Y%m%d%H%M")} {edate.strftime("%Y%m%d%H%M")} {timedelta_to_HMS(interval)}</cycledef>')
 
-        sdate = sdate + to_timedelta(interval)
+        sdate = sdate + interval
         if sdate <= edate:
-            strings.append(f'\t<cycledef group="gfs_seq">{sdate.strftime("%Y%m%d%H%M")} {edate.strftime("%Y%m%d%H%M")} {interval}</cycledef>')
+            strings.append(f'\t<cycledef group="gfs_seq">{sdate.strftime("%Y%m%d%H%M")} {edate.strftime("%Y%m%d%H%M")} {timedelta_to_HMS(interval)}</cycledef>')
 
         strings.append('')
         strings.append('')

workflow/rocoto/gfs_tasks.py

@@ -1,5 +1,6 @@
 from applications.applications import AppConfig
 from rocoto.tasks import Tasks, create_wf_task
+from wxflow import timedelta_to_HMS
 import rocoto.rocoto as rocoto
 import numpy as np
 
@@ -86,10 +87,10 @@ def prep(self):
         gfs_enkf = True if self.app_config.do_hybvar and 'gfs' in self.app_config.eupd_cdumps else False
 
         deps = []
-        dep_dict = {'type': 'metatask', 'name': 'gdaspost', 'offset': '-06:00:00'}
+        dep_dict = {'type': 'metatask', 'name': 'gdaspost', 'offset': f"-{timedelta_to_HMS(self._base['cycle_interval'])}"}
         deps.append(rocoto.add_dependency(dep_dict))
         data = f'{atm_hist_path}/[email protected]'
-        dep_dict = {'type': 'data', 'data': data, 'offset': '-06:00:00'}
+        dep_dict = {'type': 'data', 'data': data, 'offset': f"-{timedelta_to_HMS(self._base['cycle_interval'])}"}
         deps.append(rocoto.add_dependency(dep_dict))
         data = f'{dump_path}/{self.cdump}[email protected]_d'
         dep_dict = {'type': 'data', 'data': data}
@@ -116,7 +117,7 @@ def waveinit(self):
             dep_dict = {'type': 'task', 'name': f'{self.cdump}prep'}
             deps.append(rocoto.add_dependency(dep_dict))
             if self.cdump in ['gdas']:
-                dep_dict = {'type': 'cycleexist', 'condition': 'not', 'offset': '-06:00:00'}
+                dep_dict = {'type': 'cycleexist', 'condition': 'not', 'offset': f"-{timedelta_to_HMS(self._base['cycle_interval'])}"}
                 deps.append(rocoto.add_dependency(dep_dict))
             dependencies = rocoto.create_dependency(dep_condition='or', dep=deps)
             cycledef = 'gdas_half,gdas' if self.cdump in ['gdas'] else self.cdump
@@ -155,7 +156,7 @@ def aerosol_init(self):
             interval = self._base['INTERVAL_GFS']
         elif self.cdump in ['gdas']:
             interval = self._base['INTERVAL']
-        offset = f'-{interval}'
+        offset = f'-{timedelta_to_HMS(interval)}'
 
         # Files from previous cycle
         files = [f'@Y@m@[email protected]_core.res.nc'] + \
@@ -181,7 +182,7 @@ def anal(self):
         dep_dict = {'type': 'task', 'name': f'{self.cdump}prep'}
         deps.append(rocoto.add_dependency(dep_dict))
         if self.app_config.do_hybvar:
-            dep_dict = {'type': 'metatask', 'name': 'enkfgdasepmn', 'offset': '-06:00:00'}
+            dep_dict = {'type': 'metatask', 'name': 'enkfgdasepmn', 'offset': f"-{timedelta_to_HMS(self._base['cycle_interval'])}"}
             deps.append(rocoto.add_dependency(dep_dict))
             dependencies = rocoto.create_dependency(dep_condition='and', dep=deps)
         else:
@@ -223,7 +224,7 @@ def analcalc(self):
         dep_dict = {'type': 'task', 'name': f'{self.cdump}sfcanl'}
         deps.append(rocoto.add_dependency(dep_dict))
         if self.app_config.do_hybvar and self.cdump in ['gdas']:
-            dep_dict = {'type': 'task', 'name': 'enkfgdasechgres', 'offset': '-06:00:00'}
+            dep_dict = {'type': 'task', 'name': 'enkfgdasechgres', 'offset': f"-{timedelta_to_HMS(self._base['cycle_interval'])}"}
             deps.append(rocoto.add_dependency(dep_dict))
         dependencies = rocoto.create_dependency(dep_condition='and', dep=deps)
 
@@ -262,7 +263,7 @@ def atmanlinit(self):
         dep_dict = {'type': 'task', 'name': f'{self.cdump}prepatmiodaobs'}
         deps.append(rocoto.add_dependency(dep_dict))
         if self.app_config.do_hybvar:
-            dep_dict = {'type': 'metatask', 'name': 'enkfgdasepmn', 'offset': '-06:00:00'}
+            dep_dict = {'type': 'metatask', 'name': 'enkfgdasepmn', 'offset': f"-{timedelta_to_HMS(self._base['cycle_interval'])}"}
             deps.append(rocoto.add_dependency(dep_dict))
             dependencies = rocoto.create_dependency(dep_condition='and', dep=deps)
         else:
@@ -369,7 +370,7 @@ def ocnanalprep(self):
 
         deps = []
         data = f'{ocean_hist_path}/[email protected]'
-        dep_dict = {'type': 'data', 'data': data, 'offset': '-06:00:00'}
+        dep_dict = {'type': 'data', 'data': data, 'offset': f"-{timedelta_to_HMS(self._base['cycle_interval'])}"}
         deps.append(rocoto.add_dependency(dep_dict))
         dependencies = rocoto.create_dependency(dep=deps)
 
@@ -536,7 +537,7 @@ def _fcst_cycled(self):
         dependencies = rocoto.create_dependency(dep_condition='and', dep=dependencies)
 
         if self.cdump in ['gdas']:
-            dep_dict = {'type': 'cycleexist', 'condition': 'not', 'offset': '-06:00:00'}
+            dep_dict = {'type': 'cycleexist', 'condition': 'not', 'offset': f"-{timedelta_to_HMS(self._base['cycle_interval'])}"}
             dependencies.append(rocoto.add_dependency(dep_dict))
             dependencies = rocoto.create_dependency(dep_condition='or', dep=dependencies)
 
@@ -1074,7 +1075,7 @@ def eobs(self):
         deps = []
         dep_dict = {'type': 'task', 'name': f'{self.cdump.replace("enkf","")}prep'}
         deps.append(rocoto.add_dependency(dep_dict))
-        dep_dict = {'type': 'metatask', 'name': 'enkfgdasepmn', 'offset': '-06:00:00'}
+        dep_dict = {'type': 'metatask', 'name': 'enkfgdasepmn', 'offset': f"-{timedelta_to_HMS(self._base['cycle_interval'])}"}
         deps.append(rocoto.add_dependency(dep_dict))
         dependencies = rocoto.create_dependency(dep_condition='and', dep=deps)
 
@@ -1129,7 +1130,7 @@ def atmensanlinit(self):
         deps = []
         dep_dict = {'type': 'task', 'name': f'{self.cdump.replace("enkf","")}prepatmiodaobs'}
         deps.append(rocoto.add_dependency(dep_dict))
-        dep_dict = {'type': 'metatask', 'name': 'enkfgdasepmn', 'offset': '-06:00:00'}
+        dep_dict = {'type': 'metatask', 'name': 'enkfgdasepmn', 'offset': f"-{timedelta_to_HMS(self._base['cycle_interval'])}"}
         deps.append(rocoto.add_dependency(dep_dict))
         dependencies = rocoto.create_dependency(dep_condition='and', dep=deps)
 
@@ -1145,7 +1146,7 @@ def atmensanlrun(self):
         deps = []
         dep_dict = {'type': 'task', 'name': f'{self.cdump}atmensanlinit'}
         deps.append(rocoto.add_dependency(dep_dict))
-        dep_dict = {'type': 'metatask', 'name': 'enkfgdasepmn', 'offset': '-06:00:00'}
+        dep_dict = {'type': 'metatask', 'name': 'enkfgdasepmn', 'offset': f"-{timedelta_to_HMS(self._base['cycle_interval'])}"}
         deps.append(rocoto.add_dependency(dep_dict))
         dependencies = rocoto.create_dependency(dep_condition='and', dep=deps)
 
@@ -1243,7 +1244,7 @@ def efcs(self):
         dep_dict = {'type': 'task', 'name': f'{self.cdump}esfc'}
         deps.append(rocoto.add_dependency(dep_dict))
         dependencies = rocoto.create_dependency(dep_condition='and', dep=deps)
-        dep_dict = {'type': 'cycleexist', 'condition': 'not', 'offset': '-06:00:00'}
+        dep_dict = {'type': 'cycleexist', 'condition': 'not', 'offset': f"-{timedelta_to_HMS(self._base['cycle_interval'])}"}
         dependencies.append(rocoto.add_dependency(dep_dict))
         dependencies = rocoto.create_dependency(dep_condition='or', dep=dependencies)
 

workflow/rocoto/tasks.py

@@ -3,7 +3,7 @@
 import numpy as np
 from applications.applications import AppConfig
 import rocoto.rocoto as rocoto
-from wxflow import Template, TemplateConstants
+from wxflow import Template, TemplateConstants, to_timedelta
 
 __all__ = ['Tasks', 'create_wf_task']
 
@@ -34,6 +34,7 @@ def __init__(self, app_config: AppConfig, cdump: str) -> None:
         # Save dict_configs and base in the internal state (never know where it may be needed)
         self._configs = self.app_config.configs
         self._base = self._configs['base']
+        self._base['cycle_interval'] = to_timedelta(f'{self._base["assim_freq"]}H')
 
         self.n_tiles = 6  # TODO - this needs to be elsewhere