1import logging 2import sys 3 4from yt.config import ytcfg 5 6# This next bit is grabbed from: 7# http://stackoverflow.com/questions/384076/how-can-i-make-the-python-logging-output-to-be-colored 8 9 10def add_coloring_to_emit_ansi(fn): 11 # add methods we need to the class 12 def new(*args): 13 levelno = args[0].levelno 14 if levelno >= 50: 15 color = "\x1b[31m" # red 16 elif levelno >= 40: 17 color = "\x1b[31m" # red 18 elif levelno >= 30: 19 color = "\x1b[33m" # yellow 20 elif levelno >= 20: 21 color = "\x1b[32m" # green 22 elif levelno >= 10: 23 color = "\x1b[35m" # pink 24 else: 25 color = "\x1b[0m" # normal 26 ln = color + args[0].levelname + "\x1b[0m" 27 args[0].levelname = ln 28 return fn(*args) 29 30 return new 31 32 33def set_log_level(level): 34 """ 35 Select which minimal logging level should be displayed. 36 37 Parameters 38 ---------- 39 level: int or str 40 Possible values by increasing level: 41 0 or "notset" 42 1 or "all" 43 10 or "debug" 44 20 or "info" 45 30 or "warning" 46 40 or "error" 47 50 or "critical" 48 """ 49 # this is a user-facing interface to avoid importing from yt.utilities in user code. 50 51 if isinstance(level, str): 52 level = level.upper() 53 54 if level == "ALL": # non-standard alias 55 level = 1 56 ytLogger.setLevel(level) 57 ytLogger.debug("Set log level to %s", level) 58 59 60ufstring = "%(name)-3s: [%(levelname)-9s] %(asctime)s %(message)s" 61cfstring = "%(name)-3s: [%(levelname)-18s] %(asctime)s %(message)s" 62 63if ytcfg.get("yt", "stdout_stream_logging"): 64 stream = sys.stdout 65else: 66 stream = sys.stderr 67 68ytLogger = logging.getLogger("yt") 69 70 71class DuplicateFilter(logging.Filter): 72 """A filter that removes duplicated successive log entries.""" 73 74 # source 75 # https://stackoverflow.com/questions/44691558/suppress-multiple-messages-with-same-content-in-python-logging-module-aka-log-co # noqa 76 def filter(self, record): 77 current_log = (record.module, record.levelno, record.msg, record.args) 78 if current_log != getattr(self, "last_log", None): 79 self.last_log = current_log 80 return True 81 return False 82 83 84ytLogger.addFilter(DuplicateFilter()) 85 86 87class DeprecatedFieldFilter(logging.Filter): 88 """A filter that suppresses repeated logging of deprecated field warnings""" 89 90 def __init__(self, name=""): 91 self.logged_fields = [] 92 super().__init__(name=name) 93 94 def filter(self, record): 95 if not record.msg.startswith("The Derived Field"): 96 return True 97 98 field = record.args[0] 99 if field in self.logged_fields: 100 return False 101 102 self.logged_fields.append(field) 103 return True 104 105 106ytLogger.addFilter(DeprecatedFieldFilter()) 107 108 109def disable_stream_logging(): 110 if len(ytLogger.handlers) > 0: 111 ytLogger.removeHandler(ytLogger.handlers[0]) 112 h = logging.NullHandler() 113 ytLogger.addHandler(h) 114 115 116def colorize_logging(): 117 f = logging.Formatter(cfstring) 118 ytLogger.handlers[0].setFormatter(f) 119 yt_sh.emit = add_coloring_to_emit_ansi(yt_sh.emit) 120 121 122def uncolorize_logging(): 123 try: 124 f = logging.Formatter(ufstring) 125 ytLogger.handlers[0].setFormatter(f) 126 yt_sh.emit = original_emitter 127 except NameError: 128 # yt_sh and original_emitter are not defined because 129 # suppress_stream_logging is True, so we continue since there is nothing 130 # to uncolorize 131 pass 132 133 134_level = min(max(ytcfg.get("yt", "log_level"), 0), 50) 135 136if ytcfg.get("yt", "suppress_stream_logging"): 137 disable_stream_logging() 138else: 139 yt_sh = logging.StreamHandler(stream=stream) 140 # create formatter and add it to the handlers 141 formatter = logging.Formatter(ufstring) 142 yt_sh.setFormatter(formatter) 143 # add the handler to the logger 144 ytLogger.addHandler(yt_sh) 145 set_log_level(_level) 146 ytLogger.propagate = False 147 148 original_emitter = yt_sh.emit 149 150 if ytcfg.get("yt", "colored_logs"): 151 colorize_logging() 152