-
Notifications
You must be signed in to change notification settings - Fork 3
Expand file tree
/
Copy pathcommon.py
More file actions
145 lines (112 loc) · 3.78 KB
/
Copy pathcommon.py
File metadata and controls
145 lines (112 loc) · 3.78 KB
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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
import _thread
import utime
from event_message import Event, EventManager
class Lock(object):
def __init__(self):
self.lock = _thread.allocate_lock()
def __enter__(self, *args, **kwargs):
self.lock.acquire()
def __exit__(self, *args, **kwargs):
self.lock.release()
class LOG_LV(object):
"""日志等级"""
DEBUG = "DEBUG"
INFO = "INFO"
WARNING = "WARNING"
ERROR = "ERROR"
CRITICAL = "CRITICAL"
class Abstract(object):
def post_processor_after_instantiation(self, *args, **kwargs):
"""实例化后调用"""
pass
def post_processor_before_initialization(self, *args, **kwargs):
"""初始化之前调用"""
pass
def initialization(self, *args, **kwargs):
"""初始化load"""
pass
def post_processor_after_initialization(self, *args, **kwargs):
"""初始化之后调用"""
pass
class AbstractOutput(object):
def open(self):
pass
def output(self, message, **kwargs):
pass
def close(self):
pass
class PrintLogOutput(AbstractOutput):
def output(self, message, **kwargs):
fmt = "{} [ {} ] {} {}"
print(fmt.format(*message))
class LogService(Abstract):
"""
default log is async queue
you can set
"""
INSTANCE = None
REPORTER = PrintLogOutput()
LEVEL = LOG_LV.DEBUG
MODE = 1
LEVEL_MAP = {
LOG_LV.DEBUG: 0,
LOG_LV.INFO: 1,
LOG_LV.WARNING: 2,
LOG_LV.ERROR: 3,
LOG_LV.CRITICAL: 4
}
def __init__(self):
self.__event = Event(__name__)
self.__em = EventManager()
self.__em.register_event(self.__event)
@classmethod
def create(cls):
if not cls.INSTANCE:
cls.INSTANCE = LogService()
cls.INSTANCE.start()
return cls.INSTANCE
def log_send(self, tag, level, msg, mode):
if self.MODE:
mode = self.MODE
if self.LEVEL_MAP[level] >= self.LEVEL_MAP.get(self.LEVEL, 0):
output_format = "{ascdate} {asctime}"
rt = utime.localtime()
d_f = "{0:02}"
send_msg = (output_format.format(
ascdate=str(rt[0]) + "-" + d_f.format(rt[1]) + "-" + d_f.format(rt[2])
, asctime=d_f.format(rt[3]) + ":" + d_f.format(rt[4]) + ":" + d_f.format(rt[5])
), tag, level, msg)
# self.REPORTER.output(send_msg)
print("{} [ {} ] {} {}".format(*send_msg))
# self.__event.post(send_msg)
def output(self, *args, **kwargs):
emo = kwargs.get("event_message")
self.REPORTER.output(emo.msg)
def start(self):
self.__event.add_handler(self.output)
self.__em.start()
class LogAdapter(object):
"""
log adapter mode
mode used : adapter proxy single LogService
"""
def __init__(self, tag, enable=1):
self.log_service = LogService.create()
self.tag = tag
self.enable = enable
self.mode = 1
def critical(self, msg):
if self.enable:
self.log_service.log_send(self.tag, LOG_LV.CRITICAL, msg, self.mode)
def debug(self, msg):
if self.enable:
self.log_service.log_send(self.tag, LOG_LV.DEBUG, msg, self.mode)
def info(self, msg):
if self.enable:
self.log_service.log_send(self.tag, LOG_LV.INFO, msg, self.mode)
def warning(self, msg):
if self.enable:
self.log_service.log_send(self.tag, LOG_LV.WARNING, msg, self.mode)
def error(self, msg):
if self.enable:
self.log_service.log_send(self.tag, LOG_LV.ERROR, msg, self.mode)