# -*- coding: utf-8 -*-
# Virus Data Analysis Tool: a data reduction GUI for HETDEX/VIRUS data
# Copyright (C) 2015, 2016, 2017 "The HETDEX collaboration"
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Create a window representing a queue via a list of items.
The original implementation has been generated from reading ui file
'listWindow.ui'
Created: Mon Jun 15 16:25:52 2015
by: PyQt4 UI code generator 4.10.4
"""
from __future__ import (absolute_import, division, print_function,
unicode_literals)
import logging
from qtpy import QtCore, QtWidgets
from vdat.command_interpreter.core import CommandInterpreter
from vdat.command_interpreter.signals import get_signal
from vdat.command_interpreter.helpers import log_command_logger
_queue = []
[docs]class QueuedCommand(QtWidgets.QListWidgetItem):
"""Class describing the objects stored in the
:class:`ModifyableListWidget`.
Each object represent a command
Parameters
----------
command : :class:`QCommandInterpreter`
instance of the command interpreter to add to the queue
label : string
a label to appear for this command on the queue
tool_tip : string, optional
tool tip to show
parent : :class:`PyQt5.QtWidgets.QWidget` instance
the parent widget
"""
def __init__(self, command, label, tool_tip=None, parent=None):
super(QueuedCommand, self).__init__(parent=parent)
self.setText(label)
if tool_tip:
self.setToolTip(tool_tip)
self.command = command
[docs]class Queue(QtWidgets.QMainWindow):
"""A queue that stores user commands and displays them in a GUI window.
.. list-table:: **Custom signals**
:header-rows: 1
* - Name
- Signature
- Description
* - :attr:`closeSignal`
-
- emitted when the queue window is closed
* - :attr:`job_added`
-
- emitted when a job is added to the queue
* - :attr:`run_signal`
-
- emitted when a new command can be run
* - :attr:`queue_empty_signal`
-
- emitted when the queue is empty
* - :attr:`command_done`
- bool
- these five signals are a Qt re-implementation of the signals
described in the :mod:`~vdat.command_interpreter.signals`
* - :attr:`command_string`
- int, str
-
* - :attr:`global_logger`
- int, str
-
* - :attr:`n_primaries`
- int
-
* - :attr:`progress`
- int, int, int, int
-
.. list-table:: **Custom slot**
:header-rows: 1
* - Name
- Signature
- Description
* - :meth:`toggle`
- bool
- hide (False) or show (True) the panel
* - :meth:`run`
-
- grab a command from the queue and run it
* - :meth:`toggle_and_rerun`
-
- prepare to run a new command
.. list-table:: **Connections between custom signals and/or slots**
:header-rows: 1
* - Signal
- Slot/Signal
* - :attr:`job_added`
- :meth:`run`
* - :attr:`run_signal`
- :meth:`run`
* - :attr:`QCommandInterpreter.command_done`
- :attr:`command_done`
* - :attr:`QCommandInterpreter.command_string`
- :attr:`command_string`
* - :attr:`QCommandInterpreter.global_logger`
- :attr:`global_logger`
* - :attr:`QCommandInterpreter.n_primaries`
- :attr:`n_primaries`
* - :attr:`QCommandInterpreter.progress`
- :attr:`progress`
* - :attr:`PyQt5.QtCore.QThread.started`
- :meth:`QCommandInterpreter.run`
* - :attr:`QCommandInterpreter.finished`
- :meth:`toggle_and_rerun`
Parameters
----------
parent : :class:`PyQt5.QtWidgets.QWidget` instance
the parent widget
Attributes
----------
is_command_running : bool
mark whether a command is running or not in a thread
"""
closeSignal = QtCore.Signal()
job_added = QtCore.Signal()
run_signal = QtCore.Signal()
queue_empty_signal = QtCore.Signal()
# the queue re-emit the signals from the VDATCommandWorker
# This signals are connected to the worker ones and disconnected when the
# worker is done
command_done = QtCore.Signal(bool)
command_string = QtCore.Signal(int, str)
global_logger = QtCore.Signal(int, str)
n_primaries = QtCore.Signal(int)
progress = QtCore.Signal(int, int, int, int)
_reconnect_names = ['command_done', 'command_string', 'global_logger',
'n_primaries', 'progress']
def __init__(self, parent=None):
super(Queue, self).__init__(parent=parent)
self.log = logging.getLogger('logger')
self.itemList = []
self.is_command_running = False
self.setupUi()
self.job_added.connect(self.run)
self.run_signal.connect(self.run)
# save the thread and the worker into a list to remove them only
# when new ones are needed. This hopefully avoids deadlocks
self._threads_workers = []
[docs] def setupUi(self):
"""Setup the queue window """
self.setObjectName("vdatqueue")
self.resize(300, 380)
self.setWindowTitle("VDAT Command Queue")
self.centralwidget = QtWidgets.QWidget(self)
self.centralwidget.setObjectName("centralwidget")
self.listWidget = ModifyableListWidget(self.centralwidget)
self.listWidget.setGeometry(QtCore.QRect(5, 0, 290, 375))
self.listWidget.setObjectName("listwidget")
self.setCentralWidget(self.centralwidget)
[docs] def closeEvent(self, event):
"""When the user closes the window, ignore the request, hide the window
and emit the :attr:`closeSignal` signal.
Parameters
----------
event : :class:`PyQt5.QtGui.QKeyEvent`
object describing a key being pressed or released
"""
event.ignore() # Ignore the request
self.closeSignal.emit()
self.setVisible(False) # Just hide it instead
[docs] @QtCore.Slot(bool)
def toggle(self, tggl):
"""Hide or show the panel. Alias of :meth:`setVisible`.
Parameters
----------
toggle : bool
whether the window is visible or not
"""
self.setVisible(tggl)
[docs] def add_command(self, command, label, tool_tip=None):
"""Add a command to the queue. Emit the :attr:`job_added` signal.
Parameters
----------
command : :class:`QCommandInterpreter`
instance of the command interpreter to add to the queue
label : string
A label to appear for this command on the queue
tool_tip : string, optional
tool tip to show
"""
item = QueuedCommand(command, label, tool_tip=tool_tip,
parent=self.listWidget)
self.listWidget.addItem(item)
self.job_added.emit()
[docs] def get_command(self):
"""Get the top item from the queue.
Returns
-------
:class:`QCommandInterpreter` instance
command to run, or None if the list is empty
"""
item = self.listWidget.takeItem(0)
if isinstance(item, QueuedCommand):
return item.command
else:
return None
[docs] def connect_worker_signals(self, worker):
'''Connect the ``worker`` signals to the corresponding ones in this
class
Parameters
----------
worker : :class:`QCommandInterpreter`
worker instance with the signals to connect with this object ones
'''
for name in self._reconnect_names:
getattr(worker, name).connect(getattr(self, name))
[docs] def remove_old_thread_worker(self):
'''Quit the used thread and mark the it and worker for
deletion'''
for old_thread, old_worker in self._threads_workers:
old_worker.deleteLater()
old_thread.quit()
old_thread.deleteLater()
self._threads_workers = []
[docs] @QtCore.Slot()
def run(self):
"""Grab a job from the queue and run it on a
newly created :class:`~PyQt5.QtCore.QThread`.
If a command is already running, and return.
If the queue is empty notify the user and return.
The :class:`~PyQt5.QtCore.QThread` and :class:`QCommandInterpreter`
instances are saved in a local cache and removed the next time
:meth:`run` is called.
"""
if self.is_command_running:
self.log.info("Added command to queue")
return
command = self.get_command()
if not command:
self.log.info("All commands on queue completed!")
self.queue_empty_signal.emit()
else:
self.is_command_running = True
self.log.debug("Launching new command...")
# Make a worker to run the function, move it to the background
# thread. The worker cannot be a local variable otherwise it goes
# out of scope as soon as we exit this function!
thread = QtCore.QThread(self)
self.connect_worker_signals(command)
self._threads_workers.append([thread, command])
command.moveToThread(thread)
# start to run the worker when starting the thread
thread.started.connect(command.run)
# Grab the next job, toggle is_command_running
command.finished.connect(self.toggle_and_rerun)
thread.start()
[docs] @QtCore.Slot()
def toggle_and_rerun(self):
"""Remove the old thread and command, mark that the command is not
running and emit the :attr:`run_signal` signal"""
self.remove_old_thread_worker()
self.is_command_running = False
self.run_signal.emit()
# Mixing Qt classes with other classes can be tricky. See:
# http://python.6.x6.nabble.com/Issue-with-multiple-inheritance-td5207771.html
# http://pyqt.sourceforge.net/Docs/PyQt5/multiinheritance.html#support-for-cooperative-multi-inheritance
# for some reason this does not affect PyQt4 (probably because needs to cope
# with old style classes.
[docs]class QCommandInterpreter(CommandInterpreter, QtCore.QObject):
'''Create a QObject from the
:class:`~vdat.command_interpreter.core.CommandInterpreter`.
Reimplement the
:meth:`~vdat.command_interpreter.core.CommandInterpreter.make_signals` to
use only the ``command_logger`` signals from the command_interpreter and
use :class:`~PyQt5.QtCore.Signal` for the other signals
.. list-table:: **Custom signals**
:header-rows: 1
* - Name
- Signature
- Description
* - :attr:`command_done`
- bool
- these five signals are a Qt re-implementation of the signals
described in the :mod:`~vdat.command_interpreter.signals`
* - :attr:`command_string`
- int, str
-
* - :attr:`global_logger`
- int, str
-
* - :attr:`n_primaries`
- int
-
* - :attr:`progress`
- int, int, int, int
-
* - :attr:`finished`
-
- emitted when the run is finished
.. list-table:: **Custom slot**
:header-rows: 1
* - Name
- Signature
- Description
* - :meth:`run`
-
- Run the command
Parameters
----------
parent : :class:`PyQt5.QtWidgets.QWidget` instance
the parent widget
args, kwargs:
passed to the
:class:`~vdat.command_interpreter.core.CommandInterpreter`
'''
# signals from the original command interpreter
command_done = QtCore.Signal(bool)
command_string = QtCore.Signal(int, str)
global_logger = QtCore.Signal(int, str)
n_primaries = QtCore.Signal(int)
progress = QtCore.Signal(int, int, int, int)
# signal emitted when command_done(True) is emitted
finished = QtCore.Signal()
def __init__(self, *args, **kwargs):
# Extract the ``parent`` from ``kwargs`` and initialize
# :class:`~PyQt5.QtCore.QObject` and pass all the other arguments and
# kwargs to the command interpreter
parent = kwargs.pop('parent', None)
CommandInterpreter.__init__(self, *args, **kwargs)
QtCore.QObject.__init__(self, parent=parent)
# self.finished.connect(self.disconnect_signals)
[docs] def make_signals(self):
'''Use Signals instead of
:mod:`vdat.command_interpreter.signals`, except for the command_logger
one'''
self.command_logger = get_signal('command_logger')
[docs] def connect_signals(self):
self.command_logger.connect(log_command_logger)
[docs] def disconnect_signals(self):
self.command_logger.disconnect(log_command_logger)
[docs] @QtCore.Slot()
def run(self):
'''Transform the method to a Slot'''
self.connect_signals()
super(QCommandInterpreter, self).run()
self.disconnect_signals()
self.finished.emit()
[docs]class QueuAction(QtWidgets.QAction):
"""Action for the queue window.
Create the menu entry and binds signals known by the queue window to
show/hide it
.. list-table:: **Custom slot**
:header-rows: 1
* - Name
- Signature
- Description
* - :meth:`update_text`
- bool
- change the text to show in the action item
.. list-table:: **Connections between custom signals and/or slots**
:header-rows: 1
* - Signal
- Slot/Signal
* - :attr:`toggled`
- :meth:`update_text`
* - :attr:`Queue.closeSignal`
- :meth:`toggle`
* - :attr:`toggled`
- :meth:`Queue.toggle`
Parameters
----------
*args, **kargs:
arguments passed to the parent class
"""
def __init__(self, *args, **kwargs):
super(QueuAction, self).__init__(*args, **kwargs)
self.setText("Hide queue")
self.setCheckable(True)
self.toggle()
self.toggled.connect(self.update_text)
self.connect_with_queue()
[docs] def connect_with_queue(self):
'''Connect with signals and slots in :class:`Queue`'''
queue = get_queue()
queue.closeSignal.connect(self.toggle)
self.toggled.connect(queue.toggle)
[docs] @QtCore.Slot(bool)
def update_text(self, toggled):
'''Update the text in the action when the queue window is opened or
closed
Parameters
----------
toggled : bool
whether is checked or not
'''
if toggled:
self.setText("Hide queue")
else:
self.setText("Show queue")
[docs]def set_queue(parent=None):
"""Create a :class:`Queue` instance and save it. You can access it with
:func:`get_queue`
Parameters
----------
parent : :class:`PyQt5.QtWidgets.QWidget` instance
the parent widget
"""
_queue.append(Queue(parent=parent))
[docs]def get_queue():
"""Get the locally stored :class:`Queue` instance"""
return _queue[0]
