/
graceful_interrupt_handler.py
90 lines (68 loc) · 2.2 KB
/
graceful_interrupt_handler.py
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
# coding=utf-8
"""
Graceful Interrupt Handler as a context manager.
Can be nested.
From:
http://stackoverflow.com/a/10972804
https://gist.github.com/nonZero/2907502
"""
__docformat__ = 'restructuredtext en'
import signal
__all__ = ('GracefulInterruptHandler',)
class GracefulInterruptHandler(object):
"""
Example Usage::
with GracefulInterruptHandler() as h1:
while True:
print("(1)...")
time.sleep(1)
with GracefulInterruptHandler() as h2:
while True:
print("\t(2)...")
time.sleep(1)
if h2.interrupted:
print("\t(2) interrupted!")
time.sleep(2)
break
if h1.interrupted:
print("(1) interrupted!")
time.sleep(2)
break
"""
def __init__(self, sig=signal.SIGINT):
self.sig = sig
self.interrupted = False
self.released = False
self.original_handler = None
def __enter__(self):
return self.capture()
def capture(self):
"""
Capture the signal. Useful when not using the "with GracefulInterruptHandler" syntax.
:return: current GracefulInterruptHandler instance
:rtype: GracefulInterruptHandler
"""
self.interrupted = False
self.released = False
self.original_handler = signal.getsignal(self.sig)
# noinspection PyUnusedLocal
def handler(signum, frame):
"""
signal that an interrupt has occurred.
:param signum: the signal number
:param frame: unused
"""
self.release()
self.interrupted = True
signal.signal(self.sig, handler)
return self
# noinspection PyUnusedLocal,PyShadowingBuiltins
def __exit__(self, type, value, tb):
self.release()
def release(self):
"""release the signal handler"""
if self.released:
return False
signal.signal(self.sig, self.original_handler)
self.released = True
return True