"""
TODO:
- [ ] This needs to be cleaned up. Python logging is way too global for my
taste. I would prefer something more instance level, but that will require
some thought. The netharn.FitHarn has a way of accomplishing this that I
reasonably like. That might be worth generalizing and porting here. But
for now, this will remain somewhat ad-hoc.
"""
import logging.config
import warnings
from logging import CRITICAL, ERROR, WARNING, INFO, DEBUG
[docs]
def setup_logging(verbose=1):
"""
Define logging level
Args:
verbose (int):
Accepted values:
* 0: no logging
* 1: INFO level
* 2: DEBUG level
"""
log_med = "%(asctime)s-15s %(name)-32s [%(levelname)-8s] %(message)s"
log_large = "%(asctime)s-15s %(name)-32s [%(levelname)-8s] "
log_large += "(%(module)-17s) %(message)s"
log_config = {}
if verbose > 2:
warnings.warn(f'geowatch util_logging only accepts a maximum verbosity of 2. Reconfiguring {verbose} to 2.')
verbose = 2
if verbose == 0:
log_config = {
"version": 1,
"disable_existing_loggers": False,
"handlers": {
"null": {"level": "DEBUG", "class": "logging.NullHandler"}
},
"loggers": {
"watchlog": {
"handlers": ["null"],
"propagate": True,
"level": "INFO"
}
},
}
elif verbose == 1:
log_config = {
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"standard": {
"format": log_med
}
},
"handlers": {
"console": {
"level": "DEBUG",
"class": "logging.StreamHandler",
"formatter": "standard",
}
},
"loggers": {
"watchlog": {
"handlers": ["console"],
"propagate": True,
"level": "INFO",
}
},
}
elif verbose == 2:
log_config = {
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"verbose": {
"format": log_large
}
},
"handlers": {
"console": {
"level": "DEBUG",
"class": "logging.StreamHandler",
"formatter": "verbose",
}
},
"loggers": {
"watchlog": {
"handlers": ["console"],
"propagate": True,
"level": "DEBUG",
}
},
}
else:
raise ValueError("'verbose' must be one of: 0, 1, 2")
return log_config
[docs]
def get_logger(verbose=1):
logcfg = setup_logging(verbose)
logging.config.dictConfig(logcfg)
logger = logging.getLogger('watchlog')
return logger
[docs]
class PrintLogger:
"""
Simple print-based logger that duck-types the logging.Logger class and
"simply works" without configuration.
Example:
>>> from geowatch.utils.util_logging import * # NOQA
>>> logger = PrintLogger(level=logging.INFO)
>>> logger.info('hello')
>>> logger.debug('world')
hello
>>> logger = PrintLogger(level=logging.DEBUG)
>>> logger.info('hello')
>>> logger.debug('world')
hello
world
"""
def __init__(self, name='<print-logger>', level=None, verbose=1):
self.name = name
if level is None:
if verbose == 0:
level = CRITICAL
elif verbose == 1:
level = INFO
elif verbose == 2:
level = DEBUG
self.level = level
self.parent = None
self.propagate = True
self.handlers = []
self.disabled = False
self._cache = {}
[docs]
def setLevel(self, level):
raise NotImplementedError
[docs]
def debug(self, msg, *args, **kwargs):
"""
Log 'msg % args' with severity 'DEBUG'.
"""
if self.isEnabledFor(DEBUG):
self._log(DEBUG, msg, args, **kwargs)
[docs]
def info(self, msg, *args, **kwargs):
"""
Log 'msg % args' with severity 'INFO'.
"""
if self.isEnabledFor(INFO):
self._log(INFO, msg, args, **kwargs)
[docs]
def warning(self, msg, *args, **kwargs):
"""
Log 'msg % args' with severity 'WARNING'.
"""
if self.isEnabledFor(WARNING):
self._log(WARNING, msg, args, **kwargs)
[docs]
def error(self, msg, *args, **kwargs):
"""
Log 'msg % args' with severity 'ERROR'.
"""
if self.isEnabledFor(ERROR):
self._log(ERROR, msg, args, **kwargs)
[docs]
def exception(self, msg, *args, exc_info=True, **kwargs):
"""
Convenience method for logging an ERROR with exception information.
"""
self.error(msg, *args, exc_info=exc_info, **kwargs)
[docs]
def critical(self, msg, *args, **kwargs):
"""
Log 'msg % args' with severity 'CRITICAL'.
"""
if self.isEnabledFor(CRITICAL):
self._log(CRITICAL, msg, args, **kwargs)
[docs]
def log(self, level, msg, *args, **kwargs):
"""
Log 'msg % args' with the integer severity 'level'.
"""
if not isinstance(level, int):
raise TypeError("level must be an integer")
if self.isEnabledFor(level):
self._log(level, msg, args, **kwargs)
[docs]
def findCaller(self, stack_info=False, stacklevel=1):
"""
Find the stack frame of the caller so that we can note the source
file name, line number and function name.
"""
raise NotImplementedError
[docs]
def makeRecord(self, name, level, fn, lno, msg, args, exc_info,
func=None, extra=None, sinfo=None):
"""
A factory method which can be overridden in subclasses to create
specialized LogRecords.
"""
return msg
[docs]
def filter(self, record):
return True
def _log(self, level, msg, args, exc_info=None, extra=None, stack_info=False,
stacklevel=1):
"""
Low-level logging routine which creates a LogRecord and then calls
all the handlers of this logger to handle the record.
"""
import sys
sinfo = None
fn, lno, func = "(unknown file)", 0, "(unknown function)"
if exc_info:
if isinstance(exc_info, BaseException):
exc_info = (type(exc_info), exc_info, exc_info.__traceback__)
elif not isinstance(exc_info, tuple):
exc_info = sys.exc_info()
record = self.makeRecord(self.name, level, fn, lno, msg, args,
exc_info, func, extra, sinfo)
self.handle(record)
[docs]
def handle(self, record):
"""
Call the handlers for the specified record.
This method is used for unpickled records received from a socket, as
well as those created locally. Logger-level filtering is applied.
"""
if (not self.disabled) and self.filter(record):
self.callHandlers(record)
[docs]
def addHandler(self, hdlr):
"""
Add the specified handler to this logger.
"""
raise NotImplementedError
[docs]
def removeHandler(self, hdlr):
"""
Remove the specified handler from this logger.
"""
raise NotImplementedError
[docs]
def hasHandlers(self):
"""
See if this logger has any handlers configured.
Loop through all handlers for this logger and its parents in the
logger hierarchy. Return True if a handler was found, else False.
Stop searching up the hierarchy whenever a logger with the "propagate"
attribute set to zero is found - that will be the last logger which
is checked for the existence of handlers.
"""
raise NotImplementedError
[docs]
def callHandlers(self, record):
print(record)
# raise NotImplementedError
[docs]
def getEffectiveLevel(self):
"""
Get the effective level for this logger.
"""
return self.level
[docs]
def isEnabledFor(self, level):
"""
Is this logger enabled for level 'level'?
"""
if self.disabled:
return False
return level >= self.level
[docs]
def getChild(self, suffix):
raise NotImplementedError
def __repr__(self):
level = logging.getLevelName(self.getEffectiveLevel())
return '<%s %s (%s)>' % (self.__class__.__name__, self.name, level)
# def __reduce__(self):
# raise NotImplementedError