-
Notifications
You must be signed in to change notification settings - Fork 44
/
utils.py
101 lines (87 loc) · 2.95 KB
/
utils.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
try:
import intake
except ImportError:
intake = None
def google_cmip_col(catalog="main"):
"""A tiny utility function to point to the 'official' pangeo cmip6 cloud files."""
if intake is None:
raise ImportError(
"This functionality requires intake-esm. Install with `conda install -c conda-forge intake-esm"
)
if catalog == "main":
return intake.open_esm_datastore(
"https://storage.googleapis.com/cmip6/pangeo-cmip6.json"
)
# this doesnt work anymore, but ill leave it here as an example for the future
# elif catalog == "testing":
# return intake.open_esm_datastore(
# "https://storage.googleapis.com/cmip6/pangeo-cmip6-testing.json"
# )
else:
raise ValueError("Catalog not recognized. Should be `main` or `testing`")
def model_id_match(match_list, id_tuple):
"""Matches `id_tuple` to the list of tuples `exception_list`, which can contain
wildcards (match any entry) and lists (match any entry that is in the list).
Parameters
----------
match_list : list
list of tuples with id strings corresponding to e.g. `source_id`, `grid_label`...
id_tuple : tuple
single tuple with id strings.
"""
# Check the size of tuples
if any([len(t) != len(id_tuple) for t in match_list]):
raise ValueError(
"Each tuple in `match_list` must have the same number of elements as `match_id`"
)
match_list_checked = []
for ml in match_list:
ml_processed = []
for i in range(len(ml)):
match_element = ml[i]
if isinstance(match_element, str) and match_element != "*":
match_element = [match_element]
if id_tuple[i] in match_element or match_element == "*":
ml_processed.append(True)
else:
ml_processed.append(False)
match_list_checked.append(all(ml_processed))
return any(match_list_checked)
def _key_from_attrs(ds, attrs, sep="."):
return sep.join([ds.attrs[i] if i in ds.attrs.keys() else "none" for i in attrs])
def cmip6_dataset_id(
ds,
sep=".",
id_attrs=[
"activity_id",
"institution_id",
"source_id",
"experiment_id",
"variant_label",
"table_id",
"grid_label",
"version",
"variable_id",
],
):
"""Creates a unique string id for e.g. saving files to disk from CMIP6 output
Parameters
----------
ds : xr.Dataset
Input dataset
sep : str, optional
String/Symbol to seperate fields in resulting string, by default "."
Returns
-------
str
Concatenated
"""
return _key_from_attrs(ds, id_attrs, sep=sep)
def _maybe_make_list(item):
"utility function to make sure output is a list"
if isinstance(item, str):
return [item]
elif isinstance(item, list):
return item
else:
return list(item)