"""
Repeatedly perform a task until aborted by the user.
"""
import datetime
import traceback
from . import Qt
from . import QtCore
from . import QtGui
from . import QtWidgets
from . import application
from . import prompt
[docs]class LoopUntilAbort(object):
def __init__(self, *, loop_delay=0, max_iterations=None, single_shot=False,
title=None, bg_color='#DFDFDF', text_color='#20548B',
font_family='Helvetica', font_size=14):
"""Repeatedly perform a task until aborted by the user.
This class provides an interface to show the status of a task (e.g., read
a sensor value and write the value to a file) that you want to perform for
an unknown period of time (e.g., during lunch or overnight) and you want to
stop the task whenever you return. It can be regarded as a way to tell
your program to *"get as much data as possible until I get back"*.
The following example illustrates how to repeatedly write data to a
file in a loop:
.. literalinclude:: ../../msl/examples/qt/loop_until_abort.py
Examples
--------
To run the above example enter the following::
>>> from msl.examples.qt import LoopExample # doctest: +SKIP
>>> loop = LoopExample() # doctest: +SKIP
>>> loop.start() # doctest: +SKIP
Another example which uses *single-shot* mode::
>>> from msl.examples.qt import LoopExampleSleep # doctest: +SKIP
>>> loop = LoopExampleSleep() # doctest: +SKIP
>>> loop.start() # doctest: +SKIP
Parameters
----------
loop_delay : :class:`int`, optional
The delay time, in milliseconds, to wait between successive calls
to the :meth:`loop` method. For example, if `loop_delay` = ``0``
then there is no time delay between successive calls to the
:meth:`loop` method; if `loop_delay` = ``1000`` then wait 1 second
between successive calls to the :meth:`loop` method.
max_iterations : :class:`int`, optional
The maximum number of times to call the :meth:`loop` method. The
default value is :data:`None`, which means to loop until the user
aborts the program.
single_shot : :class:`bool`, optional
Whether to call the :meth:`loop` method once (and only once). If
you specify the value to be :data:`True` then you must call the
:meth:`loop_once` method in the subclass whenever you want to
run the :meth:`loop` one more time. This is useful if the
:meth:`loop` depends on external factors (e.g., waiting for an
oscilloscope to download a trace after a trigger event) and the
amount of time that the :meth:`loop` requires to complete is not
known.
title : :class:`str`, optional
The text to display in the title bar of the
:class:`QtWidgets.QMainWindow`. If :data:`None` then uses the name
of the subclass as the title.
bg_color : :class:`QtGui.QColor`, optional
The background color of the :class:`QtWidgets.QMainWindow`. Can be
any data type and value that the constructor of a :class:`QtGui.QColor`
accepts.
text_color : :class:`QtGui.QColor`, optional
The color of the **Elapsed time** and **Iteration** text.Can be
any data type and value that the constructor of a :class:`QtGui.QColor`
accepts.
font_family : :class:`QtGui.QFont`, optional
The font family to use for the text. Can be any value that the
constructor of a :class:`QtGui.QFont` accepts.
font_size : :class:`int`, optional
The font size of the text.
"""
self._iteration = 0
self._loop_error = False
self._start_time = None
self._loop_delay = loop_delay
self._single_shot = single_shot
self._max_iterations = int(max_iterations) if max_iterations else None
self._app = application()
self._central_widget = QtWidgets.QWidget()
bg_hex_color = QtGui.QColor(bg_color).name()
self._central_widget.setStyleSheet(f'background:{bg_hex_color};')
self._main_window = QtWidgets.QMainWindow()
self._main_window.setCentralWidget(self._central_widget)
self._main_window.closeEvent = self._shutdown
if title is None:
title = self.__class__.__name__
self._main_window.setWindowTitle(title)
self._main_window.setWindowFlags(
Qt.WindowType.WindowCloseButtonHint | Qt.WindowType.WindowMinimizeButtonHint)
font = QtGui.QFont(font_family, pointSize=font_size)
text_hex_color = QtGui.QColor(text_color).name()
self._elapsed_time_label = QtWidgets.QLabel()
self._elapsed_time_label.setFont(font)
self._elapsed_time_label.setStyleSheet(f'color:{text_hex_color};')
self._iteration_label = QtWidgets.QLabel()
self._iteration_label.setFont(font)
self._iteration_label.setStyleSheet(f'color:{text_hex_color};')
self._user_label = QtWidgets.QLabel()
self._user_label.setFont(font)
vbox = QtWidgets.QVBoxLayout()
vbox.addWidget(self._elapsed_time_label)
vbox.addWidget(self._iteration_label)
vbox.addWidget(self._user_label)
vbox.setStretchFactor(self._user_label, 1)
self._central_widget.setLayout(vbox)
self._elapsed_time_timer = QtCore.QTimer()
self._elapsed_time_timer.timeout.connect(self._update_elapsed_time_label) # noqa: timeout.connect
self._loop_timer = QtCore.QTimer()
self._loop_timer.timeout.connect(self._call_loop) # noqa: timeout.connect
@property
def current_time(self):
""":class:`datetime.datetime`: The current time."""
return datetime.datetime.now()
@property
def elapsed_time(self):
""":class:`datetime.datetime`: The elapsed time since the :meth:`loop` started."""
return self.current_time - self._start_time
@property
def iteration(self):
""":class:`int`: The number of times that the :meth:`loop` method has been called."""
return self._iteration
@property
def loop_delay(self):
""":class:`int`: The time delay, in milliseconds, between successive calls to the :meth:`loop`."""
return self._loop_delay
@property
def loop_timer(self):
""":class:`QtCore.QTimer`: The reference to the :meth:`loop`\'s timer."""
return self._loop_timer
@property
def main_window(self):
""":class:`QtWidgets.QMainWindow`: The reference to the main window."""
return self._main_window
@property
def max_iterations(self):
""":class:`int` or :data:`None`: The maximum number of times that the :meth:`loop` will be called."""
return self._max_iterations
@property
def start_time(self):
""":class:`datetime.datetime`: The time when the :meth:`loop` started."""
return self._start_time
@property
def user_label(self):
""":class:`QtWidgets.QLabel`: The reference to the label object that the user can modify the text of.
See Also
--------
:meth:`set_label_text`
To set the text of the :class:`QtWidgets.QLabel`.
"""
return self._user_label
[docs] def cleanup(self):
"""This method gets called when the :class:`QtWidgets.QMainWindow` is closing.
You can override this method to properly clean up any tasks.
For example, to close a file that is open.
"""
pass
[docs] def loop(self):
"""The task to perform in a loop.
.. attention::
You **MUST** override this method.
"""
raise NotImplementedError("You must override the 'loop' method.")
[docs] def loop_once(self):
"""Run the :meth:`loop` method once.
This method should be called if the :class:`LoopUntilAbort` class
was initialized with `single_shot` = :data:`True`, in order to run the
:meth:`loop` method one more time.
"""
if not self._loop_timer:
raise RuntimeError('The loop timer has stopped')
if not self._loop_timer.isSingleShot():
self._stop_timers()
raise RuntimeError('Single shots are not enabled')
self._loop_timer.start(0)
[docs] def set_label_text(self, text):
"""Update the text of the label that the user has access to.
Parameters
----------
text : :class:`str`
The text to display in the user-accessible label.
See Also
--------
:meth:`user_label`
For the reference to the :class:`QtWidgets.QLabel` object.
"""
self._user_label.setText(text)
[docs] def set_status_bar_text(self, text):
"""Set the text to display in the status bar of the :class:`QtWidgets.QMainWindow`.
Parameters
----------
text : :class:`str`
The text to display in the status bar of the :class:`QtWidgets.QMainWindow`.
"""
self._main_window.statusBar().showMessage(text)
[docs] def start(self):
"""Show the :class:`QtWidgets.QMainWindow` and start looping."""
self._start_time = self.current_time
s = self._start_time.strftime('%d %B %Y at %H:%M:%S')
self.set_status_bar_text('Started ' + s)
self._update_elapsed_time_label()
self._update_iteration_label()
self._loop_delay = max(0, int(self._loop_delay)) if not self._single_shot else 0
if self._loop_delay > 0:
self._call_loop()
self._loop_timer.setSingleShot(self._single_shot)
self._loop_timer.start(self._loop_delay)
self._elapsed_time_timer.start(1000)
self._main_window.show()
self._app.exec()
def _call_loop(self):
"""Call the loop method once."""
if self._is_max_reached():
self._stop_timers()
msg = f'Maximum number of iterations reached ({self._iteration})'
self.set_status_bar_text(msg)
prompt.information(msg)
else:
self._iteration += 1
try:
self.loop()
except: # noqa: using bare 'except'
msg = 'The following exception occurred in the loop() method:\n\n'
prompt.critical(msg + traceback.format_exc())
self._loop_error = True
self._stop_timers()
err_time = self.current_time.strftime('%d %B %Y at %H:%M:%S')
self.set_status_bar_text('Error occurred on ' + err_time)
else:
self._update_iteration_label()
def _cleanup(self):
"""Wraps the cleanup method in a try-except block."""
try:
self.cleanup()
except: # noqa: using bare 'except'
msg = 'The following exception occurred in the cleanup() method:\n\n'
prompt.critical(msg + traceback.format_exc())
self._stop_timers()
def _is_max_reached(self):
"""Whether the maximum number of iterations was reached"""
return self._max_iterations is not None and self._iteration == self._max_iterations
def _shutdown(self, event):
"""Abort the loop"""
# check that it is okay to abort
if not self._is_max_reached() and not self._loop_error:
if not prompt.yes_no('Are you sure that you want to abort the loop?'):
# check again whether max_iterations was reached while the prompt window was displayed
if self._is_max_reached():
prompt.information('The maximum number of iterations was already reached.\n'
'Loop already aborted.')
event.ignore()
return
self._stop_timers()
self._cleanup()
event.accept()
def _stop_timers(self):
"""Stop and delete the timers."""
if self._loop_timer:
self._loop_timer.stop()
self._loop_timer = None
if self._elapsed_time_timer:
self._elapsed_time_timer.stop()
self._elapsed_time_timer = None
def _update_elapsed_time_label(self):
"""update the 'Elapsed time' label"""
dt = self.elapsed_time
hours, remainder = divmod(dt.seconds, 3600)
minutes, seconds = divmod(remainder, 60)
base = f'Elapsed time {hours:02d}:{minutes:02d}:{seconds:02d}'
if dt.days == 0:
self._elapsed_time_label.setText(base)
elif dt.days == 1:
self._elapsed_time_label.setText(f'{base} (+1 day)')
else:
self._elapsed_time_label.setText(f'{base} (+{dt.days} days)')
def _update_iteration_label(self):
"""update the `Iteration` label"""
self._iteration_label.setText(f'Iteration {self._iteration}')