self-hosted/home-assistant-core
self-hosted/home-assistant-core
1
0
Fork 0
mirror of https://github.com/home-assistant/core synced 2024-10-14 17:12:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Bump pytraccar to 1.0.0 (#75671)

Browse source
This commit is contained in:
Joakim Sørensen 2022-07-26 00:45:01 +02:00 committed by GitHub
parent ea354f3d5f
commit e87c2b9e25
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 128 additions and 103 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

182
homeassistant/components/traccar/device_tracker.py
View file

@ -1,11 +1,19 @@
"""Support for Traccar device tracking."""
from __future__ import annotations
import asyncio
from collections.abc import Awaitable, Callable
from datetime import datetime, timedelta
import logging
from pytraccar.api import API
from pytraccar import (
ApiClient,
DeviceModel,
GeofenceModel,
PositionModel,
TraccarAuthenticationException,
TraccarException,
)
from stringcase import camelcase
import voluptuous as vol
@ -170,17 +178,13 @@ async def async_setup_scanner(
discovery_info: DiscoveryInfoType | None = None,
) -> bool:
"""Validate the configuration and return a Traccar scanner."""
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],
api = ApiClient(
host=config[CONF_HOST],
port=config[CONF_PORT],
ssl=config[CONF_SSL],
username=config[CONF_USERNAME],
password=config[CONF_PASSWORD],
client_session=async_get_clientsession(hass, config[CONF_VERIFY_SSL]),
)
scanner = TraccarScanner(
@ -202,15 +206,15 @@ class TraccarScanner:
def __init__(
self,
api,
hass,
async_see,
scan_interval,
max_accuracy,
skip_accuracy_on,
custom_attributes,
event_types,
):
api: ApiClient,
hass: HomeAssistant,
async_see: Callable[..., Awaitable[None]],
scan_interval: timedelta,
max_accuracy: int,
skip_accuracy_on: bool,
custom_attributes: list[str],
event_types: list[str],
) -> None:
"""Initialize."""
if EVENT_ALL_EVENTS in event_types:
@ -220,15 +224,18 @@ class TraccarScanner:
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
self._devices: list[DeviceModel] = []
self._positions: list[PositionModel] = []
self._geofences: list[GeofenceModel] = []
async def async_init(self):
"""Further initialize connection to Traccar."""
await self._api.test_connection()
if self._api.connected and not self._api.authenticated:
try:
await self._api.get_server()
except TraccarAuthenticationException:
_LOGGER.error("Authentication for Traccar failed")
return False
@ -238,57 +245,63 @@ class TraccarScanner:
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)
try:
(self._devices, self._positions, self._geofences,) = await asyncio.gather(
self._api.get_devices(),
self._api.get_positions(),
self._api.get_geofences(),
)
except TraccarException as ex:
_LOGGER.error("Error while updating device data: %s", ex)
return
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 = {}
for position in self._positions:
device = next(
(dev for dev in self._devices if dev.id == position.device_id), None
)
if not device:
continue
attr = {
ATTR_TRACKER: "traccar",
ATTR_ADDRESS: position.address,
ATTR_SPEED: position.speed,
ATTR_ALTITUDE: position.altitude,
ATTR_MOTION: position.attributes.get("motion", False),
ATTR_TRACCAR_ID: device.id,
ATTR_GEOFENCE: next(
(
geofence.name
for geofence in self._geofences
if geofence.id in (device.geofence_ids or [])
),
None,
),
ATTR_CATEGORY: device.category,
ATTR_STATUS: device.status,
}
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 device.attributes.get(custom_attr) is not None:
attr[custom_attr] = position.attributes[custom_attr]
if custom_attr in self._skip_accuracy_on:
skip_accuracy_filter = True
if position.attributes.get(custom_attr) is not None:
attr[custom_attr] = position.attributes[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"]
accuracy = position.accuracy or 0.0
if (
not skip_accuracy_filter
and self._max_accuracy > 0
@ -302,42 +315,39 @@ class TraccarScanner:
continue
await self._async_see(
dev_id=slugify(device_info["device_id"]),
gps=(device_info.get("latitude"), device_info.get("longitude")),
dev_id=slugify(device.name),
gps=(position.latitude, position.longitude),
gps_accuracy=accuracy,
battery=device_info.get("battery"),
battery=position.attributes.get("batteryLevel", -1),
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,
start_intervel = datetime.utcnow()
events = await self._api.get_reports_events(
devices=[device.id for device in self._devices],
start_time=start_intervel,
end_time=start_intervel - self._scan_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(
f"traccar_{self._event_types.get(event['type'])}",
f"traccar_{self._event_types.get(event.type)}",
{
"device_traccar_id": event["deviceId"],
"device_name": device_name,
"type": event["type"],
"serverTime": event.get("eventTime") or event.get("serverTime"),
"attributes": event["attributes"],
"device_traccar_id": event.device_id,
"device_name": next(
(
dev.name
for dev in self._devices
if dev.id == event.device_id
),
None,
),
"type": event.type,
"serverTime": event.event_time,
"attributes": event.attributes,
},
)

2
homeassistant/components/traccar/manifest.json
View file

@ -3,7 +3,7 @@
"name": "Traccar",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/traccar",
"requirements": ["pytraccar==0.10.0", "stringcase==1.2.0"],
"requirements": ["pytraccar==1.0.0", "stringcase==1.2.0"],
"dependencies": ["webhook"],
"codeowners": ["@ludeeus"],
"iot_class": "local_polling",

2
requirements_all.txt
View file

@ -1996,7 +1996,7 @@ pytomorrowio==0.3.4
pytouchline==0.7
# homeassistant.components.traccar
pytraccar==0.10.0
pytraccar==1.0.0
# homeassistant.components.tradfri
pytradfri[async]==9.0.0

2
requirements_test_all.txt
View file

@ -1340,7 +1340,7 @@ pytile==2022.02.0
pytomorrowio==0.3.4
# homeassistant.components.traccar
pytraccar==0.10.0
pytraccar==1.0.0
# homeassistant.components.tradfri
pytradfri[async]==9.0.0

43
tests/components/traccar/test_device_tracker.py
View file

@ -2,6 +2,8 @@
from datetime import datetime
from unittest.mock import AsyncMock, patch
from pytraccar import ReportsEventeModel
from homeassistant.components.device_tracker.const import DOMAIN
from homeassistant.components.traccar.device_tracker import (
PLATFORM_SCHEMA as TRACCAR_PLATFORM_SCHEMA,
@ -35,26 +37,39 @@ async def test_import_events_catch_all(hass):
device = {"id": 1, "name": "abc123"}
api_mock = AsyncMock()
api_mock.devices = [device]
api_mock.get_events.return_value = [
{
"deviceId": device["id"],
"type": "ignitionOn",
"serverTime": datetime.utcnow(),
"attributes": {},
},
{
"deviceId": device["id"],
"type": "ignitionOff",
"serverTime": datetime.utcnow(),
"attributes": {},
},
api_mock.get_reports_events.return_value = [
ReportsEventeModel(
**{
"id": 1,
"positionId": 1,
"geofenceId": 1,
"maintenanceId": 1,
"deviceId": device["id"],
"type": "ignitionOn",
"eventTime": datetime.utcnow().isoformat(),
"attributes": {},
}
),
ReportsEventeModel(
**{
"id": 2,
"positionId": 2,
"geofenceId": 1,
"maintenanceId": 1,
"deviceId": device["id"],
"type": "ignitionOff",
"eventTime": datetime.utcnow().isoformat(),
"attributes": {},
}
),
]
events_ignition_on = async_capture_events(hass, "traccar_ignition_on")
events_ignition_off = async_capture_events(hass, "traccar_ignition_off")
with patch(
"homeassistant.components.traccar.device_tracker.API", return_value=api_mock
"homeassistant.components.traccar.device_tracker.ApiClient",
return_value=api_mock,
):
assert await async_setup_component(hass, DOMAIN, conf_dict)