hass-core/homeassistant/components/folder_watcher.py
Robin df78eecc1b Adds folder_watcher component (#12918)
* Create watchdog_file_watcher.py

* Rename watchdog_file_watcher.py to folder_watcher.py

* Address a number of issues

* Adds filter

* Adds pattern matching

* Adds create_event_handler()

* Update folder_watcher.py

* Adds run_setup()

* Remove stop_watching()

* Adds shutdown()

* Update config to allow patterns on each folder

* Update to patterns from filters

* Adds watchdog

* Fix indents on schema

* Update folder_watcher.py

* Create test_file_watcher.py

* Fix lints

* Add test_invalid_path()

* Adds folder_watcher

* Update test_file_watcher.py

* Update folder_watcher.py

* Simplify config

* Adapt for new config

* Run observer.schedule() on EVENT_HOMEASSISTANT_START

* Amend Watcher removing entity and tidying startup

* Tidy config

* Rename process to on_any_event for consistency

* Rename on_any_event back to process

Using `on_any_event` resulted in 2 events being fired

* Update folder_watcher.py

* Fix return False on setup

* Update test_file_watcher.py

* Update folder_watcher.py

* Adds watchdog

* Undo adding watchdog

* Update test_file_watcher.py

* Update test_file_watcher.py

* Update test_file_watcher.py

* Update test_file_watcher.py

* Update test_file_watcher.py

* Add event

* Update test_file_watcher.py

* Update .coveragerc

* Update test_file_watcher.py

* Update test_file_watcher.py

* debug + join

* test event

* lint

* lint

* Rename test_file_watcher.py to test_folder_watcher.py

* hound

* Tidy test

* Further refine test

* Adds to test_all

* Fix test for py35

* Change test again

* Update test_folder_watcher.py

* Fix test

* Add watchdog to test

* Update folder_watcher.py

* add watchdog

* Update folder_watcher.py
2018-03-29 18:10:20 -07:00

111 lines
3.3 KiB
Python

"""
Component for monitoring activity on a folder.
For more details about this platform, refer to the documentation at
https://home-assistant.io/components/folder_watcher/
"""
import os
import logging
import voluptuous as vol
from homeassistant.const import (
EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP)
import homeassistant.helpers.config_validation as cv
REQUIREMENTS = ['watchdog==0.8.3']
_LOGGER = logging.getLogger(__name__)
CONF_FOLDER = 'folder'
CONF_PATTERNS = 'patterns'
CONF_WATCHERS = 'watchers'
DEFAULT_PATTERN = '*'
DOMAIN = "folder_watcher"
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.All(cv.ensure_list, [vol.Schema({
vol.Required(CONF_FOLDER): cv.isdir,
vol.Optional(CONF_PATTERNS, default=[DEFAULT_PATTERN]):
vol.All(cv.ensure_list, [cv.string]),
})])
}, extra=vol.ALLOW_EXTRA)
def setup(hass, config):
"""Set up the folder watcher."""
conf = config[DOMAIN]
for watcher in conf:
path = watcher[CONF_FOLDER]
patterns = watcher[CONF_PATTERNS]
if not hass.config.is_allowed_path(path):
_LOGGER.error("folder %s is not valid or allowed", path)
return False
Watcher(path, patterns, hass)
return True
def create_event_handler(patterns, hass):
""""Return the Watchdog EventHandler object."""
from watchdog.events import PatternMatchingEventHandler
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):
"""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)
self.hass.bus.fire(
DOMAIN, {
"event_type": event.event_type,
'path': event.src_path,
'file': file_name,
'folder': folder,
})
def on_modified(self, event):
"""File modified."""
self.process(event)
def on_moved(self, event):
"""File moved."""
self.process(event)
def on_created(self, event):
"""File created."""
self.process(event)
def on_deleted(self, event):
"""File deleted."""
self.process(event)
return EventHandler(patterns, hass)
class Watcher():
"""Class for starting Watchdog."""
def __init__(self, path, patterns, hass):
"""Initialise the watchdog observer."""
from watchdog.observers import Observer
self._observer = Observer()
self._observer.schedule(
create_event_handler(patterns, hass),
path,
recursive=True)
hass.bus.listen_once(EVENT_HOMEASSISTANT_START, self.startup)
hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, self.shutdown)
def startup(self, event):
"""Start the watcher."""
self._observer.start()
def shutdown(self, event):
"""Shutdown the watcher."""
self._observer.stop()
self._observer.join()