diff --git a/homeassistant/components/folder_watcher/__init__.py b/homeassistant/components/folder_watcher/__init__.py index cd979d51457c49..41a20360ff303b 100644 --- a/homeassistant/components/folder_watcher/__init__.py +++ b/homeassistant/components/folder_watcher/__init__.py @@ -1,13 +1,25 @@ """Component for monitoring activity on a folder.""" +from __future__ import annotations + import logging import os +from typing import cast import voluptuous as vol -from watchdog.events import PatternMatchingEventHandler +from watchdog.events import ( + FileClosedEvent, + FileCreatedEvent, + FileDeletedEvent, + FileModifiedEvent, + FileMovedEvent, + FileSystemEvent, + FileSystemMovedEvent, + PatternMatchingEventHandler, +) from watchdog.observers import Observer from homeassistant.const import EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP -from homeassistant.core import HomeAssistant +from homeassistant.core import Event, HomeAssistant import homeassistant.helpers.config_validation as cv from homeassistant.helpers.typing import ConfigType @@ -42,8 +54,8 @@ def setup(hass: HomeAssistant, config: ConfigType) -> bool: """Set up the folder watcher.""" conf = config[DOMAIN] for watcher in conf: - path = watcher[CONF_FOLDER] - patterns = watcher[CONF_PATTERNS] + path: str = watcher[CONF_FOLDER] + patterns: list[str] = watcher[CONF_PATTERNS] if not hass.config.is_allowed_path(path): _LOGGER.error("Folder %s is not valid or allowed", path) return False @@ -52,70 +64,72 @@ def setup(hass: HomeAssistant, config: ConfigType) -> bool: return True -def create_event_handler(patterns, hass): +def create_event_handler(patterns: list[str], hass: HomeAssistant) -> EventHandler: """Return the Watchdog EventHandler object.""" - class EventHandler(PatternMatchingEventHandler): - """Class for handling Watcher events.""" - - def __init__(self, patterns, hass): - """Initialise the EventHandler.""" - super().__init__(patterns) - self.hass = hass - - def process(self, event, moved=False): - """On Watcher event, fire HA event.""" - _LOGGER.debug("process(%s)", event) - if not event.is_directory: - folder, file_name = os.path.split(event.src_path) - fireable = { - "event_type": event.event_type, - "path": event.src_path, - "file": file_name, - "folder": folder, - } - - if moved: - dest_folder, dest_file_name = os.path.split(event.dest_path) - fireable.update( - { - "dest_path": event.dest_path, - "dest_file": dest_file_name, - "dest_folder": dest_folder, - } - ) - self.hass.bus.fire( - DOMAIN, - fireable, - ) + return EventHandler(patterns, hass) - def on_modified(self, event): - """File modified.""" - self.process(event) - def on_moved(self, event): - """File moved.""" - self.process(event, moved=True) +class EventHandler(PatternMatchingEventHandler): + """Class for handling Watcher events.""" + + def __init__(self, patterns: list[str], hass: HomeAssistant) -> None: + """Initialise the EventHandler.""" + super().__init__(patterns) + self.hass = hass + + def process(self, event: FileSystemEvent, moved: bool = False) -> None: + """On Watcher event, fire HA event.""" + _LOGGER.debug("process(%s)", event) + if not event.is_directory: + folder, file_name = os.path.split(event.src_path) + fireable = { + "event_type": event.event_type, + "path": event.src_path, + "file": file_name, + "folder": folder, + } + + if moved: + event = cast(FileSystemMovedEvent, event) + dest_folder, dest_file_name = os.path.split(event.dest_path) + fireable.update( + { + "dest_path": event.dest_path, + "dest_file": dest_file_name, + "dest_folder": dest_folder, + } + ) + self.hass.bus.fire( + DOMAIN, + fireable, + ) - def on_created(self, event): - """File created.""" - self.process(event) + def on_modified(self, event: FileModifiedEvent) -> None: + """File modified.""" + self.process(event) - def on_deleted(self, event): - """File deleted.""" - self.process(event) + def on_moved(self, event: FileMovedEvent) -> None: + """File moved.""" + self.process(event, moved=True) - def on_closed(self, event): - """File closed.""" - self.process(event) + def on_created(self, event: FileCreatedEvent) -> None: + """File created.""" + self.process(event) - return EventHandler(patterns, hass) + def on_deleted(self, event: FileDeletedEvent) -> None: + """File deleted.""" + self.process(event) + + def on_closed(self, event: FileClosedEvent) -> None: + """File closed.""" + self.process(event) class Watcher: """Class for starting Watchdog.""" - def __init__(self, path, patterns, hass): + def __init__(self, path: str, patterns: list[str], hass: HomeAssistant) -> None: """Initialise the watchdog observer.""" self._observer = Observer() self._observer.schedule( @@ -124,11 +138,11 @@ def __init__(self, path, patterns, hass): hass.bus.listen_once(EVENT_HOMEASSISTANT_START, self.startup) hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, self.shutdown) - def startup(self, event): + def startup(self, event: Event) -> None: """Start the watcher.""" self._observer.start() - def shutdown(self, event): + def shutdown(self, event: Event) -> None: """Shutdown the watcher.""" self._observer.stop() self._observer.join()