1#!/usr/bin/env python3
2"""
3
4"""
5
6# Imports:
7from __future__ import annotations
8
9# ##-- stdlib imports
10import datetime
11import enum
12import functools as ftz
13import itertools as itz
14import logging as logmod
15import os
16import pathlib as pl
17import pdb
18import re
19import signal
20import weakref
21from uuid import UUID, uuid1
22
23# ##-- end stdlib imports
24
25# ##-- types
26# isort: off
27# General
28import abc
29import collections.abc
30import typing
31import types
32from typing import cast, assert_type, assert_never
33from typing import Generic, NewType, Never
34from typing import no_type_check, final, override, overload
35# Protocols and Interfaces:
36from typing import Protocol, runtime_checkable
37# isort: on
38# ##-- end types
39
40# ##-- type checking
41# isort: off
42if typing.TYPE_CHECKING:
43 from typing import Final, ClassVar, Any, Self
44 from typing import Literal, LiteralString
45 from typing import TypeGuard
46 from collections.abc import Iterable, Iterator, Callable, Generator
47 from collections.abc import Sequence, Mapping, MutableMapping, Hashable
48
49 from jgdv import Maybe
50## isort: on
51# ##-- end type checking
52
53##-- logging
54logging = logmod.getLogger(__name__)
55##-- end logging
56
57env : dict = cast("dict", os.environ)
58PRE_COMMIT : Final[bool] = "PRE_COMMIT" in env
59BREAK_HEADER : Final[str] = "\n---- Task Interrupted ---- "
60
61##--| Body:
62
[docs]
63class SignalHandler:
64 """ Install a breakpoint to run on (by default) SIGINT
65
66 disables itself if PRE_COMMIT is in the environment.
67 Can act as a context manager
68
69 """
70
71 def __init__(self) -> None:
72 self._disabled = PRE_COMMIT
73
[docs]
74 @staticmethod
75 def handle(signum, frame) -> None:
76 breakpoint(header=BREAK_HEADER)
77 SignalHandler.install()
78
[docs]
79 @staticmethod
80 def install(sig=signal.SIGINT) -> None:
81 logging.debug("Installing Basic Interrupt Handler for: %s", signal.strsignal(sig))
82 signal.signal(sig, SignalHandler.handle)
83
[docs]
84 @staticmethod
85 def uninstall(sig=signal.SIGINT) -> None:
86 logging.debug("Uninstalling Basic Interrupt Handler for: %s", signal.strsignal(sig))
87 signal.signal(sig, signal.SIG_DFL)
88
89 def __enter__(self) -> Self:
90 if not self._disabled:
91 SignalHandler.install()
92 return self
93
94 def __exit__(self, etype:Maybe[type], err:Maybe[Exception], tb:Maybe[Traceback]) -> bool:
95 if not self._disabled:
96 SignalHandler.uninstall()
97 return False
98
[docs]
99class NullHandler:
100 """ An interrupt handler that does nothing """
101
[docs]
102 @staticmethod
103 def handle(signum, frame) -> None:
104 return
105
[docs]
106 @staticmethod
107 def install(sig=signal.SIGINT) -> None:
108 logging.debug("Installing Null Interrupt handler for: %s", signal.strsignal(sig))
109 # Install handler for Interrupt signal
110 signal.signal(sig, NullHandler.handle)
111
[docs]
112 @staticmethod
113 def uninstall(sig=signal.SIGINT) -> None:
114 logging.debug("Uninstalling Null Interrupt handler for: %s", signal.strsignal(sig))
115 signal.signal(sig, signal.SIG_DFL)
116
117 def __enter__(self) -> None:
118 if not self._disabled:
119 NullHandler.install()
120 return self
121
122 def __exit__(self, etype:Maybe[type], err:Maybe[Exception], tb:Maybe[Traceback]) -> bool:
123 if not self._disabled:
124 NullHandler.uninstall()
125 return False