jmri_scripting.watcher
ScriptWatcher — poll-based hot-reload for .py files.
Watches a directory and restarts any script whose source file mtime
changes. Polls every interval seconds (default 1s) — coarse but
zero-dependency. For production setups that need sub-second response,
swap in watchdog or inotify later.
1"""``ScriptWatcher`` — poll-based hot-reload for .py files. 2 3Watches a directory and restarts any script whose source file mtime 4changes. Polls every ``interval`` seconds (default 1s) — coarse but 5zero-dependency. For production setups that need sub-second response, 6swap in ``watchdog`` or ``inotify`` later. 7""" 8from __future__ import annotations 9 10import asyncio 11import logging 12import pathlib 13from typing import TYPE_CHECKING 14 15if TYPE_CHECKING: 16 from jmri_scripting.runner import ScriptRunner 17 18_log = logging.getLogger(__name__) 19 20 21class ScriptWatcher: 22 """Re-runs scripts in a directory when their .py files change.""" 23 24 def __init__( 25 self, 26 runner: ScriptRunner, 27 directory: str | pathlib.Path, 28 *, 29 interval: float = 1.0, 30 ) -> None: 31 self._runner = runner 32 self._dir = pathlib.Path(directory) 33 self._interval = interval 34 self._mtimes: dict[pathlib.Path, float] = {} 35 self._task: asyncio.Task[None] | None = None 36 self._stop = asyncio.Event() 37 38 async def start(self) -> None: 39 if self._task is not None: 40 return 41 # Seed the initial mtimes. 42 for p in self._dir.glob("*.py"): 43 self._mtimes[p] = p.stat().st_mtime 44 self._task = asyncio.create_task(self._poll(), name="script-watcher") 45 46 async def stop(self) -> None: 47 self._stop.set() 48 if self._task is not None: 49 import contextlib # noqa: PLC0415 50 self._task.cancel() 51 with contextlib.suppress(asyncio.CancelledError, Exception): 52 await self._task 53 self._task = None 54 55 async def _poll(self) -> None: 56 import contextlib # noqa: PLC0415 57 try: 58 while not self._stop.is_set(): 59 self._tick() 60 with contextlib.suppress(TimeoutError): 61 await asyncio.wait_for(self._stop.wait(), timeout=self._interval) 62 except asyncio.CancelledError: 63 return 64 65 def _tick(self) -> None: 66 if not self._dir.is_dir(): 67 return 68 current: dict[pathlib.Path, float] = {} 69 for p in self._dir.glob("*.py"): 70 try: 71 current[p] = p.stat().st_mtime 72 except FileNotFoundError: 73 continue 74 75 # New + changed files: restart. 76 for p, mtime in current.items(): 77 prev = self._mtimes.get(p) 78 if prev is None or mtime > prev + 1e-6: 79 _log.info("hot-reload: %s", p.name) 80 try: 81 self._runner.run_file(p, name=p.stem) 82 except Exception as e: # noqa: BLE001 83 _log.warning("failed to re-run %s: %s", p, e) 84 85 # Deleted files: stop the running script. 86 for p in list(self._mtimes): 87 if p not in current: 88 _log.info("hot-reload: %s removed, stopping", p.name) 89 self._runner.stop(p.stem) 90 91 self._mtimes = current
class
ScriptWatcher:
22class ScriptWatcher: 23 """Re-runs scripts in a directory when their .py files change.""" 24 25 def __init__( 26 self, 27 runner: ScriptRunner, 28 directory: str | pathlib.Path, 29 *, 30 interval: float = 1.0, 31 ) -> None: 32 self._runner = runner 33 self._dir = pathlib.Path(directory) 34 self._interval = interval 35 self._mtimes: dict[pathlib.Path, float] = {} 36 self._task: asyncio.Task[None] | None = None 37 self._stop = asyncio.Event() 38 39 async def start(self) -> None: 40 if self._task is not None: 41 return 42 # Seed the initial mtimes. 43 for p in self._dir.glob("*.py"): 44 self._mtimes[p] = p.stat().st_mtime 45 self._task = asyncio.create_task(self._poll(), name="script-watcher") 46 47 async def stop(self) -> None: 48 self._stop.set() 49 if self._task is not None: 50 import contextlib # noqa: PLC0415 51 self._task.cancel() 52 with contextlib.suppress(asyncio.CancelledError, Exception): 53 await self._task 54 self._task = None 55 56 async def _poll(self) -> None: 57 import contextlib # noqa: PLC0415 58 try: 59 while not self._stop.is_set(): 60 self._tick() 61 with contextlib.suppress(TimeoutError): 62 await asyncio.wait_for(self._stop.wait(), timeout=self._interval) 63 except asyncio.CancelledError: 64 return 65 66 def _tick(self) -> None: 67 if not self._dir.is_dir(): 68 return 69 current: dict[pathlib.Path, float] = {} 70 for p in self._dir.glob("*.py"): 71 try: 72 current[p] = p.stat().st_mtime 73 except FileNotFoundError: 74 continue 75 76 # New + changed files: restart. 77 for p, mtime in current.items(): 78 prev = self._mtimes.get(p) 79 if prev is None or mtime > prev + 1e-6: 80 _log.info("hot-reload: %s", p.name) 81 try: 82 self._runner.run_file(p, name=p.stem) 83 except Exception as e: # noqa: BLE001 84 _log.warning("failed to re-run %s: %s", p, e) 85 86 # Deleted files: stop the running script. 87 for p in list(self._mtimes): 88 if p not in current: 89 _log.info("hot-reload: %s removed, stopping", p.name) 90 self._runner.stop(p.stem) 91 92 self._mtimes = current
Re-runs scripts in a directory when their .py files change.
ScriptWatcher( runner: jmri_scripting.ScriptRunner, directory: str | pathlib._local.Path, *, interval: float = 1.0)
25 def __init__( 26 self, 27 runner: ScriptRunner, 28 directory: str | pathlib.Path, 29 *, 30 interval: float = 1.0, 31 ) -> None: 32 self._runner = runner 33 self._dir = pathlib.Path(directory) 34 self._interval = interval 35 self._mtimes: dict[pathlib.Path, float] = {} 36 self._task: asyncio.Task[None] | None = None 37 self._stop = asyncio.Event()