Source code for imagepypelines.Logger

# @Email:
# @Website:
# @License:
# @github:
# Copyright (c) 2018 - 2020 Jeff Maggio, Jai Mehra, Ryan Hartzell
import logging
from termcolor import colored
import sys

# --------- enable terminal colors if we are in on a windows system ---------
import os
if == 'nt':
    import colorama
    del colorama

    'debug': logging.DEBUG,
    'info': logging.INFO,
    'warning': logging.WARNING,
    'error': logging.ERROR,
    'critical': logging.CRITICAL,
"""Log level strings mapped to their numerical values"""

    'debug': 'cyan',
    'info': None,
    'warning': 'yellow',
    'error': 'red',
    'critical': 'red',
"""Module variable controlling the color of our logs, this can be modified to
suit the end user's needs or ignored entirely by setting

"""Module variable controlling the text attributes of our logs, this can be
modified to suit the end user's needs or ignored entirely by setting

# this is defined lower down in this file
"""logging.Logger subclass that is the root of all loggers instantiated in

"""logging.LoggerAdapter subclass that wraps the master logger"""
# """logging.Manager for all ImagePypelines loggers"""

# Define our new special Logger class that can be pickled
# (like the loggers of python 3.7)
[docs]class ImagepypelinesLogger( logging.getLoggerClass() ): """subclass of logging.Logger that can be pickled, also adds colored logging outputs if desired. the color, functionality and text attributes can be controlled by setting the module variables imagepypelines.LOG_COLORS, imagepypelines.ENABLE_LOG_COLOR, imagepypelines.LOG_TEXT_ATTRS Attributes: ENABLE_LOG_COLOR(bool): class level variable controlling whether or not to markup log output with ANSI color codes. True by default """ ENABLE_LOG_COLOR = True def _color_msg(self, msg, level, LEVEL): if self.isEnabledFor(LEVEL) and self.ENABLE_LOG_COLOR: return colored(msg, LOG_COLORS[level], attrs=LOG_TEXT_ATTRS[level]) return msg
[docs] def debug(self, msg, *args, **kwargs): msg = self._color_msg(msg, 'debug', logging.DEBUG) return super().debug(msg, *args, **kwargs)
[docs] def info(self, msg, *args, **kwargs): msg = self._color_msg(msg, 'info', logging.INFO) return super().info(msg, *args, **kwargs)
[docs] def warning(self, msg, *args, **kwargs): msg = self._color_msg(msg, 'warning', logging.WARNING) return super().warning(msg, *args, **kwargs)
[docs] def error(self, msg, *args, **kwargs): msg = self._color_msg(msg, 'error', logging.ERROR) return super().error(msg, *args, **kwargs)
[docs] def critical(self, msg, *args, **kwargs): msg = self._color_msg(msg, 'critical', logging.CRITICAL) return super().critical(msg, *args, **kwargs)
[docs] def getChild(self,*args,**kwargs): # make a formatter for the child logger child = super().getChild(*args,**kwargs) if not len(self.handlers): ch = logging.StreamHandler() formatter = logging.Formatter( '%(asctime)s | %(name)s [ %(levelname)8s ]: %(message)s') ch.setFormatter(formatter) child.addHandler(ch) return child
[docs] def setLevel(self, level, *args, **kwargs): level = LOG_LEVELS.get(level, level) super().setLevel(level, *args, **kwargs)
# JEFF: modified from here def __reduce__(self): if == 'ImagePypelines': return make_master, (self.level,) return get_logger, (,)
class ImagepypelinesLoggerAdapter(logging.LoggerAdapter): def getChild(self, name, *args, **kwargs): logger = self.logger.getChild(name, *args, **kwargs) adapter = ImagepypelinesLoggerAdapter(logger, self.extra) return adapter
[docs]def get_master_logger(): if MASTER_ADAPTER: return MASTER_ADAPTER metadata = {'pipeline_id':'master', 'pipeline_uuid':'master', 'pipeline_name':'master'} return ImagepypelinesLoggerAdapter(make_master(),metadata)
def make_master(level=logging.INFO): """creates the master logger if it doesn't exist, returns it if it does""" if MASTER_LOGGER: return MASTER_LOGGER # create our ImagePypelines master logger # set our subclass as the root of all child loggers master = ImagepypelinesLogger('ImagePypelines') manager = logging.Manager(master) manager.setLoggerClass(ImagepypelinesLogger) master.manager = manager ch = logging.StreamHandler() formatter = logging.Formatter( '%(asctime)s | %(name)s [ %(levelname)8s ]: %(message)s') ch.setFormatter(formatter) master.addHandler(ch) master.setLevel(level) return master MASTER_LOGGER = make_master() MASTER_ADAPTER = get_master_logger()
[docs]def get_logger(name, pipeline=None, parent=MASTER_LOGGER): """Creates a new child logging adapter from the given parent (root logger by default) Args: name(str): the name of the new child logger pipeline(Block,Pipeline): the Pipeline object associated with this logger (optional) parent(logging.LoggerAdapter,logging.Logger): parent logger to spawn a new logger off of Returns: ImagepypelinesLoggerAdapter: a new child logger adapter with conn ids for id, uuid, and name of the Pipeline (if applicable) """ if isinstance(parent, logging.LoggerAdapter): parent = parent.logger logger = parent.getChild(name) if pipeline: metadata = {'pipeline_id', 'pipeline_uuid':pipeline.uuid, 'pipeline_name'} else: metadata = {'pipeline_id':None, 'pipeline_uuid':None, 'pipeline_name':None} adapter = ImagepypelinesLoggerAdapter(logger, metadata) return adapter
[docs]def set_log_level(log_level): """sets the global master logger level""" global MASTER_LOGGER log_level = LOG_LEVELS.get(log_level, log_level) MASTER_LOGGER.setLevel(log_level)