# The LogFormatter is adapted light from tornado, which is licensed under
# Apache 2.0. See other_licenses/ in the repository directory.
import logging
import sys
try:
import colorama
colorama.init()
except ImportError:
colorama = None
try:
import curses
except ImportError:
curses = None
__all__ = ('config_bluesky_logging', 'get_handler',
'LogFormatter', 'set_handler')
def _stderr_supports_color():
try:
if hasattr(sys.stderr, 'isatty') and sys.stderr.isatty():
if curses:
curses.setupterm()
if curses.tigetnum("colors") > 0:
return True
elif colorama:
if sys.stderr is getattr(colorama.initialise, 'wrapped_stderr',
object()):
return True
except Exception:
# Very broad exception handling because it's always better to
# fall back to non-colored logs than to break at startup.
pass
return False
class ComposableLogAdapter(logging.LoggerAdapter):
def process(self, msg, kwargs):
# The logging.LoggerAdapter siliently ignores `extra` in this usage:
# log_adapter.debug(msg, extra={...})
# and passes through log_adapater.extra instead. This subclass merges
# the extra passed via keyword argument with the extra in the
# attribute, giving precedence to the keyword argument.
kwargs["extra"] = {**self.extra, **kwargs.get('extra', {})}
return msg, kwargs
plain_log_format = "[%(levelname)1.1s %(asctime)s.%(msecs)03d %(module)15s:%(lineno)5d] %(message)s"
color_log_format = ("%(color)s[%(levelname)1.1s %(asctime)s.%(msecs)03d "
"%(module)15s:%(lineno)5d]%(end_color)s %(message)s")
logger = logging.getLogger('bluesky')
doc_logger = logging.getLogger('bluesky.emit_document')
msg_logger = logging.getLogger('bluesky.RE.msg')
state_logger = logging.getLogger('bluesky.RE.state')
current_handler = None
def validate_level(level) -> int:
'''
Return a int for level comparison
'''
if isinstance(level, int):
levelno = level
elif isinstance(level, str):
levelno = logging.getLevelName(level)
if isinstance(levelno, int):
return levelno
else:
raise ValueError("Your level is illegal, please use one of python logging string")
def _set_handler_with_logger(logger_name='bluesky', file=sys.stdout, datefmt='%H:%M:%S', color=True,
level='WARNING'):
if isinstance(file, str):
handler = logging.FileHandler(file)
else:
handler = logging.StreamHandler(file)
levelno = validate_level(level)
handler.setLevel(levelno)
if color:
format = color_log_format
else:
format = plain_log_format
handler.setFormatter(
LogFormatter(format, datefmt=datefmt))
logging.getLogger(logger_name).addHandler(handler)
if logger.getEffectiveLevel() > levelno:
logger.setLevel(levelno)
[docs]def config_bluesky_logging(file=sys.stdout, datefmt='%H:%M:%S', color=True, level='WARNING'):
"""
Set a new handler on the ``logging.getLogger('bluesky')`` logger.
If this is called more than once, the handler from the previous invocation
is removed (if still present) and replaced.
Parameters
----------
file : object with ``write`` method or filename string
Default is ``sys.stdout``.
datefmt : string
Date format. Default is ``'%H:%M:%S'``.
color : boolean
Use ANSI color codes. True by default.
level : str or int
Python logging level, given as string or corresponding integer.
Default is 'WARNING'.
Returns
-------
handler : logging.Handler
The handler, which has already been added to the 'bluesky' logger.
Examples
--------
Log to a file.
>>> config_bluesky_logging(file='/tmp/what_is_happening.txt')
Include the date along with the time. (The log messages will always include
microseconds, which are configured separately, not as part of 'datefmt'.)
>>> config_bluesky_logging(datefmt="%Y-%m-%d %H:%M:%S")
Turn off ANSI color codes.
>>> config_bluesky_logging(color=False)
Increase verbosity: show level INFO or higher.
>>> config_bluesky_logging(level='INFO')
"""
global current_handler
if isinstance(file, str):
handler = logging.FileHandler(file)
else:
handler = logging.StreamHandler(file)
levelno = validate_level(level)
handler.setLevel(levelno)
if color:
format = color_log_format
else:
format = plain_log_format
handler.setFormatter(
LogFormatter(format, datefmt=datefmt))
if current_handler in logger.handlers:
logger.removeHandler(current_handler)
logger.addHandler(handler)
current_handler = handler
if logger.getEffectiveLevel() > levelno:
logger.setLevel(levelno)
return handler
set_handler = config_bluesky_logging # for back-compat
[docs]def get_handler():
"""
Return the handler configured by the most recent call to :func:`config_bluesky_logging`.
If :func:`config_bluesky_logging` has not yet been called, this returns ``None``.
"""
return current_handler