from os import kill
from collections import OrderedDict
from contextlib import suppress
from datetime import datetime
from os import path
from shutil import copyfile
from mango.constants import c
import sys
class _strings():
if c.t.does_styling:
u = c.t.underline
HEADER = u + c.t.magenta
OKBLUE = u + c.t.blue
OKGREEN = u + c.t.green
WARNING = u + c.t.yellow
FAIL = u + c.t.red
ENDC = u + c.t.normal
else:
HEADER = ''
OKBLUE = ''
OKGREEN = ''
WARNING = ''
FAIL = ''
ENDC = ''
if c.using_IP:
tab = ''
else:
tab = " "
_W = ("{}{f}Warning{e}: ".format(tab, f=WARNING, e=ENDC), 9)
_F = ("{}{f}Error{e}: ".format(tab, f=FAIL, e=ENDC), 7)
_G = ("{}{f}D'oh!{e}: ".format(tab, f=OKGREEN, e=ENDC), 7)
_B = ("{}{f}Keyboard Interrupt{e}: ".format(tab, f=OKBLUE, e=ENDC), 20)
_M = ("{}{f}Message{e}: ".format(tab, f=HEADER, e=ENDC), 9)
[docs]class error(_strings):
"""
Error class.
Handles error printing to the terminal
* All errors are written to stderr
"""
def __init__(self, op=open):
"""Init."""
self.type_e = {"W": self._W, "F": self._F,
"G": self._G, "B": self._B,
"M": self._M, "ME": self._M}
self.error_list = OrderedDict()
self.reprinter = Reprinter()
self.valuehold = 0
self.w_message = True
self.directory = "./"
self.run = 0
self.pids = []
self.open = op
self.pp = ''
[docs] def count(self, etype, msg):
"""
">" infront of error message will remove call count.
Parameters
----------
etype: string
Error type W, F, G, B, M, ME
(Warning, Fatal, Oops, Keyboard, Message, Message exit)
msg: string
message for error
"""
nocount = (etype[0] == '>')
self.etype = etype.replace(">", '')
self.message = '{}{}{}{}'.format(">" if nocount else '', self.type_e[self.etype][0],
msg.replace("\n", "\n" + " " * self.type_e[self.etype][1] + self.tab),
'\n' if self.etype in ['F', 'ME'] else '')
if self.message in self.error_list:
self.error_list[self.message] += 1
else:
self.error_list[self.message] = 1
self.print()
if self.etype in ["F", "ME"]:
self._killproc()
[docs] def setup(self, w_message, directory, run, pp=0):
"""
Set up for the error system.
Parameters
----------
w_message: bool
Turns (most) errors on or off, defaults to on
directory: string
Error save directory
run: int
run number
"""
self.directory = directory
self.w_message = w_message
self.run = run
self.pp = '_pp' if pp == 1 else ''
self.errorfile = "{}/Run{}{}_Errors".format(self.directory, self.run, self.pp)
if path.isfile(self.errorfile):
copyfile(self.errorfile, self.errorfile + "R" + datetime.now().isoformat(timespec='minutes'))
[docs] def addpid(self, pid):
"""PID collection."""
self.pids += [pid]
def _killproc(self):
"""Fatal error exit."""
self.endproc()
for i in self.pids:
with suppress(ProcessLookupError):
kill(i, 9)
exit(0)
[docs] def endproc(self):
"""Write errors to file at end of run."""
# Possibly dont write file if certain errors exist only TODO
if self.error_list and self.run != 0:
self._collector()
with self.open(self.errorfile, 'w') as errorfile:
errorfile.write(self.keyhold)
def _collector(self):
self.keyhold = ''
for key, value in self.error_list.copy().items():
if key[0] == ">":
fkey = key[1:]
elif self.etype == "F" or self.etype == "B":
fkey = key
else:
fkey = key + " x" + str(value)
self.keyhold += fkey + "\n"
[docs] def print(self):
"""Print errors to screen."""
if self.w_message or self.etype in ["F", "B", "G", "ME"]:
self.reprinter.reprint(self.message.strip(">") + "\n")
[docs]class Reprinter:
"""Rewrite terminal output using ANSI escape sequences."""
def __init__(self):
"""Init."""
self.text = ''
self.frun = True
@staticmethod
def _moveup():
print(chr(27) + '[9;1H', file=sys.stderr, end='')
[docs] def reprint(self, text):
"""
Reprinter.
Parameters
----------
text: string
text to be overwritable
"""
if self.frun:
print(file=sys.stderr)
self.frun = False
# self._moveup()
print(text, file=sys.stderr, end="")
self.text = text
def _clientError(Message):
"""Raise Kill when asked."""
# Possible race condition
# Killing threads, writing error to file killed before finished
if Message.startswith("F"):
raise ExitQuiet from None
[docs]class Quiet(Exception):
"""Base Exception for no traceback."""
pass
[docs]class ExitQuiet(Quiet):
"""Exit Program Quietly Exception."""
pass
[docs]def notraceback(kind, message, traceback):
"""Exceptionhook only print errors if not Quiet."""
if Quiet not in kind.__bases__:
if not issubclass(kind, KeyError):
message = ecol(message)
sys.__excepthook__(kind, message, traceback)
else:
message = ecol(message)
sys.stderr.write(message.args[0] + "\n")
[docs]def ecol(m):
"""Colour crash errors."""
if len(m.args) > 0:
m.args = (_strings.FAIL + f"{m.args[0]}" + _strings.ENDC,)
return m
sys.excepthook = notraceback
if __name__ == '__main__':
pass