File size: 2,490 Bytes
35c1128 | 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 | import argparse
import subprocess
import sys
import time
from pathlib import Path
from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer
class RestartOnChange(FileSystemEventHandler):
"""Restart a subprocess when watched files change."""
def __init__(self, command: list[str], watch_paths: list[str]) -> None:
self.command = command
self.watch_paths = [Path(p).resolve() for p in watch_paths]
self.process: subprocess.Popen | None = None
self.restart()
def restart(self) -> None:
if self.process and self.process.poll() is None:
self.process.terminate()
try:
self.process.wait(timeout=5)
except subprocess.TimeoutExpired:
self.process.kill()
self.process.wait()
self.process = subprocess.Popen(self.command)
def on_any_event(self, event) -> None: # pragma: no cover - runtime utility
if event.is_directory:
return
path = Path(event.src_path)
if path.suffix != ".py":
return
if any(str(path).startswith(str(p)) for p in self.watch_paths):
print(f"[watcher] {path} changed, running tests...")
subprocess.run([sys.executable, "-m", "pytest", "-q"])
print("[watcher] restarting process...")
self.restart()
def main() -> None: # pragma: no cover - CLI entry
parser = argparse.ArgumentParser(
description="Watch files and restart a command on changes",
)
parser.add_argument(
"--command",
nargs="+",
default=[sys.executable, "mcp_server.py"],
help="Command to run",
)
parser.add_argument(
"--paths",
nargs="+",
default=["bit_transformer", "mcp_server.py"],
help="Paths to watch for changes",
)
args = parser.parse_args()
observer = Observer()
handler = RestartOnChange(args.command, args.paths)
for p in args.paths:
observer.schedule(handler, p, recursive=True)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
pass
finally:
observer.stop()
handler.restart()
if handler.process and handler.process.poll() is None:
handler.process.terminate()
handler.process.wait()
observer.join()
if __name__ == "__main__": # pragma: no cover - CLI entry
main()
|