/
message.py
340 lines (277 loc) · 9.43 KB
/
message.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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
"""
Copyright (C) 2017-2024 Vanessa Sochat.
This Source Code Form is subject to the terms of the
Mozilla Public License, v. 2.0. If a copy of the MPL was not distributed
with this file, You can obtain one at http://mozilla.org/MPL/2.0/.
Modified from https://github.com/Visual-mov/Colorful-Julia (MIT License)
"""
import os
import sys
from .spinner import Spinner
ABORT = -5
CRITICAL = -4
ERROR = -3
WARNING = -2
LOG = -1
INFO = 1
CUSTOM = 1
QUIET = 0
VERBOSE = VERBOSE1 = 2
VERBOSE2 = 3
VERBOSE3 = 4
DEBUG = 5
PURPLE = "\033[95m"
YELLOW = "\033[93m"
RED = "\033[91m"
DARKRED = "\033[31m"
CYAN = "\033[36m"
class SCIFMessage:
def __init__(self, MESSAGELEVEL=None):
self.level = update_level()
self.history = []
self.errorStream = sys.stderr
self.outputStream = sys.stdout
self.colorize = self.useColor()
self.colors = {
ABORT: DARKRED,
CRITICAL: RED,
ERROR: RED,
WARNING: YELLOW,
LOG: PURPLE,
CUSTOM: PURPLE,
DEBUG: CYAN,
"OFF": "\033[0m", # end sequence
"CYAN": CYAN,
"PURPLE": PURPLE,
"RED": RED,
"DARKRED": DARKRED,
"YELLOW": YELLOW,
}
# Colors --------------------------------------------
def useColor(self):
"""useColor will determine if color should be added
to a print. Will check if being run in a terminal, and
if has support for asci"""
COLORIZE = get_user_color_preference()
if COLORIZE is not None:
return COLORIZE
streams = [self.errorStream, self.outputStream]
for stream in streams:
if not hasattr(stream, "isatty"):
return False
if not stream.isatty():
return False
return True
def addColor(self, level, text):
"""addColor to the prompt (usually prefix) if terminal
supports, and specified to do so"""
if self.colorize:
if level in self.colors:
text = "%s%s%s" % (self.colors[level], text, self.colors["OFF"])
return text
def emitError(self, level):
"""determine if a level should print to
stderr, includes all levels but INFO and QUIET"""
if level in [
ABORT,
ERROR,
WARNING,
VERBOSE,
VERBOSE1,
VERBOSE2,
VERBOSE3,
DEBUG,
]:
return True
return False
def emitOutput(self, level):
"""determine if a level should print to stdout
only includes INFO"""
if level in [LOG, INFO]:
return True
return False
def isEnabledFor(self, messageLevel):
"""check if a messageLevel is enabled to emit a level"""
if messageLevel <= self.level:
return True
return False
def emit(self, level, message, prefix=None, color=None):
"""emit is the main function to print the message
optionally with a prefix
:param level: the level of the message
:param message: the message to print
:param prefix: a prefix for the message
"""
self.level = update_level()
if color is None:
color = level
if prefix is not None:
prefix = self.addColor(color, "%s " % (prefix))
else:
prefix = ""
message = self.addColor(color, message)
# Add the prefix
message = "%s%s" % (prefix, message)
if not message.endswith("\n"):
message = "%s\n" % message
# If the level is quiet, only print to error
if self.level == QUIET:
pass
# Otherwise if in range print to stdout and stderr
elif self.isEnabledFor(level):
if self.emitError(level):
self.write(self.errorStream, message)
else:
self.write(self.outputStream, message)
# Add all log messages to history
self.history.append(message)
def write(self, stream, message):
"""write will write a message to a stream,
first checking the encoding
"""
if isinstance(message, bytes):
message = message.decode("utf-8")
stream.write(message)
def get_logs(self, join_newline=True):
"""'get_logs will return the complete history, joined by newline
(default) or as is.
"""
if join_newline:
return "\n".join(self.history)
return self.history
def show_progress(
self,
iteration,
total,
length=40,
min_level=0,
prefix=None,
carriage_return=True,
suffix=None,
symbol=None,
):
"""create a terminal progress bar, default bar shows for verbose+
:param iteration: current iteration (Int)
:param total: total iterations (Int)
:param length: character length of bar (Int)
"""
self.level = update_level()
percent = 100 * (iteration / float(total))
progress = int(length * iteration // total)
if suffix is None:
suffix = ""
if prefix is None:
prefix = "Progress"
# Download sizes can be imperfect, setting carriage_return to False
# and writing newline with caller cleans up the UI
if percent >= 100:
percent = 100
progress = length
if symbol is None:
symbol = "="
if progress < length:
bar = symbol * progress + "|" + "-" * (length - progress - 1)
else:
bar = symbol * progress + "-" * (length - progress)
# Only show progress bar for level > min_level
if self.level > min_level:
percent = "%5s" % ("{0:.1f}").format(percent)
output = "\r" + prefix + " |%s| %s%s %s" % (bar, percent, "%", suffix)
sys.stdout.write(output),
if iteration == total and carriage_return:
sys.stdout.write("\n")
sys.stdout.flush()
# Logging ------------------------------------------
def abort(self, message):
self.emit(ABORT, message, "ABORT")
def critical(self, message):
self.emit(CRITICAL, message, "CRITICAL")
def error(self, message):
self.emit(ERROR, message, "ERROR")
def exit(self, message, error_code=1):
self.emit(ERROR, message, "ERROR")
sys.exit(error_code)
def warning(self, message):
self.emit(WARNING, message, "WARNING")
def log(self, message):
self.emit(LOG, message, "LOG")
def custom(self, prefix, message="", color=PURPLE):
self.emit(CUSTOM, message, prefix, color)
def info(self, message):
self.emit(INFO, message)
def newline(self):
return self.info("")
def verbose(self, message):
self.emit(VERBOSE, message, "VERBOSE")
def verbose1(self, message):
self.emit(VERBOSE, message, "VERBOSE1")
def verbose2(self, message):
self.emit(VERBOSE2, message, "VERBOSE2")
def verbose3(self, message):
self.emit(VERBOSE3, message, "VERBOSE3")
def debug(self, message):
self.emit(DEBUG, message, "DEBUG")
def is_quiet(self):
"""is_quiet returns true if the level is under 1"""
if self.level < 1:
return False
return True
# Terminal ------------------------------------------
def table(self, rows, col_width=2):
"""table will print a table of entries. If the rows is
a dictionary, the keys are interpreted as column names. if
not, a numbered list is used.
"""
labels = [str(x) for x in range(1, len(rows) + 1)]
if isinstance(rows, dict):
labels = list(rows.keys())
rows = list(rows.values())
for row in rows:
label = labels.pop(0)
label = label.ljust(col_width)
message = "\t".join(row)
self.custom(prefix=label, message=message)
def update_level():
"""update_level will configure a logging to standard out based on the user's
selected level, which should be in an environment variable called
MESSAGELEVEL. if MESSAGELEVEL is not set, the maximum level
(5) is assumed (all messages).
"""
try:
level = int(os.environ.get("SCIF_MESSAGELEVEL", INFO))
except ValueError:
level = str(os.environ.get("SCIF_MESSAGELEVEL", INFO))
if level == "CRITICAL":
return CRITICAL
elif level == "ABORT":
return ABORT
elif level == "ERROR":
return ERROR
elif level == "WARNING":
return WARNING
elif level == "LOG":
return LOG
elif level == "INFO":
return INFO
elif level == "QUIET":
return QUIET
elif level.startswith("VERBOSE"):
return VERBOSE3
elif level == "LOG":
return LOG
elif level == "DEBUG":
return DEBUG
return level
def get_user_color_preference():
COLORIZE = os.environ.get("SCIF_COLORIZE", None)
if COLORIZE is not None:
COLORIZE = convert2boolean(COLORIZE)
return COLORIZE
def convert2boolean(arg):
"""convert2boolean is used for environmental variables that must be
returned as boolean"""
if not isinstance(arg, bool):
return arg.lower() in ("yes", "true", "t", "1", "y")
return arg
SCIFMessage.spinner = Spinner()
bot = SCIFMessage()