1from __future__ import annotations 2 3import logging 4import threading 5from typing import TYPE_CHECKING 6 7from watchdog.utils import BaseThread 8 9if TYPE_CHECKING: 10 from typing import Callable 11 12 from watchdog.events import FileSystemEvent 13 14logger = logging.getLogger(__name__) 15 16 17class EventDebouncer(BaseThread): 18 """Background thread for debouncing event handling. 19 20 When an event is received, wait until the configured debounce interval 21 passes before calling the callback. If additional events are received 22 before the interval passes, reset the timer and keep waiting. When the 23 debouncing interval passes, the callback will be called with a list of 24 events in the order in which they were received. 25 """ 26 27 def __init__( 28 self, 29 debounce_interval_seconds: int, 30 events_callback: Callable[[list[FileSystemEvent]], None], 31 ) -> None: 32 super().__init__() 33 self.debounce_interval_seconds = debounce_interval_seconds 34 self.events_callback = events_callback 35 36 self._events: list[FileSystemEvent] = [] 37 self._cond = threading.Condition() 38 39 def handle_event(self, event: FileSystemEvent) -> None: 40 with self._cond: 41 self._events.append(event) 42 self._cond.notify() 43 44 def stop(self) -> None: 45 with self._cond: 46 super().stop() 47 self._cond.notify() 48 49 def run(self) -> None: 50 with self._cond: 51 while True: 52 # Wait for first event (or shutdown). 53 self._cond.wait() 54 55 if self.debounce_interval_seconds: 56 # Wait for additional events (or shutdown) until the debounce interval passes. 57 while self.should_keep_running(): 58 if not self._cond.wait(timeout=self.debounce_interval_seconds): 59 break 60 61 if not self.should_keep_running(): 62 break 63 64 events = self._events 65 self._events = [] 66 self.events_callback(events) 67