ddeb6b6baa
* Extract const to a const file, Add battery as tracker attribute, add accuracy filter option * Update homeassistant/components/traccar/device_tracker.py Co-Authored-By: Otto Winter <otto@otto-winter.com> * Update homeassistant/components/traccar/device_tracker.py Co-Authored-By: Otto Winter <otto@otto-winter.com> * Update homeassistant/components/traccar/device_tracker.py Use [] syntax for keys that are in the validated data. Co-Authored-By: Otto Winter <otto@otto-winter.com> * Fix indentation
204 lines
9.1 KiB
Python
204 lines
9.1 KiB
Python
"""Support for Traccar device tracking."""
|
|
from datetime import datetime, timedelta
|
|
import logging
|
|
|
|
import voluptuous as vol
|
|
|
|
from homeassistant.components.device_tracker import PLATFORM_SCHEMA
|
|
from homeassistant.const import (
|
|
CONF_HOST, CONF_PORT, CONF_SSL, CONF_VERIFY_SSL,
|
|
CONF_PASSWORD, CONF_USERNAME, CONF_SCAN_INTERVAL,
|
|
CONF_MONITORED_CONDITIONS, CONF_EVENT)
|
|
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 .const import (
|
|
ATTR_ADDRESS, ATTR_CATEGORY, ATTR_GEOFENCE,
|
|
ATTR_MOTION, ATTR_SPEED, ATTR_TRACKER, ATTR_TRACCAR_ID, ATTR_STATUS,
|
|
EVENT_DEVICE_MOVING, EVENT_COMMAND_RESULT, EVENT_DEVICE_FUEL_DROP,
|
|
EVENT_GEOFENCE_ENTER, EVENT_DEVICE_OFFLINE, EVENT_DRIVER_CHANGED,
|
|
EVENT_GEOFENCE_EXIT, EVENT_DEVICE_OVERSPEED, EVENT_DEVICE_ONLINE,
|
|
EVENT_DEVICE_STOPPED, EVENT_MAINTENANCE, EVENT_ALARM, EVENT_TEXT_MESSAGE,
|
|
EVENT_DEVICE_UNKNOWN, EVENT_IGNITION_OFF, EVENT_IGNITION_ON,
|
|
EVENT_ALL_EVENTS, CONF_MAX_ACCURACY, CONF_SKIP_ACCURACY_ON)
|
|
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
DEFAULT_SCAN_INTERVAL = timedelta(seconds=30)
|
|
SCAN_INTERVAL = DEFAULT_SCAN_INTERVAL
|
|
|
|
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
|
|
vol.Required(CONF_PASSWORD): cv.string,
|
|
vol.Required(CONF_USERNAME): cv.string,
|
|
vol.Required(CONF_HOST): cv.string,
|
|
vol.Optional(CONF_PORT, default=8082): cv.port,
|
|
vol.Optional(CONF_SSL, default=False): cv.boolean,
|
|
vol.Optional(CONF_VERIFY_SSL, default=True): cv.boolean,
|
|
vol.Required(CONF_MAX_ACCURACY, default=0): vol.All(vol.Coerce(int),
|
|
vol.Range(min=0)),
|
|
vol.Optional(CONF_SKIP_ACCURACY_ON,
|
|
default=[]): vol.All(cv.ensure_list, [cv.string]),
|
|
vol.Optional(CONF_MONITORED_CONDITIONS,
|
|
default=[]): vol.All(cv.ensure_list, [cv.string]),
|
|
vol.Optional(CONF_EVENT,
|
|
default=[]): vol.All(cv.ensure_list,
|
|
[vol.Any(EVENT_DEVICE_MOVING,
|
|
EVENT_COMMAND_RESULT,
|
|
EVENT_DEVICE_FUEL_DROP,
|
|
EVENT_GEOFENCE_ENTER,
|
|
EVENT_DEVICE_OFFLINE,
|
|
EVENT_DRIVER_CHANGED,
|
|
EVENT_GEOFENCE_EXIT,
|
|
EVENT_DEVICE_OVERSPEED,
|
|
EVENT_DEVICE_ONLINE,
|
|
EVENT_DEVICE_STOPPED,
|
|
EVENT_MAINTENANCE,
|
|
EVENT_ALARM,
|
|
EVENT_TEXT_MESSAGE,
|
|
EVENT_DEVICE_UNKNOWN,
|
|
EVENT_IGNITION_OFF,
|
|
EVENT_IGNITION_ON,
|
|
EVENT_ALL_EVENTS)]),
|
|
})
|
|
|
|
|
|
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
|
|
|
|
session = async_get_clientsession(hass, config[CONF_VERIFY_SSL])
|
|
|
|
api = API(hass.loop, session, config[CONF_USERNAME], config[CONF_PASSWORD],
|
|
config[CONF_HOST], config[CONF_PORT], config[CONF_SSL])
|
|
|
|
scanner = TraccarScanner(
|
|
api, hass, async_see,
|
|
config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL),
|
|
config[CONF_MAX_ACCURACY], config[CONF_SKIP_ACCURACY_ON],
|
|
config[CONF_MONITORED_CONDITIONS], config[CONF_EVENT])
|
|
|
|
return await scanner.async_init()
|
|
|
|
|
|
class TraccarScanner:
|
|
"""Define an object to retrieve Traccar data."""
|
|
|
|
def __init__(self, api, hass, async_see, scan_interval, max_accuracy,
|
|
skip_accuracy_on, custom_attributes, event_types):
|
|
"""Initialize."""
|
|
from stringcase import camelcase
|
|
self._event_types = {camelcase(evt): evt for evt in event_types}
|
|
self._custom_attributes = custom_attributes
|
|
self._scan_interval = scan_interval
|
|
self._async_see = async_see
|
|
self._api = api
|
|
self.connected = False
|
|
self._hass = hass
|
|
self._max_accuracy = max_accuracy
|
|
self._skip_accuracy_on = skip_accuracy_on
|
|
|
|
async def async_init(self):
|
|
"""Further initialize connection to Traccar."""
|
|
await self._api.test_connection()
|
|
if self._api.connected and not self._api.authenticated:
|
|
_LOGGER.error("Authentication for Traccar failed")
|
|
return False
|
|
|
|
await self._async_update()
|
|
async_track_time_interval(self._hass,
|
|
self._async_update,
|
|
self._scan_interval)
|
|
return True
|
|
|
|
async def _async_update(self, now=None):
|
|
"""Update info from Traccar."""
|
|
if not self.connected:
|
|
_LOGGER.debug('Testing connection to Traccar')
|
|
await self._api.test_connection()
|
|
self.connected = self._api.connected
|
|
if self.connected:
|
|
_LOGGER.info("Connection to Traccar restored")
|
|
else:
|
|
return
|
|
_LOGGER.debug('Updating device data')
|
|
await self._api.get_device_info(self._custom_attributes)
|
|
self._hass.async_create_task(self.import_device_data())
|
|
if self._event_types:
|
|
self._hass.async_create_task(self.import_events())
|
|
self.connected = self._api.connected
|
|
|
|
async def import_device_data(self):
|
|
"""Import device data from Traccar."""
|
|
for device_unique_id in self._api.device_info:
|
|
device_info = self._api.device_info[device_unique_id]
|
|
device = None
|
|
attr = {}
|
|
skip_accuracy_filter = False
|
|
|
|
attr[ATTR_TRACKER] = 'traccar'
|
|
if device_info.get('address') is not None:
|
|
attr[ATTR_ADDRESS] = device_info['address']
|
|
if device_info.get('geofence') is not None:
|
|
attr[ATTR_GEOFENCE] = device_info['geofence']
|
|
if device_info.get('category') is not None:
|
|
attr[ATTR_CATEGORY] = device_info['category']
|
|
if device_info.get('speed') is not None:
|
|
attr[ATTR_SPEED] = device_info['speed']
|
|
if device_info.get('motion') is not None:
|
|
attr[ATTR_MOTION] = device_info['motion']
|
|
if device_info.get('traccar_id') is not None:
|
|
attr[ATTR_TRACCAR_ID] = device_info['traccar_id']
|
|
for dev in self._api.devices:
|
|
if dev['id'] == device_info['traccar_id']:
|
|
device = dev
|
|
break
|
|
if device is not None and device.get('status') is not None:
|
|
attr[ATTR_STATUS] = device['status']
|
|
for custom_attr in self._custom_attributes:
|
|
if device_info.get(custom_attr) is not None:
|
|
attr[custom_attr] = device_info[custom_attr]
|
|
if custom_attr in self._skip_accuracy_on:
|
|
skip_accuracy_filter = True
|
|
|
|
accuracy = 0.0
|
|
if device_info.get('accuracy') is not None:
|
|
accuracy = device_info['accuracy']
|
|
if (not skip_accuracy_filter and self._max_accuracy > 0 and
|
|
accuracy > self._max_accuracy):
|
|
_LOGGER.debug('Excluded position by accuracy filter: %f (%s)',
|
|
accuracy, attr[ATTR_TRACCAR_ID])
|
|
continue
|
|
|
|
await self._async_see(
|
|
dev_id=slugify(device_info['device_id']),
|
|
gps=(device_info.get('latitude'),
|
|
device_info.get('longitude')),
|
|
gps_accuracy=accuracy,
|
|
battery=device_info.get('battery'),
|
|
attributes=attr)
|
|
|
|
async def import_events(self):
|
|
"""Import events from Traccar."""
|
|
device_ids = [device['id'] for device in self._api.devices]
|
|
end_interval = datetime.utcnow()
|
|
start_interval = end_interval - self._scan_interval
|
|
events = await self._api.get_events(
|
|
device_ids=device_ids,
|
|
from_time=start_interval,
|
|
to_time=end_interval,
|
|
event_types=self._event_types.keys())
|
|
if events is not None:
|
|
for event in events:
|
|
device_name = next((
|
|
dev.get('name') for dev in self._api.devices
|
|
if dev.get('id') == event['deviceId']), None)
|
|
self._hass.bus.async_fire(
|
|
'traccar_' + self._event_types.get(event["type"]), {
|
|
'device_traccar_id': event['deviceId'],
|
|
'device_name': device_name,
|
|
'type': event['type'],
|
|
'serverTime': event['serverTime'],
|
|
'attributes': event['attributes']
|
|
})
|