Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Import Traccar YAML configuration to Traccar Server (#109226) 

* Import Traccar YAML configuration to Traccar Server

* Remove import
pull/109231/head
Joakim Sørensen 2024-01-31 18:16:23 +01:00 committed by GitHub
parent 0b0bf73780
commit cd96fb381f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 230 additions and 271 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,30 +1,25 @@
"""Support for Traccar device tracking."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
from typing import Any
from pytraccar import (
ApiClient,
DeviceModel,
GeofenceModel,
PositionModel,
TraccarAuthenticationException,
TraccarConnectionException,
TraccarException,
)
from stringcase import camelcase
from pytraccar import ApiClient, TraccarException
import voluptuous as vol
from homeassistant.components.device_tracker import (
CONF_SCAN_INTERVAL,
PLATFORM_SCHEMA as PARENT_PLATFORM_SCHEMA,
AsyncSeeCallback,
SourceType,
TrackerEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.components.device_tracker.legacy import (
YAML_DEVICES,
remove_device_from_config,
)
from homeassistant.config import load_yaml_config_file
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
CONF_EVENT,
CONF_HOST,
@ -34,34 +29,34 @@ from homeassistant.const import (
CONF_SSL,
CONF_USERNAME,
CONF_VERIFY_SSL,
EVENT_HOMEASSISTANT_STARTED,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.core import (
DOMAIN as HOMEASSISTANT_DOMAIN,
Event,
HomeAssistant,
callback,
)
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import dt as dt_util, slugify
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_TRACCAR_ID,
ATTR_TRACKER,
CONF_MAX_ACCURACY,
CONF_SKIP_ACCURACY_ON,
EVENT_ALARM,
@ -178,7 +173,7 @@ async def async_setup_scanner(
async_see: AsyncSeeCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> bool:
"""Validate the configuration and return a Traccar scanner."""
"""Import configuration to the new integration."""
api = ApiClient(
host=config[CONF_HOST],
port=config[CONF_PORT],
@ -188,180 +183,62 @@ async def async_setup_scanner(
client_session=async_get_clientsession(hass, config[CONF_VERIFY_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: ApiClient,
hass: HomeAssistant,
async_see: AsyncSeeCallback,
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:
event_types = EVENTS
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._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."""
async def _run_import(_: Event):
known_devices: dict[str, dict[str, Any]] = {}
try:
await self._api.get_server()
except TraccarAuthenticationException:
_LOGGER.error("Authentication for Traccar failed")
return False
except TraccarConnectionException as exception:
_LOGGER.error("Connection with Traccar failed - %s", exception)
return False
known_devices = await hass.async_add_executor_job(
load_yaml_config_file, hass.config.path(YAML_DEVICES)
)
except (FileNotFoundError, HomeAssistantError):
_LOGGER.debug(
"No valid known_devices.yaml found, "
"skip removal of devices from known_devices.yaml"
)
await self._async_update()
async_track_time_interval(
self._hass, self._async_update, self._scan_interval, cancel_on_shutdown=True
if known_devices:
traccar_devices: list[str] = []
try:
resp = await api.get_devices()
traccar_devices = [slugify(device["name"]) for device in resp]
except TraccarException as exception:
_LOGGER.error("Error while getting device data: %s", exception)
return
for dev_name in traccar_devices:
if dev_name in known_devices:
await hass.async_add_executor_job(
remove_device_from_config, hass, dev_name
)
_LOGGER.debug("Removed device %s from known_devices.yaml", dev_name)
if not hass.states.async_available(f"device_tracker.{dev_name}"):
hass.states.async_remove(f"device_tracker.{dev_name}")
hass.async_create_task(
hass.config_entries.flow.async_init(
"traccar_server",
context={"source": SOURCE_IMPORT},
data=config,
)
)
return True
async def _async_update(self, now=None):
"""Update info from Traccar."""
_LOGGER.debug("Updating device data")
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())
async def import_device_data(self):
"""Import device data from Traccar."""
for position in self._positions:
device = next(
(dev for dev in self._devices if dev["id"] == position["deviceId"]),
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 (position["geofenceIds"] or [])
),
None,
),
ATTR_CATEGORY: device["category"],
ATTR_STATUS: device["status"],
}
skip_accuracy_filter = False
for custom_attr in self._custom_attributes:
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 = 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,
attr[ATTR_TRACCAR_ID],
)
continue
await self._async_see(
dev_id=slugify(device["name"]),
gps=(position["latitude"], position["longitude"]),
gps_accuracy=accuracy,
battery=position["attributes"].get("batteryLevel", -1),
attributes=attr,
)
async def import_events(self):
"""Import events from Traccar."""
# get_reports_events requires naive UTC datetimes as of 1.0.0
start_intervel = dt_util.utcnow().replace(tzinfo=None)
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(),
async_create_issue(
hass,
HOMEASSISTANT_DOMAIN,
f"deprecated_yaml_{DOMAIN}",
breaks_in_ha_version="2024.8.0",
is_fixable=False,
issue_domain=DOMAIN,
severity=IssueSeverity.WARNING,
translation_key="deprecated_yaml",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "Traccar",
},
)
if events is not None:
for event in events:
self._hass.bus.async_fire(
f"traccar_{self._event_types.get(event['type'])}",
{
"device_traccar_id": event["deviceId"],
"device_name": next(
(
dev["name"]
for dev in self._devices
if dev["id"] == event["deviceId"]
),
None,
),
"type": event["type"],
"serverTime": event["eventTime"],
"attributes": event["attributes"],
},
)
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STARTED, _run_import)
return True
class TraccarEntity(TrackerEntity, RestoreEntity):

34
homeassistant/components/traccar_server/config_flow.py
View File

@ -1,6 +1,7 @@
"""Config flow for Traccar Server integration."""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from pytraccar import ApiClient, ServerModel, TraccarException
@ -159,6 +160,39 @@ class TraccarServerConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def async_step_import(self, import_info: Mapping[str, Any]) -> FlowResult:
"""Import an entry."""
configured_port = str(import_info[CONF_PORT])
self._async_abort_entries_match(
{
CONF_HOST: import_info[CONF_HOST],
CONF_PORT: configured_port,
}
)
if "all_events" in (imported_events := import_info.get("event", [])):
events = list(EVENTS.values())
else:
events = imported_events
return self.async_create_entry(
title=f"{import_info[CONF_HOST]}:{configured_port}",
data={
CONF_HOST: import_info[CONF_HOST],
CONF_PORT: configured_port,
CONF_SSL: import_info.get(CONF_SSL, False),
CONF_VERIFY_SSL: import_info.get(CONF_VERIFY_SSL, True),
CONF_USERNAME: import_info[CONF_USERNAME],
CONF_PASSWORD: import_info[CONF_PASSWORD],
},
options={
CONF_MAX_ACCURACY: import_info[CONF_MAX_ACCURACY],
CONF_EVENTS: events,
CONF_CUSTOM_ATTRIBUTES: import_info.get("monitored_conditions", []),
CONF_SKIP_ACCURACY_FILTER_FOR: import_info.get(
"skip_accuracy_filter_on", []
),
},
)
@staticmethod
@callback
def async_get_options_flow(

78
tests/components/traccar/test_device_tracker.py
View File

@ -1,78 +0,0 @@
"""The tests for the Traccar device tracker platform."""
from unittest.mock import AsyncMock, patch
from pytraccar import ReportsEventeModel
from homeassistant.components.device_tracker import DOMAIN
from homeassistant.components.traccar.device_tracker import (
PLATFORM_SCHEMA as TRACCAR_PLATFORM_SCHEMA,
)
from homeassistant.const import (
CONF_EVENT,
CONF_HOST,
CONF_PASSWORD,
CONF_PLATFORM,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from tests.common import async_capture_events
async def test_import_events_catch_all(hass: HomeAssistant) -> None:
"""Test importing all events and firing them in HA using their event types."""
conf_dict = {
DOMAIN: TRACCAR_PLATFORM_SCHEMA(
{
CONF_PLATFORM: "traccar",
CONF_HOST: "fake_host",
CONF_USERNAME: "fake_user",
CONF_PASSWORD: "fake_pass",
CONF_EVENT: ["all_events"],
}
)
}
device = {"id": 1, "name": "abc123"}
api_mock = AsyncMock()
api_mock.devices = [device]
api_mock.get_reports_events.return_value = [
ReportsEventeModel(
**{
"id": 1,
"positionId": 1,
"geofenceId": 1,
"maintenanceId": 1,
"deviceId": device["id"],
"type": "ignitionOn",
"eventTime": dt_util.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
"attributes": {},
}
),
ReportsEventeModel(
**{
"id": 2,
"positionId": 2,
"geofenceId": 1,
"maintenanceId": 1,
"deviceId": device["id"],
"type": "ignitionOff",
"eventTime": dt_util.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
"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.ApiClient",
return_value=api_mock,
):
assert await async_setup_component(hass, DOMAIN, conf_dict)
assert len(events_ignition_on) == 1
assert len(events_ignition_off) == 1

126
tests/components/traccar_server/test_config_flow.py
View File

@ -1,16 +1,19 @@
"""Test the Traccar Server config flow."""
from typing import Any
from unittest.mock import AsyncMock, patch
import pytest
from pytraccar import TraccarException
from homeassistant import config_entries
from homeassistant.components.traccar.device_tracker import PLATFORM_SCHEMA
from homeassistant.components.traccar_server.const import (
CONF_CUSTOM_ATTRIBUTES,
CONF_EVENTS,
CONF_MAX_ACCURACY,
CONF_SKIP_ACCURACY_FILTER_FOR,
DOMAIN,
EVENTS,
)
from homeassistant.const import (
CONF_HOST,
@ -156,6 +159,129 @@ async def test_options(
}
@pytest.mark.parametrize(
("imported", "data", "options"),
(
(
{
CONF_HOST: "1.1.1.1",
CONF_PORT: 443,
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
{
CONF_HOST: "1.1.1.1",
CONF_PORT: "443",
CONF_VERIFY_SSL: True,
CONF_SSL: False,
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
{
CONF_EVENTS: [],
CONF_CUSTOM_ATTRIBUTES: [],
CONF_SKIP_ACCURACY_FILTER_FOR: [],
CONF_MAX_ACCURACY: 0,
},
),
(
{
CONF_HOST: "1.1.1.1",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_SSL: True,
"event": ["device_online", "device_offline"],
},
{
CONF_HOST: "1.1.1.1",
CONF_PORT: "8082",
CONF_VERIFY_SSL: True,
CONF_SSL: True,
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
{
CONF_EVENTS: ["device_online", "device_offline"],
CONF_CUSTOM_ATTRIBUTES: [],
CONF_SKIP_ACCURACY_FILTER_FOR: [],
CONF_MAX_ACCURACY: 0,
},
),
(
{
CONF_HOST: "1.1.1.1",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_SSL: True,
"event": ["device_online", "device_offline", "all_events"],
},
{
CONF_HOST: "1.1.1.1",
CONF_PORT: "8082",
CONF_VERIFY_SSL: True,
CONF_SSL: True,
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
{
CONF_EVENTS: list(EVENTS.values()),
CONF_CUSTOM_ATTRIBUTES: [],
CONF_SKIP_ACCURACY_FILTER_FOR: [],
CONF_MAX_ACCURACY: 0,
},
),
),
)
async def test_import_from_yaml(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
imported: dict[str, Any],
data: dict[str, Any],
options: dict[str, Any],
) -> None:
"""Test importing configuration from YAML."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=PLATFORM_SCHEMA({"platform": "traccar", **imported}),
)
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == f"{data[CONF_HOST]}:{data[CONF_PORT]}"
assert result["data"] == data
assert result["options"] == options
async def test_abort_import_already_configured(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
) -> None:
"""Test abort for existing server while importing."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: "1.1.1.1", CONF_PORT: "8082"},
)
config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=PLATFORM_SCHEMA(
{
"platform": "traccar",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_HOST: "1.1.1.1",
CONF_PORT: "8082",
}
),
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_abort_already_configured(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,