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
|
# (C) Copyright 2014-15 Enthought, Inc., Austin, TX
# All right reserved.
#
# This software is provided without warranty under the terms of the BSD
# license included in enthought/LICENSE.txt and may be redistributed only
# under the conditions described in the aforementioned license. The license
# is also available online at http://www.enthought.com/licenses/BSD.txt
# Thanks for using Enthought open source!
import contextlib
import threading
from pyface.gui import GUI
from pyface.qt import QtCore, QtGui
from traits.api import HasStrictTraits, Instance
class ConditionTimeoutError(RuntimeError):
pass
@contextlib.contextmanager
def dont_quit_when_last_window_closed(qt_app):
"""
Suppress exit of the application when the last window is closed.
"""
flag = qt_app.quitOnLastWindowClosed()
qt_app.setQuitOnLastWindowClosed(False)
try:
yield
finally:
qt_app.setQuitOnLastWindowClosed(flag)
class EventLoopHelper(HasStrictTraits):
qt_app = Instance(QtGui.QApplication)
gui = Instance(GUI)
def event_loop_with_timeout(self, repeat=2, timeout=10.0):
"""Helper function to send all posted events to the event queue and
wait for them to be processed. This runs the real event loop and
does not emulate it with QApplication.processEvents.
Parameters
----------
repeat : int
Number of times to process events. Default is 2.
timeout: float, optional, keyword only
Number of seconds to run the event loop in the case that the trait
change does not occur. Default value is 10.0.
"""
def repeat_loop(condition, repeat):
# We sendPostedEvents to ensure that enaml events are processed
self.qt_app.sendPostedEvents()
repeat = repeat - 1
if repeat <= 0:
self.gui.invoke_later(condition.set)
else:
self.gui.invoke_later(
repeat_loop, condition=condition, repeat=repeat
)
condition = threading.Event()
self.gui.invoke_later(repeat_loop, repeat=repeat, condition=condition)
self.event_loop_until_condition(
condition=condition.is_set, timeout=timeout)
def event_loop(self, repeat=1):
"""Emulates an event loop `repeat` times with
QApplication.processEvents.
Parameters
----------
repeat : int
Number of times to process events. Default is 1.
"""
for i in range(repeat):
self.qt_app.sendPostedEvents()
self.qt_app.processEvents()
def event_loop_until_condition(self, condition, timeout=10.0):
"""Runs the real Qt event loop until the provided condition evaluates
to True.
Raises ConditionTimeoutError if the timeout occurs before the condition
is satisfied.
Parameters
----------
condition : callable
A callable to determine if the stop criteria have been met. This
should accept no arguments.
timeout : float
Number of seconds to run the event loop in the case that the trait
change does not occur.
"""
def handler():
if condition():
self.qt_app.quit()
# Make sure we don't get a premature exit from the event loop.
with dont_quit_when_last_window_closed(self.qt_app):
condition_timer = QtCore.QTimer()
condition_timer.setInterval(50)
condition_timer.timeout.connect(handler)
timeout_timer = QtCore.QTimer()
timeout_timer.setSingleShot(True)
timeout_timer.setInterval(timeout * 1000)
timeout_timer.timeout.connect(self.qt_app.quit)
timeout_timer.start()
condition_timer.start()
try:
self.qt_app.exec_()
if not condition():
raise ConditionTimeoutError(
'Timed out waiting for condition')
finally:
timeout_timer.stop()
condition_timer.stop()
@contextlib.contextmanager
def delete_widget(self, widget, timeout=1.0):
"""Runs the real Qt event loop until the widget provided has been
deleted. Raises ConditionTimeoutError on timeout.
Parameters
----------
widget : QObject
The widget whose deletion will stop the event loop.
timeout : float
Number of seconds to run the event loop in the case that the
widget is not deleted.
"""
timer = QtCore.QTimer()
timer.setSingleShot(True)
timer.setInterval(timeout * 1000)
timer.timeout.connect(self.qt_app.quit)
widget.destroyed.connect(self.qt_app.quit)
yield
timer.start()
self.qt_app.exec_()
if not timer.isActive():
# We exited the event loop on timeout
raise ConditionTimeoutError(
'Could not destroy widget before timeout: {!r}'.format(widget))
|