forked from NOAA-EMC/global-workflow
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Eliminate post groups (NOAA-EMC#2667)
Eliminates the post groups used for upp and products jobs so that each task only processes one forecast hour. This is more efficient and greatly simplifies downstream dependencies that depend on a specific forecast hour. Resolves NOAA-EMC#2666 Refs NOAA-EMC#2642
- Loading branch information
1 parent
0b810c8
commit 8993b42
Showing
9 changed files
with
56 additions
and
101 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -939,7 +939,7 @@ def _fcst_cycled(self): | |
|
|
||
| def atmanlupp(self): | ||
| postenvars = self.envars.copy() | ||
| postenvar_dict = {'FHRLST': 'f000', | ||
| postenvar_dict = {'FHR3': '000', | ||
| 'UPP_RUN': 'analysis'} | ||
| for key, value in postenvar_dict.items(): | ||
| postenvars.append(rocoto.create_envar(name=key, value=str(value))) | ||
|
|
@@ -975,7 +975,7 @@ def atmanlupp(self): | |
|
|
||
| def atmanlprod(self): | ||
| postenvars = self.envars.copy() | ||
| postenvar_dict = {'FHRLST': '-f001'} | ||
| postenvar_dict = {'FHR3': '-001'} | ||
| for key, value in postenvar_dict.items(): | ||
| postenvars.append(rocoto.create_envar(name=key, value=str(value))) | ||
|
|
||
|
|
@@ -1002,24 +1002,6 @@ def atmanlprod(self): | |
|
|
||
| return task | ||
|
|
||
| @staticmethod | ||
| def _get_ufs_postproc_grps(cdump, config, component='atmos'): | ||
|
|
||
| fhrs = Tasks._get_forecast_hours(cdump, config, component=component) | ||
|
|
||
| nfhrs_per_grp = config.get('NFHRS_PER_GROUP', 1) | ||
| ngrps = len(fhrs) // nfhrs_per_grp if len(fhrs) % nfhrs_per_grp == 0 else len(fhrs) // nfhrs_per_grp + 1 | ||
|
|
||
| fhrs = [f'f{fhr:03d}' for fhr in fhrs] | ||
| fhrs = np.array_split(fhrs, ngrps) | ||
| fhrs = [fhr.tolist() for fhr in fhrs] | ||
|
|
||
| grp = ' '.join(f'_{fhr[0]}-{fhr[-1]}' if len(fhr) > 1 else f'_{fhr[0]}' for fhr in fhrs) | ||
| dep = ' '.join([fhr[-1] for fhr in fhrs]) | ||
| lst = ' '.join(['_'.join(fhr) for fhr in fhrs]) | ||
|
|
||
| return grp, dep, lst | ||
|
|
||
| def atmupp(self): | ||
| return self._upptask(upp_run='forecast', task_id='atmupp') | ||
|
|
||
|
|
@@ -1032,32 +1014,28 @@ def _upptask(self, upp_run="forecast", task_id="atmupp"): | |
| if upp_run not in VALID_UPP_RUN: | ||
| raise KeyError(f"{upp_run} is invalid; UPP_RUN options are: {('|').join(VALID_UPP_RUN)}") | ||
|
|
||
| varname1, varname2, varname3 = 'grp', 'dep', 'lst' | ||
| varval1, varval2, varval3 = self._get_ufs_postproc_grps(self.cdump, self._configs['upp']) | ||
| var_dict = {varname1: varval1, varname2: varval2, varname3: varval3} | ||
|
|
||
| postenvars = self.envars.copy() | ||
| postenvar_dict = {'FHRLST': '#lst#', | ||
| postenvar_dict = {'FHR3': '#fhr#', | ||
| 'UPP_RUN': upp_run} | ||
| for key, value in postenvar_dict.items(): | ||
| postenvars.append(rocoto.create_envar(name=key, value=str(value))) | ||
|
|
||
| atm_hist_path = self._template_to_rocoto_cycstring(self._base["COM_ATMOS_HISTORY_TMPL"]) | ||
| deps = [] | ||
| data = f'{atm_hist_path}/{self.cdump}.t@Hz.atm#dep#.nc' | ||
| data = f'{atm_hist_path}/{self.cdump}.t@Hz.atmf#fhr#.nc' | ||
| dep_dict = {'type': 'data', 'data': data, 'age': 120} | ||
| deps.append(rocoto.add_dependency(dep_dict)) | ||
| data = f'{atm_hist_path}/{self.cdump}.t@Hz.sfc#dep#.nc' | ||
| data = f'{atm_hist_path}/{self.cdump}.t@Hz.sfcf#fhr#.nc' | ||
| dep_dict = {'type': 'data', 'data': data, 'age': 120} | ||
| deps.append(rocoto.add_dependency(dep_dict)) | ||
| data = f'{atm_hist_path}/{self.cdump}[email protected].log#dep#.txt' | ||
| data = f'{atm_hist_path}/{self.cdump}[email protected].logf#fhr#.txt' | ||
| dep_dict = {'type': 'data', 'data': data, 'age': 60} | ||
| deps.append(rocoto.add_dependency(dep_dict)) | ||
| 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_name = f'{self.cdump}{task_id}#{varname1}#' | ||
| task_name = f'{self.cdump}{task_id}_f#fhr#' | ||
| task_dict = {'task_name': task_name, | ||
| 'resources': resources, | ||
| 'dependency': dependencies, | ||
|
|
@@ -1069,9 +1047,12 @@ def _upptask(self, upp_run="forecast", task_id="atmupp"): | |
| 'maxtries': '&MAXTRIES;' | ||
| } | ||
|
|
||
| fhrs = self._get_forecast_hours(self.cdump, self._configs['upp']) | ||
| fhr_var_dict = {'fhr': ' '.join([f"{fhr:03d}" for fhr in fhrs])} | ||
|
|
||
| metatask_dict = {'task_name': f'{self.cdump}{task_id}', | ||
| 'task_dict': task_dict, | ||
| 'var_dict': var_dict | ||
| 'var_dict': fhr_var_dict | ||
| } | ||
|
|
||
| task = rocoto.create_task(metatask_dict) | ||
|
|
@@ -1091,25 +1072,21 @@ def _atmosoceaniceprod(self, component: str): | |
|
|
||
| products_dict = {'atmos': {'config': 'atmos_products', | ||
| 'history_path_tmpl': 'COM_ATMOS_MASTER_TMPL', | ||
| 'history_file_tmpl': f'{self.cdump}[email protected].grb2#dep#'}, | ||
| 'history_file_tmpl': f'{self.cdump}[email protected].grb2f#fhr#'}, | ||
| 'ocean': {'config': 'oceanice_products', | ||
| 'history_path_tmpl': 'COM_OCEAN_HISTORY_TMPL', | ||
| 'history_file_tmpl': f'{self.cdump}[email protected]_avg.#dep#.nc'}, | ||
| 'history_file_tmpl': f'{self.cdump}[email protected]_avg.f#fhr#.nc'}, | ||
| 'ice': {'config': 'oceanice_products', | ||
| 'history_path_tmpl': 'COM_ICE_HISTORY_TMPL', | ||
| 'history_file_tmpl': f'{self.cdump}[email protected]_avg.#dep#.nc'}} | ||
| 'history_file_tmpl': f'{self.cdump}[email protected]_avg.f#fhr#.nc'}} | ||
|
|
||
| component_dict = products_dict[component] | ||
| config = component_dict['config'] | ||
| history_path_tmpl = component_dict['history_path_tmpl'] | ||
| history_file_tmpl = component_dict['history_file_tmpl'] | ||
|
|
||
| varname1, varname2, varname3 = 'grp', 'dep', 'lst' | ||
| varval1, varval2, varval3 = self._get_ufs_postproc_grps(self.cdump, self._configs[config], component=component) | ||
| var_dict = {varname1: varval1, varname2: varval2, varname3: varval3} | ||
|
|
||
| postenvars = self.envars.copy() | ||
| postenvar_dict = {'FHRLST': '#lst#', 'COMPONENT': component} | ||
| postenvar_dict = {'FHR3': '#fhr#', 'COMPONENT': component} | ||
| for key, value in postenvar_dict.items(): | ||
| postenvars.append(rocoto.create_envar(name=key, value=str(value))) | ||
|
|
||
|
|
@@ -1129,7 +1106,7 @@ def _atmosoceaniceprod(self, component: str): | |
| cycledef = 'gdas_half,gdas' if self.cdump in ['gdas'] else self.cdump | ||
| resources = self.get_resource(component_dict['config']) | ||
|
|
||
| task_name = f'{self.cdump}{component}_prod#{varname1}#' | ||
| task_name = f'{self.cdump}{component}_prod_f#fhr#' | ||
| task_dict = {'task_name': task_name, | ||
| 'resources': resources, | ||
| 'dependency': dependencies, | ||
|
|
@@ -1141,9 +1118,11 @@ def _atmosoceaniceprod(self, component: str): | |
| 'maxtries': '&MAXTRIES;' | ||
| } | ||
|
|
||
| fhrs = self._get_forecast_hours(self.cdump, self._configs[config], component) | ||
| fhr_var_dict = {'fhr': ' '.join([f"{fhr:03d}" for fhr in fhrs])} | ||
| metatask_dict = {'task_name': f'{self.cdump}{component}_prod', | ||
| 'task_dict': task_dict, | ||
| 'var_dict': var_dict | ||
| 'var_dict': fhr_var_dict | ||
| } | ||
|
|
||
| task = rocoto.create_task(metatask_dict) | ||
|
|
||