Add Traccar server integration (#109002)

* Add Traccar server integration

* Add explination

* Update homeassistant/components/traccar_server/coordinator.py

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

* Add data_description

---------

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

.coveragerc

@@ -1424,6 +1424,11 @@ omit =
     homeassistant/components/tplink_omada/controller.py
     homeassistant/components/tplink_omada/update.py
     homeassistant/components/traccar/device_tracker.py
+    homeassistant/components/traccar_server/__init__.py
+    homeassistant/components/traccar_server/coordinator.py
+    homeassistant/components/traccar_server/device_tracker.py
+    homeassistant/components/traccar_server/entity.py
+    homeassistant/components/traccar_server/helpers.py
     homeassistant/components/tractive/__init__.py
     homeassistant/components/tractive/binary_sensor.py
     homeassistant/components/tractive/device_tracker.py

CODEOWNERS

@@ -1394,6 +1394,8 @@ build.json @home-assistant/supervisor
 /tests/components/tplink_omada/ @MarkGodwin
 /homeassistant/components/traccar/ @ludeeus
 /tests/components/traccar/ @ludeeus
+/homeassistant/components/traccar_server/ @ludeeus
+/tests/components/traccar_server/ @ludeeus
 /homeassistant/components/trace/ @home-assistant/core
 /tests/components/trace/ @home-assistant/core
 /homeassistant/components/tractive/ @Danielhiversen @zhulik @bieniu

homeassistant/components/traccar_server/__init__.py

@@ -0,0 +1,70 @@
+"""The Traccar Server integration."""
+from __future__ import annotations
+
+from pytraccar import ApiClient
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import (
+    CONF_HOST,
+    CONF_PASSWORD,
+    CONF_PORT,
+    CONF_SSL,
+    CONF_USERNAME,
+    CONF_VERIFY_SSL,
+    Platform,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import (
+    CONF_CUSTOM_ATTRIBUTES,
+    CONF_EVENTS,
+    CONF_MAX_ACCURACY,
+    CONF_SKIP_ACCURACY_FILTER_FOR,
+    DOMAIN,
+)
+from .coordinator import TraccarServerCoordinator
+
+PLATFORMS: list[Platform] = [Platform.DEVICE_TRACKER]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Traccar Server from a config entry."""
+    coordinator = TraccarServerCoordinator(
+        hass=hass,
+        client=ApiClient(
+            client_session=async_get_clientsession(hass),
+            host=entry.data[CONF_HOST],
+            port=entry.data[CONF_PORT],
+            username=entry.data[CONF_USERNAME],
+            password=entry.data[CONF_PASSWORD],
+            ssl=entry.data[CONF_SSL],
+            verify_ssl=entry.data[CONF_VERIFY_SSL],
+        ),
+        events=entry.options.get(CONF_EVENTS, []),
+        max_accuracy=entry.options.get(CONF_MAX_ACCURACY, 0.0),
+        skip_accuracy_filter_for=entry.options.get(CONF_SKIP_ACCURACY_FILTER_FOR, []),
+        custom_attributes=entry.options.get(CONF_CUSTOM_ATTRIBUTES, []),
+    )
+
+    await coordinator.async_config_entry_first_refresh()
+
+    hass.data.setdefault(DOMAIN, {})
+    hass.data[DOMAIN][entry.entry_id] = coordinator
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    entry.async_on_unload(entry.add_update_listener(async_reload_entry))
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        hass.data[DOMAIN].pop(entry.entry_id)
+    return unload_ok
+
+
+async def async_reload_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
+    """Handle an options update."""
+    await hass.config_entries.async_reload(entry.entry_id)

homeassistant/components/traccar_server/config_flow.py

@@ -0,0 +1,168 @@
+"""Config flow for Traccar Server integration."""
+from __future__ import annotations
+
+from typing import Any
+
+from pytraccar import ApiClient, ServerModel, TraccarException
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.const import (
+    CONF_HOST,
+    CONF_PASSWORD,
+    CONF_PORT,
+    CONF_SSL,
+    CONF_USERNAME,
+    CONF_VERIFY_SSL,
+)
+from homeassistant.core import callback
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.schema_config_entry_flow import (
+    SchemaFlowFormStep,
+    SchemaOptionsFlowHandler,
+)
+from homeassistant.helpers.selector import (
+    BooleanSelector,
+    BooleanSelectorConfig,
+    NumberSelector,
+    NumberSelectorConfig,
+    NumberSelectorMode,
+    SelectSelector,
+    SelectSelectorConfig,
+    SelectSelectorMode,
+    TextSelector,
+    TextSelectorConfig,
+    TextSelectorType,
+)
+
+from .const import (
+    CONF_CUSTOM_ATTRIBUTES,
+    CONF_EVENTS,
+    CONF_MAX_ACCURACY,
+    CONF_SKIP_ACCURACY_FILTER_FOR,
+    DOMAIN,
+    EVENTS,
+    LOGGER,
+)
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_HOST): TextSelector(
+            TextSelectorConfig(type=TextSelectorType.TEXT)
+        ),
+        vol.Optional(CONF_PORT, default="8082"): TextSelector(
+            TextSelectorConfig(type=TextSelectorType.TEXT)
+        ),
+        vol.Required(CONF_USERNAME): TextSelector(
+            TextSelectorConfig(type=TextSelectorType.EMAIL)
+        ),
+        vol.Required(CONF_PASSWORD): TextSelector(
+            TextSelectorConfig(type=TextSelectorType.PASSWORD)
+        ),
+        vol.Optional(CONF_SSL, default=False): BooleanSelector(BooleanSelectorConfig()),
+        vol.Optional(CONF_VERIFY_SSL, default=True): BooleanSelector(
+            BooleanSelectorConfig()
+        ),
+    }
+)
+
+OPTIONS_FLOW = {
+    "init": SchemaFlowFormStep(
+        schema=vol.Schema(
+            {
+                vol.Optional(CONF_MAX_ACCURACY, default=0.0): NumberSelector(
+                    NumberSelectorConfig(
+                        mode=NumberSelectorMode.BOX,
+                        min=0.0,
+                    )
+                ),
+                vol.Optional(CONF_CUSTOM_ATTRIBUTES, default=[]): SelectSelector(
+                    SelectSelectorConfig(
+                        mode=SelectSelectorMode.DROPDOWN,
+                        multiple=True,
+                        sort=True,
+                        custom_value=True,
+                        options=[],
+                    )
+                ),
+                vol.Optional(CONF_SKIP_ACCURACY_FILTER_FOR, default=[]): SelectSelector(
+                    SelectSelectorConfig(
+                        mode=SelectSelectorMode.DROPDOWN,
+                        multiple=True,
+                        sort=True,
+                        custom_value=True,
+                        options=[],
+                    )
+                ),
+                vol.Optional(CONF_EVENTS, default=[]): SelectSelector(
+                    SelectSelectorConfig(
+                        mode=SelectSelectorMode.DROPDOWN,
+                        multiple=True,
+                        sort=True,
+                        custom_value=True,
+                        options=list(EVENTS),
+                    )
+                ),
+            }
+        )
+    ),
+}
+
+
+class TraccarServerConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Traccar Server."""
+
+    async def _get_server_info(self, user_input: dict[str, Any]) -> ServerModel:
+        """Get server info."""
+        client = ApiClient(
+            client_session=async_get_clientsession(self.hass),
+            host=user_input[CONF_HOST],
+            port=user_input[CONF_PORT],
+            username=user_input[CONF_USERNAME],
+            password=user_input[CONF_PASSWORD],
+            ssl=user_input[CONF_SSL],
+            verify_ssl=user_input[CONF_VERIFY_SSL],
+        )
+        return await client.get_server()
+
+    async def async_step_user(
+        self,
+        user_input: dict[str, Any] | None = None,
+    ) -> FlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            self._async_abort_entries_match(
+                {
+                    CONF_HOST: user_input[CONF_HOST],
+                    CONF_PORT: user_input[CONF_PORT],
+                }
+            )
+            try:
+                await self._get_server_info(user_input)
+            except TraccarException as exception:
+                LOGGER.error("Unable to connect to Traccar Server: %s", exception)
+                errors["base"] = "cannot_connect"
+            except Exception:  # pylint: disable=broad-except
+                LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            else:
+                return self.async_create_entry(
+                    title=f"{user_input[CONF_HOST]}:{user_input[CONF_PORT]}",
+                    data=user_input,
+                )
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=STEP_USER_DATA_SCHEMA,
+            errors=errors,
+        )
+
+    @staticmethod
+    @callback
+    def async_get_options_flow(
+        config_entry: config_entries.ConfigEntry,
+    ) -> SchemaOptionsFlowHandler:
+        """Get the options flow for this handler."""
+        return SchemaOptionsFlowHandler(config_entry, OPTIONS_FLOW)

homeassistant/components/traccar_server/const.py

@@ -0,0 +1,39 @@
+"""Constants for the Traccar Server integration."""
+from logging import getLogger
+
+DOMAIN = "traccar_server"
+LOGGER = getLogger(__package__)
+
+ATTR_ADDRESS = "address"
+ATTR_ALTITUDE = "altitude"
+ATTR_CATEGORY = "category"
+ATTR_GEOFENCE = "geofence"
+ATTR_MOTION = "motion"
+ATTR_SPEED = "speed"
+ATTR_STATUS = "status"
+ATTR_TRACKER = "tracker"
+ATTR_TRACCAR_ID = "traccar_id"
+
+CONF_MAX_ACCURACY = "max_accuracy"
+CONF_CUSTOM_ATTRIBUTES = "custom_attributes"
+CONF_EVENTS = "events"
+CONF_SKIP_ACCURACY_FILTER_FOR = "skip_accuracy_filter_for"
+
+EVENTS = {
+    "deviceMoving": "device_moving",
+    "commandResult": "command_result",
+    "deviceFuelDrop": "device_fuel_drop",
+    "geofenceEnter": "geofence_enter",
+    "deviceOffline": "device_offline",
+    "driverChanged": "driver_changed",
+    "geofenceExit": "geofence_exit",
+    "deviceOverspeed": "device_overspeed",
+    "deviceOnline": "device_online",
+    "deviceStopped": "device_stopped",
+    "maintenance": "maintenance",
+    "alarm": "alarm",
+    "textMessage": "text_message",
+    "deviceUnknown": "device_unknown",
+    "ignitionOff": "ignition_off",
+    "ignitionOn": "ignition_on",
+}

homeassistant/components/traccar_server/coordinator.py

@@ -0,0 +1,165 @@
+"""Data update coordinator for Traccar Server."""
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime, timedelta
+from typing import TYPE_CHECKING, Any, TypedDict
+
+from pytraccar import (
+    ApiClient,
+    DeviceModel,
+    GeofenceModel,
+    PositionModel,
+    TraccarException,
+)
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+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]
+
+
+TraccarServerCoordinatorData = dict[str, 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=timedelta(seconds=30),
+        )
+        self.client = client
+        self.custom_attributes = custom_attributes
+        self.events = events
+        self.max_accuracy = max_accuracy
+        self.skip_accuracy_filter_for = skip_accuracy_filter_for
+        self._last_event_import: datetime | None = None
+
+    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.client.get_devices(),
+                self.client.get_positions(),
+                self.client.get_geofences(),
+            )
+        except TraccarException as ex:
+            raise UpdateFailed("Error while updating device data: %s") 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]
+
+        for position in positions:
+            if (device := get_device(position["deviceId"], devices)) is None:
+                continue
+
+            attr = {}
+            skip_accuracy_filter = False
+
+            for custom_attr in self.custom_attributes:
+                attr[custom_attr] = getattr(
+                    device["attributes"],
+                    custom_attr,
+                    getattr(position["attributes"], custom_attr, None),
+                )
+                if custom_attr in self.skip_accuracy_filter_for:
+                    skip_accuracy_filter = True
+
+            accuracy = position["accuracy"] or 0.0
+            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,
+                    device["id"],
+                )
+                continue
+
+            data[device["uniqueId"]] = {
+                "device": device,
+                "geofence": get_first_geofence(
+                    geofences,
+                    position["geofenceIds"] or [],
+                ),
+                "position": position,
+                "attributes": attr,
+            }
+
+        if self.events:
+            self.hass.async_create_task(self.import_events(devices))
+
+        return data
+
+    async def import_events(self, devices: list[DeviceModel]) -> None:
+        """Import events from Traccar."""
+        start_time = dt_util.utcnow().replace(tzinfo=None)
+        end_time = None
+
+        if self._last_event_import is not None:
+            end_time = start_time - (start_time - self._last_event_import)
+
+        events = await self.client.get_reports_events(
+            devices=[device["id"] for device in devices],
+            start_time=start_time,
+            end_time=end_time,
+            event_types=self.events,
+        )
+        if not events:
+            return
+
+        self._last_event_import = start_time
+        for event in events:
+            device = get_device(event["deviceId"], devices)
+            self.hass.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": getattr(device, "name", None),
+                    "type": event["type"],
+                    "serverTime": event["eventTime"],
+                    "attributes": event["attributes"],
+                },
+            )

homeassistant/components/traccar_server/device_tracker.py

@@ -0,0 +1,85 @@
+"""Support for Traccar server device tracking."""
+from __future__ import annotations
+
+from typing import Any
+
+from homeassistant.components.device_tracker import SourceType, TrackerEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .const import (
+    ATTR_ADDRESS,
+    ATTR_ALTITUDE,
+    ATTR_CATEGORY,
+    ATTR_GEOFENCE,
+    ATTR_MOTION,
+    ATTR_SPEED,
+    ATTR_STATUS,
+    ATTR_TRACCAR_ID,
+    ATTR_TRACKER,
+    DOMAIN,
+)
+from .coordinator import TraccarServerCoordinator
+from .entity import TraccarServerEntity
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up device tracker entities."""
+    coordinator: TraccarServerCoordinator = hass.data[DOMAIN][entry.entry_id]
+    async_add_entities(
+        TraccarServerDeviceTracker(coordinator, entry["device"])
+        for entry in coordinator.data.values()
+    )
+
+
+class TraccarServerDeviceTracker(TraccarServerEntity, TrackerEntity):
+    """Represent a tracked device."""
+
+    _attr_has_entity_name = True
+    _attr_name = None
+
+    @property
+    def battery_level(self) -> int:
+        """Return battery value of the device."""
+        return self.traccar_position["attributes"].get("batteryLevel", -1)
+
+    @property
+    def extra_state_attributes(self) -> dict[str, Any]:
+        """Return device specific attributes."""
+        return {
+            **self.traccar_attributes,
+            ATTR_ADDRESS: self.traccar_position["address"],
+            ATTR_ALTITUDE: self.traccar_position["altitude"],
+            ATTR_CATEGORY: self.traccar_device["category"],
+            ATTR_GEOFENCE: getattr(self.traccar_geofence, "name", None),
+            ATTR_MOTION: self.traccar_position["attributes"].get("motion", False),
+            ATTR_SPEED: self.traccar_position["speed"],
+            ATTR_STATUS: self.traccar_device["status"],
+            ATTR_TRACCAR_ID: self.traccar_device["id"],
+            ATTR_TRACKER: DOMAIN,
+        }
+
+    @property
+    def latitude(self) -> float:
+        """Return latitude value of the device."""
+        return self.traccar_position["latitude"]
+
+    @property
+    def longitude(self) -> float:
+        """Return longitude value of the device."""
+        return self.traccar_position["longitude"]
+
+    @property
+    def location_accuracy(self) -> int:
+        """Return the gps accuracy of the device."""
+        return self.traccar_position["accuracy"]
+
+    @property
+    def source_type(self) -> SourceType:
+        """Return the source type, eg gps or router, of the device."""
+        return SourceType.GPS

homeassistant/components/traccar_server/entity.py

@@ -0,0 +1,59 @@
+"""Base entity for Traccar Server."""
+from __future__ import annotations
+
+from typing import Any
+
+from pytraccar import DeviceModel, GeofenceModel, PositionModel
+
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN
+from .coordinator import TraccarServerCoordinator
+
+
+class TraccarServerEntity(CoordinatorEntity[TraccarServerCoordinator]):
+    """Base entity for Traccar Server."""
+
+    def __init__(
+        self,
+        coordinator: TraccarServerCoordinator,
+        device: DeviceModel,
+    ) -> None:
+        """Initialize the Traccar Server entity."""
+        super().__init__(coordinator)
+        self.device_id = device["uniqueId"]
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, device["uniqueId"])},
+            model=device["model"],
+            name=device["name"],
+        )
+        self._attr_unique_id = device["uniqueId"]
+
+    @property
+    def available(self) -> bool:
+        """Return True if entity is available."""
+        return (
+            self.coordinator.last_update_success
+            and self.device_id in self.coordinator.data
+        )
+
+    @property
+    def traccar_device(self) -> DeviceModel:
+        """Return the device."""
+        return self.coordinator.data[self.device_id]["device"]
+
+    @property
+    def traccar_geofence(self) -> GeofenceModel | None:
+        """Return the geofence."""
+        return self.coordinator.data[self.device_id]["geofence"]
+
+    @property
+    def traccar_position(self) -> PositionModel:
+        """Return the position."""
+        return self.coordinator.data[self.device_id]["position"]
+
+    @property
+    def traccar_attributes(self) -> dict[str, Any]:
+        """Return the attributes."""
+        return self.coordinator.data[self.device_id]["attributes"]

homeassistant/components/traccar_server/helpers.py

@@ -0,0 +1,23 @@
+"""Helper functions for the Traccar Server integration."""
+from __future__ import annotations
+
+from pytraccar import DeviceModel, GeofenceModel
+
+
+def get_device(device_id: int, devices: list[DeviceModel]) -> DeviceModel | None:
+    """Return the device."""
+    return next(
+        (dev for dev in devices if dev["id"] == device_id),
+        None,
+    )
+
+
+def get_first_geofence(
+    geofences: list[GeofenceModel],
+    target: list[int],
+) -> GeofenceModel | None:
+    """Return the geofence."""
+    return next(
+        (geofence for geofence in geofences if geofence["id"] in target),
+        None,
+    )

homeassistant/components/traccar_server/manifest.json

@@ -0,0 +1,9 @@
+{
+  "domain": "traccar_server",
+  "name": "Traccar Server",
+  "codeowners": ["@ludeeus"],
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/traccar_server",
+  "iot_class": "local_polling",
+  "requirements": ["pytraccar==2.0.0"]
+}

homeassistant/components/traccar_server/strings.json

@@ -0,0 +1,45 @@
+{
+  "config": {
+    "step": {
+      "user": {
+        "data": {
+          "host": "[%key:common::config_flow::data::host%]",
+          "password": "[%key:common::config_flow::data::password%]",
+          "port": "[%key:common::config_flow::data::port%]",
+          "ssl": "[%key:common::config_flow::data::ssl%]",
+          "username": "[%key:common::config_flow::data::username%]",
+          "verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
+        },
+        "data_description": {
+          "host": "The hostname or IP address of your Traccar Server",
+          "username": "The username (email) you use to login to your Traccar Server"
+        }
+      }
+    },
+    "error": {
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
+      "unknown": "[%key:common::config_flow::error::unknown%]"
+    },
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
+    }
+  },
+  "options": {
+    "step": {
+      "init": {
+        "data": {
+          "max_accuracy": "Max accuracy",
+          "skip_accuracy_filter_for": "Position skip filter for attributes",
+          "custom_attributes": "Custom attributes",
+          "events": "Events"
+        },
+        "data_description": {
+          "max_accuracy": "Any position reports with accuracy higher than this value will be ignored",
+          "skip_accuracy_filter_for": "Attributes defined here will bypass the accuracy filter if they are present in the update",
+          "custom_attributes": "Add any custom or calculated attributes here. These will be added to the device attributes",
+          "events": "Selected events will be fired in Home Assistant"
+        }
+      }
+    }
+  }
+}

homeassistant/generated/config_flows.py

@@ -538,6 +538,7 @@ FLOWS = {
         "tplink",
         "tplink_omada",
         "traccar",
+        "traccar_server",
         "tractive",
         "tradfri",
         "trafikverket_camera",

homeassistant/generated/integrations.json

@@ -6168,6 +6168,12 @@
       "config_flow": true,
       "iot_class": "local_polling"
     },
+    "traccar_server": {
+      "name": "Traccar Server",
+      "integration_type": "hub",
+      "config_flow": true,
+      "iot_class": "local_polling"
+    },
     "tractive": {
       "name": "Tractive",
       "integration_type": "device",

requirements_all.txt

@@ -2305,6 +2305,7 @@ pytomorrowio==0.3.6
 pytouchline==0.7
 
 # homeassistant.components.traccar
+# homeassistant.components.traccar_server
 pytraccar==2.0.0
 
 # homeassistant.components.tradfri

requirements_test_all.txt

@@ -1760,6 +1760,7 @@ pytile==2023.04.0
 pytomorrowio==0.3.6
 
 # homeassistant.components.traccar
+# homeassistant.components.traccar_server
 pytraccar==2.0.0
 
 # homeassistant.components.tradfri

tests/components/traccar_server/__init__.py

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

tests/components/traccar_server/conftest.py

@@ -0,0 +1,14 @@
+"""Common fixtures for the Traccar Server tests."""
+from collections.abc import Generator
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+
+@pytest.fixture
+def mock_setup_entry() -> Generator[AsyncMock, None, None]:
+    """Override async_setup_entry."""
+    with patch(
+        "homeassistant.components.traccar_server.async_setup_entry", return_value=True
+    ) as mock_setup_entry:
+        yield mock_setup_entry

tests/components/traccar_server/test_config_flow.py

@@ -0,0 +1,189 @@
+"""Test the Traccar Server config flow."""
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from pytraccar import TraccarException
+
+from homeassistant import config_entries
+from homeassistant.components.traccar_server.const import (
+    CONF_CUSTOM_ATTRIBUTES,
+    CONF_EVENTS,
+    CONF_MAX_ACCURACY,
+    CONF_SKIP_ACCURACY_FILTER_FOR,
+    DOMAIN,
+)
+from homeassistant.const import (
+    CONF_HOST,
+    CONF_PASSWORD,
+    CONF_PORT,
+    CONF_SSL,
+    CONF_USERNAME,
+    CONF_VERIFY_SSL,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResultType
+
+from tests.common import MockConfigEntry
+
+
+async def test_form(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> None:
+    """Test we get the form."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == FlowResultType.FORM
+    assert result["errors"] == {}
+
+    with patch(
+        "homeassistant.components.traccar_server.config_flow.ApiClient.get_server",
+        return_value={"id": "1234"},
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                CONF_HOST: "1.1.1.1",
+                CONF_USERNAME: "test-username",
+                CONF_PASSWORD: "test-password",
+            },
+        )
+        await hass.async_block_till_done()
+
+    assert result["type"] == FlowResultType.CREATE_ENTRY
+    assert result["title"] == "1.1.1.1:8082"
+    assert result["data"] == {
+        CONF_HOST: "1.1.1.1",
+        CONF_PORT: "8082",
+        CONF_USERNAME: "test-username",
+        CONF_PASSWORD: "test-password",
+        CONF_SSL: False,
+        CONF_VERIFY_SSL: True,
+    }
+    assert len(mock_setup_entry.mock_calls) == 1
+
+
+@pytest.mark.parametrize(
+    ("side_effect", "error"),
+    (
+        (TraccarException, "cannot_connect"),
+        (Exception, "unknown"),
+    ),
+)
+async def test_form_cannot_connect(
+    hass: HomeAssistant,
+    mock_setup_entry: AsyncMock,
+    side_effect: Exception,
+    error: str,
+) -> None:
+    """Test we handle cannot connect error."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "homeassistant.components.traccar_server.config_flow.ApiClient.get_server",
+        side_effect=side_effect,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                CONF_HOST: "1.1.1.1",
+                CONF_USERNAME: "test-username",
+                CONF_PASSWORD: "test-password",
+            },
+        )
+
+    assert result["type"] == FlowResultType.FORM
+    assert result["errors"] == {"base": error}
+
+    with patch(
+        "homeassistant.components.traccar_server.config_flow.ApiClient.get_server",
+        return_value={"id": "1234"},
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                CONF_HOST: "1.1.1.1",
+                CONF_USERNAME: "test-username",
+                CONF_PASSWORD: "test-password",
+            },
+        )
+        await hass.async_block_till_done()
+
+    assert result["type"] == FlowResultType.CREATE_ENTRY
+    assert result["title"] == "1.1.1.1:8082"
+    assert result["data"] == {
+        CONF_HOST: "1.1.1.1",
+        CONF_PORT: "8082",
+        CONF_USERNAME: "test-username",
+        CONF_PASSWORD: "test-password",
+        CONF_SSL: False,
+        CONF_VERIFY_SSL: True,
+    }
+    assert len(mock_setup_entry.mock_calls) == 1
+
+
+async def test_options(
+    hass: HomeAssistant,
+    mock_setup_entry: AsyncMock,
+) -> None:
+    """Test options flow."""
+
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={},
+    )
+
+    config_entry.add_to_hass(hass)
+
+    assert await hass.config_entries.async_setup(config_entry.entry_id)
+
+    assert CONF_MAX_ACCURACY not in config_entry.options
+
+    result = await hass.config_entries.options.async_init(config_entry.entry_id)
+
+    result = await hass.config_entries.options.async_configure(
+        result["flow_id"],
+        user_input={CONF_MAX_ACCURACY: 2.0},
+    )
+    await hass.async_block_till_done()
+
+    assert result["type"] == FlowResultType.CREATE_ENTRY
+    assert config_entry.options == {
+        CONF_MAX_ACCURACY: 2.0,
+        CONF_EVENTS: [],
+        CONF_CUSTOM_ATTRIBUTES: [],
+        CONF_SKIP_ACCURACY_FILTER_FOR: [],
+    }
+
+
+async def test_abort_already_configured(
+    hass: HomeAssistant,
+    mock_setup_entry: AsyncMock,
+) -> None:
+    """Test abort for existing server."""
+
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_HOST: "1.1.1.1", CONF_PORT: "8082"},
+    )
+
+    config_entry.add_to_hass(hass)
+
+    assert await hass.config_entries.async_setup(config_entry.entry_id)
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {
+            CONF_HOST: "1.1.1.1",
+            CONF_PORT: "8082",
+            CONF_USERNAME: "test-username",
+            CONF_PASSWORD: "test-password",
+        },
+    )
+
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == "already_configured"