Last active
October 15, 2023 01:06
-
-
Save andres-fr/e7d961f7e3f006d25680ba807ec494d7 to your computer and use it in GitHub Desktop.
A well-suported and pre-configured color logger for Python 3
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin python | |
# -*- coding:utf-8 -*- | |
""" | |
This module features convenient color logger (wrapping the built-in ``loging.`` | |
class), plus some helper functionality. | |
""" | |
import os | |
import sys | |
import datetime | |
import pytz | |
import logging | |
from typing import Optional | |
import socket | |
import json | |
# | |
# ignore "mypy" import error for coloredlogs. | |
# see https://mypy.readthedocs.io/en/latest/running_mypy.html#missing-imports | |
import coloredlogs # type: ignore | |
# ############################################################################## | |
# # COLOR LOGGER | |
# ############################################################################## | |
def make_timestamp(timezone="Europe/Berlin", with_tz_output=False): | |
""" | |
Output example: day, month, year, hour, min, sec, milisecs: | |
10_Feb_2018_20:10:16.151 | |
""" | |
ts = datetime.datetime.now(tz=pytz.timezone(timezone)).strftime( | |
"%Y_%m_%d_%H_%M_%S.%f")[:-3] | |
if with_tz_output: | |
return "%s(%s)" % (ts, timezone) | |
else: | |
return ts | |
class HostnameFilter(logging.Filter): | |
""" | |
Needed to include hostname into the logger. See:: | |
https://stackoverflow.com/a/55584223/4511978 | |
""" | |
def filter(self, record) -> bool: | |
record.hostname = socket.gethostname() | |
return True | |
class ColorLogger: | |
""" | |
This class: | |
1. Creates a ``logging.Logger`` with a convenient configuration. | |
2. Attaches ``coloredlogs.install`` to it for colored terminal output | |
3. Provides some wrapper methods for convenience | |
Usage example:: | |
# create 2 loggers | |
cl1 = ColorLogger("term.and.file.logger", "/tmp/test.txt") | |
cl2 = ColorLogger("JustTermLogger") | |
# use them at wish | |
cl1.logger.debug("this is a debugging message") | |
cl2.logger.info("this is an informational message") | |
cl1.logger.warning("this is a warning message") | |
cl2.logger.error("this is an error message") | |
cl1.logger.critical("this is a critical message") | |
""" | |
FORMAT_STR = ("%(asctime)s.%(msecs)03d %(hostname)s: %(name)s" + | |
"[%(process)d] %(levelname)s %(message)s") | |
DATE_FORMAT = "%Y-%m-%d %H:%M:%S" | |
def get_logger(self, logger_name, logfile_dir: Optional[str], | |
filemode: str = "a", | |
logging_level: int = logging.DEBUG) -> logging.Logger: | |
""" | |
:param filemode: In case ``logfile_dir`` is given, this specifies the | |
output mode (e.g. 'a' for append). | |
:returns: a ``logging.Logger`` configured to output all events at level | |
``self.logging_level`` or above into ``sys.stdout`` and (optionally) | |
the given ``logfile_dir``, if not None. | |
""" | |
# create logger, formatter and filter, and set desired logger level | |
logger = logging.getLogger(logger_name) | |
formatter = logging.Formatter(self.FORMAT_STR, | |
datefmt=self.DATE_FORMAT) | |
hostname_filter = HostnameFilter() | |
logger.setLevel(logging_level) | |
# create and wire stdout handler | |
stdout_handler = logging.StreamHandler(sys.stdout) | |
stdout_handler.addFilter(hostname_filter) | |
stdout_handler.setFormatter(formatter) | |
logger.addHandler(stdout_handler) | |
# optionally, create and wire file handler | |
if logfile_dir is not None: | |
basename = make_timestamp( | |
with_tz_output=False) + logger_name + ".log" | |
logfile_path = os.path.join(logfile_dir, basename) | |
# create one handler for print and one for export | |
file_handler = logging.FileHandler(logfile_path, filemode) | |
file_handler.addFilter(hostname_filter) | |
file_handler.setFormatter(formatter) | |
logger.addHandler(file_handler) | |
# | |
return logger | |
def __init__(self, logger_name: str, | |
logfile_path: Optional[str] = None, | |
filemode: str = "a", | |
logging_level: int = logging.DEBUG): | |
""" | |
:param logger_name: A process may have several loggers. This parameter | |
distinguishes them. | |
:param logfile_path: Where to write out. | |
""" | |
self.logger: logging.Logger = self.get_logger(logger_name, | |
logfile_path, filemode, | |
logging_level) | |
# | |
coloredlogs.install(logger=self.logger, | |
fmt=self.FORMAT_STR, | |
level=logging_level) | |
# a few convenience wrappers: | |
def debug(self, *args, **kwargs) -> None: | |
self.logger.debug(*args, **kwargs) | |
def info(self, *args, **kwargs) -> None: | |
self.logger.info(*args, **kwargs) | |
def warning(self, *args, **kwargs) -> None: | |
self.logger.warning(*args, **kwargs) | |
def error(self, *args, **kwargs) -> None: | |
self.logger.error(*args, **kwargs) | |
def critical(self, *args, **kwargs) -> None: | |
self.logger.critical(*args, **kwargs) | |
class JsonColorLogger(ColorLogger): | |
""" | |
Less human-friendly, more machine-friendly prompts for easier downstream | |
processing. Usage example:: | |
# log JSON data using loj as follows: | |
for step in range(50): | |
logger.loj("TEST", {"step": 50, "value": 123}) | |
# resulting logs can be then read as follows: | |
df = JsonColorlogger.read_file(log_path) | |
print(df) | |
""" | |
FORMAT_STR = ("""["%(asctime)s.%(msecs)03d", %(message)s]""") | |
DATE_FORMAT = "%Y-%m-%d_%H:%M:%S" | |
def loj(self, header, body): | |
""" | |
""" | |
self.info(json.dumps((header, body))) | |
@staticmethod | |
def read_file(path, body_fn=lambda x: x, filter_headers=None, | |
ts_colname="TIMESTAMP", header_colname="HEADER"): | |
""" | |
Load a log file into a pandas table. The file is assumed to be created | |
by repeatedly calling the ``loj`` method, hence having lines in the | |
form ``[timestamp, [header, body]]``. | |
:param filter_headers: Collection of allowed headers. If given, only | |
entries with these headers will be gathered. | |
:param body_fn: The ``body`` entry from each line will be passed to | |
this function for processing. The function must return a dictionary, | |
such that all the returned key-value pairs will be aggregated into | |
the output pandas table. If the body is already in the form of a | |
dictionary, and you want to preserve all key-value pairs, this | |
function is simply the identity. It can be used to filter out | |
entries, rename keys, etc. | |
:returns: A pandas dataframe, with ``ts_colname, header_colname`` | |
columns, as well as one column per key returned by ``body_fn``. | |
""" | |
try: | |
import pandas as pd | |
except ModuleNotFoundError: | |
print("This method requires pandas to be installed!") | |
return | |
# first gather a list of [timestamp, [header, body]] entries | |
entries = pd.read_json(path, lines=True).values.tolist() | |
result = [] | |
for i, (ts, (header, body)) in enumerate(entries): | |
if (filter_headers is not None) and (header not in filter_headers): | |
continue | |
body = body_fn(body) | |
assert ((ts_colname not in body) and | |
(header_colname not in body)), \ | |
"Colliding timestamp/header column names! {body.keys()}" | |
result.append({ts_colname: ts, header_colname: header, **body}) | |
# | |
result = pd.DataFrame(result) | |
return result |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment