-
Notifications
You must be signed in to change notification settings - Fork 1
/
reload.py
executable file
·148 lines (109 loc) · 3.55 KB
/
reload.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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
#!/usr/bin/env python
import sys, time, re, os, signal, fcntl
from subprocess import Popen, PIPE, STDOUT
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class Reloader:
def __init__(self, command, delay=0, sig=signal.SIGTERM):
self.command = command
self.delay = delay
self.sig = sig
self.started = False
def start_command(self):
self.process = Popen(self.command, preexec_fn=os.setsid, bufsize=0, stdout=PIPE, stderr=STDOUT)
fcntl.fcntl(self.process.stdout.fileno(), fcntl.F_SETFL, os.O_NONBLOCK)
self.started = True
def stop_command(self):
if self.started:
try:
os.killpg(self.process.pid, self.sig)
except OSError:
pass
self.started = False
def restart_command(self):
print("Restarting command")
self.stop_command()
self.start_command()
def read(self):
out = ''
more = self.process != None
while more:
try:
r = self.process.stdout.read()
if not r:
more = False
else:
out += r.decode("UTF-8")
except IOError:
more = False
return out
class ReloadEventHandler(FileSystemEventHandler):
def __init__(self, ignore_patterns=[]):
self._modified = False
self.ignore_patterns = [re.compile(r) for r in ignore_patterns]
def dispatch(self, event):
if event.is_directory:
return
path = os.path.basename(event.src_path)
if any(r.match(path) for r in self.ignore_patterns):
return
super(ReloadEventHandler, self).dispatch(event)
def on_modified(self, event):
print("Detected change in " + event.src_path)
self._modified = True
@property
def modified(self):
if self._modified:
self._modified = False
return True
else:
return False
def load_ignore_patterns(name):
patterns = []
if os.path.exists(name):
with open(name, "r") as f:
pass
for line in f:
p = line.strip()
if p:
patterns.append(p)
return patterns
def reload(*command, ignore_patterns=[]):
"""Reload given command"""
path = "."
sig = signal.SIGTERM
delay = 0.25
ignorefile = ".reloadignore"
ignore_patterns = ignore_patterns or load_ignore_patterns(ignorefile)
event_handler = ReloadEventHandler(ignore_patterns)
reloader = Reloader(command, signal)
observer = Observer()
observer.schedule(event_handler, path, recursive=True)
observer.start()
reloader.start_command()
try:
while True:
time.sleep(delay)
sys.stdout.write(reloader.read())
sys.stdout.flush()
if event_handler.modified:
reloader.restart_command()
except KeyboardInterrupt:
observer.stop()
observer.join()
reloader.stop_command()
sys.stdout.write(reloader.read())
sys.stdout.flush()
def reload_me(*args, ignore_patterns=[]):
"""Reload currently running command with given args"""
command = [sys.executable, sys.argv[0]]
command.extend(args)
reload(*command, ignore_patterns=ignore_patterns)
def main():
if len(sys.argv) < 2:
print("Usage: reload <command>")
exit(1)
command = sys.argv[1:]
reload(*command)
if __name__ == "__main__":
main()