forked from SciTools/iris
-
Notifications
You must be signed in to change notification settings - Fork 0
/
__init__.py
91 lines (79 loc) · 3.27 KB
/
__init__.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
# Copyright Iris contributors
#
# This file is part of Iris and is released under the LGPL license.
# See COPYING and COPYING.LESSER in the root of the repository for full
# licensing details.
"""
Scripts for generating supporting data for benchmarking.
Data generated using Iris should use :func:`run_function_elsewhere`, which
means that data is generated using a fixed version of Iris and a fixed
environment, rather than those that get changed when the benchmarking run
checks out a new commit.
Downstream use of data generated 'elsewhere' requires saving; usually in a
NetCDF file. Could also use pickling but there is a potential risk if the
benchmark sequence runs over two different Python versions.
"""
from inspect import getsource
from os import environ
from pathlib import Path
from subprocess import CalledProcessError, check_output, run
from textwrap import dedent
#: Python executable used by :func:`run_function_elsewhere`, set via env
#: variable of same name. Must be path of Python within an environment that
#: includes Iris (including dependencies and test modules) and Mule.
try:
DATA_GEN_PYTHON = environ["DATA_GEN_PYTHON"]
_ = check_output([DATA_GEN_PYTHON, "-c", "a = True"])
except KeyError:
error = "Env variable DATA_GEN_PYTHON not defined."
raise KeyError(error)
except (CalledProcessError, FileNotFoundError, PermissionError):
error = (
"Env variable DATA_GEN_PYTHON not a runnable python executable path."
)
raise ValueError(error)
default_data_dir = (Path(__file__).parents[2] / ".data").resolve()
BENCHMARK_DATA = Path(environ.get("BENCHMARK_DATA", default_data_dir))
if BENCHMARK_DATA == default_data_dir:
BENCHMARK_DATA.mkdir(exist_ok=True)
elif not BENCHMARK_DATA.is_dir():
message = f"Not a directory: {BENCHMARK_DATA} ."
raise ValueError(message)
# Manual flag to allow the rebuilding of synthetic data.
REUSE_DATA = True
def run_function_elsewhere(func_to_run, *args, **kwargs):
"""
Run a given function using the :const:`DATA_GEN_PYTHON` executable.
This structure allows the function to be written natively.
Parameters
----------
func_to_run : FunctionType
The function object to be run.
NOTE: the function must be completely self-contained, i.e. perform all
its own imports (within the target :const:`DATA_GEN_PYTHON`
environment).
*args : tuple, optional
Function call arguments. Must all be expressible as simple literals,
i.e. the ``repr`` must be a valid literal expression.
**kwargs: dict, optional
Function call keyword arguments. All values must be expressible as
simple literals (see ``*args``).
Returns
-------
str
The ``stdout`` from the run.
"""
func_string = dedent(getsource(func_to_run))
func_string = func_string.replace("@staticmethod\n", "")
func_call_term_strings = [repr(arg) for arg in args]
func_call_term_strings += [
f"{name}={repr(val)}" for name, val in kwargs.items()
]
func_call_string = (
f"{func_to_run.__name__}(" + ",".join(func_call_term_strings) + ")"
)
python_string = "\n".join([func_string, func_call_string])
result = run(
[DATA_GEN_PYTHON, "-c", python_string], capture_output=True, check=True
)
return result.stdout