Source code for nxswriter.StreamSet

#!/usr/bin/env python
#   This file is part of nexdatas - Tango Server for NeXus data writer
#
#    Copyright (C) 2012-2017 DESY, Jan Kotanski <jkotan@mail.desy.de>
#
#    nexdatas 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.
#
#    nexdatas 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 nexdatas.  If not, see <http://www.gnu.org/licenses/>.
#

""" labels to Tango Streams """

import sys
import weakref
import datetime


# (:obj:`bool`) write stream to stdout
stdoutflag = False

# (:obj:`bool`) write stream to stderrr
stderrflag = True


[docs]class StreamSet(object): def __init__(self, streams): """ streamset constractor :param streams: tango-like steamset class :type streams: :class:`StreamSet` or :class:`tango.LatestDeviceImpl` """ #: (:class:`tango.log4tango.TangoStream`) Tango fatal log stream self.log_fatal = None #: (:class:`tango.log4tango.TangoStream`) Tango error log stream self.log_error = None #: (:class:`tango.log4tango.TangoStream`) Tango warn log stream self.log_warn = None #: (:class:`tango.log4tango.TangoStream`) Tango info log stream self.log_info = None #: (:class:`tango.log4tango.TangoStream`) Tango debug log stream self.log_debug = None #: (:obj:`set <:obj:`str` >`) if tango server if not hasattr(streams, "__call__"): streams = weakref.ref(streams) \ if streams is not None else (lambda: None) if streams(): if hasattr(streams(), "log_fatal"): self.log_fatal = streams().log_fatal if hasattr(streams(), "log_error"): self.log_error = streams().log_error if hasattr(streams(), "log_warn"): self.log_warn = streams().log_warn if hasattr(streams(), "log_info"): self.log_info = streams().log_info if hasattr(streams(), "log_debug"): self.log_debug = streams().log_debug
[docs] def fatal(self, message, std=None): """ writes fatal error message :param message: error message :type message: :obj:`str` :param std: True if it writes to sys stream when log stream does not exist :type std: :obj:`bool` """ if std is None: std = stderrflag try: if self.log_fatal: self.log_fatal.write(message + '\n') if std: sys.stderr.write( "%s: FATAL: %s\n" % (datetime.datetime.now(), message)) sys.stderr.flush() except Exception: print(message)
[docs] def error(self, message, std=None): """ writes error message :param message: error message :type message: :obj:`str` :param std: True if it writes to sys stream when log stream does not exist :type std: :obj:`bool` """ if std is None: std = stderrflag try: if self.log_error: self.log_error.write(message + '\n') if std: sys.stderr.write( "%s: ERROR: %s\n" % (datetime.datetime.now(), message)) sys.stderr.flush() except Exception: print(message)
[docs] def warn(self, message, std=None): """ writes warning message :param message: warning message :type message: :obj:`str` :param std: True if it writes to sys stream when log stream does not exist :type std: :obj:`bool` """ if std is None: std = stderrflag try: if self.log_warn: self.log_warn.write(message + '\n') if std: sys.stderr.write( "%s: WARNING: %s\n" % (datetime.datetime.now(), message)) sys.stderr.flush() except Exception: print(message)
[docs] def info(self, message, std=None): """ writes info message :param message: info message :type message: :obj:`str` :param std: True if it writes to sys stream when log stream does not exist :type std: :obj:`bool` """ if std is None: std = stdoutflag try: if self.log_info: self.log_info.write(message + '\n') elif std: sys.stdout.write( "%s: INFO: %s\n" % (datetime.datetime.now(), message)) sys.stdout.flush() except Exception: print(message)
[docs] def debug(self, message, std=None): """ writes debug message :param message: debug message :type message: :obj:`str` :param std: True if it writes to sys stream when log stream does not exist :type std: :obj:`bool` """ if std is None: std = stdoutflag try: if self.log_debug: self.log_debug.write(message + '\n') elif std: sys.stdout.write( "%s: DEBUG: %s\n" % (datetime.datetime.now(), message)) sys.stdout.flush() except Exception: print(message)