coordinator.py

Go to the documentation of this file.

"""Data update coordinator for Traccar Server."""
 
from __future__ import annotations
 
import asyncio
from datetime import datetime
from logging import DEBUG as LOG_LEVEL_DEBUG
from typing import TYPE_CHECKING, Any, TypedDict
 
from pytraccar import (
    ApiClient,
    DeviceModel,
    GeofenceModel,
    PositionModel,
    SubscriptionData,
    TraccarException,
)
 
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util import dt as dt_util
 
from .const import DOMAIN, EVENTS, LOGGER
from .helpers import get_device, get_first_geofence
 
 
class TraccarServerCoordinatorDataDevice(TypedDict):
    """Traccar Server coordinator data."""
 
    device: DeviceModel
    geofence: GeofenceModel | None
    position: PositionModel
    attributes: dict[str, Any]
 
 
type TraccarServerCoordinatorData = dict[int, TraccarServerCoordinatorDataDevice]
 
 
class TraccarServerCoordinator(DataUpdateCoordinator[TraccarServerCoordinatorData]):
    """Class to manage fetching Traccar Server data."""
 
    config_entry: ConfigEntry
 
    def __init__(
        self,
        hass: HomeAssistant,
        client: ApiClient,
        *,
        events: list[str],
        max_accuracy: float,
        skip_accuracy_filter_for: list[str],
        custom_attributes: list[str],
    ) -> None:
        """Initialize global Traccar Server data updater."""
        super().__init__(
            hass=hass,
            logger=LOGGER,
            name=DOMAIN,
            update_interval=None,
        )
        self.clientclient = client
        self.custom_attributescustom_attributes = custom_attributes
        self.eventsevents = events
        self.max_accuracymax_accuracy = max_accuracy
        self.skip_accuracy_filter_forskip_accuracy_filter_for = skip_accuracy_filter_for
        self._geofences_geofences: list[GeofenceModel] = []
        self._last_event_import_last_event_import: datetime | None = None
        self._should_log_subscription_error_should_log_subscription_error: bool = True
 
    async def _async_update_data(self) -> TraccarServerCoordinatorData:
        """Fetch data from Traccar Server."""
        LOGGER.debug("Updating device data")
        data: TraccarServerCoordinatorData = {}
        try:
            (
                devices,
                positions,
                geofences,
            ) = await asyncio.gather(
                self.clientclient.get_devices(),
                self.clientclient.get_positions(),
                self.clientclient.get_geofences(),
            )
        except TraccarException as ex:
            raise UpdateFailed(f"Error while updating device data: {ex}") from ex
 
        if TYPE_CHECKING:
            assert isinstance(devices, list[DeviceModel])  # type: ignore[misc]
            assert isinstance(positions, list[PositionModel])  # type: ignore[misc]
            assert isinstance(geofences, list[GeofenceModel])  # type: ignore[misc]
 
        self._geofences_geofences = geofences
 
        if self.loggerlogger.isEnabledFor(LOG_LEVEL_DEBUG):
            self.loggerlogger.debug("Received devices: %s", devices)
            self.loggerlogger.debug("Received positions: %s", positions)
 
        for position in positions:
            device_id = position["deviceId"]
            if (device := get_device(device_id, devices)) is None:
                self.loggerlogger.debug(
                    "Device %s not found for position: %s",
                    device_id,
                    position["id"],
                )
                continue
 
            if (
                attr
                := self._return_custom_attributes_if_not_filtered_by_accuracy_configuration_return_custom_attributes_if_not_filtered_by_accuracy_configuration(
                    device, position
                )
            ) is None:
                self.loggerlogger.debug(
                    "Skipping position update %s for %s due to accuracy filter",
                    position["id"],
                    device_id,
                )
                continue
 
            data[device_id] = {
                "device": device,
                "geofence": get_first_geofence(
                    geofences,
                    position["geofenceIds"] or [],
                ),
                "position": position,
                "attributes": attr,
            }
 
        return data
 
    async def handle_subscription_data(self, data: SubscriptionData) -> None:
        """Handle subscription data."""
        self.loggerlogger.debug("Received subscription data: %s", data)
        self._should_log_subscription_error_should_log_subscription_error = True
        update_devices = set()
        for device in data.get("devices") or []:
            if (device_id := device["id"]) not in self.datadata:
                self.loggerlogger.debug("Device %s not found in data", device_id)
                continue
 
            if (
                attr
                := self._return_custom_attributes_if_not_filtered_by_accuracy_configuration_return_custom_attributes_if_not_filtered_by_accuracy_configuration(
                    device, self.datadata[device_id]["position"]
                )
            ) is None:
                continue
 
            self.datadata[device_id]["device"] = device
            self.datadata[device_id]["attributes"] = attr
            update_devices.add(device_id)
 
        for position in data.get("positions") or []:
            if (device_id := position["deviceId"]) not in self.datadata:
                self.loggerlogger.debug(
                    "Device %s for position %s not found in data",
                    device_id,
                    position["id"],
                )
                continue
 
            if (
                attr
                := self._return_custom_attributes_if_not_filtered_by_accuracy_configuration_return_custom_attributes_if_not_filtered_by_accuracy_configuration(
                    self.datadata[device_id]["device"], position
                )
            ) is None:
                self.loggerlogger.debug(
                    "Skipping position update %s for %s due to accuracy filter",
                    position["id"],
                    device_id,
                )
                continue
 
            self.datadata[device_id]["position"] = position
            self.datadata[device_id]["attributes"] = attr
            self.datadata[device_id]["geofence"] = get_first_geofence(
                self._geofences_geofences,
                position["geofenceIds"] or [],
            )
            update_devices.add(device_id)
 
        for device_id in update_devices:
            async_dispatcher_send(self.hasshass, f"{DOMAIN}_{device_id}")
 
    async def import_events(self, _: datetime) -> None:
        """Import events from Traccar."""
        start_time = dt_util.utcnow().replace(tzinfo=None)
        end_time = None
 
        if self._last_event_import_last_event_import is not None:
            end_time = start_time - (start_time - self._last_event_import_last_event_import)
 
        events = await self.clientclient.get_reports_events(
            devices=list(self.datadata),
            start_time=start_time,
            end_time=end_time,
            event_types=self.eventsevents,
        )
        if not events:
            return
 
        self._last_event_import_last_event_import = start_time
        for event in events:
            device = self.datadata[event["deviceId"]]["device"]
            self.hasshass.bus.async_fire(
                # This goes against two of the HA core guidelines:
                # 1. Event names should be prefixed with the domain name of
                #    the integration
                # 2. This should be event entities
                #
                # However, to not break it for those who currently use
                # the "old" integration, this is kept as is.
                f"traccar_{EVENTS[event['type']]}",
                {
                    "device_traccar_id": event["deviceId"],
                    "device_name": device["name"] if device else None,
                    "type": event["type"],
                    "serverTime": event["eventTime"],
                    "attributes": event["attributes"],
                },
            )
 
    async def subscribe(self) -> None:
        """Subscribe to events."""
        try:
            await self.clientclient.subscribe(self.handle_subscription_datahandle_subscription_data)
        except TraccarException as ex:
            if self._should_log_subscription_error_should_log_subscription_error:
                self._should_log_subscription_error_should_log_subscription_error = False
                LOGGER.error("Error while subscribing to Traccar: %s", ex)
            # Retry after 10 seconds
            await asyncio.sleep(10)
            await self.subscribesubscribe()
 
    def _return_custom_attributes_if_not_filtered_by_accuracy_configuration(
        self,
        device: DeviceModel,
        position: PositionModel,
    ) -> dict[str, Any] | None:
        """Return a dictionary of custom attributes if not filtered by accuracy configuration."""
        attr = {}
        skip_accuracy_filter = False
 
        for custom_attr in self.custom_attributescustom_attributes:
            if custom_attr in self.skip_accuracy_filter_forskip_accuracy_filter_for:
                skip_accuracy_filter = True
            attr[custom_attr] = device["attributes"].get(
                custom_attr,
                position["attributes"].get(custom_attr, None),
            )
 
        accuracy = position["accuracy"] or 0.0
        if (
            not skip_accuracy_filter
            and self.max_accuracymax_accuracy > 0
            and accuracy > self.max_accuracymax_accuracy
        ):
            return None
        return attr