jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-25 06:07:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Doorbird Refactor (#23892)

Browse Source 
* Remove schedule management. Allow custom HTTP events defined in the configuration

* Consolidate doorbird request handling.  Make token a per device configuration item.

* Lint fixes

* Do not register dummy listener

* Remove punctuation
This commit is contained in:
Matt Snyder 2019-05-18 14:46:00 -05:00 committed by Martin Hjelmare
parent a8286535eb
commit 10a1b156e3
1 changed files with 66 additions and 201 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

267
homeassistant/components/doorbird/__init__.py
View File

@ -6,8 +6,8 @@ import voluptuous as vol
from homeassistant.components.http import HomeAssistantView
from homeassistant.const import (
CONF_DEVICES, CONF_HOST, CONF_MONITORED_CONDITIONS, CONF_NAME,
CONF_PASSWORD, CONF_TOKEN, CONF_USERNAME)
CONF_DEVICES, CONF_HOST, CONF_NAME, CONF_PASSWORD, CONF_TOKEN,
CONF_USERNAME)
import homeassistant.helpers.config_validation as cv
from homeassistant.util import dt as dt_util, slugify
@ -18,25 +18,7 @@ DOMAIN = 'doorbird'
API_URL = '/api/{}'.format(DOMAIN)
CONF_CUSTOM_URL = 'hass_url_override'
CONF_DOORBELL_EVENTS = 'doorbell_events'
CONF_DOORBELL_NUMS = 'doorbell_numbers'
CONF_RELAY_NUMS = 'relay_numbers'
CONF_MOTION_EVENTS = 'motion_events'
SENSOR_TYPES = {
'doorbell': {
'name': 'Button',
'device_class': 'occupancy',
},
'motion': {
'name': 'Motion',
'device_class': 'motion',
},
'relay': {
'name': 'Relay',
'device_class': 'relay',
}
}
CONF_EVENTS = 'events'
RESET_DEVICE_FAVORITES = 'doorbird_reset_favorites'
@ -44,19 +26,15 @@ DEVICE_SCHEMA = vol.Schema({
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Optional(CONF_DOORBELL_NUMS, default=[1]): vol.All(
cv.ensure_list, [cv.positive_int]),
vol.Optional(CONF_RELAY_NUMS, default=[1]): vol.All(
cv.ensure_list, [cv.positive_int]),
vol.Required(CONF_TOKEN): cv.string,
vol.Optional(CONF_EVENTS, default=[]): vol.All(
cv.ensure_list, [cv.string]),
vol.Optional(CONF_CUSTOM_URL): cv.string,
vol.Optional(CONF_NAME): cv.string,
vol.Optional(CONF_MONITORED_CONDITIONS, default=[]):
vol.All(cv.ensure_list, [vol.In(SENSOR_TYPES)]),
vol.Optional(CONF_NAME): cv.string
})
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Required(CONF_TOKEN): cv.string,
vol.Required(CONF_DEVICES): vol.All(cv.ensure_list, [DEVICE_SCHEMA])
}),
}, extra=vol.ALLOW_EXTRA)
@ -66,13 +44,8 @@ def setup(hass, config):
"""Set up the DoorBird component."""
from doorbirdpy import DoorBird
token = config[DOMAIN].get(CONF_TOKEN)
# Provide an endpoint for the doorstations to call to trigger events
hass.http.register_view(DoorBirdRequestView(token))
# Provide an endpoint for the user to call to clear device changes
hass.http.register_view(DoorBirdCleanupView(token))
hass.http.register_view(DoorBirdRequestView)
doorstations = []
@ -80,10 +53,9 @@ def setup(hass, config):
device_ip = doorstation_config.get(CONF_HOST)
username = doorstation_config.get(CONF_USERNAME)
password = doorstation_config.get(CONF_PASSWORD)
doorbell_nums = doorstation_config.get(CONF_DOORBELL_NUMS)
relay_nums = doorstation_config.get(CONF_RELAY_NUMS)
custom_url = doorstation_config.get(CONF_CUSTOM_URL)
events = doorstation_config.get(CONF_MONITORED_CONDITIONS)
events = doorstation_config.get(CONF_EVENTS)
token = doorstation_config.get(CONF_TOKEN)
name = (doorstation_config.get(CONF_NAME)
or 'DoorBird {}'.format(index + 1))
@ -92,7 +64,7 @@ def setup(hass, config):
if status[0]:
doorstation = ConfiguredDoorBird(device, name, events, custom_url,
doorbell_nums, relay_nums, token)
token)
doorstations.append(doorstation)
_LOGGER.info('Connected to DoorBird "%s" as %s@%s',
doorstation.name, username, device_ip)
@ -108,7 +80,7 @@ def setup(hass, config):
# Subscribe to doorbell or motion events
if events:
try:
doorstation.update_schedule(hass)
doorstation.register_events(hass)
except HTTPError:
hass.components.persistent_notification.create(
'Doorbird configuration failed. Please verify that API '
@ -124,15 +96,15 @@ def setup(hass, config):
def _reset_device_favorites_handler(event):
"""Handle clearing favorites on device."""
slug = event.data.get('slug')
token = event.data.get('token')
if slug is None:
if token is None:
return
doorstation = get_doorstation_by_slug(hass, slug)
doorstation = get_doorstation_by_token(hass, token)
if doorstation is None:
_LOGGER.error('Device not found %s', format(slug))
_LOGGER.error('Device not found for provided token.')
# Clear webhooks
favorites = doorstation.device.favorites()
@ -146,30 +118,22 @@ def setup(hass, config):
return True
def get_doorstation_by_slug(hass, slug):
def get_doorstation_by_token(hass, token):
"""Get doorstation by slug."""
for doorstation in hass.data[DOMAIN]:
if slugify(doorstation.name) in slug:
if token == doorstation.token:
return doorstation
def handle_event(event):
"""Handle dummy events."""
return None
class ConfiguredDoorBird():
"""Attach additional information to pass along with configured device."""
def __init__(self, device, name, events, custom_url, doorbell_nums,
relay_nums, token):
def __init__(self, device, name, events, custom_url, token):
"""Initialize configured device."""
self._name = name
self._device = device
self._custom_url = custom_url
self._monitored_events = events
self._doorbell_nums = doorbell_nums
self._relay_nums = relay_nums
self._events = events
self._token = token
@property
@ -187,14 +151,13 @@ class ConfiguredDoorBird():
"""Get custom url for device."""
return self._custom_url
def update_schedule(self, hass):
"""Register monitored sensors and deregister others."""
from doorbirdpy import DoorBirdScheduleEntrySchedule
# Create a new schedule (24/7)
schedule = DoorBirdScheduleEntrySchedule()
schedule.add_weekday(0, 604800) # seconds in a week
@property
def token(self):
"""Get token for device."""
return self._token
def register_events(self, hass):
"""Register events on device."""
# Get the URL of this server
hass_url = hass.config.api.base_url
@ -202,98 +165,39 @@ class ConfiguredDoorBird():
if self.custom_url is not None:
hass_url = self.custom_url
# For all sensor types (enabled + disabled)
for sensor_type in SENSOR_TYPES:
name = '{} {}'.format(self.name, SENSOR_TYPES[sensor_type]['name'])
slug = slugify(name)
for event in self._events:
event = self._get_event_name(event)
url = '{}{}/{}?token={}'.format(hass_url, API_URL, slug,
self._token)
if sensor_type in self._monitored_events:
# Enabled -> register
self._register_event(url, sensor_type, schedule)
_LOGGER.info('Registered for %s pushes from DoorBird "%s". '
'Use the "%s_%s" event for automations.',
sensor_type, self.name, DOMAIN, slug)
self._register_event(hass_url, event)
# Register a dummy listener so event is listed in GUI
hass.bus.listen('{}_{}'.format(DOMAIN, slug), handle_event)
else:
# Disabled -> deregister
self._deregister_event(url, sensor_type)
_LOGGER.info('Deregistered %s pushes from DoorBird "%s". '
'If any old favorites or schedules remain, '
'follow the instructions in the component '
'documentation to clear device registrations.',
sensor_type, self.name)
_LOGGER.info('Successfully registered URL for %s on %s',
event, self.name)
def _register_event(self, hass_url, event, schedule):
@property
def slug(self):
"""Get device slug."""
return slugify(self._name)
def _get_event_name(self, event):
return '{}_{}'.format(self.slug, event)
def _register_event(self, hass_url, event):
"""Add a schedule entry in the device for a sensor."""
from doorbirdpy import DoorBirdScheduleEntryOutput
url = '{}{}/{}?token={}'.format(hass_url, API_URL, event, self._token)
# Register HA URL as webhook if not already, then get the ID
if not self.webhook_is_registered(hass_url):
self.device.change_favorite('http', 'Home Assistant ({} events)'
.format(event), hass_url)
if not self.webhook_is_registered(url):
self.device.change_favorite('http', 'Home Assistant ({})'
.format(event), url)
fav_id = self.get_webhook_id(hass_url)
fav_id = self.get_webhook_id(url)
if not fav_id:
_LOGGER.warning('Could not find favorite for URL "%s". '
'Skipping sensor "%s".', hass_url, event)
'Skipping sensor "%s"', url, event)
return
# Add event handling to device schedule
output = DoorBirdScheduleEntryOutput(event='http',
param=fav_id,
schedule=schedule)
if event == 'doorbell':
# Repeat edit for each monitored doorbell number
for doorbell in self._doorbell_nums:
entry = self.device.get_schedule_entry(event, str(doorbell))
entry.output.append(output)
self.device.change_schedule(entry)
elif event == 'relay':
# Repeat edit for each monitored doorbell number
for relay in self._relay_nums:
entry = self.device.get_schedule_entry(event, str(relay))
entry.output.append(output)
else:
entry = self.device.get_schedule_entry(event)
entry.output.append(output)
self.device.change_schedule(entry)
def _deregister_event(self, hass_url, event):
"""Remove the schedule entry in the device for a sensor."""
# Find the right favorite and delete it
fav_id = self.get_webhook_id(hass_url)
if not fav_id:
return
self._device.delete_favorite('http', fav_id)
if event == 'doorbell':
# Delete the matching schedule for each doorbell number
for doorbell in self._doorbell_nums:
self._delete_schedule_action(event, fav_id, str(doorbell))
else:
self._delete_schedule_action(event, fav_id)
def _delete_schedule_action(self, sensor, fav_id, param=""):
"""Remove the HA output from a schedule."""
entries = self._device.schedule()
for entry in entries:
if entry.input != sensor or entry.param != param:
continue
for action in entry.output:
if action.event == 'http' and action.param == fav_id:
entry.output.remove(action)
self._device.change_schedule(entry)
def webhook_is_registered(self, ha_url, favs=None) -> bool:
def webhook_is_registered(self, url, favs=None) -> bool:
"""Return whether the given URL is registered as a device favorite."""
favs = favs if favs else self.device.favorites()
@ -301,12 +205,12 @@ class ConfiguredDoorBird():
return False
for fav in favs['http'].values():
if fav['value'] == ha_url:
if fav['value'] == url:
return True
return False
def get_webhook_id(self, ha_url, favs=None) -> str or None:
def get_webhook_id(self, url, favs=None) -> str or None:
"""
Return the device favorite ID for the given URL.
@ -318,7 +222,7 @@ class ConfiguredDoorBird():
return None
for fav_id in favs['http']:
if favs['http'][fav_id]['value'] == ha_url:
if favs['http'][fav_id]['value'] == url:
return fav_id
return None
@ -340,72 +244,33 @@ class DoorBirdRequestView(HomeAssistantView):
requires_auth = False
url = API_URL
name = API_URL[1:].replace('/', ':')
extra_urls = [API_URL + '/{sensor}']
def __init__(self, token):
"""Initialize view."""
HomeAssistantView.__init__(self)
self._token = token
extra_urls = [API_URL + '/{event}']
# pylint: disable=no-self-use
async def get(self, request, sensor):
async def get(self, request, event):
"""Respond to requests from the device."""
from aiohttp import web
hass = request.app['hass']
request_token = request.query.get('token')
token = request.query.get('token')
authenticated = request_token == self._token
device = get_doorstation_by_token(hass, token)
if request_token == '' or not authenticated:
return web.Response(status=401, text='Unauthorized')
if device is None:
return web.Response(status=401, text='Invalid token provided.')
doorstation = get_doorstation_by_slug(hass, sensor)
if doorstation:
event_data = doorstation.get_event_data()
if device:
event_data = device.get_event_data()
else:
event_data = {}
hass.bus.async_fire('{}_{}'.format(DOMAIN, sensor), event_data)
if event == 'clear':
hass.bus.async_fire(RESET_DEVICE_FAVORITES,
{'token': token})
message = 'HTTP Favorites cleared for {}'.format(device.slug)
return web.Response(status=200, text=message)
hass.bus.async_fire('{}_{}'.format(DOMAIN, event), event_data)
return web.Response(status=200, text='OK')
class DoorBirdCleanupView(HomeAssistantView):
"""Provide a URL to call to delete ALL webhooks/schedules."""
requires_auth = False
url = API_URL + '/clear/{slug}'
name = 'DoorBird Cleanup'
def __init__(self, token):
"""Initialize view."""
HomeAssistantView.__init__(self)
self._token = token
# pylint: disable=no-self-use
async def get(self, request, slug):
"""Act on requests."""
from aiohttp import web
hass = request.app['hass']
request_token = request.query.get('token')
authenticated = request_token == self._token
if request_token == '' or not authenticated:
return web.Response(status=401, text='Unauthorized')
device = get_doorstation_by_slug(hass, slug)
# No matching device
if device is None:
return web.Response(status=404,
text='Device slug {} not found'.format(slug))
hass.bus.async_fire(RESET_DEVICE_FAVORITES,
{'slug': slug})
message = 'Clearing schedule for {}'.format(slug)
return web.Response(status=200, text=message)