f892c3394b
* Add Canary component * Made some change to how canary data is updated and stored * Updated to use py-canary:0.1.2 * Addressed flake8 warnings * Import canary API locally * Import canary API locally again * Addressed pylint errors * Updated requirements_all.txt * Fixed incorrect unit of measurement for air quality sensor * Added tests for Canary component and sensors * Updated canary component to handle exception better when initializing * Fixed tests * Fixed tests again * Addressed review comments * Fixed houndci error * Addressed comment about camera force update * Addressed comment regarding timeout when fetching camera image * Updated to use py-canary==0.2.2 * Increased update frequency to 30 seconds * Added support for Canary alarm control panel * Address review comments * Fixed houndci error * Fixed lint errors * Updated test to only test setup component / platform * Fixed flake error * Fixed failing test * Uptake py-canary:0.2.3 * canary.alarm_control_panel DISARM is now mapped to canary PRIVACY mode * Fixed failing tests * Removed unnecessary methods * Removed polling in canary camera component and update camera info when getting camera image * Added more tests to cover Canary sensors * Address review comments * Addressed review comment in tests * Fixed pylint errors * Excluded canary alarm_control_panel and camera from coverage calculation
117 lines
3.7 KiB
Python
117 lines
3.7 KiB
Python
"""
|
|
Support for Canary.
|
|
|
|
For more details about this component, please refer to the documentation at
|
|
https://home-assistant.io/components/canary/
|
|
"""
|
|
import logging
|
|
from datetime import timedelta
|
|
|
|
import voluptuous as vol
|
|
from requests import ConnectTimeout, HTTPError
|
|
|
|
import homeassistant.helpers.config_validation as cv
|
|
from homeassistant.const import CONF_USERNAME, CONF_PASSWORD, CONF_TIMEOUT
|
|
from homeassistant.helpers import discovery
|
|
from homeassistant.util import Throttle
|
|
|
|
REQUIREMENTS = ['py-canary==0.2.3']
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
NOTIFICATION_ID = 'canary_notification'
|
|
NOTIFICATION_TITLE = 'Canary Setup'
|
|
|
|
DOMAIN = 'canary'
|
|
DATA_CANARY = 'canary'
|
|
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=30)
|
|
DEFAULT_TIMEOUT = 10
|
|
|
|
CONFIG_SCHEMA = vol.Schema({
|
|
DOMAIN: vol.Schema({
|
|
vol.Required(CONF_USERNAME): cv.string,
|
|
vol.Required(CONF_PASSWORD): cv.string,
|
|
vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int,
|
|
}),
|
|
}, extra=vol.ALLOW_EXTRA)
|
|
|
|
CANARY_COMPONENTS = [
|
|
'alarm_control_panel', 'camera', 'sensor'
|
|
]
|
|
|
|
|
|
def setup(hass, config):
|
|
"""Set up the Canary component."""
|
|
conf = config[DOMAIN]
|
|
username = conf.get(CONF_USERNAME)
|
|
password = conf.get(CONF_PASSWORD)
|
|
timeout = conf.get(CONF_TIMEOUT)
|
|
|
|
try:
|
|
hass.data[DATA_CANARY] = CanaryData(username, password, timeout)
|
|
except (ConnectTimeout, HTTPError) as ex:
|
|
_LOGGER.error("Unable to connect to Canary service: %s", str(ex))
|
|
hass.components.persistent_notification.create(
|
|
'Error: {}<br />'
|
|
'You will need to restart hass after fixing.'
|
|
''.format(ex),
|
|
title=NOTIFICATION_TITLE,
|
|
notification_id=NOTIFICATION_ID)
|
|
return False
|
|
|
|
for component in CANARY_COMPONENTS:
|
|
discovery.load_platform(hass, component, DOMAIN, {}, config)
|
|
|
|
return True
|
|
|
|
|
|
class CanaryData(object):
|
|
"""Get the latest data and update the states."""
|
|
|
|
def __init__(self, username, password, timeout):
|
|
"""Init the Canary data object."""
|
|
from canary.api import Api
|
|
self._api = Api(username, password, timeout)
|
|
|
|
self._locations_by_id = {}
|
|
self._readings_by_device_id = {}
|
|
self._entries_by_location_id = {}
|
|
|
|
self.update()
|
|
|
|
@Throttle(MIN_TIME_BETWEEN_UPDATES)
|
|
def update(self, **kwargs):
|
|
"""Get the latest data from py-canary."""
|
|
for location in self._api.get_locations():
|
|
location_id = location.location_id
|
|
|
|
self._locations_by_id[location_id] = location
|
|
self._entries_by_location_id[location_id] = self._api.get_entries(
|
|
location_id, entry_type="motion", limit=1)
|
|
|
|
for device in location.devices:
|
|
if device.is_online:
|
|
self._readings_by_device_id[device.device_id] = \
|
|
self._api.get_latest_readings(device.device_id)
|
|
|
|
@property
|
|
def locations(self):
|
|
"""Return a list of locations."""
|
|
return self._locations_by_id.values()
|
|
|
|
def get_motion_entries(self, location_id):
|
|
"""Return a list of motion entries based on location_id."""
|
|
return self._entries_by_location_id.get(location_id, [])
|
|
|
|
def get_location(self, location_id):
|
|
"""Return a location based on location_id."""
|
|
return self._locations_by_id.get(location_id, [])
|
|
|
|
def get_readings(self, device_id):
|
|
"""Return a list of readings based on device_id."""
|
|
return self._readings_by_device_id.get(device_id, [])
|
|
|
|
def set_location_mode(self, location_id, mode_name, is_private=False):
|
|
"""Set location mode."""
|
|
self._api.set_location_mode(location_id, mode_name, is_private)
|
|
self.update(no_throttle=True)
|