# MagellanMapper logging
"""Logging utilities."""
import logging
from logging import handlers
import pathlib
[docs]
class LogWriter:
"""File-like object to write standard output to logging functions.
Attributes:
fn_logging (func): Logging function
buffer (list[str]): String buffer.
"""
def __init__(self, fn_logging):
"""Create a writer for a logging function."""
self.fn_logger = fn_logging
self.buffer = []
[docs]
def write(self, msg):
"""Write to logging function with buffering.
Args:
msg (str): Line to write, from which trailing newlines will be
removed.
"""
if msg.endswith("\n"):
# remove trailing newlines in buffer and pass to logging function
self.buffer.append(msg.rstrip("\n"))
self.fn_logger("".join(self.buffer))
self.buffer = []
else:
self.buffer.append(msg)
[docs]
def flush(self):
"""Empty function, deferring to logging handler's flush."""
pass
[docs]
def setup_logger():
"""Set up a basic root logger with a stream handler.
Returns:
:class:`logging.Logger`: Root logger for the application.
"""
logger = logging.getLogger()
logger.setLevel(logging.INFO)
# set up handler for console
handler_stream = logging.StreamHandler()
handler_stream.setLevel(logging.INFO)
handler_stream.setFormatter(logging.Formatter(
"%(name)s - %(levelname)s - %(message)s"))
logger.addHandler(handler_stream)
return logger
[docs]
def update_log_level(logger, level):
"""Update the logging level.
Args:
logger (:class:`logging.Logger`): Logger to update.
level (Union[str, int]): Level given either as a string corresponding
to ``Logger`` levels, or their corresponding integers, ranging
from 0 (``NOTSET``) to 50 (``CRITICAL``). For convenience,
values can be given from 0-5, which will be multiplied by 10.
Returns:
:class:`logging.Logger`: The logger for chained calls.
"""
if isinstance(level, str):
# specify level by level name
level = level.upper()
elif isinstance(level, int):
# specify by level integer (0-50)
if level < 10:
# for convenience, assume values under 10 are 10-fold
level *= 10
else:
return
try:
# set level for the logger and all its handlers
logger.setLevel(level)
for handler in logger.handlers:
handler.setLevel(level)
if level < logging.INFO:
# avoid lots of debugging messages from Matplotlib
logging.getLogger("matplotlib").setLevel(logging.INFO)
except (TypeError, ValueError) as e:
logger.error(e, exc_info=True)
return logger
[docs]
def add_file_handler(
logger: logging.Logger, path: str, backups: int = 5) -> pathlib.Path:
"""Add a rotating log file handler with a new log file.
Rotates the file each time this function is called for the given number
of backups rather than by size or time. Avoids file conflicts from
multiple instances when permission errors occur (eg on Windows), instead
creating a log filed named with an incremented number (eg ``out1.log``).
Args:
logger: Logger to update.
path: Path to log. Increments to ``path<n>.<ext>`` if the
file at ``path`` cannot be rotated.
backups: Number of backups to maintain; defaults to 5.
Returns:
The log output path.
"""
# check if log file already exists
pathl = pathlib.Path(path)
path_log = pathl
roll = pathl.is_file()
# create a rotations file handler to manage number of backups while
# manually managing rollover based on file presence rather than size
pathl.parent.mkdir(parents=True, exist_ok=True)
i = 0
handler_file = None
while handler_file is None:
try:
# if the existing file at path cannot be rotated, increment the
# filename to create a new series of rotating log files
path_log = pathl if i == 0 else pathlib.Path(
f"{pathl.parent / pathl.stem}{i}{pathl.suffix}")
logger.debug(f"Trying logger path: {path_log}")
handler_file = handlers.RotatingFileHandler(
path_log, backupCount=backups, encoding="utf-8")
if roll:
# create a new log file if exists, backing up the old one
handler_file.doRollover()
except (PermissionError, FileNotFoundError) as e:
# permission errors occur on Windows if the log is opened by
# another application instance; file not found errors occur if a
# backup file is skipped in backup series
logger.debug(e)
handler_file = None
i += 1
handler_file.setLevel(logger.level)
handler_file.setFormatter(logging.Formatter(
"%(asctime)s - %(name)s - %(levelname)s - %(message)s"))
logger.addHandler(handler_file)
return path_log