MCPcopy Create free account
hub / github.com/Ishabdullah/Codey-v2 / FileWatchManager

Class FileWatchManager

core/background.py:244–369  ·  view source on GitHub ↗

Manages file system watches using watchdog. Features: - Watch files/directories for changes - Callback on file events - Debouncing to avoid excessive triggers

Source from the content-addressed store, hash-verified

242
243
244class FileWatchManager:
245 """
246 Manages file system watches using watchdog.
247
248 Features:
249 - Watch files/directories for changes
250 - Callback on file events
251 - Debouncing to avoid excessive triggers
252 """
253 def __init__(self):
254 self._watches: Dict[str, dict] = {}
255 self._observer = None
256 self._running = False
257
258 def start(self):
259 """Start the file watch observer."""
260 try:
261 from watchdog.observers import Observer
262 self._observer = Observer()
263 self._observer.start()
264 self._running = True
265 info("FileWatch: observer started")
266 except ImportError:
267 warning("FileWatch: watchdog not installed, file watches disabled")
268 except Exception as e:
269 error(f"FileWatch: failed to start observer: {e}")
270
271 def stop(self):
272 """Stop the file watch observer."""
273 if self._observer and self._running:
274 self._observer.stop()
275 self._observer.join()
276 self._running = False
277 info("FileWatch: observer stopped")
278
279 def add_watch(self, path: str, callback: Callable,
280 patterns: List[str] = None,
281 recursive: bool = True) -> str:
282 """
283 Add a file watch.
284
285 Args:
286 path: Path to watch
287 callback: Function to call on events
288 patterns: File patterns to watch (e.g., ['*.py'])
289 recursive: Watch subdirectories
290
291 Returns:
292 Watch ID
293 """
294 watch_id = f"watch_{path}_{int(time.time())}"
295
296 self._watches[watch_id] = {
297 "path": path,
298 "callback": callback,
299 "patterns": patterns,
300 "recursive": recursive,
301 }

Callers 1

get_file_watch_managerFunction · 0.85

Calls

no outgoing calls

Tested by

no test coverage detected