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
|
#
# Copyright 2017 Ettus Research (National Instruments)
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
"""
MPM Logging
"""
from __future__ import print_function
import copy
import logging
from logging import CRITICAL, ERROR, WARNING, INFO, DEBUG
from builtins import str
# Colors
BOLD = str('\033[1m')
RED = str('\x1b[31m')
YELLOW = str('\x1b[33m')
GREEN = str('\x1b[32m')
PINK = str('\x1b[35m')
GREY = str('\x1b[90m')
RESET = str('\x1b[0m')
# Additional log level
TRACE = 1
class ColorStreamHandler(logging.StreamHandler):
"""
StreamHandler that prints colored output
"""
def emit(self, record):
"""
Prints record with colors.
record is not modified.
"""
record_ = copy.copy(record)
levelno = record_.levelno
if levelno >= CRITICAL:
color = RED
elif levelno >= ERROR:
color = RED
elif levelno >= WARNING:
color = YELLOW
elif levelno >= INFO:
color = GREEN
elif levelno >= DEBUG:
color = PINK
elif levelno >= TRACE:
color = ''
else: # NOTSET and anything else
color = RESET
record_.msg = BOLD + color + str(record_.msg) + RESET
logging.StreamHandler.emit(self, record_)
class MPMLogger(logging.getLoggerClass()):
"""
Extends the regular Python logging with level 'trace' (like UHD)
"""
def __init__(self, *args, **kwargs):
logging.Logger.__init__(self, *args, **kwargs)
def trace(self, *args, **kwargs):
""" Extends logging for super-high verbosity """
self.log(TRACE, *args, **kwargs)
LOGGER = None # Logger singleton
def get_main_logger(use_console=True, use_journal=False, console_color=True):
"""
Returns the top-level logger object. This is the only API call from this
file that should be used outside.
"""
global LOGGER
if LOGGER is not None:
return LOGGER
logging.addLevelName(TRACE, 'TRACE')
logging.setLoggerClass(MPMLogger)
LOGGER = logging.getLogger('MPM')
if use_console:
console_handler = ColorStreamHandler() if console_color else logging.StreamHandler()
console_formatter = logging.Formatter("[%(name)s] [%(levelname)s] %(message)s")
console_handler.setFormatter(console_formatter)
LOGGER.addHandler(console_handler)
if use_journal:
from systemd.journal import JournalHandler
journal_handler = JournalHandler(SYSLOG_IDENTIFIER='usrp_hwd')
journal_formatter = logging.Formatter('[%(levelname)s] [%(module)s] %(message)s')
journal_handler.setFormatter(journal_formatter)
LOGGER.addHandler(journal_handler)
# Set default level:
default_log_level = TRACE
LOGGER.setLevel(default_log_level)
return LOGGER
def get_logger(child_name):
"""docstring for get_logger"""
assert LOGGER is not None
return get_main_logger().getChild(child_name)
if __name__ == "__main__":
print("Testing logger: ")
LOG = get_main_logger().getChild('test')
LOG.setLevel(TRACE)
LOG.trace("trace message")
LOG.debug("debug message")
LOG.info("info message")
LOG.warning("warning message")
LOG.error("error message")
LOG.critical("critical message")
LOG2 = get_main_logger()
LOG3 = get_main_logger()
assert LOG2 is LOG3
|
