Add Geofency device tracker (#9106)
* Added Geofency device tracker Added Geofency device tracker * fix pylint error * review fixes * merge coroutinespull/9258/head
parent
b72b7cfe0e
commit
c240d907d2
|
@ -0,0 +1,127 @@
|
|||
"""
|
||||
Support for the Geofency platform.
|
||||
|
||||
For more details about this platform, please refer to the documentation at
|
||||
https://home-assistant.io/components/device_tracker.geofency/
|
||||
"""
|
||||
import asyncio
|
||||
from functools import partial
|
||||
import logging
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.device_tracker import PLATFORM_SCHEMA
|
||||
from homeassistant.components.http import HomeAssistantView
|
||||
from homeassistant.const import (
|
||||
ATTR_LATITUDE, ATTR_LONGITUDE, HTTP_UNPROCESSABLE_ENTITY, STATE_NOT_HOME)
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.util import slugify
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
DEPENDENCIES = ['http']
|
||||
|
||||
BEACON_DEV_PREFIX = 'beacon'
|
||||
CONF_MOBILE_BEACONS = 'mobile_beacons'
|
||||
|
||||
LOCATION_ENTRY = '1'
|
||||
LOCATION_EXIT = '0'
|
||||
|
||||
URL = '/api/geofency'
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
|
||||
vol.Optional(CONF_MOBILE_BEACONS): vol.All(
|
||||
cv.ensure_list, [cv.string]),
|
||||
})
|
||||
|
||||
|
||||
def setup_scanner(hass, config, see, discovery_info=None):
|
||||
"""Set up an endpoint for the Geofency application."""
|
||||
mobile_beacons = config.get(CONF_MOBILE_BEACONS) or []
|
||||
|
||||
hass.http.register_view(GeofencyView(see, mobile_beacons))
|
||||
|
||||
return True
|
||||
|
||||
|
||||
class GeofencyView(HomeAssistantView):
|
||||
"""View to handle Geofency requests."""
|
||||
|
||||
url = URL
|
||||
name = 'api:geofency'
|
||||
|
||||
def __init__(self, see, mobile_beacons):
|
||||
"""Initialize Geofency url endpoints."""
|
||||
self.see = see
|
||||
self.mobile_beacons = [slugify(beacon) for beacon in mobile_beacons]
|
||||
|
||||
@asyncio.coroutine
|
||||
def post(self, request):
|
||||
"""Handle Geofency requests."""
|
||||
data = yield from request.post()
|
||||
hass = request.app['hass']
|
||||
|
||||
data = self._validate_data(data)
|
||||
if not data:
|
||||
return ("Invalid data", HTTP_UNPROCESSABLE_ENTITY)
|
||||
|
||||
if self._is_mobile_beacon(data):
|
||||
return (yield from self._set_location(hass, data, None))
|
||||
else:
|
||||
if data['entry'] == LOCATION_ENTRY:
|
||||
location_name = data['name']
|
||||
else:
|
||||
location_name = STATE_NOT_HOME
|
||||
|
||||
return (yield from self._set_location(hass, data, location_name))
|
||||
|
||||
@staticmethod
|
||||
def _validate_data(data):
|
||||
"""Validate POST payload."""
|
||||
data = data.copy()
|
||||
|
||||
required_attributes = ['address', 'device', 'entry',
|
||||
'latitude', 'longitude', 'name']
|
||||
|
||||
valid = True
|
||||
for attribute in required_attributes:
|
||||
if attribute not in data:
|
||||
valid = False
|
||||
_LOGGER.error("'%s' not specified in message", attribute)
|
||||
|
||||
if not valid:
|
||||
return False
|
||||
|
||||
data['address'] = data['address'].replace('\n', ' ')
|
||||
data['device'] = slugify(data['device'])
|
||||
data['name'] = slugify(data['name'])
|
||||
|
||||
data[ATTR_LATITUDE] = float(data[ATTR_LATITUDE])
|
||||
data[ATTR_LONGITUDE] = float(data[ATTR_LONGITUDE])
|
||||
|
||||
return data
|
||||
|
||||
def _is_mobile_beacon(self, data):
|
||||
"""Check if we have a mobile beacon."""
|
||||
return 'beaconUUID' in data and data['name'] in self.mobile_beacons
|
||||
|
||||
@staticmethod
|
||||
def _device_name(data):
|
||||
"""Return name of device tracker."""
|
||||
if 'beaconUUID' in data:
|
||||
return "{}_{}".format(BEACON_DEV_PREFIX, data['name'])
|
||||
else:
|
||||
return data['device']
|
||||
|
||||
@asyncio.coroutine
|
||||
def _set_location(self, hass, data, location_name):
|
||||
"""Fire HA event to set location."""
|
||||
device = self._device_name(data)
|
||||
|
||||
yield from hass.async_add_job(
|
||||
partial(self.see, dev_id=device,
|
||||
gps=(data[ATTR_LATITUDE], data[ATTR_LONGITUDE]),
|
||||
location_name=location_name,
|
||||
attributes=data))
|
||||
|
||||
return "Setting location for {}".format(device)
|
|
@ -0,0 +1,230 @@
|
|||
"""The tests for the Geofency device tracker platform."""
|
||||
# pylint: disable=redefined-outer-name
|
||||
import asyncio
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components import zone
|
||||
import homeassistant.components.device_tracker as device_tracker
|
||||
from homeassistant.components.device_tracker.geofency import (
|
||||
CONF_MOBILE_BEACONS, URL)
|
||||
from homeassistant.const import (
|
||||
CONF_PLATFORM, HTTP_OK, HTTP_UNPROCESSABLE_ENTITY, STATE_HOME,
|
||||
STATE_NOT_HOME)
|
||||
from homeassistant.setup import async_setup_component
|
||||
from homeassistant.util import slugify
|
||||
|
||||
HOME_LATITUDE = 37.239622
|
||||
HOME_LONGITUDE = -115.815811
|
||||
|
||||
NOT_HOME_LATITUDE = 37.239394
|
||||
NOT_HOME_LONGITUDE = -115.763283
|
||||
|
||||
GPS_ENTER_HOME = {
|
||||
'latitude': HOME_LATITUDE,
|
||||
'longitude': HOME_LONGITUDE,
|
||||
'device': '4A7FE356-2E9D-4264-A43F-BF80ECAEE416',
|
||||
'name': 'Home',
|
||||
'radius': 100,
|
||||
'id': 'BAAD384B-A4AE-4983-F5F5-4C2F28E68205',
|
||||
'date': '2017-08-19T10:53:53Z',
|
||||
'address': 'Testing Trail 1',
|
||||
'entry': '1'
|
||||
}
|
||||
|
||||
GPS_EXIT_HOME = {
|
||||
'latitude': HOME_LATITUDE,
|
||||
'longitude': HOME_LONGITUDE,
|
||||
'device': '4A7FE356-2E9D-4264-A43F-BF80ECAEE416',
|
||||
'name': 'Home',
|
||||
'radius': 100,
|
||||
'id': 'BAAD384B-A4AE-4983-F5F5-4C2F28E68205',
|
||||
'date': '2017-08-19T10:53:53Z',
|
||||
'address': 'Testing Trail 1',
|
||||
'entry': '0'
|
||||
}
|
||||
|
||||
BEACON_ENTER_HOME = {
|
||||
'latitude': HOME_LATITUDE,
|
||||
'longitude': HOME_LONGITUDE,
|
||||
'beaconUUID': 'FFEF0E83-09B2-47C8-9837-E7B563F5F556',
|
||||
'minor': '36138',
|
||||
'major': '8629',
|
||||
'device': '4A7FE356-2E9D-4264-A43F-BF80ECAEE416',
|
||||
'name': 'Home',
|
||||
'radius': 100,
|
||||
'id': 'BAAD384B-A4AE-4983-F5F5-4C2F28E68205',
|
||||
'date': '2017-08-19T10:53:53Z',
|
||||
'address': 'Testing Trail 1',
|
||||
'entry': '1'
|
||||
}
|
||||
|
||||
BEACON_EXIT_HOME = {
|
||||
'latitude': HOME_LATITUDE,
|
||||
'longitude': HOME_LONGITUDE,
|
||||
'beaconUUID': 'FFEF0E83-09B2-47C8-9837-E7B563F5F556',
|
||||
'minor': '36138',
|
||||
'major': '8629',
|
||||
'device': '4A7FE356-2E9D-4264-A43F-BF80ECAEE416',
|
||||
'name': 'Home',
|
||||
'radius': 100,
|
||||
'id': 'BAAD384B-A4AE-4983-F5F5-4C2F28E68205',
|
||||
'date': '2017-08-19T10:53:53Z',
|
||||
'address': 'Testing Trail 1',
|
||||
'entry': '0'
|
||||
}
|
||||
|
||||
BEACON_ENTER_CAR = {
|
||||
'latitude': NOT_HOME_LATITUDE,
|
||||
'longitude': NOT_HOME_LONGITUDE,
|
||||
'beaconUUID': 'FFEF0E83-09B2-47C8-9837-E7B563F5F556',
|
||||
'minor': '36138',
|
||||
'major': '8629',
|
||||
'device': '4A7FE356-2E9D-4264-A43F-BF80ECAEE416',
|
||||
'name': 'Car 1',
|
||||
'radius': 100,
|
||||
'id': 'BAAD384B-A4AE-4983-F5F5-4C2F28E68205',
|
||||
'date': '2017-08-19T10:53:53Z',
|
||||
'address': 'Testing Trail 1',
|
||||
'entry': '1'
|
||||
}
|
||||
|
||||
BEACON_EXIT_CAR = {
|
||||
'latitude': NOT_HOME_LATITUDE,
|
||||
'longitude': NOT_HOME_LONGITUDE,
|
||||
'beaconUUID': 'FFEF0E83-09B2-47C8-9837-E7B563F5F556',
|
||||
'minor': '36138',
|
||||
'major': '8629',
|
||||
'device': '4A7FE356-2E9D-4264-A43F-BF80ECAEE416',
|
||||
'name': 'Car 1',
|
||||
'radius': 100,
|
||||
'id': 'BAAD384B-A4AE-4983-F5F5-4C2F28E68205',
|
||||
'date': '2017-08-19T10:53:53Z',
|
||||
'address': 'Testing Trail 1',
|
||||
'entry': '0'
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def geofency_client(loop, hass, test_client):
|
||||
"""Geofency mock client."""
|
||||
assert loop.run_until_complete(async_setup_component(
|
||||
hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'geofency',
|
||||
CONF_MOBILE_BEACONS: ['Car 1']
|
||||
}}))
|
||||
|
||||
with patch('homeassistant.components.device_tracker.update_config'):
|
||||
yield loop.run_until_complete(test_client(hass.http.app))
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup_zones(loop, hass):
|
||||
"""Setup Zone config in HA."""
|
||||
assert loop.run_until_complete(async_setup_component(
|
||||
hass, zone.DOMAIN, {
|
||||
'zone': {
|
||||
'name': 'Home',
|
||||
'latitude': HOME_LATITUDE,
|
||||
'longitude': HOME_LONGITUDE,
|
||||
'radius': 100,
|
||||
}}))
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_data_validation(geofency_client):
|
||||
"""Test data validation."""
|
||||
# No data
|
||||
req = yield from geofency_client.post(URL)
|
||||
assert req.status == HTTP_UNPROCESSABLE_ENTITY
|
||||
|
||||
missing_attributes = ['address', 'device',
|
||||
'entry', 'latitude', 'longitude', 'name']
|
||||
|
||||
# missing attributes
|
||||
for attribute in missing_attributes:
|
||||
copy = GPS_ENTER_HOME.copy()
|
||||
del copy[attribute]
|
||||
req = yield from geofency_client.post(URL, data=copy)
|
||||
assert req.status == HTTP_UNPROCESSABLE_ENTITY
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_gps_enter_and_exit_home(hass, geofency_client):
|
||||
"""Test GPS based zone enter and exit."""
|
||||
# Enter the Home zone
|
||||
req = yield from geofency_client.post(URL, data=GPS_ENTER_HOME)
|
||||
assert req.status == HTTP_OK
|
||||
device_name = slugify(GPS_ENTER_HOME['device'])
|
||||
state_name = hass.states.get('{}.{}'.format(
|
||||
'device_tracker', device_name)).state
|
||||
assert STATE_HOME == state_name
|
||||
|
||||
# Exit the Home zone
|
||||
req = yield from geofency_client.post(URL, data=GPS_EXIT_HOME)
|
||||
assert req.status == HTTP_OK
|
||||
device_name = slugify(GPS_EXIT_HOME['device'])
|
||||
state_name = hass.states.get('{}.{}'.format(
|
||||
'device_tracker', device_name)).state
|
||||
assert STATE_NOT_HOME == state_name
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_beacon_enter_and_exit_home(hass, geofency_client):
|
||||
"""Test iBeacon based zone enter and exit - a.k.a stationary iBeacon."""
|
||||
# Enter the Home zone
|
||||
req = yield from geofency_client.post(URL, data=BEACON_ENTER_HOME)
|
||||
assert req.status == HTTP_OK
|
||||
device_name = slugify("beacon_{}".format(BEACON_ENTER_HOME['name']))
|
||||
state_name = hass.states.get('{}.{}'.format(
|
||||
'device_tracker', device_name)).state
|
||||
assert STATE_HOME == state_name
|
||||
|
||||
# Exit the Home zone
|
||||
req = yield from geofency_client.post(URL, data=BEACON_EXIT_HOME)
|
||||
assert req.status == HTTP_OK
|
||||
device_name = slugify("beacon_{}".format(BEACON_ENTER_HOME['name']))
|
||||
state_name = hass.states.get('{}.{}'.format(
|
||||
'device_tracker', device_name)).state
|
||||
assert STATE_NOT_HOME == state_name
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_beacon_enter_and_exit_car(hass, geofency_client):
|
||||
"""Test use of mobile iBeacon."""
|
||||
# Enter the Car away from Home zone
|
||||
req = yield from geofency_client.post(URL, data=BEACON_ENTER_CAR)
|
||||
assert req.status == HTTP_OK
|
||||
device_name = slugify("beacon_{}".format(BEACON_ENTER_CAR['name']))
|
||||
state_name = hass.states.get('{}.{}'.format(
|
||||
'device_tracker', device_name)).state
|
||||
assert STATE_NOT_HOME == state_name
|
||||
|
||||
# Exit the Car away from Home zone
|
||||
req = yield from geofency_client.post(URL, data=BEACON_EXIT_CAR)
|
||||
assert req.status == HTTP_OK
|
||||
device_name = slugify("beacon_{}".format(BEACON_ENTER_CAR['name']))
|
||||
state_name = hass.states.get('{}.{}'.format(
|
||||
'device_tracker', device_name)).state
|
||||
assert STATE_NOT_HOME == state_name
|
||||
|
||||
# Enter the Car in the Home zone
|
||||
data = BEACON_ENTER_CAR.copy()
|
||||
data['latitude'] = HOME_LATITUDE
|
||||
data['longitude'] = HOME_LONGITUDE
|
||||
req = yield from geofency_client.post(URL, data=data)
|
||||
assert req.status == HTTP_OK
|
||||
device_name = slugify("beacon_{}".format(data['name']))
|
||||
state_name = hass.states.get('{}.{}'.format(
|
||||
'device_tracker', device_name)).state
|
||||
assert STATE_HOME == state_name
|
||||
|
||||
# Exit the Car in the Home zone
|
||||
req = yield from geofency_client.post(URL, data=data)
|
||||
assert req.status == HTTP_OK
|
||||
device_name = slugify("beacon_{}".format(data['name']))
|
||||
state_name = hass.states.get('{}.{}'.format(
|
||||
'device_tracker', device_name)).state
|
||||
assert STATE_HOME == state_name
|
Loading…
Reference in New Issue