Added clog package modules
This commit is a continuation from commit [3746b773f4] which is a separation of library modules into its own package from a larger project. Version control jump is significant in feature implementations due simply to under-using commit control after minor feature implementations. Project will receive better commit messages going forward on changes to the package.
This commit is contained in:
parent
b91572874e
commit
5e922c5515
GPG Key ID:
4E7D436CE1A0BAF1
4
__init__.py
Normal file
4
__init__.py
Normal file
|
|
@ -0,0 +1,4 @@
|
||||||
|
from ._logger import *
|
||||||
|
from .utils.common import *
|
||||||
|
#from .utils.printfmt import *
|
||||||
|
from .utils import printfmt
|
||||||
252
_logger.py
Normal file
252
_logger.py
Normal file
|
|
@ -0,0 +1,252 @@
|
||||||
|
from collections import namedtuple
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
import time
|
||||||
|
from io import TextIOWrapper
|
||||||
|
from typing import TextIO, Union, TYPE_CHECKING
|
||||||
|
|
||||||
|
from utils import common
|
||||||
|
from utils.printfmt import *
|
||||||
|
|
||||||
|
class Logger: ... # class delcaration
|
||||||
|
class Logger: # class redeclaration & initialisation
|
||||||
|
## set the PIPE to STDOUT by default
|
||||||
|
__stdpipe: TextIO = sys.stdout
|
||||||
|
## detect if there's a redirect
|
||||||
|
__IS_STDOUT_REDIR: bool = os.isatty(sys.stdout.fileno())
|
||||||
|
__IS_STDERR_REDIR: bool = os.isatty(sys.stderr.fileno())
|
||||||
|
## the default log out file
|
||||||
|
__DEFAULT_OUT_FILE: str = "dump.log"
|
||||||
|
## log message constant format template
|
||||||
|
__LOG_TMPL = "[{DATE}] [{0}] {TYPE} "
|
||||||
|
## string time format (ISO: 8601, long-form)
|
||||||
|
__TIMESTAMP_FMT = "%Y-%m-%dT%H:%M:%S%z"
|
||||||
|
## log info namedtuple for storing class states
|
||||||
|
__LOG_INFO_TUPLE = namedtuple('LogInfo',
|
||||||
|
['isatty', 'lv', 'msg', 'sep', 'end'])
|
||||||
|
## create instance attribute for class singleton
|
||||||
|
__instance__ = None
|
||||||
|
## create instance attribute as read-only for log location
|
||||||
|
log = None
|
||||||
|
|
||||||
|
|
||||||
|
def __new__(cls, *, out_f: str = ...) -> Logger:
|
||||||
|
if cls.__instance__ is None:
|
||||||
|
cls.__instance__ = super(Logger, cls).__new__(cls) # establish singleton instance
|
||||||
|
cls.log = cls.__DEFAULT_OUT_FILE # initialise attribute to default
|
||||||
|
## handle if a custom file pathspec was given
|
||||||
|
if out_f is not Ellipsis and isinstance(out_f, str):
|
||||||
|
## verify path and convert to real pathspec.
|
||||||
|
if common.isPathspec(out_f):
|
||||||
|
## redefine the default log out attribute and create
|
||||||
|
## public attribute for the currect log location
|
||||||
|
cls.log = cls.__DEFAULT_OUT_FILE = \
|
||||||
|
os.path.realpath(out_f).strip('"')
|
||||||
|
|
||||||
|
cls.__loginfo = None # default the namedtuple to None on first instance
|
||||||
|
|
||||||
|
cls.printLog2File("----[New instance of script has been started]----",
|
||||||
|
file=cls.__DEFAULT_OUT_FILE, mode='w')
|
||||||
|
|
||||||
|
return cls.__instance__
|
||||||
|
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def debug(cls, *value: object, sep: Union[str, None] = None,
|
||||||
|
end: Union[str, None] = None, wrapping: bool = True,
|
||||||
|
strace: bool = True) -> Logger:
|
||||||
|
""""""
|
||||||
|
Logger.printLog2File(*value, sep=sep, end=end,
|
||||||
|
wrapping=wrapping, strace=strace)
|
||||||
|
|
||||||
|
|
||||||
|
Logger.__stdpipe = sys.stderr
|
||||||
|
cls.__loginfo: namedtuple = Logger.__LOG_INFO_TUPLE(
|
||||||
|
Logger.__IS_STDERR_REDIR, LogLevel.DEBUG, value, sep, end
|
||||||
|
)
|
||||||
|
|
||||||
|
return cls
|
||||||
|
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def warn(cls, *value: object, sep: Union[str, None] = None,
|
||||||
|
end: Union[str, None] = None, wrapping: bool = True,
|
||||||
|
strace: bool = True) -> Logger:
|
||||||
|
""""""
|
||||||
|
Logger.printLog2File(*value, level=LogLevel.WARN, sep=sep,
|
||||||
|
end=end, wrapping=wrapping, strace=strace)
|
||||||
|
|
||||||
|
Logger.__stdpipe = sys.stderr
|
||||||
|
cls.__loginfo: namedtuple = Logger.__LOG_INFO_TUPLE(
|
||||||
|
Logger.__IS_STDERR_REDIR, LogLevel.WARN, value, sep, end
|
||||||
|
)
|
||||||
|
|
||||||
|
return cls
|
||||||
|
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def error(cls, *value: object, sep: Union[str, None] = None,
|
||||||
|
end: Union[str, None] = None, wrapping: bool = True,
|
||||||
|
strace: bool = True) -> Logger:
|
||||||
|
""""""
|
||||||
|
Logger.printLog2File(*value, level=LogLevel.ERROR, sep=sep,
|
||||||
|
end=end, wrapping=wrapping, strace=strace)
|
||||||
|
|
||||||
|
Logger.__stdpipe = sys.stderr
|
||||||
|
cls.__loginfo: namedtuple = Logger.__LOG_INFO_TUPLE(
|
||||||
|
Logger.__IS_STDERR_REDIR, LogLevel.ERROR, value, sep, end
|
||||||
|
)
|
||||||
|
|
||||||
|
return cls
|
||||||
|
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def withConsole() -> None:
|
||||||
|
if not Logger.__loginfo is None:
|
||||||
|
Logger.__printLog__(Logger.__loginfo.isatty,
|
||||||
|
Logger.__loginfo.lv, Logger.__loginfo.msg,
|
||||||
|
Logger.__loginfo.sep, Logger.__loginfo.end)
|
||||||
|
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def genLogHeader(_type: Union[LogLevel, int, str]) -> str:
|
||||||
|
## if the _type represents a LogLevel value, convert it to str
|
||||||
|
if not isinstance(_type, str):
|
||||||
|
_type = loglevel_as_str(_type)
|
||||||
|
|
||||||
|
## return newly formatted log message header
|
||||||
|
return Logger.__LOG_TMPL.format('{CALLER}', DATE=\
|
||||||
|
time.strftime(Logger.__TIMESTAMP_FMT, time.localtime()),
|
||||||
|
TYPE=_type) + "{0}"
|
||||||
|
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def printLog2File(*value: object,
|
||||||
|
level: Union[LogLevel, int] = LogLevel.DEBUG, mode: str = 'a',
|
||||||
|
file: Union[TextIOWrapper, str] = ..., sep: Union[str, None] = None,
|
||||||
|
end: Union[str, None] = None, wrapping: bool = True,
|
||||||
|
strace: bool = True, header: bool = True) -> None:
|
||||||
|
""""""
|
||||||
|
## handle if no file parameter was given
|
||||||
|
if file is Ellipsis or not os.path.exists(file):
|
||||||
|
file = Logger.__DEFAULT_OUT_FILE
|
||||||
|
|
||||||
|
_frame = sys._getframe(2) if sys._getframe(1).f_code.co_name in \
|
||||||
|
dir(Logger) else sys._getframe(1)
|
||||||
|
## get the executing filename of where log was called
|
||||||
|
_fname = _frame.f_code.co_filename.removeprefix(
|
||||||
|
os.getcwd()).strip('\\\/')
|
||||||
|
## generate new header for log file and construct new message
|
||||||
|
if header:
|
||||||
|
msg = Logger.genLogHeader(level).format(
|
||||||
|
" ".join(value), CALLER="{0}:{1}[{2}]".format(
|
||||||
|
_fname, _frame.f_code.co_name, _frame.f_lineno
|
||||||
|
).replace("module", "global") if strace else "LOGGER"
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
msg = " ".join(value)
|
||||||
|
|
||||||
|
# perform wrapping of message and indent wrapped lines
|
||||||
|
if wrapping:
|
||||||
|
msg = wrap(msg).replace('\n', '\n\t')
|
||||||
|
|
||||||
|
|
||||||
|
if isinstance(file, str):
|
||||||
|
with open(file, mode, encoding="utf-8") as log:
|
||||||
|
Logger.__stdpipe = log # pre-requisite to write PIPE to file
|
||||||
|
Logger.__printLog__(False, level, (msg,), sep, end, False)
|
||||||
|
elif isinstance(file, TextIOWrapper):
|
||||||
|
Logger.__stdpipe = log
|
||||||
|
Logger.__printLog__(False, level, (msg,), sep, end, False)
|
||||||
|
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def printLog(*value: object, level: Union[int, LogLevel] = LogLevel.NORMAL,
|
||||||
|
sep: Union[str, None] = None, end: Union[str, None] = None,
|
||||||
|
file: Union[TextIOWrapper, None] = None,
|
||||||
|
flush: bool = True) -> None:
|
||||||
|
"""Wrapper method over the built-in `print()` function defined
|
||||||
|
using 3.x syntax. All Familiar functionality can be passed to
|
||||||
|
the method as found when calling `print()`, but comes with added
|
||||||
|
features.
|
||||||
|
|
||||||
|
`IO.printLog` is designed for purpose of logging information to
|
||||||
|
the console window or to a file, either via an explicit write by
|
||||||
|
passing a compatible `SupportsWrite[str]` value to `file=`, or
|
||||||
|
by redirecting the standard PIPE streams to an external file.
|
||||||
|
In addition, different levels of logging will result in output
|
||||||
|
to standard PIPE streams to have appropriate highlighting to the
|
||||||
|
message displayed. If standard PIPE streams are to be redirected
|
||||||
|
to an external file via a PIPE redirect, the highlighting syntax
|
||||||
|
is dropped do prevent ANSI escape code sequences from being
|
||||||
|
written to file.
|
||||||
|
|
||||||
|
The standard log level is `NORMAL`, referring to standard
|
||||||
|
formatted text to the standard stream. Log level can be
|
||||||
|
elevated by either passing an integer to represent the log level,
|
||||||
|
or pass an enum variable from `class LogLevel` from the
|
||||||
|
`utils/printfmt.py` module.
|
||||||
|
|
||||||
|
Examples of logging:
|
||||||
|
```
|
||||||
|
>>> import utils
|
||||||
|
>>> utils.IO.printLog("Hello, World!")
|
||||||
|
Hello, World!
|
||||||
|
>>> utils.IO.printLog("Hello,", "World" + "!", level=LogLevel.DEBUG)
|
||||||
|
\033[94mHello, World!\033[0m
|
||||||
|
>>> utils.IO.printLog("Hello,", end=" ") ; utils.IO.printLog("World!", level=1)
|
||||||
|
Hello, \033[92mWorld!\033[0m
|
||||||
|
>>> with open("dump.log", 'a') as log_file:
|
||||||
|
... utils.IO.printLog("Hello, Log File!", file=log_file)
|
||||||
|
...
|
||||||
|
>>>
|
||||||
|
```
|
||||||
|
"""
|
||||||
|
### :@Ethan: whilst `print` disables force flushing of the
|
||||||
|
# stream, from testing, it's best to forcibly flush the stream
|
||||||
|
# so the default behaviour is to do exactly this.
|
||||||
|
|
||||||
|
## configure PIPE to STDERR if logging is high enough
|
||||||
|
if file is None:
|
||||||
|
if level >= LogLevel.DEBUG:
|
||||||
|
Logger.__stdpipe = sys.stderr
|
||||||
|
else:
|
||||||
|
Logger.__stdpipe = sys.stdout
|
||||||
|
## handle if the file is a TextIOWrapper
|
||||||
|
elif isinstance(file, TextIOWrapper):
|
||||||
|
Logger.__stdpipe = file # `print` will handle this as is.
|
||||||
|
## otherwise the method was given an invalid argument
|
||||||
|
else:
|
||||||
|
Logger.printLog("Warning: logging function was called with a",
|
||||||
|
"file specifier parameter which is not a valid option.",
|
||||||
|
level=LogLevel.WARN)
|
||||||
|
return
|
||||||
|
|
||||||
|
|
||||||
|
## display message to console with appropriate colouring
|
||||||
|
### :@NOTE: if there's a PIPE redirect, don't use colour
|
||||||
|
### for that redirect PIPE
|
||||||
|
|
||||||
|
if level < LogLevel.WARN: ## handle output for STDOUT
|
||||||
|
Logger.__printLog__(Logger.__IS_STDOUT_REDIR, level, value,
|
||||||
|
sep, end, flush)
|
||||||
|
else: ## handle output for STDERR
|
||||||
|
Logger.__printLog__(Logger.__IS_STDERR_REDIR, level, value,
|
||||||
|
sep, end, flush)
|
||||||
|
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def __printLog__(isatty: bool, lv: LogLevel, msg: object,
|
||||||
|
s: Union[str, None] = None, e: Union[str, None] = None,
|
||||||
|
flsh: bool = True) -> None:
|
||||||
|
## handle if we have a redirect
|
||||||
|
if isatty and (Logger.__stdpipe is sys.stdout or \
|
||||||
|
Logger.__stdpipe is sys.stderr):
|
||||||
|
## write ANSI code to start coloured text
|
||||||
|
print(logAsCol(lv), end="", file=Logger.__stdpipe, flush=flsh)
|
||||||
|
## unpack the object and pass to print
|
||||||
|
print(*msg, sep=s, end="", file=Logger.__stdpipe, flush=flsh)
|
||||||
|
## reset the colour sequence back to normal
|
||||||
|
print(Colours.NORMAL, end=e, file=Logger.__stdpipe, flush=flsh)
|
||||||
|
else:
|
||||||
|
print(*msg, sep=s, end=e, file=Logger.__stdpipe, flush=flsh)
|
||||||
10
main.py
Normal file
10
main.py
Normal file
|
|
@ -0,0 +1,10 @@
|
||||||
|
if __name__ == "__main__":
|
||||||
|
import _logger
|
||||||
|
|
||||||
|
logger = _logger.Logger()
|
||||||
|
|
||||||
|
logger.debug("This is a debug message").withConsole()
|
||||||
|
logger.warn("This is a warning message").withConsole()
|
||||||
|
logger.error("This is an error message").withConsole()
|
||||||
|
|
||||||
|
logger.debug("A message to file but not TTY")
|
||||||
2
utils/__init__.py
Normal file
2
utils/__init__.py
Normal file
|
|
@ -0,0 +1,2 @@
|
||||||
|
from .common import *
|
||||||
|
from .printfmt import *
|
||||||
21
utils/common.py
Normal file
21
utils/common.py
Normal file
|
|
@ -0,0 +1,21 @@
|
||||||
|
import os
|
||||||
|
import re
|
||||||
|
|
||||||
|
class LogLevel:
|
||||||
|
NORMAL = 0
|
||||||
|
PASS = 1
|
||||||
|
DEBUG = 2
|
||||||
|
WARN = 3
|
||||||
|
ERROR = 4
|
||||||
|
|
||||||
|
def isPathspec(path_spec: str) -> bool:
|
||||||
|
__OS: str = os.name
|
||||||
|
# define the path spec pattern depending on OS
|
||||||
|
__REGEX_PAT = re.compile(r'^(.+)[\\]([^\\]+)\\*$') \
|
||||||
|
if __OS == "nt" else re.compile(r'^(.+)[\/]([^\/]+)$')
|
||||||
|
## perform a regex match to ensure that the given path is a
|
||||||
|
## valid pathspec for the system.
|
||||||
|
return bool(re.match(
|
||||||
|
__REGEX_PAT, path_spec.strip(r"\"'")) or \
|
||||||
|
re.match(r'^[\w\d\-_]+$', path_spec.strip(r"\"'"))
|
||||||
|
)
|
||||||
41
utils/printfmt.py
Normal file
41
utils/printfmt.py
Normal file
|
|
@ -0,0 +1,41 @@
|
||||||
|
import textwrap
|
||||||
|
from typing import Union
|
||||||
|
from .common import LogLevel
|
||||||
|
|
||||||
|
class Colours:
|
||||||
|
NORMAL = '\033[0m'
|
||||||
|
RED = '\033[91m'
|
||||||
|
GREEN = '\033[92m'
|
||||||
|
YELLOW = '\033[93m'
|
||||||
|
BLUE = '\033[94m'
|
||||||
|
|
||||||
|
|
||||||
|
def loglevel_as_str(level: Union[LogLevel, int]) -> str:
|
||||||
|
## perform a lookup of each attribute of LogLevel and its associate
|
||||||
|
## value, and search for the attribute that has the matching level
|
||||||
|
## value passed to the function.
|
||||||
|
_attrs = [_dir for _dir in dir(LogLevel) \
|
||||||
|
if not _dir.startswith('__')] # list of all attributes of LogLevel
|
||||||
|
for _ in _attrs: # iterate over all attributes defined
|
||||||
|
if getattr(LogLevel, _) == level: # if the level matches
|
||||||
|
return "{0: <5}".format(_) # return the attribute name
|
||||||
|
|
||||||
|
return " "
|
||||||
|
|
||||||
|
|
||||||
|
def logAsCol(level: Union[int, LogLevel]) -> Colours:
|
||||||
|
## perform a colour lookup based on level number
|
||||||
|
if level == LogLevel.DEBUG:
|
||||||
|
return Colours.BLUE
|
||||||
|
elif level == LogLevel.WARN:
|
||||||
|
return Colours.YELLOW
|
||||||
|
elif level == LogLevel.ERROR:
|
||||||
|
return Colours.RED
|
||||||
|
elif level == LogLevel.PASS:
|
||||||
|
return Colours.GREEN
|
||||||
|
else:
|
||||||
|
return Colours.NORMAL
|
||||||
|
|
||||||
|
|
||||||
|
def wrap(value: str, *, width: int = 120, tb_size: int = 4) -> str:
|
||||||
|
return "\n".join(textwrap.wrap(value, width=width, tabsize=tb_size))
|
||||||
Loading…
Reference in New Issue
Block a user