Skip to content

Base

area_weighting(var)

Apply area weighting to the variable var using the cosine of latitude: \(\cos (\phi)\).

Parameters:

Name Type Description Default
var DataArray

Variable to weight e.g. ds.t_surf to weight the surface temperature, where ds is the dataset for the experiment which contains all variables.

 required

Returns:

Type Description
DataArrayWeighted

Area weighted version of var.
Source code in isca_tools/utils/base.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def area_weighting(var: xr.DataArray) -> DataArrayWeighted:
    """
    Apply area weighting to the variable `var` using the `cosine` of latitude: $\cos (\phi)$.

    Args:
        var: Variable to weight e.g. `ds.t_surf` to weight the surface temperature, where
            `ds` is the dataset for the experiment which contains all variables.

    Returns:
        Area weighted version of `var`.
    """
    weights = np.cos(np.deg2rad(var.lat))
    weights.name = "weights"
    return var.weighted(weights)

get_memory_usage()

Get current process’s memory in MB.

Returns:

Name Type Description
mem_mb float

Memory usage in MB
Source code in isca_tools/utils/base.py 
101
102
103
104
105
106
107
108
109
110
def get_memory_usage() -> float:
    """
    Get current process’s memory in MB.

    Returns:
        mem_mb: Memory usage in MB
    """
    process = psutil.Process(os.getpid())
    mem_mb = process.memory_info().rss / (1024 * 1024)
    return mem_mb

len_safe(x)

Return length of x which can have multiple values, or just be a number.

Parameters:

Name Type Description Default
x

Variable to return length of.

 required

Returns:

Type Description
int

Number of elements in x.
Source code in isca_tools/utils/base.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def len_safe(x) -> int:
    """
    Return length of `x` which can have multiple values, or just be a number.

    Args:
        x: Variable to return length of.

    Returns:
        Number of elements in `x`.
    """
    if isinstance(x, numbers.Number):
        return 1
    try:
        return len(x)
    except TypeError:
        raise TypeError(f"Unsupported type with no length: {type(x)}")

parse_int_list(value, format_func=lambda x: str(x), all_values=None)

Takes in a value or list of values e.g. [1, 2, 3] and converts it into a list of strings where each string has the format given by format_func e.g. ['1', '2', '3'] for the default case.

There are three string options for value: * value='x:y', will return all integers between x and y inclusive. * value='firstX' will return first X values of all_values. * value='firstY' will return first Y values of all_values.

Parameters:

Name Type Description Default
value Union[str, int, List]

Variable to convert into list of strings

 required
format_func Callable

How to format each integer within the string.

 lambda x: str(x)
all_values Optional[List]

List of all possible integers, must be provided if value='firstX' or value='firstY'.

 None

Returns:

Type Description
List

List, where each integer in value is converted using format_func.
Source code in isca_tools/utils/base.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def parse_int_list(value: Union[str, int, List], format_func: Callable = lambda x: str(x),
                   all_values: Optional[List] = None) -> List:
    """
    Takes in a value or list of values e.g. `[1, 2, 3]` and converts it into a list of strings where
    each string has the format given by `format_func` e.g. `['1', '2', '3']` for the default case.

    There are three string options for `value`:
    * `value='x:y'`, will return all integers between `x` and `y` inclusive.
    * `value='firstX'` will return first X values of `all_values`.
    * `value='firstY'` will return first Y values of `all_values`.

    Args:
        value: Variable to convert into list of strings
        format_func: How to format each integer within the string.
        all_values: List of all possible integers, must be provided if `value='firstX'` or `value='firstY'`.

    Returns:
        List, where each integer in `value` is converted using `format_func`.
    """
    if isinstance(value, list):
        pass
    elif isinstance(value, int):
        value = [value]
    elif isinstance(value, str):
        value = value.strip()       # remove blank space
        # Can specify just first or last n years
        if re.search(r'^first(\d+)', value):
            if all_values is None:
                raise ValueError(f'With value={value}, must provide all_values')
            n_req = int(re.search(r'^first(\d+)', value).group(1))
            if n_req > len(all_values):
                warnings.warn(f"Requested {value} but there are only "
                              f"{len(all_values)} available:\n{all_values}")
            value = all_values[:n_req]
        elif re.search(r'^last(\d+)', value):
            if all_values is None:
                raise ValueError(f'With value={value}, must provide all_values')
            n_req = int(re.search(r'^last(\d+)', all_values).group(1))
            if n_req > len(all_values):
                warnings.warn(f"Requested {value} but there are only "
                              f"{len(all_values)} available:\n{all_values}")
            value = all_values[-n_req:]
        elif ':' in value:
            # If '1979:2023' returns all integers from 1979 to 2023
            start, end = map(int, value.split(':'))
            value = list(range(start, end + 1))
        else:
            value = [int(value)]
    else:
        raise ValueError(f"Unsupported format: {value}")
    return [format_func(i) for i in value]

print_log(text, logger=None)

Quick function to add to log if log exists, otherwise print it.

Parameters:

Name Type Description Default
text str

Text to be printed.

 required
logger Optional[Logger]
None

Returns:

Source code in isca_tools/utils/base.py 
33
34
35
36
37
38
39
40
41
42
43
44
def print_log(text: str, logger:Optional[logging.Logger] = None) -> None:
    """
    Quick function to add to log if log exists, otherwise print it.

    Args:
        text: Text to be printed.
        logger:

    Returns:

    """
    logger.info(text) if logger else print(text)

round_any(x, base, round_type='round')

Rounds x to the nearest multiple of base with the rounding done according to round_type.

Parameters:

Name Type Description Default
x Union[float, ndarray]

Number or array to round.

 required
base float

Rounds x to nearest integer multiple of value of base.

 required
round_type str

One of the following, indicating how to round x -

  • 'round'
  • 'ceil'
  • 'float'
 'round'

Returns:

Type Description
Union[float, ndarray]

Rounded version of x.
Example 
round_any(3, 5) = 5
round_any(3, 5, 'floor') = 0
Source code in isca_tools/utils/base.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def round_any(x: Union[float, np.ndarray], base: float, round_type: str = 'round') -> Union[float, np.ndarray]:
    """
    Rounds `x` to the nearest multiple of `base` with the rounding done according to `round_type`.

    Args:
        x: Number or array to round.
        base: Rounds `x` to nearest integer multiple of value of `base`.
        round_type: One of the following, indicating how to round `x` -

            - `'round'`
            - `'ceil'`
            - `'float'`

    Returns:
        Rounded version of `x`.

    Example:
        ```
        round_any(3, 5) = 5
        round_any(3, 5, 'floor') = 0
        ```
    """
    if round_type == 'round':
        return base * np.round(x / base)
    elif round_type == 'ceil':
        return base * np.ceil(x / base)
    elif round_type == 'floor':
        return base * np.floor(x / base)
    else:
        raise ValueError(f"round_type specified was {round_type} but it should be one of the following:\n"
                         f"round, ceil, floor")

run_func_loop(func, max_wait_time=300, wait_interval=20, func_check=None, logger=None)

Safe way to run a function, such that if hit error, will try again every wait_interval seconds up to a maximum of max_wait_time seconds.

If func_check is given and returns True at any point, it will exit the loop without executing func.

Most obvious usage is for creating a directory e.g. os.makedirs, especially to a server where connection cuts in and out.

Parameters:

Name Type Description Default
func Callable

Function to run. Must have no arguments.

 required
max_wait_time int

Maximum number of seconds to try and run func.

 300
wait_interval int

Interval in seconds to wait between running func.

 20
func_check Optional[Callable]

Function that returns a boolean. If it returns True at any point, the loop will exit the loop without executing func.

 None
logger Optional[Logger]

Logger to record information

 None

Returns:

Type Description

Whatever func returns.
Source code in isca_tools/utils/base.py 
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
def run_func_loop(func: Callable, max_wait_time: int = 300, wait_interval: int = 20,
                  func_check: Optional[Callable] = None, logger: Optional[logging.Logger] = None):
    """
    Safe way to run a function, such that if hit error, will try again every `wait_interval` seconds up to a
    maximum of `max_wait_time` seconds.

    If `func_check` is given and returns `True` at any point, it will exit the loop without executing `func`.

    Most obvious usage is for creating a directory e.g. `os.makedirs`, especially to a server where connection
    cuts in and out.

    Args:
        func: Function to run. Must have no arguments.
        max_wait_time: Maximum number of seconds to try and run `func`.
        wait_interval: Interval in seconds to wait between running `func`.
        func_check: Function that returns a boolean. If it returns `True` at any point, the loop
            will exit the loop without executing `func`.
        logger: Logger to record information

    Returns:
        Whatever `func` returns.
    """
    i = 0
    j = 0
    success = False
    start_time = time.time()
    output = None
    while not success and (time.time() - start_time) < max_wait_time:
        if func_check is not None:
            if func_check():
                print_log("func_check passed so did not exectute func", logger)
                success = True
                break
        try:
            output = func()
            success = True
        except PermissionError as e:
            i += 1
            if i == 1:
                # Only print on first instance of error
                print_log(f'Permission Error: {e}', logger)
            time.sleep(wait_interval)
        except Exception as e:
            j += 1
            if j == 1:
                # Only print on first instance of error
                print_log(f'Unexpected Error: {e}', logger)
            time.sleep(wait_interval)
    if not success:
        raise ValueError(f"Making output directory - Failed to run function after {max_wait_time} seconds.")
    return output

split_list_max_n(lst, n)

Split lst into balanced chunks with at most n elements each.

Parameters:

Name Type Description Default
lst Union[List, ndarray]

List to split.

 required
n int

Maximum number of elements in each chunk of lst.

 required

Returns:

Type Description
List

List of n chunks of lst
Source code in isca_tools/utils/base.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def split_list_max_n(lst: Union[List, np.ndarray], n: int) -> List:
    """
    Split `lst` into balanced chunks with at most `n` elements each.

    Args:
        lst: List to split.
        n: Maximum number of elements in each chunk of `lst`.

    Returns:
        List of `n` chunks of `lst`
    """
    k = int(np.ceil(len(lst) / n))  # Number of chunks needed
    avg = int(np.ceil(len(lst) / k))
    return [lst[i * avg : (i + 1) * avg] for i in range(k)]