Last active
March 28, 2022 14:13
-
-
Save ckesanapalli/2cbdbf75c690726ec0762753f0cfdbb8 to your computer and use it in GitHub Desktop.
Logging configuration code with wrapper that log entry, exit and runtime of functions with a decorator using logging module
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
import logging | |
from datetime import datetime | |
from pathlib import Path | |
import functools | |
from pprint import pformat | |
def get_func_name(func): | |
"""Return the name of the given function including module name""" | |
if func.__class__.__module__ == "builtins": | |
mod_name = func.__module__ if hasattr(func, '__module__') and \ | |
func.__module__ != None else "" | |
qual_name = func.__qualname__ if hasattr(func, '__qualname__') and \ | |
func.__qualname__ != None else func.__name__ | |
else: | |
class_func = func.__class__ | |
mod_name = class_func.__module__ if hasattr(class_func, '__module__') \ | |
and class_func.__module__ != None else "" | |
qual_name = (class_func.__qualname__ if hasattr(class_func, '__qualname__') \ | |
and class_func.__qualname__ != None else class_func.__name__) \ | |
+ '.' + func.__name__ | |
return '.'.join([mod_name, qual_name]) | |
def get_func_small_doc(func): | |
"""Return first two lines from the function documentation""" | |
try: | |
split_doc = func.__doc__.split("\n") | |
if len(split_doc) < 2: | |
doc = ' '.join(split_doc).strip() | |
else: | |
doc = ' '.join(split_doc[:2]).strip() | |
except: | |
doc = None | |
return doc | |
def get_logger_wrap(logger, indent=4, sort_dicts=False, **kwargs): | |
var2str = functools.partial(pformat, indent=indent, | |
sort_dicts=sort_dicts, **kwargs) | |
def logger_wrap(*, func_entry=True, | |
func_exit=True, level=logging.DEBUG, debug_io=False, debug_args=False, | |
debug_kwargs=False, debug_result=False, log_full_doc=False): | |
""" | |
A wrapper function to log the function or class entry, exit and runtime | |
outputting arguments and results | |
Parameters | |
---------- | |
logger: logging.Logger | |
Logger | |
func_entry : bool, optional | |
Whether to log the entry of function. The default is True. | |
func_exit : bool, optional | |
Whether to log the exit of function. The default is True. | |
level : int, optional | |
logging level. The default is 0 (DEBUG). | |
debug_io : bool, optional | |
Whether to log function inputs and output parameters. The default is False. | |
debug_args : bool, optional | |
Whether to log function arguments. The default is False. | |
debug_kwargs : bool, optional | |
Whether to log function keyword arguments. The default is False. | |
debug_result : bool, optional | |
Whether to log function result. The default is False. | |
is_full_doc : bool, optional | |
Whether to log full documentation of function. | |
If False, first two lines of function documentation are logged. | |
The default is False. | |
Returns | |
------- | |
decorator: | |
An object that can be used to decorate a function or a class. | |
""" | |
def wrapper(func): | |
name = get_func_name(func) | |
doc = func.__doc__ if log_full_doc else get_func_small_doc(func) | |
@functools.wraps(func) | |
def wrapped(*args, **kwargs): | |
if func_entry: | |
entry_dict = {"State": "ENTRY", "Name": name, "Description": doc} | |
if debug_args or debug_io: | |
entry_dict["Args"] = args | |
if debug_kwargs or debug_io: | |
entry_dict["Kwargs"] = kwargs | |
entry_str = var2str(entry_dict) | |
logger.log(level, entry_str, stacklevel=2) | |
start = datetime.now() | |
result = func(*args, **kwargs) | |
end = datetime.now() | |
runtime = str(end - start) | |
if func_exit: | |
exit_dict = {"State": "EXIT", "Name": name, "Runtime": runtime} | |
if debug_result or debug_io: | |
exit_dict["Result"] = result | |
exit_str = var2str(exit_dict) | |
logger.log(level, exit_str, stacklevel=2) | |
return result | |
return wrapped | |
return wrapper | |
return logger_wrap | |
class CustomFormatter(logging.Formatter): | |
"""Logging colored formatter, adapted from https://stackoverflow.com/a/56944256/3638629""" | |
grey = '\x1b[38;21m' | |
blue = '\x1b[38;5;39m' | |
yellow = '\x1b[38;5;226m' | |
red = '\x1b[38;5;196m' | |
bold_red = '\x1b[31;1m' | |
reset = '\x1b[0m' | |
def __init__(self, fmt): | |
super().__init__() | |
self.fmt = fmt | |
self.FORMATS = { | |
logging.DEBUG: self.grey + self.fmt + self.reset, | |
logging.INFO: self.blue + self.fmt + self.reset, | |
logging.WARNING: self.yellow + self.fmt + self.reset, | |
logging.ERROR: self.red + self.fmt + self.reset, | |
logging.CRITICAL: self.bold_red + self.fmt + self.reset | |
} | |
def format(self, record): | |
log_fmt = self.FORMATS.get(record.levelno) | |
formatter = logging.Formatter(log_fmt) | |
return formatter.format(record) | |
def updated_logger(logger, file_name, log_dir, level=logging.DEBUG, timed_path=True, **kwargs): | |
""" | |
Return updated logger with the log file path and the logging level | |
Parameters | |
---------- | |
logger : logging.logger | |
Logger. | |
file_name : str | |
name of the logger file. | |
log_dir : str or Path | |
Path of the log directory. | |
level : int, optional | |
Level of the logging. The default is logging.DEBUG. | |
timed_path : bool, optional | |
whether to add datetime to the file suffix. The default is True. | |
Returns | |
------- | |
logger: logging.logger | |
Updated Logger | |
logger_wrap: decorator | |
Log decorator for functions | |
""" | |
log_dir = Path(log_dir) | |
log_dir.mkdir(parents=True, exist_ok=True) | |
logger.setLevel(level) | |
time_str = datetime.strftime(datetime.now(), '%Y_%m_%d_%H_%M_%S') | |
log_file_name = f"{file_name}_{time_str}.log" if timed_path else f"{file_name}.log" | |
log_file_path = log_dir / log_file_name | |
fmt_str = "%(asctime)s | %(levelname)-8s | \ | |
%(module)s:%(name)s:%(funcName)s:%(lineno)d - %(message)s" | |
formatter = logging.Formatter(fmt_str) | |
# create console handler with a higher log level | |
ch = logging.StreamHandler() | |
ch.setLevel(logging.DEBUG) | |
ch.setFormatter(CustomFormatter(fmt_str)) | |
logger.addHandler(ch) | |
# create file handler which logs even debug messages | |
fh = logging.FileHandler(log_file_path, mode='w') | |
fh.setLevel(logging.DEBUG) | |
fh.setFormatter(formatter) | |
logger.addHandler(fh) | |
logger_wrap = get_logger_wrap(logger, **kwargs) | |
return logger, logger_wrap | |
def get_heading(logger, title, level=0, total_len=79): | |
""" | |
Converts 'string' to a heading ('== string ==') | |
Parameters | |
---------- | |
logger : logging.logger | |
logger object | |
title : str | |
title string before formatting | |
level : int, optional | |
Heading type ranging from 0 to 2, by default 0 | |
total_len : int, optional | |
Maximum desired length of heading, by default 79 | |
Returns | |
------- | |
logger.info | |
Outputs logger.info(string) | |
""" | |
title_len = len(title) | |
if title_len > total_len: | |
logger.warning( | |
f"Title length ({title_len}) is greater than Total length ({total_len})" | |
) | |
title_len = total_len - 2 | |
line1_len = (total_len - title_len) // 2 | |
line2_len = total_len - (title_len + line1_len) | |
if level == 0: | |
line_chr = "=" | |
elif level == 1: | |
line_chr = "-" | |
elif level == 2: | |
line_chr = ":" | |
else: | |
line_chr = "." | |
line1 = line_chr * line1_len | |
line2 = line_chr * line2_len | |
formated_heading = " ".join([line1, title, line2]) | |
return logger.info(formated_heading) | |
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
import time | |
import logging | |
from log_conf import updated_logger | |
from dataclasses import dataclass | |
logger, logger_wrap = updated_logger(logging.getLogger(__name__), "test", | |
indent=4, sort_dicts=False) | |
@logger_wrap(debug_io=True) | |
def my_func(a, b, c=1, d="Text"): | |
"""A test function""" | |
logger.info("Sleeping for 1 sec") | |
time.sleep(1) | |
return str([a, b, c, d]) | |
@logger_wrap(debug_io=True) | |
@dataclass | |
class my_class(): | |
"""A test class""" | |
a : int | |
area: float | |
test_class = my_class(2, 3) | |
my_func("hello", "logger", c="print this", d="keyword arg") | |
# Logging Existing library functions | |
import numpy as np | |
logger_wrap(debug_io=True)(np.random.rand)(10,10) | |
logger_wrap(log_full_doc=True)(np.sin)(np.random.rand(10,10)) | |
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
2022-03-28 02:00:01,643 | DEBUG | main:__main__:<module>:23 - { State: ENTRY, | |
Name: __main__.my_class, | |
Description: A test class, | |
Args: (2, 3), | |
Kwargs: {}} | |
2022-03-28 02:00:01,644 | DEBUG | main:__main__:<module>:23 - { State: EXIT, | |
Name: __main__.my_class, | |
Runtime: 0:00:00, | |
Result: my_class(a=2, area=3)} | |
2022-03-28 02:00:01,644 | WARNING | main:__main__:<module>:24 - Not Sleeping | |
2022-03-28 02:00:01,644 | DEBUG | main:__main__:<module>:26 - { State: ENTRY, | |
Name: __main__.my_func, | |
Description: A test function, | |
Args: (hello, logger), | |
Kwargs: {c: print this, d: keyword arg}} | |
2022-03-28 02:00:01,645 | INFO | main:__main__:my_func:12 - Sleeping for 1 sec | |
2022-03-28 02:00:02,653 | DEBUG | main:__main__:<module>:26 - { State: EXIT, | |
Name: __main__.my_func, | |
Runtime: 0:00:01.008160, | |
Result: [hello, logger, print this, keyword arg]} | |
2022-03-28 02:00:02,654 | DEBUG | main:__main__:<module>:30 - { State: ENTRY, | |
Name: .RandomState.rand, | |
Description: rand(d0, d1, ..., dn), | |
Args: (10, 10), | |
Kwargs: {}} | |
2022-03-28 02:00:02,655 | DEBUG | main:__main__:<module>:30 - { State: EXIT, | |
Name: .RandomState.rand, | |
Runtime: 0:00:00, | |
Result: [[0.2363461 0.48550162 0.13387681 0.87350022 0.73049686 | |
0.92323771 | |
0.26544497 0.62579515 0.64877539 0.85464577] | |
[0.64606151 0.83008665 0.49724045 0.5762315 0.21503162 | |
0.75508816 | |
0.36645548 0.97247661 0.35752656 0.66461037] | |
[0.52730874 0.77483066 0.43904472 0.76365511 0.60058024 | |
0.89924877 | |
0.36866417 0.38233081 0.68954406 0.25537451] | |
[0.3137653 0.87982569 0.05163364 0.97376399 0.26522001 | |
0.2569054 | |
0.39716299 0.36478841 0.15514908 0.9514152 ] | |
[0.17787851 0.75302585 0.53634242 0.93452775 0.32741251 | |
0.6441435 | |
0.67265579 0.21157967 0.71253889 0.50326617] | |
[0.84149008 0.37644909 0.42836803 0.99342584 0.73844244 | |
0.25168303 | |
0.29965424 0.49836129 0.81708351 0.19190296] | |
[0.70483546 0.66344476 0.44611424 0.49046865 0.97799542 | |
0.92617753 | |
0.04034599 0.83111146 0.26974899 0.47880171] | |
[0.0913788 0.2426281 0.33022209 0.25033408 0.9195304 | |
0.82793309 | |
0.8876626 0.71317787 0.14346796 0.33335825] | |
[0.34448748 0.62689749 0.64079311 0.51878668 0.9918069 | |
0.85796244 | |
0.70466754 0.55603669 0.95307835 0.92907495] | |
[0.43503599 0.16428707 0.41947888 0.10935243 0.32389202 | |
0.59091364 | |
0.29246639 0.44738868 0.56157866 0.38954625]]} | |
2022-03-28 02:00:02,656 | DEBUG | main:__main__:<module>:31 - { State: ENTRY, | |
Name: numpy.ufunc.sin, | |
Description: sin(x, /, out=None, *, where=True, casting=same_kind, | |
order=K, dtype=None, subok=True[, signature, extobj]) | |
Trigonometric sine, element-wise. | |
Parameters | |
---------- | |
x : array_like | |
Angle, in radians (:math:`2 \\pi` rad equals 360 | |
degrees). | |
out : ndarray, None, or tuple of ndarray and None, | |
optional | |
A location into which the result is stored. If | |
provided, it must have | |
a shape that the inputs broadcast to. If not provided | |
or None, | |
a freshly-allocated array is returned. A tuple | |
(possible only as a | |
keyword argument) must have length equal to the number | |
of outputs. | |
where : array_like, optional | |
This condition is broadcast over the input. At | |
locations where the | |
condition is True, the `out` array will be set to the | |
ufunc result. | |
Elsewhere, the `out` array will retain its original | |
value. | |
Note that if an uninitialized `out` array is created | |
via the default | |
``out=None``, locations within it where the condition | |
is False will | |
remain uninitialized. | |
**kwargs | |
For other keyword-only arguments, see the | |
:ref:`ufunc docs <ufuncs.kwargs>`. | |
Returns | |
------- | |
y : array_like | |
The sine of each element of x. | |
This is a scalar if `x` is a scalar. | |
See Also | |
-------- | |
arcsin, sinh, cos | |
Notes | |
----- | |
The sine is one of the fundamental functions of | |
trigonometry (the | |
mathematical study of triangles). Consider a circle of | |
radius 1 | |
centered on the origin. A ray comes in from the | |
:math:`+x` axis, makes | |
an angle at the origin (measured counter-clockwise from | |
that axis), and | |
departs from the origin. The :math:`y` coordinate of the | |
outgoing | |
rays intersection with the unit circle is the sine of | |
that angle. It | |
ranges from -1 for :math:`x=3\\pi / 2` to +1 for | |
:math:`\\pi / 2.` The | |
function has zeroes where the angle is a multiple of | |
:math:`\\pi`. | |
Sines of angles between :math:`\\pi` and :math:`2\\pi` are | |
negative. | |
The numerous properties of the sine and related functions | |
are included | |
in any standard trigonometry text. | |
Examples | |
-------- | |
Print sine of one angle: | |
>>> np.sin(np.pi/2.) | |
1.0 | |
Print sines of an array of angles given in degrees: | |
>>> np.sin(np.array((0., 30., 45., 60., 90.)) * np.pi / | |
180. ) | |
array([ 0. , 0.5 , 0.70710678, 0.8660254 | |
, 1. ]) | |
Plot the sine function: | |
>>> import matplotlib.pylab as plt | |
>>> x = np.linspace(-np.pi, np.pi, 201) | |
>>> plt.plot(x, np.sin(x)) | |
>>> plt.xlabel(Angle [rad]) | |
>>> plt.ylabel(sin(x)) | |
>>> plt.axis(tight) | |
>>> plt.show()} | |
2022-03-28 02:00:02,658 | DEBUG | main:__main__:<module>:31 - {State: EXIT, Name: numpy.ufunc.sin, Runtime: 0:00:00} |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment