Handle the new JSON payload from traccar clients (#147254)

This commit is contained in:
Joakim Sørensen 2025-06-21 10:53:17 +01:00 committed by GitHub
parent 7442f7af28
commit 79a9f34150
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 82 additions and 8 deletions

View File

@ -1,9 +1,12 @@
"""Support for Traccar Client.""" """Support for Traccar Client."""
from http import HTTPStatus from http import HTTPStatus
from json import JSONDecodeError
import logging
from aiohttp import web from aiohttp import web
import voluptuous as vol import voluptuous as vol
from voluptuous.humanize import humanize_error
from homeassistant.components import webhook from homeassistant.components import webhook
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
@ -20,7 +23,6 @@ from .const import (
ATTR_LATITUDE, ATTR_LATITUDE,
ATTR_LONGITUDE, ATTR_LONGITUDE,
ATTR_SPEED, ATTR_SPEED,
ATTR_TIMESTAMP,
DOMAIN, DOMAIN,
) )
@ -29,6 +31,7 @@ PLATFORMS = [Platform.DEVICE_TRACKER]
TRACKER_UPDATE = f"{DOMAIN}_tracker_update" TRACKER_UPDATE = f"{DOMAIN}_tracker_update"
LOGGER = logging.getLogger(__name__)
DEFAULT_ACCURACY = 200 DEFAULT_ACCURACY = 200
DEFAULT_BATTERY = -1 DEFAULT_BATTERY = -1
@ -49,21 +52,50 @@ WEBHOOK_SCHEMA = vol.Schema(
vol.Optional(ATTR_BATTERY, default=DEFAULT_BATTERY): vol.Coerce(float), vol.Optional(ATTR_BATTERY, default=DEFAULT_BATTERY): vol.Coerce(float),
vol.Optional(ATTR_BEARING): vol.Coerce(float), vol.Optional(ATTR_BEARING): vol.Coerce(float),
vol.Optional(ATTR_SPEED): vol.Coerce(float), vol.Optional(ATTR_SPEED): vol.Coerce(float),
vol.Optional(ATTR_TIMESTAMP): vol.Coerce(int),
}, },
extra=vol.REMOVE_EXTRA, extra=vol.REMOVE_EXTRA,
) )
def _parse_json_body(json_body: dict) -> dict:
"""Parse JSON body from request."""
location = json_body.get("location", {})
coords = location.get("coords", {})
battery_level = location.get("battery", {}).get("level")
return {
"id": json_body.get("device_id"),
"lat": coords.get("latitude"),
"lon": coords.get("longitude"),
"accuracy": coords.get("accuracy"),
"altitude": coords.get("altitude"),
"batt": battery_level * 100 if battery_level is not None else DEFAULT_BATTERY,
"bearing": coords.get("heading"),
"speed": coords.get("speed"),
}
async def handle_webhook( async def handle_webhook(
hass: HomeAssistant, webhook_id: str, request: web.Request hass: HomeAssistant,
webhook_id: str,
request: web.Request,
) -> web.Response: ) -> web.Response:
"""Handle incoming webhook with Traccar Client request.""" """Handle incoming webhook with Traccar Client request."""
if not (requestdata := dict(request.query)):
try: try:
data = WEBHOOK_SCHEMA(dict(request.query)) requestdata = _parse_json_body(await request.json())
except vol.MultipleInvalid as error: except JSONDecodeError as error:
LOGGER.error("Error parsing JSON body: %s", error)
return web.Response( return web.Response(
text=error.error_message, status=HTTPStatus.UNPROCESSABLE_ENTITY text="Invalid JSON",
status=HTTPStatus.UNPROCESSABLE_ENTITY,
)
try:
data = WEBHOOK_SCHEMA(requestdata)
except vol.MultipleInvalid as error:
LOGGER.warning(humanize_error(requestdata, error))
return web.Response(
text=error.error_message,
status=HTTPStatus.UNPROCESSABLE_ENTITY,
) )
attrs = { attrs = {

View File

@ -17,7 +17,6 @@ ATTR_LONGITUDE = "lon"
ATTR_MOTION = "motion" ATTR_MOTION = "motion"
ATTR_SPEED = "speed" ATTR_SPEED = "speed"
ATTR_STATUS = "status" ATTR_STATUS = "status"
ATTR_TIMESTAMP = "timestamp"
ATTR_TRACKER = "tracker" ATTR_TRACKER = "tracker"
ATTR_TRACCAR_ID = "traccar_id" ATTR_TRACCAR_ID = "traccar_id"

View File

@ -146,8 +146,12 @@ async def test_enter_and_exit(
assert len(entity_registry.entities) == 1 assert len(entity_registry.entities) == 1
async def test_enter_with_attrs(hass: HomeAssistant, client, webhook_id) -> None: async def test_enter_with_attrs_as_query(
"""Test when additional attributes are present.""" hass: HomeAssistant,
client,
webhook_id,
) -> None:
"""Test when additional attributes are present URL query."""
url = f"/api/webhook/{webhook_id}" url = f"/api/webhook/{webhook_id}"
data = { data = {
"timestamp": 123456789, "timestamp": 123456789,
@ -197,6 +201,45 @@ async def test_enter_with_attrs(hass: HomeAssistant, client, webhook_id) -> None
assert state.attributes["altitude"] == 123 assert state.attributes["altitude"] == 123
async def test_enter_with_attrs_as_payload(
hass: HomeAssistant, client, webhook_id
) -> None:
"""Test when additional attributes are present in JSON payload."""
url = f"/api/webhook/{webhook_id}"
data = {
"location": {
"coords": {
"heading": "105.32",
"latitude": "1.0",
"longitude": "1.1",
"accuracy": 10.5,
"altitude": 102.0,
"speed": 100.0,
},
"extras": {},
"manual": True,
"is_moving": False,
"_": "&id=123&lat=1.0&lon=1.1&timestamp=2013-09-17T07:32:51Z&",
"odometer": 0,
"activity": {"type": "still"},
"timestamp": "2013-09-17T07:32:51Z",
"battery": {"level": 0.1, "is_charging": False},
},
"device_id": "123",
}
req = await client.post(url, json=data)
await hass.async_block_till_done()
assert req.status == HTTPStatus.OK
state = hass.states.get(f"{DEVICE_TRACKER_DOMAIN}.{data['device_id']}")
assert state.state == STATE_NOT_HOME
assert state.attributes["gps_accuracy"] == 10.5
assert state.attributes["battery_level"] == 10.0
assert state.attributes["speed"] == 100.0
assert state.attributes["bearing"] == 105.32
assert state.attributes["altitude"] == 102.0
async def test_two_devices(hass: HomeAssistant, client, webhook_id) -> None: async def test_two_devices(hass: HomeAssistant, client, webhook_id) -> None:
"""Test updating two different devices.""" """Test updating two different devices."""
url = f"/api/webhook/{webhook_id}" url = f"/api/webhook/{webhook_id}"