-
Notifications
You must be signed in to change notification settings - Fork 2
/
Copy pathlogging.py
131 lines (104 loc) · 3.12 KB
/
logging.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
import utime
import sys
import uio as io
import _thread
class Level(object):
DEBUG = 0
INFO = 1
WARN = 2
ERROR = 3
CRITICAL = 4
_levelToName = {
Level.CRITICAL: "CRITICAL",
Level.ERROR: "ERROR",
Level.WARN: "WARN",
Level.INFO: "INFO",
Level.DEBUG: "DEBUG"
}
_nameToLevel = {
"CRITICAL": Level.CRITICAL,
"ERROR": Level.ERROR,
"WARN": Level.WARN,
"INFO": Level.INFO,
"DEBUG": Level.DEBUG,
}
def getLevelName(level):
if level not in _levelToName:
raise ValueError("unknown level \"{}\", choose from <class Level>.".format(level))
return _levelToName[level]
def getNameLevel(name):
temp = name.upper()
if temp not in _nameToLevel:
raise ValueError("\"{}\" is not valid. choose from [{}]".format(name, list(_nameToLevel.keys())))
return _nameToLevel[temp]
class BasicConfig(object):
logger_register_table = {}
basic_configure = {
"level": Level.WARN,
"debug": True,
"stream": sys.stdout
}
@classmethod
def getLogger(cls, name):
if name not in cls.logger_register_table:
logger = Logger(name)
cls.logger_register_table[name] = logger
else:
logger = cls.logger_register_table[name]
return logger
@classmethod
def update(cls, **kwargs):
level = kwargs.pop("level", None)
if level is not None:
kwargs["level"] = getNameLevel(level)
return cls.basic_configure.update(kwargs)
@classmethod
def get(cls, key):
return cls.basic_configure[key]
@classmethod
def set(cls, key, value):
if key == "level":
value = getNameLevel(value)
cls.basic_configure[key] = value
class Logger(object):
lock = _thread.allocate_lock()
def __init__(self, name):
self.name = name
@staticmethod
def __getFormattedTime():
# (2023, 9, 30, 11, 11, 41, 5, 273)
cur_time_tuple = utime.localtime()
return "{:04d}-{:02d}-{:02d} {:02d}:{:02d}:{:02d}".format(
cur_time_tuple[0],
cur_time_tuple[1],
cur_time_tuple[2],
cur_time_tuple[3],
cur_time_tuple[4],
cur_time_tuple[5]
)
def log(self, level, *message):
if not BasicConfig.get("debug"):
if level < BasicConfig.get("level"):
return
stream = BasicConfig.get("stream")
prefix = "[{}][{}][{}]".format(
self.__getFormattedTime(),
getLevelName(level),
self.name,
)
with self.lock:
print(prefix, *message, file=stream)
if isinstance(stream, io.TextIOWrapper):
stream.flush()
def debug(self, *message):
self.log(Level.DEBUG, *message)
def info(self, *message):
self.log(Level.INFO, *message)
def warn(self, *message):
self.log(Level.WARN, *message)
def error(self, *message):
self.log(Level.ERROR, *message)
def critical(self, *message):
self.log(Level.CRITICAL, *message)
def getLogger(name):
return BasicConfig.getLogger(name)