From ef6827dd6abdab2996d1e19f9e0ff5d3071e0fdd Mon Sep 17 00:00:00 2001 From: Walter Kolczynski - NOAA Date: Mon, 8 Jan 2024 09:43:12 -0500 Subject: [PATCH] Refactor rocoto task XML creation (#2189) Refactors the rocoto task generation to be recursive. This will allow nested metatasks to loop over multiple variables, which is needed for GEFS product generation. As part of this refactor, there is no longer separate arguments to designate metatasks. Instead, task dicts can include a nested 'task_dict' as well as a 'var_dict' containing the variables to loop over. The nested task dict can then either have another layer, or be the innermost task. To accommodate the new recursive nature, some defaults that were previously defined in create_wf_task() had to be pushed down into the function that creates the innermost task. Also, former keywords have been absorbed by the task dict. Refs #823 Refs #827 --- jobs/JGFS_ATMOS_GEMPAK_META | 2 +- workflow/rocoto/gefs_tasks.py | 71 +- workflow/rocoto/gfs_tasks.py | 1339 +++++++++++++++++++++++++++++---- workflow/rocoto/rocoto.py | 117 ++- workflow/rocoto/tasks.py | 38 +- 5 files changed, 1341 insertions(+), 226 deletions(-) diff --git a/jobs/JGFS_ATMOS_GEMPAK_META b/jobs/JGFS_ATMOS_GEMPAK_META index b7786b1f49..8e1c05763f 100755 --- a/jobs/JGFS_ATMOS_GEMPAK_META +++ b/jobs/JGFS_ATMOS_GEMPAK_META @@ -6,7 +6,7 @@ # GFS GEMPAK META PRODUCT GENERATION ############################################ source "${HOMEgfs}/ush/preamble.sh" -source "${HOMEgfs}/ush/jjob_header.sh" -e "gempak_meta" -e "base" +source "${HOMEgfs}/ush/jjob_header.sh" -e "gempak_meta" -c "base" ############################################### diff --git a/workflow/rocoto/gefs_tasks.py b/workflow/rocoto/gefs_tasks.py index 680c7d8686..c46d9ad452 100644 --- a/workflow/rocoto/gefs_tasks.py +++ b/workflow/rocoto/gefs_tasks.py @@ -1,5 +1,5 @@ from applications.applications import AppConfig -from rocoto.tasks import Tasks, create_wf_task +from rocoto.tasks import Tasks import rocoto.rocoto as rocoto @@ -57,41 +57,74 @@ def stage_ic(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('stage_ic') - task = create_wf_task('stage_ic', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'stage_ic' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': 'gefs', + 'command': f'{self.HOMEgfs}/jobs/rocoto/stage_ic.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + task = rocoto.create_task(task_dict) return task def waveinit(self): resources = self.get_resource('waveinit') - task = create_wf_task('waveinit', resources, cdump=self.cdump, envar=self.envars, dependency=None) + task_name = f'waveinit' + task_dict = {'task_name': task_name, + 'resources': resources, + 'envars': self.envars, + 'cycledef': 'gefs', + 'command': f'{self.HOMEgfs}/jobs/rocoto/waveinit.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + task = rocoto.create_task(task_dict) return task def fcst(self): + # TODO: Add real dependencies dependencies = [] - dep_dict = {'type': 'task', 'name': f'{self.cdump}stage_ic'} + dep_dict = {'type': 'task', 'name': f'stage_ic'} dependencies.append(rocoto.add_dependency(dep_dict)) if self.app_config.do_wave: - dep_dict = {'type': 'task', 'name': f'{self.cdump}waveinit'} + dep_dict = {'type': 'task', 'name': f'waveinit'} dependencies.append(rocoto.add_dependency(dep_dict)) dependencies = rocoto.create_dependency(dep_condition='and', dep=dependencies) resources = self.get_resource('fcst') - task = create_wf_task('fcst', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'fcst' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': 'gefs', + 'command': f'{self.HOMEgfs}/jobs/rocoto/fcst.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + task = rocoto.create_task(task_dict) return task def efcs(self): dependencies = [] - dep_dict = {'type': 'task', 'name': f'{self.cdump}stage_ic'} + dep_dict = {'type': 'task', 'name': f'stage_ic'} dependencies.append(rocoto.add_dependency(dep_dict)) if self.app_config.do_wave: - dep_dict = {'type': 'task', 'name': f'{self.cdump}waveinit'} + dep_dict = {'type': 'task', 'name': f'waveinit'} dependencies.append(rocoto.add_dependency(dep_dict)) dependencies = rocoto.create_dependency(dep_condition='and', dep=dependencies) @@ -100,9 +133,27 @@ def efcs(self): efcsenvars.append(rocoto.create_envar(name='ENSGRP', value='#grp#')) groups = self._get_hybgroups(self._base['NMEM_ENS'], self._configs['efcs']['NMEM_EFCSGRP']) + var_dict = {'grp': groups} resources = self.get_resource('efcs') - task = create_wf_task('efcs', resources, cdump=self.cdump, envar=efcsenvars, dependency=dependencies, - metatask='efmn', varname='grp', varval=groups, cycledef='gefs') + + task_name = f'efcs#grp#' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': efcsenvars, + 'cycledef': 'gefs', + 'command': f'{self.HOMEgfs}/jobs/rocoto/efcs.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + metatask_dict = {'task_name': 'efmn', + 'var_dict': var_dict, + 'task_dict': task_dict + } + + task = rocoto.create_task(metatask_dict) return task diff --git a/workflow/rocoto/gfs_tasks.py b/workflow/rocoto/gfs_tasks.py index 18208983b8..ba27dff207 100644 --- a/workflow/rocoto/gfs_tasks.py +++ b/workflow/rocoto/gfs_tasks.py @@ -1,5 +1,5 @@ from applications.applications import AppConfig -from rocoto.tasks import Tasks, create_wf_task +from rocoto.tasks import Tasks from wxflow import timedelta_to_HMS import rocoto.rocoto as rocoto import numpy as np @@ -71,7 +71,19 @@ def stage_ic(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('stage_ic') - task = create_wf_task('stage_ic', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}stage_ic' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump, + 'command': f'{self.HOMEgfs}/jobs/rocoto/stage_ic.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -102,8 +114,19 @@ def prep(self): cycledef = 'gdas' resources = self.get_resource('prep') - task = create_wf_task('prep', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies, - cycledef=cycledef) + task_name = f'{self.cdump}prep' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': cycledef, + 'command': f'{self.HOMEgfs}/jobs/rocoto/prep.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -111,7 +134,7 @@ def waveinit(self): resources = self.get_resource('waveinit') dependencies = None - cycledef = None + cycledef = 'gdas_half,gdas' if self.cdump in ['gdas'] else self.cdump if self.app_config.mode in ['cycled']: deps = [] dep_dict = {'type': 'task', 'name': f'{self.cdump}prep'} @@ -120,8 +143,20 @@ def waveinit(self): 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 - task = create_wf_task('waveinit', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies, cycledef=cycledef) + + task_name = f'{self.cdump}waveinit' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': cycledef, + 'command': f'{self.HOMEgfs}/jobs/rocoto/waveinit.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -133,7 +168,19 @@ def waveprep(self): dependencies = rocoto.create_dependency(dep=deps) cycledef = 'gdas_half,gdas' if self.cdump in ['gdas'] else self.cdump resources = self.get_resource('waveprep') - task = create_wf_task('waveprep', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies, cycledef=cycledef) + task_name = f'{self.cdump}waveprep' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': cycledef, + 'command': f'{self.HOMEgfs}/jobs/rocoto/waveprep.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -172,8 +219,19 @@ def aerosol_init(self): cycledef = 'gfs_seq' resources = self.get_resource('aerosol_init') - task = create_wf_task('aerosol_init', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies, - cycledef=cycledef) + task_name = f'{self.cdump}aerosol_init' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': cycledef, + 'command': f'{self.HOMEgfs}/jobs/rocoto/aerosol_init.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -189,7 +247,19 @@ def anal(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('anal') - task = create_wf_task('anal', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}anal' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/anal.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -209,7 +279,19 @@ def sfcanl(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('sfcanl') - task = create_wf_task('sfcanl', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}sfcanl' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/sfcanl.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -229,7 +311,19 @@ def analcalc(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('analcalc') - task = create_wf_task('analcalc', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}analcalc' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/analcalc.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -241,7 +335,19 @@ def analdiag(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('analdiag') - task = create_wf_task('analdiag', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}analdiag' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/analdiag.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -253,7 +359,19 @@ def prepatmiodaobs(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('prepatmiodaobs') - task = create_wf_task('prepatmiodaobs', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}prepatmiodaobs' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/prepatmiodaobs.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -277,8 +395,19 @@ def atmanlinit(self): cycledef = 'gdas' resources = self.get_resource('atmanlinit') - task = create_wf_task('atmanlinit', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies, - cycledef=cycledef) + task_name = f'{self.cdump}atmanlinit' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': cycledef, + 'command': f'{self.HOMEgfs}/jobs/rocoto/atmanlinit.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -290,7 +419,19 @@ def atmanlrun(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('atmanlrun') - task = create_wf_task('atmanlrun', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}atmanlrun' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/atmanlrun.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -302,7 +443,19 @@ def atmanlfinal(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('atmanlfinal') - task = create_wf_task('atmanlfinal', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}atmanlfinal' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/atmanlfinal.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -314,7 +467,20 @@ def aeroanlinit(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('aeroanlinit') - task = create_wf_task('aeroanlinit', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}aeroanlinit' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/aeroanlinit.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) + return task def aeroanlrun(self): @@ -325,7 +491,19 @@ def aeroanlrun(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('aeroanlrun') - task = create_wf_task('aeroanlrun', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}aeroanlrun' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/aeroanlrun.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -337,7 +515,19 @@ def aeroanlfinal(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('aeroanlfinal') - task = create_wf_task('aeroanlfinal', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}aeroanlfinal' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/aeroanlfinal.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -349,7 +539,19 @@ def preplandobs(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('preplandobs') - task = create_wf_task('preplandobs', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}preplandobs' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/preplandobs.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -361,7 +563,19 @@ def landanl(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('landanl') - task = create_wf_task('landanl', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}landanl' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/landanl.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task def ocnanalprep(self): @@ -375,11 +589,19 @@ def ocnanalprep(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('ocnanalprep') - task = create_wf_task('ocnanalprep', - resources, - cdump=self.cdump, - envar=self.envars, - dependency=dependencies) + task_name = f'{self.cdump}ocnanalprep' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/ocnanalprep.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -391,11 +613,19 @@ def ocnanalbmat(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('ocnanalbmat') - task = create_wf_task('ocnanalbmat', - resources, - cdump=self.cdump, - envar=self.envars, - dependency=dependencies) + task_name = f'{self.cdump}ocnanalbmat' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/ocnanalbmat.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -407,11 +637,19 @@ def ocnanalrun(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('ocnanalrun') - task = create_wf_task('ocnanalrun', - resources, - cdump=self.cdump, - envar=self.envars, - dependency=dependencies) + task_name = f'{self.cdump}ocnanlrun' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/ocnanlrun.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -427,11 +665,19 @@ def ocnanalchkpt(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('ocnanalchkpt') - task = create_wf_task('ocnanalchkpt', - resources, - cdump=self.cdump, - envar=self.envars, - dependency=dependencies) + task_name = f'{self.cdump}ocnanalchkpt' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/ocnanalchkpt.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -443,11 +689,19 @@ def ocnanalpost(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('ocnanalpost') - task = create_wf_task('ocnanalpost', - resources, - cdump=self.cdump, - envar=self.envars, - dependency=dependencies) + task_name = f'{self.cdump}ocnanalpost' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/ocnanalpost.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -459,11 +713,19 @@ def ocnanalvrfy(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('ocnanalvrfy') - task = create_wf_task('ocnanalvrfy', - resources, - cdump=self.cdump, - envar=self.envars, - dependency=dependencies) + task_name = f'{self.cdump}ocnanalvrfy' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/ocnanalvrfy.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -510,7 +772,19 @@ def _fcst_forecast_only(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=dependencies) resources = self.get_resource('fcst') - task = create_wf_task('fcst', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}fcst' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/fcst.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -547,8 +821,19 @@ def _fcst_cycled(self): cycledef = 'gdas_half,gdas' if self.cdump in ['gdas'] else self.cdump resources = self.get_resource('fcst') - task = create_wf_task('fcst', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies, - cycledef=cycledef) + task_name = f'{self.cdump}fcst' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': cycledef, + 'command': f'{self.HOMEgfs}/jobs/rocoto/fcst.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -572,8 +857,19 @@ def atmanlupp(self): deps.append(rocoto.add_dependency(dep_dict)) dependencies = rocoto.create_dependency(dep=deps, dep_condition='and') resources = self.get_resource('upp') - task = create_wf_task('atmanlupp', resources, cdump=self.cdump, envar=postenvars, dependency=dependencies, - cycledef=self.cdump, command='&JOBS_DIR;/upp.sh') + task_name = f'{self.cdump}atmanlupp' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': postenvars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/upp.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -590,8 +886,19 @@ def atmanlprod(self): deps.append(rocoto.add_dependency(dep_dict)) dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('atmos_products') - task = create_wf_task('atmanlprod', resources, cdump=self.cdump, envar=postenvars, dependency=dependencies, - cycledef=self.cdump, command='&JOBS_DIR;/atmos_products.sh') + task_name = f'{self.cdump}atmanlprod' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': postenvars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/atmos_products.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -632,7 +939,7 @@ def atmupp(self): varname1, varname2, varname3 = 'grp', 'dep', 'lst' varval1, varval2, varval3 = self._get_ufs_postproc_grps(self.cdump, self._configs['upp']) - vardict = {varname2: varval2, varname3: varval3} + var_dict = {varname1: varval1, varname2: varval2, varname3: varval3} postenvars = self.envars.copy() postenvar_dict = {'FHRLST': '#lst#', @@ -654,9 +961,25 @@ def atmupp(self): dependencies = rocoto.create_dependency(dep=deps, dep_condition='and') cycledef = 'gdas_half,gdas' if self.cdump in ['gdas'] else self.cdump resources = self.get_resource('upp') - task = create_wf_task('atmupp', resources, cdump=self.cdump, envar=postenvars, dependency=dependencies, - metatask='atmupp', varname=varname1, varval=varval1, vardict=vardict, cycledef=cycledef, - command='&JOBS_DIR;/upp.sh') + + task_name = f'{self.cdump}atmupp#{varname1}#' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': postenvars, + 'cycledef': cycledef, + 'command': f'{self.HOMEgfs}/jobs/rocoto/upp.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + metatask_dict = {'task_name': f'{self.cdump}atmupp', + 'task_dict': task_dict, + 'var_dict': var_dict + } + + task = rocoto.create_task(metatask_dict) return task @@ -664,7 +987,7 @@ def atmprod(self): varname1, varname2, varname3 = 'grp', 'dep', 'lst' varval1, varval2, varval3 = self._get_ufs_postproc_grps(self.cdump, self._configs['atmos_products']) - vardict = {varname2: varval2, varname3: varval3} + var_dict = {varname1: varval1, varname2: varval2, varname3: varval3} postenvars = self.envars.copy() postenvar_dict = {'FHRLST': '#lst#'} @@ -679,9 +1002,25 @@ def atmprod(self): dependencies = rocoto.create_dependency(dep=deps) cycledef = 'gdas_half,gdas' if self.cdump in ['gdas'] else self.cdump resources = self.get_resource('atmos_products') - task = create_wf_task('atmprod', resources, cdump=self.cdump, envar=postenvars, dependency=dependencies, - metatask='atmprod', varname=varname1, varval=varval1, vardict=vardict, cycledef=cycledef, - command='&JOBS_DIR;/atmos_products.sh') + + task_name = f'{self.cdump}atmprod#{varname1}#' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': postenvars, + 'cycledef': cycledef, + 'command': f'{self.HOMEgfs}/jobs/rocoto/atmos_products.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + metatask_dict = {'task_name': f'{self.cdump}atmprod', + 'task_dict': task_dict, + 'var_dict': var_dict + } + + task = rocoto.create_task(metatask_dict) return task @@ -689,11 +1028,11 @@ def ocnpost(self): varname1, varname2, varname3 = 'grp', 'dep', 'lst' varval1, varval2, varval3 = self._get_ufs_postproc_grps(self.cdump, self._configs['ocnpost']) - vardict = {varname2: varval2, varname3: varval3} + var_dict = {varname1: varval1, varname2: varval2, varname3: varval3} postenvars = self.envars.copy() postenvar_dict = {'FHRLST': '#lst#', - 'ROTDIR': self._base.get('ROTDIR')} + 'ROTDIR': self.rotdir} for key, value in postenvar_dict.items(): postenvars.append(rocoto.create_envar(name=key, value=str(value))) @@ -707,8 +1046,25 @@ def ocnpost(self): dependencies = rocoto.create_dependency(dep_condition='or', dep=deps) cycledef = 'gdas_half,gdas' if self.cdump in ['gdas'] else self.cdump resources = self.get_resource('ocnpost') - task = create_wf_task('ocnpost', resources, cdump=self.cdump, envar=postenvars, dependency=dependencies, - metatask='ocnpost', varname=varname1, varval=varval1, vardict=vardict, cycledef=cycledef) + + task_name = f'{self.cdump}ocnpost#{varname1}#' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': postenvars, + 'cycledef': cycledef, + 'command': f'{self.HOMEgfs}/jobs/rocoto/ocnpost.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + metatask_dict = {'task_name': f'{self.cdump}ocnpost', + 'task_dict': task_dict, + 'var_dict': var_dict + } + + task = rocoto.create_task(metatask_dict) return task @@ -722,7 +1078,19 @@ def wavepostsbs(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('wavepostsbs') - task = create_wf_task('wavepostsbs', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}wavepostsbs' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/wavepostsbs.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -733,7 +1101,19 @@ def wavepostbndpnt(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('wavepostbndpnt') - task = create_wf_task('wavepostbndpnt', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}wavepostbndpnt' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/wavepostbndpnt.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -746,8 +1126,19 @@ def wavepostbndpntbll(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('wavepostbndpntbll') - task = create_wf_task('wavepostbndpntbll', resources, cdump=self.cdump, envar=self.envars, - dependency=dependencies) + task_name = f'{self.cdump}wavepostbndpntbll' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/wavepostbndpntbll.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -761,7 +1152,19 @@ def wavepostpnt(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('wavepostpnt') - task = create_wf_task('wavepostpnt', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}wavepostpnt' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/wavepostpnt.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -772,7 +1175,19 @@ def wavegempak(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('wavegempak') - task = create_wf_task('wavegempak', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}wavegempak' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/wavegempak.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -785,7 +1200,19 @@ def waveawipsbulls(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('waveawipsbulls') - task = create_wf_task('waveawipsbulls', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}waveawipsbulls' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/waveawipsbulls.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -796,8 +1223,19 @@ def waveawipsgridded(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('waveawipsgridded') - task = create_wf_task('waveawipsgridded', resources, cdump=self.cdump, envar=self.envars, - dependency=dependencies) + task_name = f'{self.cdump}waveawipsgridded' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/waveawipsgridded.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -808,7 +1246,19 @@ def postsnd(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('postsnd') - task = create_wf_task('postsnd', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}postsnd' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/postsnd.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -828,13 +1278,24 @@ def fbwind(self): dependencies = rocoto.create_dependency(dep=deps, dep_condition='and') resources = self.get_resource('awips') - # TODO: It would be better to use task dependencies on the # individual post jobs rather than data dependencies to avoid # prematurely starting with partial files. Unfortunately, the # ability to "group" post would make this more convoluted than # it should be and not worth the complexity. - task = create_wf_task('fbwind', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}fbwind' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/fbwind.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -885,17 +1346,34 @@ def awips_20km_1p0deg(self): awipsenvars = self.envars.copy() awipsenvar_dict = {'FHRGRP': '#grp#', 'FHRLST': '#lst#', - 'ROTDIR': self._base.get('ROTDIR')} + 'ROTDIR': self.rotdir} for key, value in awipsenvar_dict.items(): awipsenvars.append(rocoto.create_envar(name=key, value=str(value))) varname1, varname2, varname3 = 'grp', 'dep', 'lst' varval1, varval2, varval3 = self._get_awipsgroups(self.cdump, self._configs['awips']) - vardict = {varname2: varval2, varname3: varval3} + var_dict = {varname1: varval1, varname2: varval2, varname3: varval3} resources = self.get_resource('awips') - task = create_wf_task('awips_20km_1p0deg', resources, cdump=self.cdump, envar=awipsenvars, dependency=dependencies, - metatask='awips_20km_1p0deg', varname=varname1, varval=varval1, vardict=vardict) + + task_name = f'{self.cdump}awips_20km_1p0deg#{varname1}#' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': awipsenvars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/awips_20km_1p0deg.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + metatask_dict = {'task_name': f'{self.cdump}awips_20km_1p0deg', + 'task_dict': task_dict, + 'var_dict': var_dict + } + + task = rocoto.create_task(metatask_dict) return task @@ -909,17 +1387,34 @@ def awips_g2(self): awipsenvars = self.envars.copy() awipsenvar_dict = {'FHRGRP': '#grp#', 'FHRLST': '#lst#', - 'ROTDIR': self._base.get('ROTDIR')} + 'ROTDIR': self.rotdir} for key, value in awipsenvar_dict.items(): awipsenvars.append(rocoto.create_envar(name=key, value=str(value))) varname1, varname2, varname3 = 'grp', 'dep', 'lst' varval1, varval2, varval3 = self._get_awipsgroups(self.cdump, self._configs['awips']) - vardict = {varname2: varval2, varname3: varval3} + var_dict = {varname1: varval1, varname2: varval2, varname3: varval3} resources = self.get_resource('awips') - task = create_wf_task('awips_g2', resources, cdump=self.cdump, envar=awipsenvars, dependency=dependencies, - metatask='awips_g2', varname=varname1, varval=varval1, vardict=vardict) + + task_name = f'{self.cdump}awips_g2#{varname1}#' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': awipsenvars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/awips_g2.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + metatask_dict = {'task_name': f'{self.cdump}awips_g2', + 'task_dict': task_dict, + 'var_dict': var_dict + } + + task = rocoto.create_task(metatask_dict) return task @@ -931,7 +1426,19 @@ def gempak(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('gempak') - task = create_wf_task('gempak', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}gempak' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/gempak.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -942,7 +1449,19 @@ def gempakmeta(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('gempak') - task = create_wf_task('gempakmeta', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}gempakmeta' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/gempakmeta.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -953,7 +1472,19 @@ def gempakmetancdc(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('gempak') - task = create_wf_task('gempakmetancdc', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}gempakmetancdc' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/gempakmetancdc.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -964,17 +1495,42 @@ def gempakncdcupapgif(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('gempak') - task = create_wf_task('gempakncdcupapgif', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}gempakncdcupapgif' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/gempakncdcupapgif.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task def gempakpgrb2spec(self): deps = [] dep_dict = {'type': 'task', 'name': f'{self.cdump}npoess_pgrb2_0p5deg'} + deps.append(rocoto.add_dependency(dep_dict)) dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('gempak') - task = create_wf_task('gempakpgrb2spec', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}gempakgrb2spec' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/gempakgrb2spec.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -986,7 +1542,19 @@ def npoess_pgrb2_0p5deg(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('npoess') - task = create_wf_task('npoess_pgrb2_0p5deg', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}npoess_pgrb2_0p5deg' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/npoess_pgrb2_0p5deg.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -997,7 +1565,19 @@ def verfozn(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('verfozn') - task = create_wf_task('verfozn', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}verfozn' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/verfozn.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1008,7 +1588,19 @@ def verfrad(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('verfrad') - task = create_wf_task('verfrad', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}verfrad' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/verfrad.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1019,7 +1611,19 @@ def vminmon(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('vminmon') - task = create_wf_task('vminmon', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}vminmon' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/vminmon.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1030,7 +1634,19 @@ def tracker(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('tracker') - task = create_wf_task('tracker', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}tracker' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/tracker.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1041,7 +1657,19 @@ def genesis(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('genesis') - task = create_wf_task('genesis', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}genesis' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/genesis.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1052,7 +1680,19 @@ def genesis_fsu(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('genesis_fsu') - task = create_wf_task('genesis_fsu', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}genesis_fsu' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/genesis_fsu.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1063,7 +1703,19 @@ def fit2obs(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('fit2obs') - task = create_wf_task('fit2obs', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}fit2obs' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/fit2obs.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1084,10 +1736,28 @@ def metp(self): varname1 = 'metpcase' varval1 = 'g2g1 g2o1 pcp1' + var_dict = {varname1: varval1} resources = self.get_resource('metp') - task = create_wf_task('metp', resources, cdump=self.cdump, envar=metpenvars, dependency=dependencies, - metatask='metp', varname=varname1, varval=varval1) + + task_name = f'{self.cdump}metp#{varname1}#' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': metpenvars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/metp.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + metatask_dict = {'task_name': f'{self.cdump}metp', + 'task_dict': task_dict, + 'var_dict': var_dict + } + + task = rocoto.create_task(metatask_dict) return task @@ -1098,7 +1768,19 @@ def mos_stn_prep(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('mos_stn_prep') - task = create_wf_task('mos_stn_prep', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}mos_stn_prep' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/mos_stn_prep.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1109,7 +1791,19 @@ def mos_grd_prep(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('mos_grd_prep') - task = create_wf_task('mos_grd_prep', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}mos_grd_prep' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/mos_grd_prep.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1120,7 +1814,19 @@ def mos_ext_stn_prep(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('mos_ext_stn_prep') - task = create_wf_task('mos_ext_stn_prep', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}mos_ext_stn_prep' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/mos_ext_stn_prep.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1131,7 +1837,19 @@ def mos_ext_grd_prep(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('mos_ext_grd_prep') - task = create_wf_task('mos_ext_grd_prep', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}mos_ext_grd_prep' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/mos_ext_grd_prep.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1142,7 +1860,19 @@ def mos_stn_fcst(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('mos_stn_fcst') - task = create_wf_task('mos_stn_fcst', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}mos_stn_fcst' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/mos_stn_fcst.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1156,7 +1886,19 @@ def mos_grd_fcst(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('mos_grd_fcst') - task = create_wf_task('mos_grd_fcst', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}mos_grd_fcst' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/mos_grd_fcst.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1170,7 +1912,19 @@ def mos_ext_stn_fcst(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('mos_ext_stn_fcst') - task = create_wf_task('mos_ext_stn_fcst', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}mos_ext_stn_fcst' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/mos_ext_stn_fcst.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1187,7 +1941,19 @@ def mos_ext_grd_fcst(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('mos_ext_grd_fcst') - task = create_wf_task('mos_ext_grd_fcst', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}mos_ext_grd_fcst' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/mos_ext_grd_fcst.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1198,7 +1964,19 @@ def mos_stn_prdgen(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('mos_stn_prdgen') - task = create_wf_task('mos_stn_prdgen', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}mos_stn_prdgen' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/mos_stn_prdgen.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1212,7 +1990,19 @@ def mos_grd_prdgen(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('mos_grd_prdgen') - task = create_wf_task('mos_grd_prdgen', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}mos_grd_prdgen' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/mos_grd_prdgen.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1226,7 +2016,19 @@ def mos_ext_stn_prdgen(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('mos_ext_stn_prdgen') - task = create_wf_task('mos_ext_stn_prdgen', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}mos_ext_stn_prdgen' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/mos_ext_stn_prdgen.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1243,7 +2045,19 @@ def mos_ext_grd_prdgen(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('mos_ext_grd_prdgen') - task = create_wf_task('mos_ext_grd_prdgen', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}mos_ext_grd_prdgen' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/mos_ext_grd_prdgen.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1254,7 +2068,19 @@ def mos_wx_prdgen(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('mos_wx_prdgen') - task = create_wf_task('mos_wx_prdgen', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}mos_wx_prdgen' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/mos_wx_prdgen.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1268,7 +2094,19 @@ def mos_wx_ext_prdgen(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('mos_wx_ext_prdgen') - task = create_wf_task('mos_wx_ext_prdgen', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}mos_wx_ext_prdgen' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/mos_wx_ext_prdgen.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1341,8 +2179,19 @@ def arch(self): cycledef = 'gdas_half,gdas' if self.cdump in ['gdas'] else self.cdump resources = self.get_resource('arch') - task = create_wf_task('arch', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies, - cycledef=cycledef) + task_name = f'{self.cdump}arch' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': cycledef, + '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;' + } + + task = rocoto.create_task(task_dict) return task @@ -1359,7 +2208,19 @@ def cleanup(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('cleanup') - task = create_wf_task('cleanup', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}cleanup' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/cleanup.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1373,7 +2234,19 @@ def eobs(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('eobs') - task = create_wf_task('eobs', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}eobs' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/eobs.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1388,9 +2261,27 @@ def eomg(self): groups = self._get_hybgroups(self._base['NMEM_ENS'], self._configs['eobs']['NMEM_EOMGGRP']) + var_dict = {'grp': groups} + resources = self.get_resource('eomg') - task = create_wf_task('eomg', resources, cdump=self.cdump, envar=eomgenvars, dependency=dependencies, - metatask='eomn', varname='grp', varval=groups) + task_name = f'{self.cdump}eomg#grp#' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': eomgenvars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/eomg.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + metatask_dict = {'task_name': f'{self.cdump}eomn', + 'var_dict': var_dict, + 'task_dict': task_dict, + } + + task = rocoto.create_task(metatask_dict) return task @@ -1401,7 +2292,19 @@ def ediag(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('ediag') - task = create_wf_task('ediag', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}ediag' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/ediag.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1415,7 +2318,19 @@ def eupd(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('eupd') - task = create_wf_task('eupd', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}eupd' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/eupd.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1429,8 +2344,19 @@ def atmensanlinit(self): cycledef = "gdas" resources = self.get_resource('atmensanlinit') - task = create_wf_task('atmensanlinit', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies, - cycledef=cycledef) + task_name = f'{self.cdump}atmensanlinit' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': cycledef, + 'command': f'{self.HOMEgfs}/jobs/rocoto/atmensanlinit.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1444,7 +2370,19 @@ def atmensanlrun(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('atmensanlrun') - task = create_wf_task('atmensanlrun', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}atmensanlrun' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/atmensanlrun.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1456,7 +2394,19 @@ def atmensanlfinal(self): dependencies = rocoto.create_dependency(dep=deps) resources = self.get_resource('atmensanlfinal') - task = create_wf_task('atmensanlfinal', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}atmensanlfinal' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/atmensanlfinal.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1503,11 +2453,28 @@ def _get_ecengroups(): varname1, varname2, varname3 = 'grp', 'dep', 'lst' varval1, varval2, varval3 = _get_ecengroups() - vardict = {varname2: varval2, varname3: varval3} + var_dict = {varname1: varval1, varname2: varval2, varname3: varval3} resources = self.get_resource('ecen') - task = create_wf_task('ecen', resources, cdump=self.cdump, envar=ecenenvars, dependency=dependencies, - metatask='ecmn', varname=varname1, varval=varval1, vardict=vardict) + + task_name = f'{self.cdump}ecen#{varname1}#' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': ecenenvars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/ecen.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + metatask_dict = {'task_name': f'{self.cdump}ecmn', + 'var_dict': var_dict, + 'task_dict': task_dict + } + + task = rocoto.create_task(metatask_dict) return task def esfc(self): @@ -1525,7 +2492,19 @@ def esfc(self): dependencies = rocoto.create_dependency(dep_condition='and', dep=deps) resources = self.get_resource('esfc') - task = create_wf_task('esfc', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies) + task_name = f'{self.cdump}esfc' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': self.cdump.replace('enkf', ''), + 'command': f'{self.HOMEgfs}/jobs/rocoto/esfc.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1550,8 +2529,27 @@ def efcs(self): groups = self._get_hybgroups(self._base['NMEM_ENS_GFS'], self._configs['efcs']['NMEM_EFCSGRP_GFS']) cycledef = 'gdas_half,gdas' if self.cdump in ['enkfgdas'] else self.cdump.replace('enkf', '') resources = self.get_resource('efcs') - task = create_wf_task('efcs', resources, cdump=self.cdump, envar=efcsenvars, dependency=dependencies, - metatask='efmn', varname='grp', varval=groups, cycledef=cycledef) + + var_dict = {'grp': groups} + + task_name = f'{self.cdump}efcs#grp#' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': efcsenvars, + 'cycledef': cycledef, + 'command': f'{self.HOMEgfs}/jobs/rocoto/efcs.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + metatask_dict = {'task_name': f'{self.cdump}efmn', + 'var_dict': var_dict, + 'task_dict': task_dict + } + + task = rocoto.create_task(metatask_dict) return task @@ -1569,8 +2567,19 @@ def echgres(self): cycledef = 'gdas_half,gdas' if self.cdump in ['enkfgdas'] else self.cdump resources = self.get_resource('echgres') - task = create_wf_task('echgres', resources, cdump=self.cdump, envar=self.envars, dependency=dependencies, - cycledef=cycledef) + task_name = f'{self.cdump}echgres' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': self.envars, + 'cycledef': cycledef, + 'command': f'{self.HOMEgfs}/jobs/rocoto/echgres.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + task = rocoto.create_task(task_dict) return task @@ -1611,13 +2620,30 @@ def _get_eposgroups(epos): varname1, varname2, varname3 = 'grp', 'dep', 'lst' varval1, varval2, varval3 = _get_eposgroups(self._configs['epos']) - vardict = {varname2: varval2, varname3: varval3} + var_dict = {varname1: varval1, varname2: varval2, varname3: varval3} cycledef = 'gdas_half,gdas' if self.cdump in ['enkfgdas'] else self.cdump.replace('enkf', '') resources = self.get_resource('epos') - task = create_wf_task('epos', resources, cdump=self.cdump, envar=eposenvars, dependency=dependencies, - metatask='epmn', varname=varname1, varval=varval1, vardict=vardict, cycledef=cycledef) + + task_name = f'{self.cdump}epos#{varname1}#' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': eposenvars, + 'cycledef': cycledef, + 'command': f'{self.HOMEgfs}/jobs/rocoto/epos.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + metatask_dict = {'task_name': f'{self.cdump}epmn', + 'var_dict': var_dict, + 'task_dict': task_dict + } + + task = rocoto.create_task(metatask_dict) return task @@ -1636,7 +2662,26 @@ def earc(self): cycledef = 'gdas_half,gdas' if self.cdump in ['enkfgdas'] else self.cdump.replace('enkf', '') resources = self.get_resource('earc') - task = create_wf_task('earc', resources, cdump=self.cdump, envar=earcenvars, dependency=dependencies, - metatask='eamn', varname='grp', varval=groups, cycledef=cycledef) + + var_dict = {'grp': groups} + + task_name = f'{self.cdump}earc#grp#' + task_dict = {'task_name': task_name, + 'resources': resources, + 'dependency': dependencies, + 'envars': earcenvars, + 'cycledef': cycledef, + 'command': f'{self.HOMEgfs}/jobs/rocoto/earc.sh', + 'job_name': f'{self.pslot}_{task_name}_@H', + 'log': f'{self.rotdir}/logs/@Y@m@d@H/{task_name}.log', + 'maxtries': '&MAXTRIES;' + } + + metatask_dict = {'task_name': f'{self.cdump}eamn', + 'var_dict': var_dict, + 'task_dict': task_dict + } + + task = rocoto.create_task(metatask_dict) return task diff --git a/workflow/rocoto/rocoto.py b/workflow/rocoto/rocoto.py index c4240622d4..679c0952ed 100644 --- a/workflow/rocoto/rocoto.py +++ b/workflow/rocoto/rocoto.py @@ -10,62 +10,109 @@ Helper module to create tasks, metatasks, and dependencies for Rocoto ''' -__all__ = ['create_task', 'create_metatask', +__all__ = ['create_task', 'add_dependency', 'create_dependency', 'create_envar', 'create_entity', 'create_cycledef'] -def create_metatask(task_dict: Dict[str, Any], metatask_dict: Dict[str, Any]) -> List[str]: +def create_task(task_dict: Dict[str, Any]) -> List[str]: """ - create a Rocoto metatask given a dictionary containing task and metatask information - :param metatask_dict: metatask key-value parameters - :type metatask_dict: dict - :param task_dict: task key-value parameters - :type task_dict: dict - :return: Rocoto metatask - :rtype: list + Create XML for a rocoto task or metatask + + Creates the XML required to define a task and returns the lines + as a list of strings. Tasks can be nested to create metatasks by + defining a key 'task_dict' within the task_dict. When including + a nested task, you also need to provide a 'var_dict' key that + contains a dictionary of variables to loop over. + + All task dicts must include a 'task_name'. + + Innermost tasks (regular tasks) additionally require a 'resources' + key containing a dict of values defining the HPC settings. + + Parameters + ---------- + task_dict: dict + Dictionary of task definitions + + Returns + ------- + str + Strings containing the XML code defining the task + + Raises + ------ + KeyError + If a required key is missing + """ - # Grab metatask info from the metatask_dict - metataskname = metatask_dict.get('metataskname', 'demometatask') - varname = metatask_dict.get('varname', 'demovar') - varval = metatask_dict.get('varval', 1) - vardict = metatask_dict.get('vardict', None) + inner_task_dict = task_dict.pop('task_dict', None) - strings = [f'\n', - '\n', - f'\t{str(varval)}\n'] + if inner_task_dict is None: + strings = _create_innermost_task(task_dict) + + else: + # There is a nested task_dict, so this is a metatask + metataskname = f"{task_dict.get('task_name', 'demometatask')}" + var_dict = task_dict.get('var_dict', None) - if vardict is not None: - for key in vardict.keys(): - value = str(vardict[key]) + strings = [f'\n', + '\n'] + + if var_dict is None: + msg = f'Task {metataskname} has a nested task dict, but has no var_dict' + raise KeyError(msg) + + for key in var_dict.keys(): + value = str(var_dict[key]) strings.append(f'\t{value}\n') - strings.append('\n') - tasklines = create_task(task_dict) - for tl in tasklines: - strings.append(f'{tl}') if tl == '\n' else strings.append(f'\t{tl}') - strings.append('\n') - strings.append('\n') - return strings + strings.append('\n') + task_dict.update(inner_task_dict) + tasklines = create_task(task_dict).splitlines(True) + for tl in tasklines: + strings.append(f'{tl}') if tl == '\n' else strings.append(f'\t{tl}') + strings.append('\n') + strings.append('\n') + return ''.join(strings) -def create_task(task_dict: Dict[str, Any]) -> List[str]: + +def _create_innermost_task(task_dict: Dict[str, Any]) -> List[str]: """ - create a Rocoto task given a dictionary containing task information - :param task_dict: task key-value parameters - :type task_dict: dict - :return: Rocoto task - :rtype: list + Create XML for a regular rocoto task + + Creates the XML required to define a task and returns the lines + as a list of strings. + + All task dicts must include a 'task_name' and a 'resources' + key containing a dict of values defining the HPC settings. + + Parameters + ---------- + task_dict: dict + Dictionary of task definitions + + Returns + ------- + List[str] + List of strings containing the XML code defining the task + + Raises + ------ + KeyError + If a required key is missing + """ # Grab task info from the task_names - taskname = task_dict.get('taskname', 'demotask') + taskname = task_dict.get('task_name', 'demotask') cycledef = task_dict.get('cycledef', 'democycle') maxtries = task_dict.get('maxtries', 3) final = task_dict.get('final', False) command = task_dict.get('command', 'sleep 10') - jobname = task_dict.get('jobname', 'demojob') + jobname = task_dict.get('job_name', 'demojob') resources_dict = task_dict['resources'] account = resources_dict.get('account', 'batch') queue = resources_dict.get('queue', 'debug') diff --git a/workflow/rocoto/tasks.py b/workflow/rocoto/tasks.py index 2d44c00d4d..dee86d0d04 100644 --- a/workflow/rocoto/tasks.py +++ b/workflow/rocoto/tasks.py @@ -5,7 +5,7 @@ import rocoto.rocoto as rocoto from wxflow import Template, TemplateConstants, to_timedelta -__all__ = ['Tasks', 'create_wf_task'] +__all__ = ['Tasks'] class Tasks: @@ -42,12 +42,15 @@ 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.HOMEgfs = self._base['HOMEgfs'] + self.rotdir = self._base['ROTDIR'] + self.pslot = self._base['PSLOT'] self._base['cycle_interval'] = to_timedelta(f'{self._base["assim_freq"]}H') self.n_tiles = 6 # TODO - this needs to be elsewhere envar_dict = {'RUN_ENVIR': self._base.get('RUN_ENVIR', 'emc'), - 'HOMEgfs': self._base.get('HOMEgfs'), + 'HOMEgfs': self.HOMEgfs, 'EXPDIR': self._base.get('EXPDIR'), 'NET': self._base.get('NET'), 'CDUMP': self.cdump, @@ -197,34 +200,3 @@ def get_task(self, task_name, *args, **kwargs): raise AttributeError(f'"{task_name}" is not a valid task.\n' + 'Valid tasks are:\n' + f'{", ".join(Tasks.VALID_TASKS)}') - - -def create_wf_task(task_name, resources, - cdump='gdas', cycledef=None, envar=None, dependency=None, - metatask=None, varname=None, varval=None, vardict=None, - final=False, command=None): - tasknamestr = f'{cdump}{task_name}' - metatask_dict = None - if metatask is not None: - tasknamestr = f'{tasknamestr}#{varname}#' - metatask_dict = {'metataskname': f'{cdump}{metatask}', - 'varname': f'{varname}', - 'varval': f'{varval}', - 'vardict': vardict} - - cycledefstr = cdump.replace('enkf', '') if cycledef is None else cycledef - - task_dict = {'taskname': f'{tasknamestr}', - 'cycledef': f'{cycledefstr}', - 'maxtries': '&MAXTRIES;', - 'command': f'&JOBS_DIR;/{task_name}.sh' if command is None else command, - 'jobname': f'&PSLOT;_{tasknamestr}_@H', - 'resources': resources, - 'log': f'&ROTDIR;/logs/@Y@m@d@H/{tasknamestr}.log', - 'envars': envar, - 'dependency': dependency, - 'final': final} - - task = rocoto.create_task(task_dict) if metatask is None else rocoto.create_metatask(task_dict, metatask_dict) - - return ''.join(task)