Add navigation actions to Tesla Fleet (#183114)

Co-authored-by: Alex Whiteside <1505496+alexw23@users.noreply.github.com>
This commit is contained in:
Matti Hiljanen
2026-10-02 22:11:24 +01:00
committed by GitHub
co-authored by Alex Whiteside
parent b9b0a959bc
commit 6528860ce1
7 changed files with 370 additions and 6 deletions
@@ -33,6 +33,7 @@ from homeassistant.helpers.config_entry_oauth2_flow import (
async_get_config_entry_implementation,
)
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN, LOGGER
from .coordinator import (
@@ -44,6 +45,7 @@ from .coordinator import (
_stale_site_info_error,
)
from .models import TeslaFleetData, TeslaFleetEnergyData, TeslaFleetVehicleData
from .services import async_setup_services
PLATFORMS: Final = [
Platform.BINARY_SENSOR,
@@ -65,6 +67,12 @@ type TeslaFleetConfigEntry = ConfigEntry[TeslaFleetData]
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Tesla Fleet integration."""
async_setup_services(hass)
return True
async def _async_get_products(tesla: TeslaFleetApi) -> list[dict]:
"""Get products from Tesla Fleet API with region fallback handling."""
try:
@@ -332,5 +332,13 @@
"default": "mdi:speedometer-slow"
}
}
},
"services": {
"navigate_to_coordinates": {
"service": "mdi:crosshairs-gps"
},
"navigate_to_destination": {
"service": "mdi:map-marker"
}
}
}
@@ -1,8 +1,6 @@
rules:
# Bronze
action-setup:
status: exempt
comment: Integration does not register custom service actions in async_setup.
action-setup: done
appropriate-polling: done
brands: done
common-modules: done
@@ -13,9 +11,7 @@ rules:
status: todo
comment: Review feedback questioned whether the custom OAuth flow implementation can be simplified, including whether `CONFIG_SCHEMA` and the custom `OAuth2FlowHandler` are both needed.
dependency-transparency: done
docs-actions:
status: exempt
comment: Integration does not provide custom service actions beyond standard entity services.
docs-actions: done
docs-conditions:
status: exempt
comment: This integration does not have any conditions.
@@ -0,0 +1,106 @@
"""Service calls for the Tesla Fleet integration."""
from typing import TYPE_CHECKING
import probatio
from tesla_fleet_api.const import Scope
from homeassistant.const import CONF_DEVICE_ID, CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers import (
config_validation as cv,
device_registry as dr,
service,
)
from .const import DOMAIN
from .helpers import handle_vehicle_command, wake_up_vehicle
from .models import TeslaFleetVehicleData
if TYPE_CHECKING:
from . import TeslaFleetConfigEntry
# Attributes
ATTR_DESTINATION = "destination"
ATTR_GPS = "gps"
# Services
SERVICE_NAVIGATE_TO_DESTINATION = "navigate_to_destination"
SERVICE_NAVIGATE_TO_COORDINATES = "navigate_to_coordinates"
def async_get_vehicle_for_service_call(
hass: HomeAssistant, call: ServiceCall
) -> TeslaFleetVehicleData:
"""Get the vehicle targeted by a service call and ensure it accepts commands."""
device: dr.DeviceEntry
config: TeslaFleetConfigEntry
# Vehicles are matched by serial number, which only a main device has
device, config = service.async_get_device_and_config_entry(
hass, DOMAIN, call.data[CONF_DEVICE_ID], include_child_devices=False
)
if Scope.VEHICLE_CMDS not in config.runtime_data.scopes:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="missing_scope_vehicle_cmds",
)
for vehicle in config.runtime_data.vehicles:
if vehicle.vin == device.serial_number:
return vehicle
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="no_vehicle_data_for_device",
translation_placeholders={"device_id": device.id},
)
@callback
def async_setup_services(hass: HomeAssistant) -> None:
"""Set up the Tesla Fleet services."""
async def navigate_to_destination(call: ServiceCall) -> None:
"""Send an address, place name or map link to a vehicle."""
vehicle = async_get_vehicle_for_service_call(hass, call)
await wake_up_vehicle(vehicle)
await handle_vehicle_command(
vehicle.api.navigation_request(call.data[ATTR_DESTINATION])
)
hass.services.async_register(
DOMAIN,
SERVICE_NAVIGATE_TO_DESTINATION,
navigate_to_destination,
schema=probatio.Schema(
{
probatio.Required(CONF_DEVICE_ID): cv.string,
probatio.Required(ATTR_DESTINATION): cv.string,
}
),
)
async def navigate_to_coordinates(call: ServiceCall) -> None:
"""Send coordinates to a vehicle."""
vehicle = async_get_vehicle_for_service_call(hass, call)
await wake_up_vehicle(vehicle)
await handle_vehicle_command(
vehicle.api.navigation_gps_request(
lat=call.data[ATTR_GPS][CONF_LATITUDE],
lon=call.data[ATTR_GPS][CONF_LONGITUDE],
)
)
hass.services.async_register(
DOMAIN,
SERVICE_NAVIGATE_TO_COORDINATES,
navigate_to_coordinates,
schema=probatio.Schema(
{
probatio.Required(CONF_DEVICE_ID): cv.string,
probatio.Required(ATTR_GPS): {
probatio.Required(CONF_LATITUDE): cv.latitude,
probatio.Required(CONF_LONGITUDE): cv.longitude,
},
}
),
)
@@ -0,0 +1,28 @@
navigate_to_destination:
fields:
device_id:
required: true
selector:
device:
filter:
- integration: tesla_fleet
destination:
required: true
example: "1600 Amphitheatre Parkway, Mountain View, CA"
selector:
text:
navigate_to_coordinates:
fields:
device_id:
required: true
selector:
device:
filter:
- integration: tesla_fleet
gps:
required: true
example: '{"latitude": -27.9699373, "longitude": 153.4081865}'
selector:
location:
radius: false
@@ -634,6 +634,9 @@
"no_cable": {
"message": "Charge cable will lock automatically when connected"
},
"no_vehicle_data_for_device": {
"message": "No vehicle data for device ID: {device_id}"
},
"update_failed": {
"message": "{endpoint} data request failed: {message}"
},
@@ -651,5 +654,35 @@
"na": "North America & Asia-Pacific"
}
}
},
"services": {
"navigate_to_coordinates": {
"description": "Sets vehicle navigation to the provided latitude/longitude coordinates.",
"fields": {
"device_id": {
"description": "Vehicle to send the destination to.",
"name": "Vehicle"
},
"gps": {
"description": "Location to navigate to.",
"name": "Location"
}
},
"name": "Navigate to coordinates"
},
"navigate_to_destination": {
"description": "Sets vehicle navigation to the provided destination, which can be an address, a place name or a map link.",
"fields": {
"destination": {
"description": "Address, place name or map link to navigate to.",
"name": "Destination"
},
"device_id": {
"description": "Vehicle to send the destination to.",
"name": "Vehicle"
}
},
"name": "Navigate to destination"
}
}
}
@@ -0,0 +1,185 @@
"""Test the Tesla Fleet services."""
from unittest.mock import patch
import pytest
from homeassistant.components.tesla_fleet.const import DOMAIN
from homeassistant.components.tesla_fleet.services import (
ATTR_DESTINATION,
ATTR_GPS,
SERVICE_NAVIGATE_TO_COORDINATES,
SERVICE_NAVIGATE_TO_DESTINATION,
)
from homeassistant.const import CONF_DEVICE_ID, CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import device_registry as dr
from . import setup_platform
from .const import COMMAND_ERROR, COMMAND_OK, VEHICLE_DATA
from tests.common import MockConfigEntry
LAT = -27.9699373
LON = 153.3726526
def get_vehicle_device_id(
device_registry: dr.DeviceRegistry, config_entry: MockConfigEntry
) -> str:
"""Return the device registry ID of the test vehicle."""
device = device_registry.async_get_device_by_identifier(
(DOMAIN, VEHICLE_DATA["response"]["vin"]), config_entry.entry_id
)
assert device
return device.id
async def test_navigate_to_destination(
hass: HomeAssistant,
normal_config_entry: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test sending a destination to the vehicle."""
await setup_platform(hass, normal_config_entry)
with patch(
"tesla_fleet_api.tesla.VehicleFleet.navigation_request",
return_value=COMMAND_OK,
) as navigation_request:
await hass.services.async_call(
DOMAIN,
SERVICE_NAVIGATE_TO_DESTINATION,
{
CONF_DEVICE_ID: get_vehicle_device_id(
device_registry, normal_config_entry
),
ATTR_DESTINATION: "1600 Amphitheatre Parkway, Mountain View, CA",
},
blocking=True,
)
navigation_request.assert_called_once_with(
"1600 Amphitheatre Parkway, Mountain View, CA"
)
async def test_navigate_to_coordinates(
hass: HomeAssistant,
normal_config_entry: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test sending coordinates to the vehicle."""
await setup_platform(hass, normal_config_entry)
with patch(
"tesla_fleet_api.tesla.VehicleFleet.navigation_gps_request",
return_value=COMMAND_OK,
) as navigation_gps_request:
await hass.services.async_call(
DOMAIN,
SERVICE_NAVIGATE_TO_COORDINATES,
{
CONF_DEVICE_ID: get_vehicle_device_id(
device_registry, normal_config_entry
),
ATTR_GPS: {CONF_LATITUDE: LAT, CONF_LONGITUDE: LON},
},
blocking=True,
)
navigation_gps_request.assert_called_once_with(lat=LAT, lon=LON)
async def test_navigate_to_destination_command_error(
hass: HomeAssistant,
normal_config_entry: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test a command error from the vehicle is raised."""
await setup_platform(hass, normal_config_entry)
with (
patch(
"tesla_fleet_api.tesla.VehicleFleet.navigation_request",
return_value=COMMAND_ERROR,
),
pytest.raises(HomeAssistantError) as exc_info,
):
await hass.services.async_call(
DOMAIN,
SERVICE_NAVIGATE_TO_DESTINATION,
{
CONF_DEVICE_ID: get_vehicle_device_id(
device_registry, normal_config_entry
),
ATTR_DESTINATION: "Home",
},
blocking=True,
)
assert exc_info.value.translation_key == "command_error"
async def test_missing_vehicle_cmds_scope(
hass: HomeAssistant,
readonly_config_entry: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test services refuse to run without the vehicle commands scope."""
await setup_platform(hass, readonly_config_entry)
with pytest.raises(ServiceValidationError) as exc_info:
await hass.services.async_call(
DOMAIN,
SERVICE_NAVIGATE_TO_DESTINATION,
{
CONF_DEVICE_ID: get_vehicle_device_id(
device_registry, readonly_config_entry
),
ATTR_DESTINATION: "Home",
},
blocking=True,
)
assert exc_info.value.translation_key == "missing_scope_vehicle_cmds"
async def test_energy_site_device(
hass: HomeAssistant,
normal_config_entry: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test services reject a device that is not a vehicle."""
await setup_platform(hass, normal_config_entry)
energy_site = normal_config_entry.runtime_data.energysites[0]
device = device_registry.async_get_device_by_identifier(
(DOMAIN, str(energy_site.id)), normal_config_entry.entry_id
)
assert device
with pytest.raises(ServiceValidationError) as exc_info:
await hass.services.async_call(
DOMAIN,
SERVICE_NAVIGATE_TO_DESTINATION,
{CONF_DEVICE_ID: device.id, ATTR_DESTINATION: "Home"},
blocking=True,
)
assert exc_info.value.translation_key == "no_vehicle_data_for_device"
async def test_unknown_device(
hass: HomeAssistant, normal_config_entry: MockConfigEntry
) -> None:
"""Test services reject an unknown device."""
await setup_platform(hass, normal_config_entry)
with pytest.raises(ServiceValidationError) as exc_info:
await hass.services.async_call(
DOMAIN,
SERVICE_NAVIGATE_TO_COORDINATES,
{
CONF_DEVICE_ID: "nope",
ATTR_GPS: {CONF_LATITUDE: LAT, CONF_LONGITUDE: LON},
},
blocking=True,
)
assert exc_info.value.translation_key == "service_device_not_found"