133 lines
3.9 KiB
Python
133 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import enum
|
|
import inspect
|
|
import logging
|
|
import os
|
|
import typing
|
|
from typing import Callable
|
|
|
|
_TRACE = (logging.DEBUG + logging.NOTSET) // 2
|
|
logging.addLevelName(_TRACE, "TRACE")
|
|
|
|
rootLogger = logging.getLogger()
|
|
|
|
loglevel = os.getenv("LOGLEVEL", None)
|
|
if loglevel is not None:
|
|
rootLogger.setLevel(loglevel.upper())
|
|
rootLogger.debug("Set loglevel from LOGLEVEL environment variable: %s", loglevel)
|
|
|
|
rootLogger.handlers.clear()
|
|
|
|
|
|
class LogLevel(enum.IntEnum):
|
|
Critical = logging.CRITICAL
|
|
Error = logging.ERROR
|
|
Warning = logging.WARNING
|
|
Info = logging.INFO
|
|
Debug = logging.DEBUG
|
|
Trace = _TRACE
|
|
Notset = logging.NOTSET
|
|
|
|
|
|
class Logger:
|
|
def __init__(self, name: str, level: LogLevel | None = None):
|
|
self.name = name
|
|
if level is None:
|
|
level = rootLogger.level
|
|
self.level = level
|
|
self.logger = Logger._make_logger(self.name, self.level)
|
|
|
|
@staticmethod
|
|
def _make_logger(name: str, level: LogLevel) -> logging.Logger:
|
|
_logger = logging.getLogger(name)
|
|
_logger.setLevel(level)
|
|
_logger.propagate = False
|
|
|
|
# create console handler and set level to debug
|
|
ch = logging.StreamHandler()
|
|
ch.setLevel(level)
|
|
# create formatter
|
|
formatter = logging.Formatter('%(name)s - %(levelname)s - %(message)s')
|
|
ch.setFormatter(formatter)
|
|
# add ch to logger
|
|
_logger.addHandler(ch)
|
|
|
|
return _logger
|
|
|
|
def log(self, level: LogLevel, *args, exc_info: bool = False, **kwargs):
|
|
assert args[0] != "%s"
|
|
message = " ".join(arg if type(arg) == str else repr(arg) for arg in args)
|
|
self.logger.log(level, "%s",
|
|
message,
|
|
exc_info=exc_info, **kwargs)
|
|
|
|
def critical(self, *args):
|
|
self.log(LogLevel.Critical, *args)
|
|
|
|
def error(self, *args):
|
|
self.log(LogLevel.Error, *args)
|
|
|
|
def warning(self, *args):
|
|
self.log(LogLevel.Warning, *args)
|
|
|
|
def info(self, *args):
|
|
self.log(LogLevel.Info, *args)
|
|
|
|
def debug(self, *args):
|
|
self.log(LogLevel.Debug, *args)
|
|
|
|
def trace(self, *args):
|
|
self.log(LogLevel.Trace, *args)
|
|
|
|
|
|
T = typing.TypeVar("T")
|
|
P = typing.ParamSpec("P")
|
|
|
|
class Tracer:
|
|
def __init__(self, logger: Logger, level: LogLevel = LogLevel.Trace, max_depth: int = 15):
|
|
self.depth: int = 0
|
|
self.level = level
|
|
self.logger = logger
|
|
self.max_depth: int = max_depth
|
|
|
|
def trace_method(self, func_or_nothing: Callable[P, T] | None = None, /, **kwargs) -> Callable[P, T]:
|
|
if func_or_nothing is not None:
|
|
return self._trace_method(func=func_or_nothing, **kwargs)
|
|
|
|
return lambda func: self._trace_method(func, **kwargs)
|
|
|
|
def _trace_method(self, func: Callable[P, T], level: LogLevel = None) -> Callable[P, T]:
|
|
if level is None:
|
|
level = self.level
|
|
|
|
if self.logger.level > level:
|
|
return func
|
|
|
|
func_info = inspect.signature(func)
|
|
|
|
def _wrapped(*args: P.args, **kwargs: P.kwargs) -> T:
|
|
bound_args = func_info.bind(*args, **kwargs)
|
|
signature = ", ".join(
|
|
key if key == "self" else f"{key}={repr(value)}"
|
|
for key, value in bound_args.arguments.items()
|
|
)
|
|
prefix = f"[{self.depth: >4}]" + " " * self.depth
|
|
|
|
if self.depth == self.max_depth:
|
|
self.logger.log(level, f"{prefix} > ... (max depth of {self.max_depth} reached)")
|
|
if self.depth < self.max_depth:
|
|
self.logger.log(level, f"{prefix} > Entering {func.__name__}({signature})")
|
|
|
|
self.depth += 1
|
|
r: T = func(*args, **kwargs)
|
|
self.depth -= 1
|
|
|
|
if self.depth < self.max_depth:
|
|
self.logger.log(level, f"{prefix} < Leaving {func.__name__}({signature})")
|
|
return r
|
|
|
|
return _wrapped
|
|
|
|
|
|
logger = Logger("compiler")
|