Coverage for mango/errors.py : 59%
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from os import kill
2from collections import OrderedDict
3from contextlib import suppress
4from datetime import datetime
5from os import path
6from shutil import copyfile
9from mango.constants import c
11import sys
14class _strings():
16 if c.t.does_styling: 16 ↛ 17line 16 didn't jump to line 17, because the condition on line 16 was never true
17 u = c.t.underline
18 HEADER = u + c.t.magenta
19 OKBLUE = u + c.t.blue
20 OKGREEN = u + c.t.green
21 WARNING = u + c.t.yellow
22 FAIL = u + c.t.red
23 ENDC = u + c.t.normal
24 else:
25 HEADER = ''
26 OKBLUE = ''
27 OKGREEN = ''
28 WARNING = ''
29 FAIL = ''
30 ENDC = ''
32 if c.using_IP: 32 ↛ 33line 32 didn't jump to line 33, because the condition on line 32 was never true
33 tab = ''
34 else:
35 tab = " "
37 _W = ("{}{f}Warning{e}: ".format(tab, f=WARNING, e=ENDC), 9)
38 _F = ("{}{f}Error{e}: ".format(tab, f=FAIL, e=ENDC), 7)
39 _G = ("{}{f}D'oh!{e}: ".format(tab, f=OKGREEN, e=ENDC), 7)
40 _B = ("{}{f}Keyboard Interrupt{e}: ".format(tab, f=OKBLUE, e=ENDC), 20)
41 _M = ("{}{f}Message{e}: ".format(tab, f=HEADER, e=ENDC), 9)
44class error(_strings):
45 """
46 Error class.
48 Handles error printing to the terminal
50 * All errors are written to stderr
51 """
53 def __init__(self, op=open):
54 """Init."""
55 self.type_e = {"W": self._W, "F": self._F,
56 "G": self._G, "B": self._B,
57 "M": self._M, "ME": self._M}
59 self.error_list = OrderedDict()
60 self.reprinter = Reprinter()
61 self.valuehold = 0
62 self.w_message = True
63 self.directory = "./"
64 self.run = 0
65 self.pids = []
66 self.open = op
67 self.pp = ''
69 def count(self, etype, msg):
70 """
71 ">" infront of error message will remove call count.
73 Parameters
74 ----------
75 etype: string
76 Error type W, F, G, B, M, ME
77 (Warning, Fatal, Oops, Keyboard, Message, Message exit)
78 msg: string
79 message for error
81 """
82 nocount = (etype[0] == '>')
84 self.etype = etype.replace(">", '')
85 self.message = '{}{}{}{}'.format(">" if nocount else '', self.type_e[self.etype][0],
86 msg.replace("\n", "\n" + " " * self.type_e[self.etype][1] + self.tab),
87 '\n' if self.etype in ['F', 'ME'] else '')
89 if self.message in self.error_list: 89 ↛ 90line 89 didn't jump to line 90, because the condition on line 89 was never true
90 self.error_list[self.message] += 1
91 else:
92 self.error_list[self.message] = 1
94 self.print()
96 if self.etype in ["F", "ME"]: 96 ↛ exitline 96 didn't return from function 'count', because the condition on line 96 was never false
97 self._killproc()
99 def setup(self, w_message, directory, run, pp=0):
100 """
101 Set up for the error system.
103 Parameters
104 ----------
105 w_message: bool
106 Turns (most) errors on or off, defaults to on
107 directory: string
108 Error save directory
109 run: int
110 run number
112 """
113 self.directory = directory
114 self.w_message = w_message
115 self.run = run
116 self.pp = '_pp' if pp == 1 else ''
118 self.errorfile = "{}/Run{}{}_Errors".format(self.directory, self.run, self.pp)
120 if path.isfile(self.errorfile):
121 copyfile(self.errorfile, self.errorfile + "R" + datetime.now().isoformat(timespec='minutes'))
123 def addpid(self, pid):
124 """PID collection."""
125 self.pids += [pid]
127 def _killproc(self):
128 """Fatal error exit."""
129 self.endproc()
130 for i in self.pids: 130 ↛ 131line 130 didn't jump to line 131, because the loop on line 130 never started
131 with suppress(ProcessLookupError):
132 kill(i, 9)
134 exit(0)
136 def endproc(self):
137 """Write errors to file at end of run."""
138 # Possibly dont write file if certain errors exist only TODO
139 if self.error_list and self.run != 0: 139 ↛ 140line 139 didn't jump to line 140, because the condition on line 139 was never true
140 self._collector()
141 with self.open(self.errorfile, 'w') as errorfile:
142 errorfile.write(self.keyhold)
144 def _collector(self):
145 self.keyhold = ''
146 for key, value in self.error_list.copy().items():
147 if key[0] == ">":
148 fkey = key[1:]
149 elif self.etype == "F" or self.etype == "B":
150 fkey = key
151 else:
152 fkey = key + " x" + str(value)
153 self.keyhold += fkey + "\n"
155 def print(self):
156 """Print errors to screen."""
157 if self.w_message or self.etype in ["F", "B", "G", "ME"]: 157 ↛ exitline 157 didn't return from function 'print', because the condition on line 157 was never false
158 self.reprinter.reprint(self.message.strip(">") + "\n")
161class Reprinter:
162 """Rewrite terminal output using ANSI escape sequences."""
164 def __init__(self):
165 """Init."""
166 self.text = ''
167 self.frun = True
169 @staticmethod
170 def _moveup():
171 print(chr(27) + '[9;1H', file=sys.stderr, end='')
173 def reprint(self, text):
174 """
175 Reprinter.
177 Parameters
178 ----------
179 text: string
180 text to be overwritable
182 """
183 if self.frun: 183 ↛ 187line 183 didn't jump to line 187, because the condition on line 183 was never false
184 print(file=sys.stderr)
185 self.frun = False
186 # self._moveup()
187 print(text, file=sys.stderr, end="")
188 self.text = text
191def _clientError(Message):
192 """Raise Kill when asked."""
193 # Possible race condition
194 # Killing threads, writing error to file killed before finished
195 if Message.startswith("F"):
196 raise ExitQuiet from None
199class Quiet(Exception):
200 """Base Exception for no traceback."""
202 pass
205class ExitQuiet(Quiet):
206 """Exit Program Quietly Exception."""
208 pass
211def notraceback(kind, message, traceback):
212 """Exceptionhook only print errors if not Quiet."""
213 if Quiet not in kind.__bases__:
214 if not issubclass(kind, KeyError):
215 message = ecol(message)
216 sys.__excepthook__(kind, message, traceback)
217 else:
218 message = ecol(message)
219 sys.stderr.write(message.args[0] + "\n")
222def ecol(m):
223 """Colour crash errors."""
224 if len(m.args) > 0:
225 m.args = (_strings.FAIL + f"{m.args[0]}" + _strings.ENDC,)
226 return m
229sys.excepthook = notraceback
231if __name__ == '__main__':
232 pass