| | 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: |
| | 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: |
| | 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__": |
| | main() |
| |
|
