Source code for schrodinger.infra.exception_handler
"""A top level Python exception handler that prevents uncaught exceptions in
Python scripts from crashing Maestro.
To activate the exception handler use
`exception_handler.set_exception_handler()`. That will activate the
appropriate exception handler.
If SCHRODINGER_DEV_DEBUG or SCHRODINGER_SRC are defined, we install a handler
that simply prints tracebacks to the terminal. Otherwise (generally on customer
machines), we install a handler that writes uncaught exceptions to a folder in
.schrodinger. The user is informed of the error and told to contact customer
service.
"""
import errno
import os
import sys
import time
import traceback
from schrodinger import get_maestro
from schrodinger import in_dev_env
from schrodinger.job import jobcontrol
from schrodinger.utils import fileutils
from schrodinger.utils import thread_utils
if sys.platform.startswith("linux") and "DISPLAY" not in os.environ:
exception_dialog = None
else:
try:
from schrodinger.infra.exception_handler_dir import exception_dialog
except ImportError:
exception_dialog = None
[docs]class ExceptionRecorder:
"""
A top level exception handler that writes uncaught exceptions to a folder in
.schrodinger. The user is informed of the error and told to contact
customer service.
This handler can be activated by `sys.excepthook = ExceptionRecorder()` or
by calling the `enable_handler()` convenience function.
:cvar _OPENFLAGS: The flags used when opening a file. These flags are set
to ensure that files are opened in a thread-safe manner.
:vartype _OPENFLAGS: int
:cvar _EXCEPTIONS_DIR: The directory where exceptions are stored
:vartype _EXCEPTIONS_DIR: str
:cvar _MAX_EXCEP_FILES: The maximum number of files allowed in the
exceptions directory. Once this number of files is hit, the oldest files
will be erased after recording the next exception.
:vartype _MAX_EXCEP_FILES: int
"""
_OPENFLAGS = os.O_WRONLY | os.O_CREAT | os.O_EXCL
_EXCEPTIONS_DIR = os.path.join(
fileutils.get_directory_path(fileutils.LOCAL_APPDATA), "exceptions")
_MAX_EXCEP_FILES = 40
_MAX_EXCEP_PER_SECOND = 50
# The maximum number of files that are allowed with the same time stamp.
# Once this number of files is hit, an exception will be raised when
# trying to record the next exception with the same time stamp. This is
# intended as a fail-safe to avoid infinite loops when searching for
# filenames.
[docs] def __init__(self):
self.ignored_exceptions = set()
def __call__(self, etype, value, tb):
"""
Write the specified exception to disk and print a helpful error message to
the user. If anything goes wrong while recording the exception, print both
the original exception and the new error to stderr.
:param etype: The exception type
:type etype: type
:param value: The exception that was raised
:type value: BaseException
:param tb: The traceback that led to the exception
:type tb: traceback
"""
try:
self._recordException(etype, value, tb)
except:
traceback.print_exception(etype, value, tb)
msg = ("\nAdditionally, the following exception occurred while "
"recording the above exception:")
print(msg, file=sys.stderr)
traceback.print_exc()
def _recordException(self, etype, value, tb):
"""
Write the specified exception to disk and display a helpful error
message to the user in the terminal and, if running maestro, in the gui
:param etype: The exception type
:type etype: type
:param value: The exception that was raised
:type value: BaseException
:param tb: The traceback that led to the exception
:type tb: traceback
"""
self._createExcepDir()
self._cleanupExcepDir()
out, filename = self._getExcepFile()
traceback.print_exception(etype, value, tb, file=out)
out.close()
msg_plaintext = self._getMessage(filename)
# Without the flush, the error message turns up in a random place in
# the log file
sys.stdout.flush()
print(msg_plaintext, file=sys.stderr)
sys.stderr.flush()
with open(filename) as f:
exception_msg = f.read()
if get_maestro() and exception_dialog and thread_utils.in_main_thread():
if exception_msg in self.ignored_exceptions:
return
msg_html = self._getMessage(filename, html=True)
dialog = exception_dialog.ExceptionDialog(msg_html, exception_msg)
try:
result = dialog.exec()
finally:
# Save state of the checkbox even if exception handler crashes
if dialog.ignore_in_future:
self.ignored_exceptions.add(exception_msg)
def _getMessage(self, filepath, html=False):
"""
Returns an message formatted either as plain text or html
:param filepath: The full path of the file containing the traceback
:type filepath: str
:param html: Whether to return the message formatted in html or plain
text
:type html: bool
:return: A formatted message
:rtype: str
"""
msg_template = "An error has occurred. Information about the error" +\
" has been written to {newline}{filepath} {newline}" +\
"{newline}Please contact {contact} for assistance." +\
" Include {filename} and a description of what you" + \
" were doing when the error occurred."
filename = os.path.basename(filepath)
newline = "\n"
contact = "help@schrodinger.com"
if html:
filepath = '<a href="file:{filepath}">{filepath}</a>'.format(
filepath=filepath)
newline = "<br />"
contact = '<a href="mailto:{contact}">{contact}</a>'.format(
contact=contact)
msg = msg_template.format(newline=newline,
filepath=filepath,
filename=filename,
contact=contact)
return msg
def _getExcepFile(self):
"""
Return the file that the exception should be written to.
:return: A tuple of
- An open filehandle to write the exception to
- The filename corresponding to the filehandle
:rtype: tuple
:raise Exception: If the desired output file cannot be opened or if no
acceptable filename can be found.
"""
for cur_filename in self._genFilename():
try:
fd = os.open(cur_filename, self._OPENFLAGS)
handle = os.fdopen(fd, 'w')
return handle, cur_filename
except OSError as err:
if err.errno != errno.EEXIST:
raise
raise Exception("No acceptable exception filename could be found. "
"Last tried %s" % cur_filename)
def _genFilename(self):
"""
Generate potential output filenames to write the exception to
:return: A generator that iterates through potential output filenames, where
each filename is a fully qualified path.
:rtype: generator
"""
cur_time = time.strftime("%Y%m%d-%H%M%S")
basename = "error%s" % cur_time
filename_no_ext = os.path.join(self._EXCEPTIONS_DIR, basename)
ext = ".txt"
yield filename_no_ext + ext
for i in range(1, self._MAX_EXCEP_PER_SECOND):
yield filename_no_ext + f"-{i}{ext}"
def _createExcepDir(self):
"""
Create the exceptions directory if it does not already exist.
:raise Exception: If we cannot create the exceptions directory
"""
try:
os.makedirs(self._EXCEPTIONS_DIR, exist_ok=True)
except OSError as err:
if not os.path.isdir(self._EXCEPTIONS_DIR):
raise Exception("Exceptions directory %s exists but it is not "
"a directory." % self._EXCEPTIONS_DIR)
def _cleanupExcepDir(self):
"""
If there are more than _MAX_EXCEP_FILES - 1 files in the exceptions
directory, remove the oldest files.
"""
files = os.listdir(self._EXCEPTIONS_DIR)
num_to_delete = len(files) - self._MAX_EXCEP_FILES + 1
if num_to_delete <= 0:
return
files.sort()
for cur_file in files[:num_to_delete]:
cur_file_full = os.path.join(self._EXCEPTIONS_DIR, cur_file)
fileutils.force_remove(cur_file_full)
customer_handler = ExceptionRecorder()
[docs]def get_exception_handler():
"""
Returns the appropriate exception handler, depending on values in the user's
environment.
"""
if in_dev_env() or jobcontrol.get_backend():
# Developers prefer to have exceptions simply printed to the terminal.
# Using this exception handler prevents Maestro from crashing when an
# exception occurs during signal handling.
return traceback.print_exception
else:
return customer_handler
[docs]def set_exception_handler(handler=None):
"""
Sets the appropriate top-level Python exception handler. We use one for
customers and another for developers.
There is no effect if there is already a custom exception handler.
"""
# Someone has already set an exception handler (example, pytest);
# we should respect that
if sys.excepthook != sys.__excepthook__:
return
if handler is None:
handler = get_exception_handler()
sys.excepthook = handler