# Library with high level interface to HET's TCS
# Copyright (C) 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/>.
'''Proxy for the python interface to the HET Telescope
Control System (TCS).
This module tries to import ``tcssubsystem`` and ``TCSLog``:
* if it succeeds, initialize all the requested TCS instances.
* otherwise use some mock classes that log every command are used
'''
from __future__ import (absolute_import, division, print_function,
unicode_literals)
try:
# Python 3 or older, or with the ``backport.configparser`` package
import configparser
except ImportError: # python 2
import ConfigParser as configparser
import logging
import logging.config as logconf
import traceback
import warnings
import six
try:
import TCSLog
except ImportError:
TCSLog = None
try:
import tcssubsystem
except ImportError:
tcssubsystem = None
[docs]class TCSProxy(object):
'''Proxy implementation. One instance is exposed as :mod:`tcs_lib.tcs_proxy`.
The only method public method is :meth:`init` and must be called before the
proxy attributes can be used.
Parameters
----------
mod_name : string, optional
name of the module to use by the mock ``TCSLog`` and ``tcssubsystem``
Attributes
----------
mod_name
tcs_log : :class:`TCSLog.TCSLog` or :class:`_MockTCSLog`
instance of the TCSLog class or of the corresponding mock version
{name} : :class:`tcssubsystem.TCSSubSystem` or :class:`_MockTCSSubsystem`
for each name in the ``subsystem_names`` configuration option, instance
of the TCSSubSystem class or of the corresponding mock version
errors
proxy objects for TCS exceptions. If the :mod:`tcssubsystem` module is
found, this attribute it's an alias of the module, otherwise it is an
instances of :class:`_MockErrors`
'''
def __init__(self, mod_name=__name__):
self._mod_name = mod_name
self._initialized = False
self._tcs_systems = {}
@property
def mod_name(self):
'''Name of the module used to initialize the proxy. Can be modified
only before calling :meth:`init`'''
return self._mod_name
@mod_name.setter
def mod_name(self, value):
if self._initialized:
raise AttributeError("can't set attribute")
else:
self._mod_name = value
[docs] def init(self, conf, section='urls'):
'''Initialize the TCSLog and tcssubsystems, or their mock counterpart.
The following configuration entries from the ``section`` are used:
* ``tcs_log``, optional: url to use to initialise the
:class:`TCSLog.TCSLog` class. If not given, empty or the
:mod:`TCSLog` module is not found, initialize a mock log class
* ``tcs_log_mock_path``, optional: if the :class:`TCSLog.TCSLog` cannot
be initialized, a mock class is used instead. This class is a proxy
for a standard python logger with name ``{mod_name}.tcs_log``. By
default the :class:`~logging.NullHandler` is attached. However it is
possible to customize the loggers, using a configuration file
according to `the logger configuration file format
<https://docs.python.org/3/library/logging.config.html#configuration-file-format>`_.
The name of this configuration file can be passed using the
``tcs_log_mock_path`` option.
* ``subsystem_names``, optional: list of comma separated names of TCS
subsystems to initialize. For each ``name`` the following options are
used
* ``{name}``, optional: url to use to initialise the
:class:`tcssubsystem.TCSSubSystem` class. If not given, empty or
the :mod:`tcssubsystem` module is not found, initialize a mock
subsystem class
* ``{name}_mock_path``, optional: the option has the same meaning of
the ``tcs_log_mock_path`` one, with the difference that the python
logger name use is ``{mod_name}.{name}``
.. note::
If the logger configuration file contains the configurations for
the mock logger and subsystems, there is no need to assign it to
every ``*_mock_path``, but only to the first one used, e.g.
``tcs_log_mock_path``.
Parameters
----------
conf : :class:`configparser.ConfigParser` instance
configuration files necessary to initialize the TCS log and
subsystems.
section : string, optional
name of the section containing the configuration
Raises
------
'''
if tcssubsystem:
self._tcs_systems['errors'] = tcssubsystem
else:
self._tcs_systems['errors'] = _MockErrors()
try:
self._tcs_systems['tcs_log'] = self._init_tcs_log(conf, section)
self.tcs_log.log_info('TCSLog initialized')
try:
subsystem_names = conf.get(section, 'subsystem_names')
subsystem_names = [s.strip()
for s in subsystem_names.split(',')]
except configparser.NoOptionError:
subsystem_names = []
for name in subsystem_names:
subsys = self._init_tcs_subystem(name, conf, section)
self._tcs_systems[name] = subsys
self.tcs_log.log_info('TCS subsystem {} initialized', name)
except configparser.NoSectionError as e:
msg = ('The input section "[{}]" must be present in the'
' input configuration'.format(section))
six.raise_from(configparser.NoSectionError(msg), e)
else:
self._initialized = True
[docs] def clear(self):
'''Clear the attributes created by :meth:`init` and reset the status to
uninitialized.
'''
self._tcs_systems.clear()
self._initialized = False
[docs] def _init_tcs_log(self, conf, section):
'''Initialize the tcs logging
Parameters
----------
conf : :class:`configparser.ConfigParser` instance
configuration files necessary to initialize the TCS log and
subsystems.
section : string, optional
name of the section containing the configuration
Returns
-------
:class:`TCSLog.TCSLog` or :class:`_MockTCSLog`
'''
name = 'tcs_log'
try:
url = conf.get(section, 'tcs_log')
except configparser.NoOptionError:
url = None
try:
mock_conf = conf.get(section, 'tcs_log_mock_path')
except configparser.NoOptionError:
mock_conf = None
if TCSLog and url:
_tcs_log = TCSLog.TCSLog(self.mod_name + '.' + name, url)
else:
_tcs_log = _MockTCSLog(name, self.mod_name, mock_conf)
return _tcs_log
[docs] def _init_tcs_subystem(self, name, conf, section):
'''Initialize the TCS subsystem or mock subsystem and return it.
Parameters
----------
name : string
name of the subsystem
conf : :class:`configparser.ConfigParser` instance
configuration files necessary to initialize the TCS log and
subsystems.
section : string, optional
name of the section containing the configuration
Returns
-------
:class:`tcssubsystem.TCSSubSystem` or :class:`_MockTCSSubsystem`
'''
try:
url = conf.get(section, name)
except configparser.NoOptionError:
url = None
try:
mock_conf = conf.get(section, '{}_mock_path'.format(name))
except configparser.NoOptionError:
mock_conf = None
if tcssubsystem and url:
_tcs_sub = tcssubsystem.TCSSubSystem(self.mod_name + '.' + name,
url)
else:
_tcs_sub = _MockTCSSubsystem(name, self.mod_name, mock_conf)
return _tcs_sub
[docs] def __getattr__(self, name):
'''Try to get TCS log/subsystem called ``name``. For sure this will
fail if used before calling :meth:`init`
Parameters
----------
name : string
attribute name
Raises
------
AttributeError
if name is not a valid attribute
'''
try:
return self._tcs_systems[name]
except KeyError:
msg = "'TCSLog' object has no attribute '{}'"
raise AttributeError(msg.format(name))
[docs]class _MockErrors(object):
'''Instances of this class return an :exc:`Exception` as attribute.
'''
def __getattr__(self, name):
return Exception
[docs]class _MockTCSSubsystem(object):
'''Class that log the methods called and their positional and keyword
arguments.
The name of the logger used is ``{mod_name}.{name}`` where name is the name
to the constructor.
In order to customize this logger, use the ocd configuration file as
described `here
<https://docs.python.org/3/library/logging.config.html#configuration-file-format>`_.
If ``log_conf_file`` is an empty string or if the configuration parsing
fails, the :class:`logging.NullHandler` is added to the logger used here.
In case of a parsing failure a warning will be printed.
Parameters
----------
name: string
name of the instance created
mod_name : string
name of module to associate to the mock proxy objects; used to create
the logger name
log_conf_file : string
name of the configuration files necessary to initialize the mock tcs
log
'''
def __init__(self, name, mod_name, log_conf_file):
self._name = name
log_config_fail = False
logger_name = mod_name + '.' + name
if log_conf_file:
# if all the basic sections are there, try to load the
# configuration. If it fails warn the user about this
try:
logconf.fileConfig(log_conf_file)
except Exception:
log_config_fail = True
msg = 'The configuration of the logger failed because of:\n'
warnings.warn(msg + traceback.format_exc())
if not log_conf_file or log_config_fail:
log = logging.getLogger(name=logger_name)
log.addHandler(logging.NullHandler())
self._log = logging.getLogger(name=logger_name)
[docs] def __getattr__(self, name):
'''Return a function that logs the call
Parameters
----------
name : string
attribute name
'''
return _LogCall(self._log, self._name, name)
[docs]class _MockTCSLog(_MockTCSSubsystem):
'''Replacement for the :class:`TCSLog.TCSLog` when the :mod:`TCSLog` module
is not available.
This class translates the :class:`TCSLog.TCSLog` ``log_*`` calls to the
appropriate python logging levels:
+--------------+---------------+
| method | log level |
+==============+===============+
| log_debug | 10 (DEBUG) |
+--------------+---------------+
| log_info | 20 (INFO) |
+--------------+---------------+
| log_warn | 30 (WARNING) |
+--------------+---------------+
| log_error | 40 (ERROR) |
+--------------+---------------+
| log_fatal | 50 (CRITICAL) |
+--------------+---------------+
| log_alarm | 60 |
+--------------+---------------+
The :meth:`TCSLog.TCSLog.log_*` methods have the following signature::
log_info(msg, *args)
The log message is formatted with ``msg.format(*args)`` before emitting it.
.. warning::
Because of the formatting, log messages are emitted by a function in
**this** module and not in the place where the ``log_*`` method is
called.
The name of the logger used is ``{mod_name}.{name}``. When initialised by
:class:`TCSProxy` ``{name}`` is ``tcs_log``.
Parameters
----------
name: string
name of the logger to use
mod_name : string
name of module to associate to the mock proxy objects; used to create
the logger name
log_conf_file : string
name of the configuration files necessary to initialize the mock tcs
log
'''
def __init__(self, name, mod_name, log_conf_file):
super(_MockTCSLog, self).__init__(name, mod_name, log_conf_file)
# add an ALARM level to the logger
logging.addLevelName(60, 'ALARM')
# map between TCSLog methods and logging levels
self._method_to_level = {'log_debug': logging.DEBUG,
'log_info': logging.INFO,
'log_warn': logging.WARNING,
'log_error': logging.ERROR,
'log_fatal': logging.CRITICAL,
'log_alarm': 60,
}
[docs] def __getattr__(self, name):
'''convert calls to log_*.
Parameters
----------
name : string
attribute name
Raises
------
AttributeError
if name is not one of the valid log_* names
'''
try:
level = self._method_to_level[name]
def _log(msg, *args):
self._log.log(level, msg.format(*args))
return _log
except KeyError:
msg = "'TCSLog' object has no attribute '{}'"
raise AttributeError(msg.format(name))
[docs]class _LogCall(object):
'''Callable class log the call to function with all its positional and
keyword arguments.
The function is logged as info.
Parameters
----------
log : :class:`logging.Logger`
logger
obj, function : string
name of the object and the function called
'''
def __init__(self, log, obj, function):
self._log = log
self._obj = obj
self._function = function
[docs] def __call__(self, *args, **kwargs):
'''Log the call of the function as ``"{obj}.{function}({args},
{kwargs})"``.
Returns
-------
:class:`_MockResponse`
current instance that mimic the dictionary returned by each
TCSSubSystem method
'''
args_ = ', '.join(str(i) for i in args)
kwargs_ = ', '.join('{}={}'.format(k, v) for k, v in kwargs.items())
separator = ', ' if (args and kwargs) else ''
msg = '{o}.{f}({a}{s}{k})'
self._log.info(msg.format(o=self._obj, f=self._function, a=args_,
s=separator, k=kwargs_))
return _MockResponse(args=args, **kwargs)
[docs]class _MockResponse(dict):
'''Extension of a dictionary that returns the key if not present.
Examples
--------
>>> d = _MockResponse(a=42)
>>> d['a']
42
>>> key = 'b'
>>> d[key] == key
True
'''
[docs] def __getitem__(self, key):
'''If the key is not found, return it, otherwise returns the associated
value'''
try:
return super(_MockResponse, self).__getitem__(key)
except KeyError:
return key