github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Webhook for Traccar (#24762) 

* add initial traccar webhook support

* remove unused import

* add tests but disabled atm

* remove translations

* add timestamp parameter

* use post for tests

* rename config_flow

* format using black

* format tests using black

* Use str instead of float

* fix most comments

* check id

* add two device test

* reformat

* fix failuers

* Update tests/components/traccar/test_init.py

Co-Authored-By: Martin Hjelmare <marhje52@kth.se>

* Update tests/components/traccar/test_init.py

Co-Authored-By: Martin Hjelmare <marhje52@kth.se>

* Update tests/components/traccar/test_init.py

Co-Authored-By: Martin Hjelmare <marhje52@kth.se>

* Update tests/components/traccar/test_init.py

Co-Authored-By: Martin Hjelmare <marhje52@kth.se>

* black
This commit is contained in:
escoand 2019-08-10 00:14:03 +02:00 committed by Martin Hjelmare
parent 60dfa38717
commit dc5c1783dc
8 changed files with 556 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

111
homeassistant/components/traccar/__init__.py
View File

@ -1 +1,110 @@
"""The traccar component."""
"""Support for Traccar."""
import logging
import voluptuous as vol
from aiohttp import web
import homeassistant.helpers.config_validation as cv
from homeassistant.const import HTTP_UNPROCESSABLE_ENTITY, HTTP_OK, CONF_WEBHOOK_ID
from homeassistant.helpers import config_entry_flow
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.components.device_tracker import DOMAIN as DEVICE_TRACKER
from .const import (
ATTR_ACCURACY,
ATTR_ALTITUDE,
ATTR_BATTERY,
ATTR_BEARING,
ATTR_ID,
ATTR_LATITUDE,
ATTR_LONGITUDE,
ATTR_SPEED,
ATTR_TIMESTAMP,
DOMAIN,
)
_LOGGER = logging.getLogger(__name__)
TRACKER_UPDATE = "{}_tracker_update".format(DOMAIN)
DEFAULT_ACCURACY = 200
DEFAULT_BATTERY = -1
def _id(value: str) -> str:
"""Coerce id by removing '-'."""
return value.replace("-", "")
WEBHOOK_SCHEMA = vol.Schema(
{
vol.Required(ATTR_ID): vol.All(cv.string, _id),
vol.Required(ATTR_LATITUDE): cv.latitude,
vol.Required(ATTR_LONGITUDE): cv.longitude,
vol.Optional(ATTR_ACCURACY, default=DEFAULT_ACCURACY): vol.Coerce(float),
vol.Optional(ATTR_ALTITUDE): vol.Coerce(float),
vol.Optional(ATTR_BATTERY, default=DEFAULT_BATTERY): vol.Coerce(float),
vol.Optional(ATTR_BEARING): vol.Coerce(float),
vol.Optional(ATTR_SPEED): vol.Coerce(float),
vol.Optional(ATTR_TIMESTAMP): vol.Coerce(int),
}
)
async def async_setup(hass, hass_config):
"""Set up the Traccar component."""
hass.data[DOMAIN] = {"devices": set(), "unsub_device_tracker": {}}
return True
async def handle_webhook(hass, webhook_id, request):
"""Handle incoming webhook with Traccar request."""
try:
data = WEBHOOK_SCHEMA(dict(request.query))
except vol.MultipleInvalid as error:
return web.Response(text=error.error_message, status=HTTP_UNPROCESSABLE_ENTITY)
attrs = {
ATTR_ALTITUDE: data.get(ATTR_ALTITUDE),
ATTR_BEARING: data.get(ATTR_BEARING),
ATTR_SPEED: data.get(ATTR_SPEED),
}
device = data[ATTR_ID]
async_dispatcher_send(
hass,
TRACKER_UPDATE,
device,
data[ATTR_LATITUDE],
data[ATTR_LONGITUDE],
data[ATTR_BATTERY],
data[ATTR_ACCURACY],
attrs,
)
return web.Response(text="Setting location for {}".format(device), status=HTTP_OK)
async def async_setup_entry(hass, entry):
"""Configure based on config entry."""
hass.components.webhook.async_register(
DOMAIN, "Traccar", entry.data[CONF_WEBHOOK_ID], handle_webhook
)
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, DEVICE_TRACKER)
)
return True
async def async_unload_entry(hass, entry):
"""Unload a config entry."""
hass.components.webhook.async_unregister(entry.data[CONF_WEBHOOK_ID])
hass.data[DOMAIN]["unsub_device_tracker"].pop(entry.entry_id)()
await hass.config_entries.async_forward_entry_unload(entry, DEVICE_TRACKER)
return True
# pylint: disable=invalid-name
async_remove_entry = config_entry_flow.webhook_async_remove_entry

10
homeassistant/components/traccar/config_flow.py Normal file
View File

@ -0,0 +1,10 @@
"""Config flow for Traccar."""
from homeassistant.helpers import config_entry_flow
from .const import DOMAIN
config_entry_flow.register_webhook_flow(
DOMAIN,
"Traccar Webhook",
{"docs_url": "https://www.home-assistant.io/components/traccar/"},
)

12
homeassistant/components/traccar/const.py
View File

@ -1,16 +1,26 @@
"""Constants for Traccar integration."""
DOMAIN = "traccar"
CONF_MAX_ACCURACY = "max_accuracy"
CONF_SKIP_ACCURACY_ON = "skip_accuracy_filter_on"
ATTR_ACCURACY = "accuracy"
ATTR_ADDRESS = "address"
ATTR_ALTITUDE = "altitude"
ATTR_BATTERY = "batt"
ATTR_BEARING = "bearing"
ATTR_CATEGORY = "category"
ATTR_GEOFENCE = "geofence"
ATTR_ID = "id"
ATTR_LATITUDE = "lat"
ATTR_LONGITUDE = "lon"
ATTR_MOTION = "motion"
ATTR_SPEED = "speed"
ATTR_STATUS = "status"
ATTR_TIMESTAMP = "timestamp"
ATTR_TRACKER = "tracker"
ATTR_TRACCAR_ID = "traccar_id"
ATTR_STATUS = "status"
EVENT_DEVICE_MOVING = "device_moving"
EVENT_COMMAND_RESULT = "command_result"

177
homeassistant/components/traccar/device_tracker.py
View File

@ -4,7 +4,7 @@ import logging
import voluptuous as vol
from homeassistant.components.device_tracker import PLATFORM_SCHEMA
from homeassistant.core import callback
from homeassistant.const import (
CONF_HOST,
CONF_PORT,
@ -16,19 +16,33 @@ from homeassistant.const import (
CONF_MONITORED_CONDITIONS,
CONF_EVENT,
)
from homeassistant.components.device_tracker import PLATFORM_SCHEMA, SOURCE_TYPE_GPS
from homeassistant.components.device_tracker.config_entry import TrackerEntity
from homeassistant.helpers import device_registry
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import HomeAssistantType
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.util import slugify
from . import DOMAIN, TRACKER_UPDATE
from .const import (
ATTR_ACCURACY,
ATTR_ADDRESS,
ATTR_ALTITUDE,
ATTR_BATTERY,
ATTR_BEARING,
ATTR_CATEGORY,
ATTR_GEOFENCE,
ATTR_LATITUDE,
ATTR_LONGITUDE,
ATTR_MOTION,
ATTR_SPEED,
ATTR_STATUS,
ATTR_TRACKER,
ATTR_TRACCAR_ID,
ATTR_STATUS,
EVENT_DEVICE_MOVING,
EVENT_COMMAND_RESULT,
EVENT_DEVICE_FUEL_DROP,
@ -101,6 +115,45 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
)
async def async_setup_entry(hass: HomeAssistantType, entry, async_add_entities):
"""Configure a dispatcher connection based on a config entry."""
@callback
def _receive_data(device, latitude, longitude, battery, accuracy, attrs):
"""Receive set location."""
if device in hass.data[DOMAIN]["devices"]:
return
hass.data[DOMAIN]["devices"].add(device)
async_add_entities(
[TraccarEntity(device, latitude, longitude, battery, accuracy, attrs)]
)
hass.data[DOMAIN]["unsub_device_tracker"][
entry.entry_id
] = async_dispatcher_connect(hass, TRACKER_UPDATE, _receive_data)
# Restore previously loaded devices
dev_reg = await device_registry.async_get_registry(hass)
dev_ids = {
identifier[1]
for device in dev_reg.devices.values()
for identifier in device.identifiers
if identifier[0] == DOMAIN
}
if not dev_ids:
return
entities = []
for dev_id in dev_ids:
hass.data[DOMAIN]["devices"].add(dev_id)
entity = TraccarEntity(dev_id, None, None, None, None, None)
entities.append(entity)
async_add_entities(entities)
async def async_setup_scanner(hass, config, async_see, discovery_info=None):
"""Validate the configuration and return a Traccar scanner."""
from pytraccar.api import API
@ -273,3 +326,123 @@ class TraccarScanner:
"attributes": event["attributes"],
},
)
class TraccarEntity(TrackerEntity, RestoreEntity):
"""Represent a tracked device."""
def __init__(self, device, latitude, longitude, battery, accuracy, attributes):
"""Set up Geofency entity."""
self._accuracy = accuracy
self._attributes = attributes
self._name = device
self._battery = battery
self._latitude = latitude
self._longitude = longitude
self._unsub_dispatcher = None
self._unique_id = device
@property
def battery_level(self):
"""Return battery value of the device."""
return self._battery
@property
def device_state_attributes(self):
"""Return device specific attributes."""
return self._attributes
@property
def latitude(self):
"""Return latitude value of the device."""
return self._latitude
@property
def longitude(self):
"""Return longitude value of the device."""
return self._longitude
@property
def location_accuracy(self):
"""Return the gps accuracy of the device."""
return self._accuracy
@property
def name(self):
"""Return the name of the device."""
return self._name
@property
def should_poll(self):
"""No polling needed."""
return False
@property
def unique_id(self):
"""Return the unique ID."""
return self._unique_id
@property
def device_info(self):
"""Return the device info."""
return {"name": self._name, "identifiers": {(DOMAIN, self._unique_id)}}
@property
def source_type(self):
"""Return the source type, eg gps or router, of the device."""
return SOURCE_TYPE_GPS
async def async_added_to_hass(self):
"""Register state update callback."""
await super().async_added_to_hass()
self._unsub_dispatcher = async_dispatcher_connect(
self.hass, TRACKER_UPDATE, self._async_receive_data
)
# don't restore if we got created with data
if self._latitude is not None or self._longitude is not None:
return
state = await self.async_get_last_state()
if state is None:
self._latitude = None
self._longitude = None
self._accuracy = None
self._attributes = {
ATTR_ALTITUDE: None,
ATTR_BEARING: None,
ATTR_SPEED: None,
}
self._battery = None
return
attr = state.attributes
self._latitude = attr.get(ATTR_LATITUDE)
self._longitude = attr.get(ATTR_LONGITUDE)
self._accuracy = attr.get(ATTR_ACCURACY)
self._attributes = {
ATTR_ALTITUDE: attr.get(ATTR_ALTITUDE),
ATTR_BEARING: attr.get(ATTR_BEARING),
ATTR_SPEED: attr.get(ATTR_SPEED),
}
self._battery = attr.get(ATTR_BATTERY)
async def async_will_remove_from_hass(self):
"""Clean up after entity before removal."""
await super().async_will_remove_from_hass()
self._unsub_dispatcher()
@callback
def _async_receive_data(
self, device, latitude, longitude, battery, accuracy, attributes
):
"""Mark the device as seen."""
if device != self.name:
return
self._latitude = latitude
self._longitude = longitude
self._battery = battery
self._accuracy = accuracy
self._attributes.update(attributes)
self.async_write_ha_state()

7
homeassistant/components/traccar/manifest.json
View File

@ -1,13 +1,16 @@
{
"domain": "traccar",
"name": "Traccar",
"config_flow": true,
"documentation": "https://www.home-assistant.io/components/traccar",
"requirements": [
"pytraccar==0.9.0",
"stringcase==1.2.0"
],
"dependencies": [],
"dependencies": [
"webhook"
],
"codeowners": [
"@ludeeus"
]
}
}

1
homeassistant/generated/config_flows.py
View File

@ -51,6 +51,7 @@ FLOWS = [
"tellduslive",
"toon",
"tplink",
"traccar",
"tradfri",
"twentemilieu",
"twilio",

1
tests/components/traccar/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the Traccar component."""

243
tests/components/traccar/test_init.py Normal file
View File

@ -0,0 +1,243 @@
"""The tests the for Traccar device tracker platform."""
from unittest.mock import Mock, patch
import pytest
from homeassistant import data_entry_flow
from homeassistant.components import traccar, zone
from homeassistant.components.device_tracker import DOMAIN as DEVICE_TRACKER_DOMAIN
from homeassistant.components.traccar import DOMAIN, TRACKER_UPDATE
from homeassistant.const import (
HTTP_OK,
HTTP_UNPROCESSABLE_ENTITY,
STATE_HOME,
STATE_NOT_HOME,
)
from homeassistant.helpers.dispatcher import DATA_DISPATCHER
from homeassistant.setup import async_setup_component
HOME_LATITUDE = 37.239622
HOME_LONGITUDE = -115.815811
@pytest.fixture(autouse=True)
def mock_dev_track(mock_device_tracker_conf):
"""Mock device tracker config loading."""
pass
@pytest.fixture(name="client")
async def traccar_client(loop, hass, aiohttp_client):
"""Mock client for Traccar (unauthenticated)."""
assert await async_setup_component(hass, "persistent_notification", {})
assert await async_setup_component(hass, DOMAIN, {DOMAIN: {}})
await hass.async_block_till_done()
with patch("homeassistant.components.device_tracker.legacy.update_config"):
return await aiohttp_client(hass.http.app)
@pytest.fixture(autouse=True)
async def setup_zones(loop, hass):
"""Set up Zone config in HA."""
assert await async_setup_component(
hass,
zone.DOMAIN,
{
"zone": {
"name": "Home",
"latitude": HOME_LATITUDE,
"longitude": HOME_LONGITUDE,
"radius": 100,
}
},
)
await hass.async_block_till_done()
@pytest.fixture(name="webhook_id")
async def webhook_id_fixture(hass, client):
"""Initialize the Traccar component and get the webhook_id."""
hass.config.api = Mock(base_url="http://example.com")
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM, result
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
await hass.async_block_till_done()
return result["result"].data["webhook_id"]
async def test_missing_data(hass, client, webhook_id):
"""Test missing data."""
url = "/api/webhook/{}".format(webhook_id)
data = {"lat": "1.0", "lon": "1.1", "id": "123"}
# No data
req = await client.post(url)
await hass.async_block_till_done()
assert req.status == HTTP_UNPROCESSABLE_ENTITY
# No latitude
copy = data.copy()
del copy["lat"]
req = await client.post(url, params=copy)
await hass.async_block_till_done()
assert req.status == HTTP_UNPROCESSABLE_ENTITY
# No device
copy = data.copy()
del copy["id"]
req = await client.post(url, params=copy)
await hass.async_block_till_done()
assert req.status == HTTP_UNPROCESSABLE_ENTITY
async def test_enter_and_exit(hass, client, webhook_id):
"""Test when there is a known zone."""
url = "/api/webhook/{}".format(webhook_id)
data = {"lat": str(HOME_LATITUDE), "lon": str(HOME_LONGITUDE), "id": "123"}
# Enter the Home
req = await client.post(url, params=data)
await hass.async_block_till_done()
assert req.status == HTTP_OK
state_name = hass.states.get(
"{}.{}".format(DEVICE_TRACKER_DOMAIN, data["id"])
).state
assert STATE_HOME == state_name
# Enter Home again
req = await client.post(url, params=data)
await hass.async_block_till_done()
assert req.status == HTTP_OK
state_name = hass.states.get(
"{}.{}".format(DEVICE_TRACKER_DOMAIN, data["id"])
).state
assert STATE_HOME == state_name
data["lon"] = 0
data["lat"] = 0
# Enter Somewhere else
req = await client.post(url, params=data)
await hass.async_block_till_done()
assert req.status == HTTP_OK
state_name = hass.states.get(
"{}.{}".format(DEVICE_TRACKER_DOMAIN, data["id"])
).state
assert STATE_NOT_HOME == state_name
dev_reg = await hass.helpers.device_registry.async_get_registry()
assert len(dev_reg.devices) == 1
ent_reg = await hass.helpers.entity_registry.async_get_registry()
assert len(ent_reg.entities) == 1
async def test_enter_with_attrs(hass, client, webhook_id):
"""Test when additional attributes are present."""
url = "/api/webhook/{}".format(webhook_id)
data = {
"timestamp": 123456789,
"lat": "1.0",
"lon": "1.1",
"id": "123",
"accuracy": "10.5",
"batt": 10,
"speed": 100,
"bearing": "105.32",
"altitude": 102,
}
req = await client.post(url, params=data)
await hass.async_block_till_done()
assert req.status == HTTP_OK
state = hass.states.get("{}.{}".format(DEVICE_TRACKER_DOMAIN, data["id"]))
assert state.state == STATE_NOT_HOME
assert state.attributes["gps_accuracy"] == 10.5
assert state.attributes["battery_level"] == 10.0
assert state.attributes["speed"] == 100.0
assert state.attributes["bearing"] == 105.32
assert state.attributes["altitude"] == 102.0
data = {
"lat": str(HOME_LATITUDE),
"lon": str(HOME_LONGITUDE),
"id": "123",
"accuracy": 123,
"batt": 23,
"speed": 23,
"bearing": 123,
"altitude": 123,
}
req = await client.post(url, params=data)
await hass.async_block_till_done()
assert req.status == HTTP_OK
state = hass.states.get("{}.{}".format(DEVICE_TRACKER_DOMAIN, data["id"]))
assert state.state == STATE_HOME
assert state.attributes["gps_accuracy"] == 123
assert state.attributes["battery_level"] == 23
assert state.attributes["speed"] == 23
assert state.attributes["bearing"] == 123
assert state.attributes["altitude"] == 123
async def test_two_devices(hass, client, webhook_id):
"""Test updating two different devices."""
url = "/api/webhook/{}".format(webhook_id)
data_device_1 = {"lat": "1.0", "lon": "1.1", "id": "device_1"}
# Exit Home
req = await client.post(url, params=data_device_1)
await hass.async_block_till_done()
assert req.status == HTTP_OK
state = hass.states.get("{}.{}".format(DEVICE_TRACKER_DOMAIN, data_device_1["id"]))
assert state.state == "not_home"
# Enter Home
data_device_2 = dict(data_device_1)
data_device_2["lat"] = str(HOME_LATITUDE)
data_device_2["lon"] = str(HOME_LONGITUDE)
data_device_2["id"] = "device_2"
req = await client.post(url, params=data_device_2)
await hass.async_block_till_done()
assert req.status == HTTP_OK
state = hass.states.get("{}.{}".format(DEVICE_TRACKER_DOMAIN, data_device_2["id"]))
assert state.state == "home"
state = hass.states.get("{}.{}".format(DEVICE_TRACKER_DOMAIN, data_device_1["id"]))
assert state.state == "not_home"
@pytest.mark.xfail(
reason="The device_tracker component does not support unloading yet."
)
async def test_load_unload_entry(hass, client, webhook_id):
"""Test that the appropriate dispatch signals are added and removed."""
url = "/api/webhook/{}".format(webhook_id)
data = {"lat": str(HOME_LATITUDE), "lon": str(HOME_LONGITUDE), "id": "123"}
# Enter the Home
req = await client.post(url, params=data)
await hass.async_block_till_done()
assert req.status == HTTP_OK
state_name = hass.states.get(
"{}.{}".format(DEVICE_TRACKER_DOMAIN, data["id"])
).state
assert STATE_HOME == state_name
assert len(hass.data[DATA_DISPATCHER][TRACKER_UPDATE]) == 1
entry = hass.config_entries.async_entries(DOMAIN)[0]
assert await traccar.async_unload_entry(hass, entry)
await hass.async_block_till_done()
assert not hass.data[DATA_DISPATCHER][TRACKER_UPDATE]