Skip to content

Commit d495b93

Browse files
ceblantonceblanton
authored
Merge pull request #50 from NOAA-GFDL/48.wf-script-edits
STATIC FILES: Read combined yaml directly for static source information
2 parents 6a26a90 + fc5518b commit d495b93

File tree

9 files changed

+460
-362
lines changed

9 files changed

+460
-362
lines changed

‎Jinja2Filters/form_remap_dep.py‎

100644100755
Lines changed: 112 additions & 87 deletions
Original file line numberDiff line numberDiff line change
@@ -1,41 +1,56 @@
1-
import re
2-
import os
3-
import metomi.rose.config
4-
import ast
5-
"""! form_remap_dep parses the remap_pp_components rose-app.conf and uses input from rose-suite.conf in the form of
6-
env variables and returns the pp component and source name dependencies for remap_pp_components task execution. For
7-
instance, for an atmos PP component that requires the regridded atmos_month and regridded atmos_daily history
8-
files, this JinjaFilter when called within flow.cylc helps identify this dependency to complete the corresponding
9-
task graph. This JinjaFilter ensures a remap-pp-component only waits for the dependent make-timeseries tasks such
10-
that the succeeded components output are made available in the final destination.
1+
"""
2+
form_remap_dep:
3+
- parses the remap_pp_components rose-app.conf
4+
- uses input from rose-suite.conf in the form of env variables
5+
- returns the pp component and source name dependencies for
6+
remap_pp_components task execution.
7+
8+
For instance, for an atmos PP component that requires the regridded
9+
atmos_month and regridded atmos_daily history files, this JinjaFilter
10+
when called within flow.cylc helps identify this dependency to
11+
complete the corresponding task graph.
12+
13+
This JinjaFilter ensures a remap-pp-component only waits for the
14+
dependent make-timeseries tasks such that the succeeded components
15+
output are made available in the final destination.
16+
1117
See form_remap_dep invocations from flow.cylc
1218
"""
13-
# @file form_remap_dep.py
14-
# Author(s)
15-
# Created by A.Radhakrishnan on 06/27/2022
16-
# Credit MSD workflow team
17-
1819
# Function parameter type hints, PEP 484
19-
20-
def form_remap_dep(grid_type: str, temporal_type: str, chunk: str, pp_components_str: str, output_type: str, history_segment: str=None) -> str:
2120

22-
""" Form the task parameter list based on the grid type, the temporal type, and the desired pp component(s)
21+
import os
22+
from pathlib import Path
23+
import yaml
24+
25+
def form_remap_dep(grid_type: str,
26+
temporal_type: str,
27+
chunk: str,
28+
pp_components_str: str,
29+
output_type: str,
30+
yamlfile: str,
31+
history_segment: str=None) -> str:
32+
"""
33+
Form the task parameter list based on the grid type,
34+
the temporal type, and the desired pp component(s)
2335
2436
Arguments:
2537
@param grid_type (str): One of: native or regrid-xy
2638
@param temporal_type (str): One of: temporal or static
2739
@param chunk (str): e.g P5Y for 5-year time series
2840
@param pp_component (str): all, or a space-separated list
2941
@param output_type (str): ts or av
30-
@param history_segment (str): if given, handle special case where history segment equals pp-chunk-a
42+
@param yamlfile (str): yaml configuration file passed through workflow
43+
@param history_segment (str): if given, handle special case where history
44+
segment equals pp-chunk-a
3145
32-
@return remap_dep (multiline str) with Jinja formatting listing source-pp dependencies
46+
@return remap_dep (multiline str) with Jinja formatting listing source-pp
47+
dependencies
3348
"""
3449
pp_components = pp_components_str.split(' ')
35-
if(grid_type == "regrid-xy"):
36-
grid = "regrid"
50+
if grid_type == "regrid-xy":
51+
grid = "regrid"
3752
else:
38-
grid = grid_type
53+
grid = grid_type
3954

4055
# Determine the task needed to run before remap-pp-components
4156
# Note: history_segment should be specified for primary chunk generation,
@@ -51,94 +66,104 @@ def form_remap_dep(grid_type: str, temporal_type: str, chunk: str, pp_components
5166
raise Exception("output type not supported")
5267

5368
#print(pp_components)
54-
#print(chunk)
69+
#print(chunk)
5570
########################
5671
dict_group_source={}
57-
remap_comp = None
72+
remap_comp = None
5873
#print("DEBUG: Passed args ",grid_type, temporal_type, chunk, pp_components_str)
59-
remap_dep = ""
74+
remap_dep = ""
6075
#print("DEBUG: desired pp components:", pp_components)
61-
path_to_conf = os.path.dirname(os.path.abspath(__file__)) + '/../app/remap-pp-components/rose-app.conf'
62-
node = metomi.rose.config.load(path_to_conf)
63-
results = []
76+
77+
# Path to yaml configuration
78+
exp_dir = Path(__file__).resolve().parents[1]
79+
path_to_yamlconfig = os.path.join(exp_dir, yamlfile)
80+
81+
# Load and read yaml configuration
82+
with open(path_to_yamlconfig,'r') as yml:
83+
yml_info = yaml.safe_load(yml)
84+
6485
makets_stmt = ""
65-
regex_pp_comp = re.compile('^\w+')
66-
for keys, sub_node in node.walk():
67-
# only target the keys
68-
if len(keys) != 1:
69-
continue
86+
# Loop through pp components; check components passd in script are defined in yaml config
87+
for comp_info in yml_info["postprocess"]["components"]:
88+
comp = comp_info.get("type")
7089

71-
# skip env and command keys
72-
item = keys[0]
73-
if item == "env" or item == "command":
74-
continue
75-
comp = regex_pp_comp.match(item).group()
7690
#print("DEBUG: Examining", item, comp)
77-
#res = [pp_comp for pp_comp in pp_components if(pp_comp in comp)]
7891
if comp not in pp_components:
79-
#print(comp, " not in", pp_components)
80-
continue
81-
#print("DEBUG: Examining", item, comp)
92+
#print(comp, " not in", pp_components)
93+
continue
8294

8395
# skip if grid type is not desired
84-
# some grid types (i.e. regrid-xy) have subtypes (i.e. 1deg, 2deg)
85-
# in remap-pp-components/rose-app.conf the grid type and subgrid is specified as "regrid-xy/1deg" (e.g.).
86-
# So we will strip off after the slash and the remainder is the grid type
87-
candidate_grid_type = re.sub('\/.*', '', node.get_value(keys=[item, 'grid']))
96+
# Set grid type if component has xyInterp defined or not
97+
if "xyInterp" not in comp_info.keys():
98+
candidate_grid_type = "native"
99+
else:
100+
candidate_grid_type = "regrid-xy"
101+
88102
if candidate_grid_type != grid_type:
89103
#print("DEBUG: Skipping as not right grid; got", candidate_grid_type, "and wanted", grid_type)
90104
continue
91105

92-
# skip if temporal type is not desired
93-
# freq is optional, so if it does not exist then continue on
94-
freq = node.get_value(keys=[item, 'freq'])
106+
##########NOT SURE YET
107+
# # skip if temporal type is not desired
108+
# # freq is optional, so if it does not exist then continue on
109+
# freq = node.get_value(keys=[item, 'freq'])
110+
freq = comp_info.get("freq")
95111
if temporal_type == "static":
96-
if freq and 'P0Y' not in freq:
112+
if freq is not None and 'P0Y' not in freq:
97113
#print("DEBUG: Skipping as static is requested, no P0Y here", freq)
98114
continue
99-
elif (temporal_type == "temporal"):
100-
if freq and 'P0Y' in freq:
115+
elif temporal_type == "temporal":
116+
if freq is not None and 'P0Y' in freq:
101117
#print("DEBUG: Skipping as temporal is requested, P0Y here", freq)
102118
continue
103119
else:
104-
raise Exception("Unknown temporal type:", temporal_type)
105-
# chunk is optional, so if it does not exist then continue on
106-
chunk_from_config = node.get_value(keys=[item, 'chunk'])
107-
if chunk_from_config and chunk not in chunk_from_config:
108-
#print("DEBUG: Skipping as {} is requested, but not in rose-app config {}:".format(chunk, chunk_from_config))
109-
continue
110-
111-
results = ast.literal_eval(node.get_value(keys=[item, 'sources']))
120+
raise Exception(f"Unknown temporal type: {temporal_type}")
121+
#########
122+
# # chunk is optional, so if it does not exist then continue on
123+
chunk_from_config = comp_info.get("chunk")
124+
if chunk_from_config is not None and chunk not in chunk_from_config:
125+
#print("DEBUG: Skipping as {} is requested, but not in rose-app config {}:".format(chunk, comp_info["chunk"]))
126+
continue
127+
##########
128+
results=[]
129+
130+
# Get source list; append to results
131+
for hist_file in comp_info["sources"]:
132+
results.append(hist_file.get("history_file"))
133+
112134
remap_comp = comp
113135
answer = sorted(list(set(results)))
136+
114137
if remap_comp is not None:
115-
#If the same PP component is mapped to several sources per rose-app.conf, we make it a list and append values so we don't replace the key's value
116-
if remap_comp in dict_group_source.keys():
117-
dict_group_source[remap_comp].append(answer[0])
118-
else:
119-
dict_group_source[remap_comp] = answer
138+
#If the same PP component is mapped to several sources per rose-app.conf, we make it a list and append values so we don't replace the key's value
139+
if remap_comp in dict_group_source.keys():
140+
dict_group_source[remap_comp].append(answer[0])
141+
else:
142+
dict_group_source[remap_comp] = answer
143+
120144
if dict_group_source:
121-
for key, value in dict_group_source.items():
122-
makets_stmt = ""
123-
for src in value:
124-
if(makets_stmt != ''):
125-
# make-timeseries and make-timeavgs tasks have the chunksize in the task name,
126-
# but rename-split-to-pp does not
127-
if prereq_task == 'rename-split-to-pp':
128-
makets_stmt = f"{makets_stmt} & {prereq_task}-{grid}_{src}"
129-
else:
130-
makets_stmt = f"{makets_stmt} & {prereq_task}-{grid}-{chunk}_{src}"
131-
else:
132-
if prereq_task == 'rename-split-to-pp':
133-
makets_stmt = f"{prereq_task}-{grid}_{src}"
134-
else:
135-
makets_stmt = f"{prereq_task}-{grid}-{chunk}_{src}"
136-
137-
remap_stmt = f"remap-pp-components-{output_type}-{chunk}_{key}"
138-
remap_dep_stmt = f"{makets_stmt} => {remap_stmt}"
139-
remap_dep += f"{remap_dep_stmt}\n"
145+
for key, value in dict_group_source.items():
146+
makets_stmt = ""
147+
for src in value:
148+
if makets_stmt != '':
149+
# make-timeseries and make-timeavgs tasks have the chunksize in the task name,
150+
# but rename-split-to-pp does not
151+
if prereq_task == 'rename-split-to-pp':
152+
makets_stmt = f"{makets_stmt} & {prereq_task}-{grid}_{src}"
153+
else:
154+
makets_stmt = f"{makets_stmt} & {prereq_task}-{grid}-{chunk}_{src}"
155+
else:
156+
if prereq_task == 'rename-split-to-pp':
157+
makets_stmt = f"{prereq_task}-{grid}_{src}"
158+
else:
159+
makets_stmt = f"{prereq_task}-{grid}-{chunk}_{src}"
160+
161+
remap_stmt = f"remap-pp-components-{output_type}-{chunk}_{key}"
162+
remap_dep_stmt = f"{makets_stmt} => {remap_stmt}"
163+
remap_dep += f"{remap_dep_stmt}\n"
140164
# Possibly, no tasks are needed for the given request (grid type, temporal/static, chunk, components).
141165
# When that happens just exit with an empty string and exit normally.
142-
return(remap_dep)
166+
return remap_dep
143167

144-
#print(form_remap_dep('regrid-xy', 'temporal', 'P4D', 'land atmos land_cubic'))
168+
# Testing #
169+
#print(form_remap_dep('regrid-xy', 'temporal', 'P4D', 'atmos_cmip atmos', 'ts', 'c96L65_am5f7b12r1_amip_TEST_GRAPH.yaml'))

‎Jinja2Filters/form_task_parameters.py‎

Lines changed: 53 additions & 55 deletions
Original file line numberDiff line numberDiff line change
@@ -1,91 +1,89 @@
1-
import re
1+
"""
2+
Form the task parameter list based on the grid type,
3+
the temporal type,and the desired pp component(s)
4+
"""
5+
26
import os
3-
import metomi.rose.config
4-
import ast
57
from pathlib import Path
8+
import logging
69
import yaml
710

811
# set up logging
9-
import logging
1012
logging.basicConfig()
1113
logger = logging.getLogger(__name__)
1214
logger.setLevel(logging.INFO)
1315

1416
def form_task_parameters(grid_type, temporal_type, pp_components_str, yamlfile):
15-
"""Form the task parameter list based on the grid type, the temporal type,
16-
and the desired pp component(s)
17-
17+
"""
1818
Arguments:
1919
grid_type (str): One of: native or regrid-xy
2020
temporal_type (str): One of: temporal or static
2121
pp_component (str): all, or a space-separated list
22-
"""
23-
logger.debug(f"Desired pp components: {pp_components_str}")
22+
"""
23+
logger.debug("Desired pp components: %s", pp_components_str)
2424
pp_components = pp_components_str.split()
25-
path_to_conf = os.path.dirname(os.path.abspath(__file__)) + '/../app/remap-pp-components/rose-app.conf'
26-
node = metomi.rose.config.load(path_to_conf)
2725

2826
# Path to yaml configuration
2927
exp_dir = Path(__file__).resolve().parents[1]
3028
path_to_yamlconfig = os.path.join(exp_dir, yamlfile)
31-
# Load and read yaml configuration
29+
# Load and read yaml configuration
3230
with open(path_to_yamlconfig,'r') as yml:
3331
yml_info = yaml.safe_load(yml)
3432

33+
#
3534
results = []
36-
regex_pp_comp = re.compile('^\w+')
37-
for keys, sub_node in node.walk():
38-
# only target the keys
39-
if len(keys) != 1:
40-
continue
35+
for comp_info in yml_info["postprocess"]["components"]:
36+
comp = comp_info.get("type")
4137

42-
# skip env and command keys
43-
item = keys[0]
44-
if item == "env" or item == "command":
38+
# Check that pp_components defined matches those in the yaml file
39+
# Skip component if they don't match
40+
# skip if pp component not desired
41+
if comp not in pp_components:
4542
continue
46-
comp = regex_pp_comp.match(item).group()
47-
logger.debug(f"Examining item '{item}' comp '{comp}'")
4843

49-
# skip if pp component not desired
50-
logger.debug(f"Is {comp} in {pp_components}?")
51-
if comp in pp_components:
52-
logger.debug('Yes')
44+
# Set grid type if component has xyInterp defined or not
45+
if "xyInterp" not in comp_info.keys():
46+
candidate_grid_type = "native"
5347
else:
54-
logger.debug('No')
55-
continue
48+
candidate_grid_type = "regrid-xy"
5649

57-
# skip if grid type is not desired
58-
# some grid types (i.e. regrid-xy) have subtypes (i.e. 1deg, 2deg)
59-
# in remap-pp-components/rose-app.conf the grid type and subgrid is specified as "regrid-xy/1deg" (e.g.).
60-
# So we will strip off after the slash and the remainder is the grid type
61-
candidate_grid_type = re.sub('\/.*', '', node.get_value(keys=[item, 'grid']))
50+
# Check that candidate_grid_type matches grid type passed in function
51+
# If not, skip post-processing of component
6252
if candidate_grid_type != grid_type:
63-
logger.debug(f"Skipping as not right grid; got '{candidate_grid_type}' and wanted '{grid_type}'")
53+
logger.debug("Skipping as not right grid; got '%s' and wanted '%s'", candidate_grid_type, grid_type)
6454
continue
6555

66-
# filter static and temporal
67-
# if freq is not set => temporal
68-
# if freq includes "P0Y" => static
69-
# if freq does not include "P0Y" => temporal
70-
freq = node.get_value(keys=[item, 'freq'])
71-
if freq is not None and 'P0Y' in freq and temporal_type == 'temporal':
72-
logger.debug("Skipping static when temporal is requested")
73-
continue
56+
# Filter static and temporal
7457
if temporal_type == "static":
75-
if freq is not None and 'P0Y' not in freq:
76-
logger.debug("Skipping as static is requested, no P0Y here", freq)
77-
continue
78-
elif (temporal_type == "temporal"):
79-
if freq is not None and 'P0Y' in freq:
80-
logger.debug("Skipping as temporal is requested, P0Y here", freq)
58+
#print(comp_info["static"]["freq"])
59+
if "static" not in comp_info.keys():
60+
logger.debug("Skipping static as there are no static sources defined")
8161
continue
82-
else:
83-
raise Exception("Unknown temporal type:", temporal_type)
8462

85-
# convert array in string form to array
86-
sources = ast.literal_eval(node.get_value(keys=[item, 'sources']))
87-
results.extend(sources)
63+
for static_info in comp_info["static"]:
64+
if static_info.get("source") is not None:
65+
results.append(static_info.get("source"))
66+
## to-do: assess offline diagnostics
67+
# elif:
68+
# results = results + static_info.get("offline_sources")
69+
70+
elif temporal_type == "temporal":
71+
for hist_file in comp_info["sources"]:
72+
results.append(hist_file.get("history_file"))
73+
#results = results + comp_info.get("sources")
8874

75+
else:
76+
raise Exception(f"Unknown temporal type: {temporal_type}")
77+
78+
# results list --> set --> list: checks for repetitive sources listed
8979
answer = sorted(list(set(results)))
90-
logger.debug("Returning string" + ', '.join(answer))
91-
return(', '.join(answer))
80+
81+
# Returns a comma separated list of sources
82+
logger.debug("Returning string %s", ', '.join(answer))
83+
return ', '.join(answer)
84+
85+
## OWN TESTING ##
86+
#print(form_task_parameters('regrid-xy', 'temporal', 'ocean_cobalt_sfc ocean_cobalt_btm', 'COBALT_postprocess.yaml')))
87+
#print(form_task_parameters('regrid-xy', 'static', 'ocean_cobalt_sfc ocean_cobalt_btm', 'COBALT_postprocess.yaml'))
88+
#print(form_task_parameters('native', 'temporal', 'ocean_cobalt_sfc ocean_cobalt_btm', 'COBALT_postprocess.yaml'))
89+
#print(form_task_parameters('native', 'static', 'ocean_cobalt_sfc ocean_cobalt_btm', 'COBALT_postprocess.yaml'))

0 commit comments

Comments
 (0)