-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathdev_watcher.py
More file actions
250 lines (203 loc) Β· 7.19 KB
/
dev_watcher.py
File metadata and controls
250 lines (203 loc) Β· 7.19 KB
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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
#!/usr/bin/env python3
"""
Development file watcher for Xbox Backup Manager
Automatically restarts the application when Python files change
"""
import os
import sys
import time
import subprocess
import signal
from pathlib import Path
# Try to import watchdog, fall back to polling if not available
try:
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
WATCHDOG_AVAILABLE = True
except ImportError:
WATCHDOG_AVAILABLE = False
print("β οΈ Watchdog not installed. Using polling method.")
print(" Install with: pip install watchdog")
class AppReloader:
"""Handles reloading the application when files change"""
def __init__(self, main_script="main.py", watch_patterns=None):
self.main_script = main_script
self.watch_patterns = watch_patterns or [".py"]
self.process = None
self.watch_dir = Path.cwd()
# Files to ignore
self.ignore_patterns = [
"__pycache__",
".pyc",
".git",
".vscode",
"build",
"dist",
".venv",
"venv",
"env",
".pytest_cache",
".coverage",
"*.log",
"dev_watcher.py", # Don't watch this file
"view_palette_colors.py",
"palette_viewer_gui.py",
]
def should_watch_file(self, file_path):
"""Check if file should trigger a reload"""
path_str = str(file_path)
# Ignore certain patterns
for pattern in self.ignore_patterns:
if pattern in path_str:
return False
# Only watch files with specified extensions
return any(path_str.endswith(ext) for ext in self.watch_patterns)
def start_app(self):
"""Start the main application"""
print(f"π Starting {self.main_script}...")
# Kill existing process if running
if self.process:
self.kill_app()
# Get Python executable
python_exe = sys.executable
# Start new process
try:
self.process = subprocess.Popen(
[python_exe, self.main_script],
cwd=self.watch_dir,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
universal_newlines=True,
bufsize=1,
)
print(f"β
Started with PID: {self.process.pid}")
return True
except Exception as e:
print(f"β Failed to start: {e}")
return False
def kill_app(self):
"""Kill the running application"""
if self.process:
try:
if sys.platform == "win32":
# On Windows, use taskkill to kill the process tree
subprocess.run(
["taskkill", "/F", "/T", "/PID", str(self.process.pid)],
check=False,
capture_output=True,
)
else:
# On Unix systems
self.process.terminate()
try:
self.process.wait(timeout=3)
except subprocess.TimeoutExpired:
self.process.kill()
self.process.wait()
print("π Application stopped")
except Exception as e:
print(f"β οΈ Error stopping app: {e}")
finally:
self.process = None
def restart_app(self, reason="File changed"):
"""Restart the application"""
print(f"π {reason}, restarting...")
self.start_app()
class FileChangeHandler(FileSystemEventHandler):
"""Handles file system events using watchdog"""
def __init__(self, reloader):
self.reloader = reloader
self.last_restart = 0
self.debounce_delay = 1.0 # Seconds to wait before restart
def on_modified(self, event):
if event.is_directory:
return
if not self.reloader.should_watch_file(event.src_path):
return
# Debounce rapid file changes
current_time = time.time()
if current_time - self.last_restart < self.debounce_delay:
return
self.last_restart = current_time
file_name = os.path.basename(event.src_path)
self.reloader.restart_app(f"File '{file_name}' changed")
def watch_with_watchdog(reloader):
"""Watch files using the watchdog library"""
event_handler = FileChangeHandler(reloader)
observer = Observer()
observer.schedule(event_handler, str(reloader.watch_dir), recursive=True)
print(f"π Watching {reloader.watch_dir} for changes...")
print(" Press Ctrl+C to stop")
observer.start()
try:
while True:
time.sleep(1)
# Check if app is still running
if reloader.process and reloader.process.poll() is not None:
print("β Application exited")
break
except KeyboardInterrupt:
print("\nπ Stopping watcher...")
finally:
observer.stop()
observer.join()
def watch_with_polling(reloader):
"""Watch files using simple polling (fallback method)"""
print(f"π Polling {reloader.watch_dir} for changes...")
print(" Press Ctrl+C to stop")
file_times = {}
def scan_files():
"""Scan for file changes"""
for file_path in reloader.watch_dir.rglob("*"):
if file_path.is_file() and reloader.should_watch_file(file_path):
try:
mtime = file_path.stat().st_mtime
if file_path in file_times:
if mtime > file_times[file_path]:
file_times[file_path] = mtime
return file_path
else:
file_times[file_path] = mtime
except OSError:
pass
return None
# Initial scan
scan_files()
try:
while True:
time.sleep(1)
# Check if app is still running
if reloader.process and reloader.process.poll() is not None:
print("β Application exited")
break
# Check for file changes
changed_file = scan_files()
if changed_file:
reloader.restart_app(f"File '{changed_file.name}' changed")
time.sleep(2) # Give time for restart
except KeyboardInterrupt:
print("\nπ Stopping watcher...")
def main():
"""Main entry point"""
reloader = AppReloader()
# Handle Ctrl+C gracefully
def signal_handler(sig, frame):
print("\nπ Shutting down...")
reloader.kill_app()
sys.exit(0)
signal.signal(signal.SIGINT, signal_handler)
# Start the application initially
if not reloader.start_app():
print("β Failed to start application")
return 1
# Start watching for changes
try:
if WATCHDOG_AVAILABLE:
watch_with_watchdog(reloader)
else:
watch_with_polling(reloader)
finally:
reloader.kill_app()
return 0
if __name__ == "__main__":
sys.exit(main())