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
|
#!/usr/bin/env python3
import functools
import logging
import os
import sys
# This module is commonly used by others in here and should avoid
# taking any unnecessary dependencies back on them.
from argparse_utils import ActionNoYes
import config
import logging_utils
logger = logging.getLogger(__name__)
args = config.add_commandline_args(
f'Bootstrap ({__file__})',
'Args related to python program bootstrapper and Swiss army knife')
args.add_argument(
'--debug_unhandled_exceptions',
action=ActionNoYes,
default=False,
help='Break into pdb on top level unhandled exceptions.'
)
args.add_argument(
'--show_random_seed',
action=ActionNoYes,
default=False,
help='Should we display (and log.debug) the global random seed?'
)
args.add_argument(
'--set_random_seed',
type=int,
nargs=1,
default=None,
metavar='SEED_INT',
help='Override the global random seed with a particular number.'
)
original_hook = sys.excepthook
def handle_uncaught_exception(exc_type, exc_value, exc_tb):
"""
Top-level exception handler for exceptions that make it past any exception
handlers in the python code being run. Logs the error and stacktrace then
maybe attaches a debugger.
"""
global original_hook
msg = f'Unhandled top level exception {exc_type}'
logger.exception(msg)
print(msg, file=sys.stderr)
if issubclass(exc_type, KeyboardInterrupt):
sys.__excepthook__(exc_type, exc_value, exc_tb)
return
else:
if (
not sys.stderr.isatty() or
not sys.stdin.isatty()
):
# stdin or stderr is redirected, just do the normal thing
original_hook(exc_type, exc_value, exc_tb)
else:
# a terminal is attached and stderr is not redirected, maybe debug.
import traceback
traceback.print_exception(exc_type, exc_value, exc_tb)
if config.config['debug_unhandled_exceptions']:
import pdb
logger.info("Invoking the debugger...")
pdb.pm()
else:
original_hook(exc_type, exc_value, exc_tb)
def initialize(entry_point):
"""
Remember to initialize config, initialize logging, set/log a random
seed, etc... before running main.
"""
@functools.wraps(entry_point)
def initialize_wrapper(*args, **kwargs):
# Hook top level unhandled exceptions, maybe invoke debugger.
if sys.excepthook == sys.__excepthook__:
sys.excepthook = handle_uncaught_exception
# Try to figure out the name of the program entry point. Then
# parse configuration (based on cmdline flags, environment vars
# etc...)
if (
'__globals__' in entry_point.__dict__ and
'__file__' in entry_point.__globals__
):
config.parse(entry_point.__globals__['__file__'])
else:
config.parse(None)
# Initialize logging... and log some remembered messages from
# config module.
logging_utils.initialize_logging(logging.getLogger())
config.late_logging()
# Allow programs that don't bother to override the random seed
# to be replayed via the commandline.
import random
random_seed = config.config['set_random_seed']
if random_seed is not None:
random_seed = random_seed[0]
else:
random_seed = int.from_bytes(os.urandom(4), 'little')
if config.config['show_random_seed']:
msg = f'Global random seed is: {random_seed}'
print(msg)
logger.debug(msg)
random.seed(random_seed)
# Do it, invoke the user's code. Pay attention to how long it takes.
logger.debug(f'Starting {entry_point.__name__} (program entry point)')
ret = None
import stopwatch
with stopwatch.Timer() as t:
ret = entry_point(*args, **kwargs)
logger.debug(
f'{entry_point.__name__} (program entry point) returned {ret}.'
)
walltime = t()
(utime, stime, cutime, cstime, elapsed_time) = os.times()
logger.debug('\n'
f'user: {utime}s\n'
f'system: {stime}s\n'
f'child user: {cutime}s\n'
f'child system: {cstime}s\n'
f'machine uptime: {elapsed_time}s\n'
f'walltime: {walltime}s')
# If it doesn't return cleanly, call attention to the return value.
if ret is not None and ret != 0:
logger.error(f'Exit {ret}')
else:
logger.debug(f'Exit {ret}')
sys.exit(ret)
return initialize_wrapper
|
