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
|
import sys
from datetime import datetime
from threading import Lock, Thread
from time import sleep
class Term:
spinner = ["/", "-", "\\", "|", "/", "-", "\\", "|"]
logfile = f"logs/{datetime.now().isoformat()}.log"
term = sys.stdout
_th: Thread
_process_name = None
_process_state = None
_process_spin = False
_print_lock = Lock()
_logfile = None
@staticmethod
def begin():
Term._th = Thread(target=Term._term_thread)
Term._th.daemon = True
Term._th.start()
Term._logfile = open(Term.logfile, "a")
@staticmethod
def _term_thread():
spin_i = 0
while True:
sleep(0.2)
spin_i += 1
with Term._print_lock:
if Term._process_name is not None:
Term.mov_ls()
Term.write("\u001b[36m[")
Term.write(Term.spinner[spin_i % len(Term.spinner)] if Term._process_spin else " ")
Term.write(f"] {Term._process_name}")
if Term._process_state is not None:
Term.write(f"\u001b[35m - {Term._process_state}")
Term.write("\u001b[0m")
Term.flush()
@staticmethod
def flush(): Term.term.flush()
@staticmethod
def write(s: str): Term.term.write(s)
@staticmethod
def mov_ls(): Term.write("\u001b[2K\r")
@staticmethod
def newln(): Term.write("\n")
@staticmethod
def log(logmsg: str, display=True):
logmsg = str(logmsg)
with Term._print_lock:
Term._logfile.write(logmsg + "\n")
Term._logfile.flush()
if not display:
return
Term.mov_ls()
if Term._process_name is not None:
Term.write(" ")
Term.write(logmsg)
Term.newln()
Term.flush()
@staticmethod
def proc_start(procname: str, spin: bool = False):
with Term._print_lock:
Term._logfile.write(f"start [{procname}]\n")
Term._logfile.flush()
if Term._process_name is not None:
Term.proc_end("New process started (old may or may no actually have ended)", result=1)
if spin:
Term._process_spin = True
Term._process_name = procname
@staticmethod
def proc_state(state: str):
with Term._print_lock:
Term._logfile.write(f"[{Term._process_name}] {state}\n")
Term._logfile.flush()
if Term._process_name is not None:
Term._process_state = state
else:
Term.log(f"State for unknown process: {state}")
@staticmethod
def proc_end(msg=None, result=0):
""" Result 0=success, 1=error, 2=warning. """
colour = ["\u001b[32m", "\u001b[31m", "\u001b[33m"]
with Term._print_lock:
if Term._process_name is None:
raise Exception("No process to end...")
Term._logfile.write(f"end [{Term._process_name}] {msg}\n")
Term._logfile.flush()
Term.mov_ls()
Term.write(colour[result])
symb = " "
if result == 0: symb = "+"
if result == 1: symb = "-"
if result == 2: symb = "*"
Term.write(f"[{symb}] ")
Term.write("\u001b[0m")
Term.write(Term._process_name)
if msg is not None:
Term.write(f" - {msg}")
Term.write("\u001b[0m")
Term.newln()
Term.flush()
Term._process_name = None
Term._process_state = None
Term._process_spin = False
@staticmethod
def proc_spin(): Term._process_spin = True
@staticmethod
def proc_nospin(): Term._process_spin = False
if __name__ == "__main__":
Term.begin()
sleep(2)
Term.proc_start("TEST")
Term.log("LGOGGGG")
sleep(2)
Term.proc_spin()
Term.log("LGOGGGG")
|
