# Virus Data Analysis Tool: a data reduction GUI for HETDEX/VIRUS data
# Copyright (C) 2015, 2016 "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/>.
""":class:`~vdat.command_interpreter.core.CommandInterpreter` uses `PyQt like
signals
<http://pyqt.sourceforge.net/Docs/PyQt5/signals_slots.html#PyQt5.QtCore.pyqtSignal>`_
to communicate with the external word.
The names of the ``emit`` methods arguments are the type of the parameter
followed by an underscore and optionally by an explanatory name.
Available signals are:
* ``command_string``: accept an int and a string;
.. automethod:: CICommandString.emit
:noindex:
* ``progress``: accept four numbers, the total expected number, the number of
done, of skipped and of failures;
.. automethod:: CIProgress.emit
:noindex:
* ``command_done``: accept a boolean: ``True`` for the end of the command
interpreter, ``False`` for the end of one single command;
.. automethod:: CICommandDone.emit
:noindex:
* ``global_logger``: accept an integer and a string
.. automethod:: CIGlobalLogger.emit
:noindex:
* ``n_primaries``: accept an integer
.. automethod:: CINPrimaries.emit
:noindex:
* ``commands``: accept six strings and a dictionary. The first string is the
primary value; the second the command with all the substitutions in
place, if the execution finished, or with the placeholder, if some exception
has been raised; the third and fourth are the stdout and stderr of the
executed command; the fifth is non empty if the command has a non-null return
code; the sixth is non empty is the execution of the command crashes for some
reason; the seventh is the configuration dictionary passed to the
:class:`~vdat.command_interpreter.core.CommandInterpreter`
.. automethod:: CINPrimaries.emit
:noindex:
"""
from __future__ import (absolute_import, division, print_function,
unicode_literals)
import abc
import six
_ci_signals = {}
[docs]def register(name):
"""Initialise the decorated class and save it into the signals
dictionary with ``name`` as key
Parameters
----------
name : string
name under which the class or instance must be registered
Returns
-------
decorator : function
class decorator
"""
def decorator(cls):
"save the class into the ``_registered`` dictionary"
_ci_signals[name] = cls()
return cls
return decorator
[docs]@six.add_metaclass(abc.ABCMeta)
class BaseCISignal(object):
"""Base implementation for the signals used in the
:class:`~vdat.command_interpreter.core.CommandInterpreter`
Attributes
----------
connected : list
list of connected callables
"""
def __init__(self):
self._connected = []
@property
def connected(self):
"list of the connected callables"
return self._connected
[docs] def connect(self, callable_):
"""Connect the callable with the signal
Parameters
----------
callable_ :
function/method/callable instance to connect with the signal
"""
self._connected.append(callable_)
[docs] def disconnect(self, callable_):
"""Remove the first instance of callable from the signal
Parameters
----------
callable_ :
function/method/callable instance to disconnect from the signal
Raises
------
ValueError
if the callable is not already connected
"""
self._connected.remove(callable_)
[docs] def clear(self):
"""Remove all the associated signals"""
self._connected = []
[docs] @abc.abstractmethod
def emit(self, *args, **kwargs):
"""Abstract method that emit the signal, executing all the registered
callables.
The default implementation loop through the :attr:`connected` and call
them.
Reimplement to give it a proper signature.
"""
for c in self.connected:
c(*args, **kwargs)
[docs]@register("command_string")
class CICommandString(BaseCISignal):
"""Emit the string of the command to execute after substituting the
keywords"""
[docs] def emit(self, int_, string_):
"""
Parameters
----------
int_ : int
loop number
string_ : string
string of the command
"""
super(CICommandString, self).emit(int_, string_)
[docs]@register("progress")
class CIProgress(BaseCISignal):
"""Emit informations about execution progress"""
[docs] def emit(self, int_tot, int_done, int_skipped, int_fail):
"""
Parameters
----------
int_tot : int
total number of jobs
int_done : int
number of finished jobs; the number of successful jobs is
``int_done - int_skipped - int_fail``
int_skipped : int
number of skipped jobs
int_fail : int
number of failed jobs
"""
super(CIProgress, self).emit(int_tot, int_done, int_skipped, int_fail)
[docs]@register("command_done")
class CICommandDone(BaseCISignal):
"""Signal emitted when a command has been executed or the run of the
interpreter is done"""
[docs] def emit(self, bool_global):
"""
Parameters
----------
bool_global : boolean
if ``True``, the command interpreter is done, if ``False`` a single
command is done
"""
super(CICommandDone, self).emit(bool_global)
[docs]@register("global_logger")
class CIGlobalLogger(BaseCISignal):
"""Log information about the command interpreter execution. This channel is
not used to log the execution of the commands.
"""
[docs] def emit(self, int_level, string_msg):
"""
Parameters
----------
int_level : integer
logging level; see the `logging documentation
<https://docs.python.org/3/library/logging.html#logging-levels>`_
for more information
string_msg : string
string to log
"""
super(CIGlobalLogger, self).emit(int_level, string_msg)
[docs]@register('n_primaries')
class CINPrimaries(BaseCISignal):
"""Emit the number of primary files collected
"""
[docs] def emit(self, int_):
"""
Parameters
----------
int_ : integer
number of primary files
"""
super(CINPrimaries, self).emit(int_)
[docs]@register('command_logger')
class CICommandLogger(BaseCISignal):
"""For each primary emits the primary name(s), the actual command string
executed in the subprocess, the log messages and the full configuration
dictionary.
"""
[docs] def emit(self, string_primary, string_command, string_info, string_warning,
string_error, string_critical, dict_config):
"""
Parameters
----------
int_ : integer
number of primary files
"""
super(CICommandLogger, self).emit(string_primary, string_command,
string_info, string_warning,
string_error, string_critical,
dict_config)
[docs]def get_signal(name):
"""Get the signal associated with ``name``
Parameters
----------
name : string
name of the requested relay
Returns
-------
one of the registered signal instances
"""
return _ci_signals[name]
[docs]def get_signal_names():
"""Get the full list of signals"""
return list(_ci_signals.keys())
