Source code for mdt.gui.utils
import os
import time
import sys
import traceback
from PyQt5 import QtCore
from contextlib import contextmanager
from functools import wraps
from PyQt5.QtCore import QObject, pyqtSignal
from PyQt5.QtCore import QTimer
from PyQt5.QtWidgets import QApplication
from mdt.__version__ import __version__
from mdt.lib.log_handlers import LogListenerInterface
__author__ = 'Robbert Harms'
__date__ = "2015-08-20"
__maintainer__ = "Robbert Harms"
__email__ = "robbert@xkls.nl"
[docs]class QtManager:
windows = []
[docs] @staticmethod
def get_qt_application_instance():
q_app = QApplication.instance()
if q_app is None:
q_app = QApplication([])
q_app.lastWindowClosed.connect(QtManager.empty_windows_list)
return q_app
[docs] @staticmethod
def exec_():
if QtManager.windows:
QtManager.get_qt_application_instance().exec_()
[docs] @staticmethod
def add_window(window):
QtManager.windows.append(window)
[docs] @staticmethod
def empty_windows_list():
QtManager.windows = []
[docs]def center_window(window):
"""Center the given window on the screen.
Args:
q_app (QApplication): for desktop information
window (QMainWindow): the window to center
"""
q_app = QApplication.instance()
frame_gm = window.frameGeometry()
screen = q_app.desktop().screenNumber(q_app.desktop().cursor().pos())
center_point = q_app.desktop().screenGeometry(screen).center()
frame_gm.moveCenter(center_point)
window.move(frame_gm.topLeft())
[docs]def function_message_decorator(header, footer):
"""This creates and returns a decorator that prints a header and footer before executing the function.
Args:
header (str): the header text, we will add extra decoration to it
footer (str): the footer text, we will add extra decoration to it
Returns:
decorator function
"""
def _called_decorator(dec_func):
@wraps(dec_func)
def _decorator(*args, **kwargs):
print('')
print(header)
print('-'*20)
response = dec_func(*args, **kwargs)
print('-'*20)
print(footer)
return response
return _decorator
return _called_decorator
[docs]@contextmanager
def blocked_signals(*widgets):
"""Small context in which the signals of the given widget are blocked.
Args:
widgets (QWidget): one or more widgets
"""
def apply_block(bool_val):
for w in widgets:
w.blockSignals(bool_val)
apply_block(True)
yield
apply_block(False)
[docs]def print_welcome_message():
"""Prints a small welcome message for after the GUI has loaded.
This prints to stdout. We expect the GUI to catch the stdout events and redirect them to the GUI.
"""
from mdt import VERSION
print('Welcome to MDT version {}.'.format(VERSION))
print('')
print('This area is reserved for log output.')
print('-------------------------------------')
[docs]class ForwardingListener(LogListenerInterface):
def __init__(self, queue):
"""Forwards all incoming messages to the given _logging_update_queue.
Instances of this class can be used as a log listener to the MDT LogDispatchHandler and as a
sys.stdout replacement.
Args:
queue (Queue): the _logging_update_queue to forward the messages to
"""
self._queue = queue
[docs] def emit(self, record, formatted_message):
self._queue.put(formatted_message + "\n")
[docs] def write(self, string):
self._queue.put(string)
image_files_filters = ['All files (*)',
'Nifti (*.nii *.nii.gz)',
'IMG, HDR (*.img)']
protocol_files_filters = ['MDT protocol (*.prtcl)',
'Text files (*.txt)',
'All files (*)']
[docs]class UpdateDescriptor:
def __init__(self, attribute_name):
"""Descriptor that will emit a state_updated_signal at each update.
This accesses from the instance the attribute name prepended with an underscore (_).
"""
self._attribute_name = attribute_name
def __get__(self, instance, owner):
return getattr(instance, '_' + self._attribute_name)
def __set__(self, instance, value):
setattr(instance, '_' + self._attribute_name, value)
instance.state_updated_signal.emit(self._attribute_name)
[docs]class MessageReceiver(QObject):
text_message_signal = pyqtSignal(str)
finished = pyqtSignal()
def __init__(self, queue, *args, **kwargs):
"""A QObject (to be run in a QThread) which sits waiting for data to come through a Queue.Queue().
It blocks until data is available, and one it has got something from the _logging_update_queue, it sends
it to the "MainThread" by emitting a Qt Signal.
Attributes:
is_running (boolean): set to False to stop the receiver.
"""
super().__init__(*args, **kwargs)
self.queue = queue
self.is_running = True
[docs] def run(self):
while self.is_running:
if not self.queue.empty():
self.text_message_signal.emit(self.queue.get())
time.sleep(0.001)
self.finished.emit()
[docs]class MainTab:
[docs] def tab_opened(self):
"""Called when this tab is selected by the user."""
class TimedUpdate(QTimer):
def __init__(self, update_cb):
"""Creates a timer that can delay running a given callback function.
Every time the user adds a delayed callback the timer gets reset to the new value and we will
wait that new value until calling the callback with the last data given.
Args:
update_cb (function): the function we would like to run after a timer has run out
"""
super().__init__()
self._cb_values = []
self._update_cb = update_cb
self.timeout.connect(self._call_update_cb)
self.timeout.connect(self.stop)
def add_delayed_callback(self, delay, *cb_values):
"""Pushes a new delay to calling the callback function.
Args:
delay (int): the time in ms to wait
cb_values (*list): the list of values to use as arguments to the callback function. Leave empty to disable.
"""
self._cb_values = cb_values
self.start(delay)
def _call_update_cb(self):
if self._cb_values:
self._update_cb(*self._cb_values)
else:
self._update_cb()
[docs]def split_long_path_elements(original_path, max_single_element_length=25):
"""Split long path elements into smaller ones using spaces
Args:
original_path (str): the path you want to split
max_single_element_length (int): the maximum length allowed per path component (folders and filename).
Returns:
str: the same path but with spaces in long path elements. The result will no longer be a valid path.
"""
def split(p):
listing = []
def _split(el):
if el:
head, tail = os.path.split(el)
if not tail:
listing.append(head)
else:
_split(head)
listing.append(tail)
_split(p)
return listing
elements = list(split(original_path))
new_elements = []
for el in elements:
if len(el) > max_single_element_length:
item = ''
for i in range(0, len(el), max_single_element_length):
item += el[i:i + max_single_element_length] + ' '
item = item[:-1]
new_elements.append(item)
else:
new_elements.append(el)
return os.path.join(*new_elements)
[docs]def enable_pyqt_exception_hook():
"""Enable the PyQt exception handling hook for PyQt versions larger than 5.5.
If this is not enabled, exceptions will be handled silently and will not be printed to the user. This
makes it harder to solve the issue.
"""
if QtCore.QT_VERSION >= 0x50501:
old_stdout = sys.stdout
old_stderr = sys.stderr
def excepthook(type_, value, traceback_):
sys.stdout = old_stdout
sys.stderr = old_stderr
traceback.print_exception(type_, value, traceback_)
QtCore.qFatal('')
sys.excepthook = excepthook