# Virus Data Analysis Tool: a data reduction GUI for HETDEX/VIRUS data
# Copyright (C) 2016, 2017, 2018 "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/>.
'''This describe the central VDAT widget'''
from __future__ import (absolute_import, division, print_function,
unicode_literals)
import traceback
from pyhetdex.het.fplane import FPlane
from qtpy import QtCore, QtWidgets
import six
from . import fplane as vfplane
from . import tasks as vtasks
from . import overlay as voverlay
from . import queue as vqueue
import vdat.config as vconf
import vdat.command_interpreter as vci
import vdat.database as vdb
[docs]class VDATCentral(QtWidgets.QWidget):
"""This widget groups together the focal plane widget and the buttons into
a Qsplitter, to allow resizing and decouple them from the rest of the
gui.
.. list-table:: **Custom signals**
:header-rows: 1
* - Name
- Signature
- Description
* - :attr:`sig_ifuSelected`
- str, bool
- emitted when a user selects/deselects an IFU; the parameters are
the SLOTID of the IFU and whether the IFU has been selected
* - :attr:`sig_selectAllIFUs`
-
- emitted when a user selects all IFUs
* - :attr:`sig_deselectAllIFUs`
-
- emitted when a user deselects all IFUs
.. list-table:: **Custom slot**
:header-rows: 1
* - Name
- Signature
- Description
* - :meth:`change_target`
- str, str
- Change the view to the path passed as first argument; the second
argument is the type of files associated to the path.
* - :meth:`change_task`
- str, dict
- Switch the change the focal plane to the view for the ``task``
passed as first argument; the second argument contains the
dictionary describing the task tabs and buttons.
* - :meth:`ifuToggled`
- str, bool
- Mark ``ifuslot``, given in the first argument as selected, if the
second argument is True.
* - :meth:`selectAllIFUs`
-
- Select all the IFUs.
* - :meth:`deselectAllIFUs`
-
- Deselect all the IFUs.
* - :meth:`runCommand`
- str, list
- When clicking a button create the commands and submit them to the
queue
.. list-table:: **Connections between custom signals and/or slots**
:header-rows: 1
* - Signal
- Slot
* - :attr:`vdat.gui.tasks.VDATTaskStack.sig_taskChanged`
- :meth:`change_task`
* - :attr:`vdat.gui.tasks.VDATTaskStack.sig_runCommand`
- :meth:`runCommand`
* - :attr:`sig_selectAllIFUs`
- :meth:`vdat.gui.fplane.FplaneWidget.selectAllIFUs`
* - :attr:`sig_deselectAllIFUs`
- :meth:`vdat.gui.fplane.FplaneWidget.deselectAllIFUs`
* - :attr:`sig_ifuSelected`
- :meth:`vdat.gui.fplane.FplaneWidget.ifuSelected`
* - :attr:`vdat.gui.fplane.FplaneWidget.sig_ifuToggled`
- :meth:`ifuToggled`
Parameters
----------
parent : :class:`PyQt5.QtWidgets.QWidget` or derivate
parent object of the tree view model
"""
sig_ifuSelected = QtCore.Signal(str, bool)
sig_selectAllIFUs = QtCore.Signal()
sig_deselectAllIFUs = QtCore.Signal()
def __init__(self, parent=None):
super(VDATCentral, self).__init__(parent=parent)
conf = vconf.get_config('main')
self.tasks = vtasks.setup_tasks()
# self.tasks.setFixedWidth(150)
self.tasks.setCurrentWidgetByName('zro')
self.tasks.sig_taskChanged.connect(self.change_task)
self.tasks.sig_runCommand.connect(self.runCommand)
self.selectedIFUs = {}
self.fplane = vfplane.FplaneWidget()
self.fplane.setMinimumWidth(300)
exclude_ifuslot = conf.get_list('fplane', 'exclude_ifuslot',
use_default=True)
fp = FPlane(conf.get("fplane", "fp_filename"),
exclude_ifuslot=exclude_ifuslot)
for ifu in fp.ifuslots:
self.selectedIFUs[ifu] = False
self.sig_selectAllIFUs.connect(self.fplane.selectAllIFUs)
self.sig_deselectAllIFUs.connect(self.fplane.deselectAllIFUs)
self.sig_ifuSelected.connect(self.fplane.ifuSelected)
self.fplane.sig_ifuToggled.connect(self.ifuToggled)
self.layout = QtWidgets.QHBoxLayout(self)
self.layout.setContentsMargins(0, 0, 0, 0)
self.hsplit = QtWidgets.QSplitter(QtCore.Qt.Horizontal, parent=self)
self.hsplit.addWidget(self.tasks)
self.hsplit.addWidget(self.fplane)
self.hsplit.setSizes([100, 200])
self.hsplit.setStretchFactor(self.hsplit.indexOf(self.fplane), 1)
self.hsplit.setStretchFactor(self.hsplit.indexOf(self.tasks), 0)
self.layout.addWidget(self.hsplit)
self.setLayout(self.layout)
self.target = None
self.overlay = voverlay.Overlay('Please select a night on '
'the left to begin...', parent=self)
self.change_target(None, None)
[docs] @QtCore.Slot(str, str)
def change_target(self, target, typ):
'''Change the view to the path ``target`` of ``typ``. If the selected
type has not tasks associated, show the overlay.
This method is also a pyqt slot with signature ``str, str``.
Parameters
----------
target : string
path of the selected directory
typ : string
type of the ``target``
'''
self.setEnabled(True)
if target is not None:
self.overlay.hide()
self.target = target
try:
self.tasks.setCurrentWidgetByName(typ)
# self.tasks.currentWidget().selectFirst()
return
except vtasks.TaskError:
self.fplane.empty_fplane()
self.overlay.setText('No tasks defined in config, '
'please select another night '
'on the left')
self.setEnabled(False)
self.overlay.show()
[docs] @QtCore.Slot(str, dict)
def change_task(self, task, task_dict):
'''Switch the change the focal plane to the view for the ``task``
This method is also a pyqt slot with signature ``str, dict``.
Parameters
----------
task : string
name of the task to view
task_dict : dict
dictionary with the configuration for the tabs and buttons to
display
'''
self.fplane.change_fplane(self.target, task_dict)
# Loop over all selected IFUs and reselect them after the new Fplane
# was created
for k, v in six.iteritems(self.selectedIFUs):
if v:
# print('Selecting ', k, v)
self.sig_ifuSelected.emit(k, v)
[docs] @QtCore.Slot(str, bool)
def ifuToggled(self, ifuslot, val):
'''Mark ``ifuslot`` as selected or not
This method is also a pyqt slot with signature ``str, bool``.
Parameters
----------
ifuslot : str
SLOTID of one IFU
val : bool
True to select the IFU, False otherwise
'''
self.selectedIFUs[ifuslot] = val
self.sig_ifuSelected.emit(ifuslot, val)
[docs] @QtCore.Slot()
def selectAllIFUs(self):
'''Select all the IFU
This method is also a pyqt slot.
'''
self.sig_selectAllIFUs.emit()
[docs] @QtCore.Slot()
def deselectAllIFUs(self):
'''Deselect all the IFU
This method is also a pyqt slot.
'''
self.sig_deselectAllIFUs.emit()
[docs] @QtCore.Slot(str, list)
def runCommand(self, s, l):
'''When clicking a button create the commands and submit them to the
queue
This method is also a pyqt slot with signature ``str, list``
Parameters
----------
s : string
name of the button clicked or cumulative name of the commands to
execute
l : list
list of command to execute
'''
ifuslots = []
for k, v in six.iteritems(self.selectedIFUs):
if v:
ifuslots.append(k)
if not len(ifuslots):
box = QtWidgets.QMessageBox(parent=self)
box.setText("Please select the IFUs by clicking on them or via "
" the 'Select' menu")
box.setIcon(QtWidgets.QMessageBox.Warning)
box.exec_()
return
conf = vconf.get_config('main', section='general')
use_multiprocessing = conf.getboolean("use_multiprocessing")
n_processors = conf.getint("n_processors")
extra_kwargs = {'multiprocessing': use_multiprocessing,
'processes': n_processors}
submit_to_queue = []
for cmdstr in l:
try:
command = cmdstr.split()[0]
command_conf = self._config_with_dirs(command)
cmd_interpreter = vqueue.QCommandInterpreter(cmdstr,
command_conf,
selected=ifuslots,
**extra_kwargs)
submit_to_queue.append([cmd_interpreter, s + ": " + command,
cmdstr])
except Exception as e:
self._error_dialog(e, cmdstr)
return
queue = vqueue.get_queue()
for sq in submit_to_queue:
queue.add_command(*sq)
[docs] def _config_with_dirs(self, command_name):
"""Get the command configuration and add ``target_dir``, ``zero_dir``
and ``cal_dir`` to the configuration.
Parameters
----------
command_name : string
name of the command
Returns
-------
command_conf : dict
configuration dictionary for the command at hand
"""
command_conf = vconf.get_config('commands', section=command_name)
dirs_conf = vconf.get_config('main', section='redux_dirs')
# the target, or selected, directory is selected
command_conf['target_dir'] = dirs_conf['selected_dir']
# if no calibration and/or zero directory are selected, get them from
# the database
with vdb.connect():
db_entry = (vdb.VDATDir.select()
.where(vdb.VDATDir.path == command_conf['target_dir'])
.get())
db_zero_dir = getattr(db_entry.zero_dir, 'path', '')
db_cal_dir = getattr(db_entry.cal_dir, 'path', '')
command_conf['zero_dir'] = dirs_conf.get('zro_dir', db_zero_dir)
command_conf['cal_dir'] = dirs_conf.get('cal_dir', db_cal_dir)
return command_conf
[docs] def _error_dialog(self, error, command):
"""Create the dialog to show the error
Parameters
----------
error : :class:`Exception` instance
error raised by the constructor
command : string
full command string
"""
box = QtWidgets.QMessageBox(parent=self)
text = ("The following exceptions has been raised while"
" parsing the command: '{}'\n:{}".format(command, error))
box.setText(text)
if isinstance(error, (vci.exceptions.CIError,
vconf.MissingConfigurationSectionError)):
informative_text = ('Make sure that the command configuration'
' file and the command that you want to run'
' are in sync')
icon = QtWidgets.QMessageBox.Warning
else:
informative_text = ("The error is unexpected and it is probably a"
" bug. If you want to inspect the error click"
" on the 'Show details' button to see the full"
" traceback. If you want to report the error"
" to the developers, please ")
icon = QtWidgets.QMessageBox.Critical
box.setInformativeText(informative_text)
box.setIcon(icon)
tb_text = traceback.format_exc()
box.setDetailedText(tb_text)
box.exec_()
