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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
|
#!/usr/bin/env python3
# emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-
# vi: set ft=python sts=4 ts=4 sw=4 et:
DESCRIP = 'Run and log examples'
EPILOG = \
""" Run examples in directory
Typical usage is:
run_log_examples.py nipy/examples --log-path=~/tmp/eg_logs
to run the examples and log the result, or
run_log_examples.py nipy/examples/some_example.py
to run a single example.
"""
import os
import re
import sys
from argparse import ArgumentParser, RawDescriptionHelpFormatter
from os.path import abspath, dirname, expanduser, isfile
from os.path import join as pjoin
from os.path import sep as psep
from subprocess import PIPE, Popen
PYTHON=sys.executable
NEED_SHELL = True
class ProcLogger:
def __init__(self, log_path, working_path):
self.log_path = log_path
self.working_path = working_path
self._names = []
def cmd_str_maker(self, cmd, args):
return " ".join([cmd] + list(args))
def __call__(self, cmd_name, cmd, args=(), cwd=None):
# Mqke log files
if cmd_name in self._names:
raise ValueError(f'Command name {cmd_name} not unique')
self._names.append(cmd_name)
if cwd is None:
cwd = self.working_path
cmd_out_path = pjoin(self.log_path, cmd_name)
stdout_log = open(cmd_out_path + '.stdout', "w")
stderr_log = open(cmd_out_path + '.stderr', "w")
try:
# Start subprocess
cmd_str = self.cmd_str_maker(cmd, args)
proc = Popen(cmd_str,
cwd = cwd,
stdout = stdout_log,
stderr = stderr_log,
shell = NEED_SHELL)
# Execute
retcode = proc.wait()
finally:
if proc.poll() is None: # In case we get killed
proc.terminate()
stdout_log.close()
stderr_log.close()
return retcode
def run_pipes(self, cmd, args=(), cwd=None):
if cwd is None:
cwd = self.working_path
try:
# Start subprocess
cmd_str = self.cmd_str_maker(cmd, args)
proc = Popen(cmd_str,
cwd = cwd,
stdout = PIPE,
stderr = PIPE,
shell = NEED_SHELL)
# Execute
stdout, stderr = proc.communicate()
finally:
if proc.poll() is None: # In case we get killed
proc.terminate()
return stdout.decode(), stderr.decode(), proc.returncode
class PyProcLogger(ProcLogger):
def cmd_str_maker(self, cmd, args):
""" Execute python script `cmd`
Reject any `args` because we're using ``exec`` to execute the script.
Prepend some matplotlib setup to suppress figures
"""
if len(args) != 0:
raise ValueError(f"Cannot use args with {self.__class__}")
return(f"""{PYTHON} -c "import matplotlib as mpl; mpl.use('agg'); """
f"""exec(open('{cmd}', 'rt').read())" """)
def _record(result, fname, fileobj):
print(result)
fileobj.write(f'{fname}: {result}\n')
def main():
parser = ArgumentParser(description=DESCRIP,
epilog=EPILOG,
formatter_class=RawDescriptionHelpFormatter)
parser.add_argument('examples_path', type=str,
help='filename of example or directory containing '
'examples to run')
parser.add_argument('--log-path', type=str, default='',
help='path for output logs (default is cwd)')
parser.add_argument('--excludex', type=str, action='append', default=[],
help='regex for files to exclude (add more than one '
'--excludex option for more than one regex filter')
args = parser.parse_args()
# Proc runner
eg_path = abspath(expanduser(args.examples_path))
if args.log_path == '':
log_path = abspath(os.getcwd())
else:
log_path = abspath(expanduser(args.log_path))
excludexes = [re.compile(s) for s in args.excludex]
if isfile(eg_path): # example was a file
proc_logger = PyProcLogger(log_path=log_path,
working_path=dirname(eg_path))
print("Running " + eg_path)
stdout, stderr, code = proc_logger.run_pipes(eg_path)
print('==== Stdout ====')
print(stdout)
print('==== Stderr ====')
print(stderr)
sys.exit(code)
# Multi-run with logging to file
proc_logger = PyProcLogger(log_path=log_path,
working_path=eg_path)
fails = 0
with open(pjoin(log_path, 'summary.txt'), "w") as f:
for dirpath, dirnames, filenames in os.walk(eg_path):
for fname in filenames:
full_fname = pjoin(dirpath, fname)
if fname.endswith(".py"):
print(fname, end=': ')
sys.stdout.flush()
for excludex in excludexes:
if excludex.search(fname):
_record('SKIP', fname, f)
break
else: # run test
cmd_name = full_fname.replace(eg_path + psep, '')
cmd_name = cmd_name.replace(psep, '-')
code = proc_logger(cmd_name, full_fname, cwd=dirpath)
if code == 0:
_record('OK', fname, f)
else:
fails += 1
_record('FAIL', fname, f)
sys.exit(min(255, fails))
if __name__ == '__main__':
main()
|
