1
|
|
|
""" |
2
|
|
|
Low-level tools (e.g. memory management). |
3
|
|
|
""" |
4
|
|
|
import atexit |
5
|
|
|
import signal |
6
|
|
|
import sys |
7
|
|
|
import traceback |
8
|
|
|
from collections import Callable |
|
|
|
|
9
|
|
|
from dataclasses import dataclass |
10
|
|
|
from typing import Any, Union, Sequence, Mapping |
11
|
|
|
|
12
|
|
|
from pocketutils.core.input_output import Writeable |
|
|
|
|
13
|
|
|
|
14
|
|
|
from pocketutils.tools.base_tools import BaseTools |
15
|
|
|
|
16
|
|
|
|
17
|
|
|
@dataclass(frozen=True, repr=True) |
|
|
|
|
18
|
|
|
class SignalHandler: |
19
|
|
|
name: str |
20
|
|
|
code: int |
21
|
|
|
desc: str |
22
|
|
|
sink: Union[Writeable, Callable[[str], Any]] |
23
|
|
|
|
24
|
|
|
def __call__(self): |
25
|
|
|
sys.stderr.write(f"~~{self.name}[{self.code}] ({self.desc})~~") |
26
|
|
|
traceback.print_stack(file=sys.stderr) |
27
|
|
|
for line in traceback.format_stack(): |
28
|
|
|
sys.stderr.write(line) |
29
|
|
|
|
30
|
|
|
|
31
|
|
|
@dataclass(frozen=True, repr=True) |
|
|
|
|
32
|
|
|
class ExitHandler: |
33
|
|
|
sink: Writeable |
34
|
|
|
|
35
|
|
|
def __call__(self): |
36
|
|
|
self.sink.write(f"~~EXIT~~") |
|
|
|
|
37
|
|
|
traceback.print_stack(file=sys.stderr) |
38
|
|
|
for line in traceback.format_stack(): |
39
|
|
|
self.sink.write(line) |
40
|
|
|
|
41
|
|
|
|
42
|
|
|
class SystemTools(BaseTools): |
|
|
|
|
43
|
|
|
@classmethod |
44
|
|
|
def traceback_to_dict(cls, e: BaseException) -> Sequence[Mapping[str, Union[str, int]]]: |
|
|
|
|
45
|
|
|
tb = [] |
|
|
|
|
46
|
|
|
current = None |
47
|
|
|
tbe = traceback.TracebackException.from_exception(e) |
48
|
|
|
last, count = None, 0 |
49
|
|
|
for i, s in enumerate(tbe.stack): |
|
|
|
|
50
|
|
|
current = dict(depth=i, filename=s.filename, line=s.line, name=s.name, repeats=None) |
51
|
|
|
if current == last: |
52
|
|
|
count += 1 |
53
|
|
|
else: |
54
|
|
|
current.repeats = count |
55
|
|
|
tb.append(current) |
56
|
|
|
count = 0 |
57
|
|
|
last = current |
58
|
|
|
if current is not None and current == last: |
59
|
|
|
tb.append(current) |
60
|
|
|
return tb |
61
|
|
|
|
62
|
|
|
@classmethod |
63
|
|
|
def trace_signals(cls, sink: Writeable = sys.stderr) -> None: |
64
|
|
|
""" |
65
|
|
|
Registers signal handlers for all signals that log the traceback. |
66
|
|
|
Uses ``signal.signal``. |
67
|
|
|
""" |
68
|
|
|
for sig in signal.valid_signals(): |
|
|
|
|
69
|
|
|
handler = SignalHandler(sig.name, sig.value, signal.strsignal(sig), sink) |
|
|
|
|
70
|
|
|
signal.signal(sig.value, handler) |
71
|
|
|
|
72
|
|
|
@classmethod |
73
|
|
|
def trace_exit(cls, sink: Writeable = sys.stderr) -> None: |
74
|
|
|
""" |
75
|
|
|
Registers an exit handler via ``atexit.register`` that logs the traceback. |
76
|
|
|
""" |
77
|
|
|
atexit.register(ExitHandler(sink)) |
78
|
|
|
|
79
|
|
|
|
80
|
|
|
__all__ = ["SignalHandler", "ExitHandler", "SystemTools"] |
81
|
|
|
|