1
0
mirror of https://github.com/home-assistant/core.git synced 2026-05-08 17:49:37 +01:00

add services to google_travel_time (#160740)

Co-authored-by: Joostlek <joostlek@outlook.com>
This commit is contained in:
Kevin Stillhammer
2026-02-12 22:31:36 +01:00
committed by GitHub
parent 4801dcaded
commit acf739df81
12 changed files with 951 additions and 179 deletions
@@ -5,14 +5,25 @@ import logging
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import ConfigType
from homeassistant.util import dt as dt_util
from .const import CONF_TIME
from .const import CONF_TIME, DOMAIN
from .services import async_setup_services
PLATFORMS = [Platform.SENSOR]
_LOGGER = logging.getLogger(__name__)
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Google Travel Time component."""
async_setup_services(hass)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Google Maps Travel Time from a config entry."""
@@ -24,9 +24,7 @@ from homeassistant.helpers.selector import (
from homeassistant.util.unit_system import US_CUSTOMARY_SYSTEM
from .const import (
ALL_LANGUAGES,
ARRIVAL_TIME,
AVOID_OPTIONS,
CONF_ARRIVAL_TIME,
CONF_AVOID,
CONF_DEPARTURE_TIME,
@@ -41,12 +39,7 @@ from .const import (
DEFAULT_NAME,
DEPARTURE_TIME,
DOMAIN,
TIME_TYPES,
TRAFFIC_MODELS,
TRANSIT_PREFS,
TRANSPORT_TYPES,
TRAVEL_MODES,
UNITS,
UNITS_IMPERIAL,
UNITS_METRIC,
)
@@ -56,6 +49,15 @@ from .helpers import (
UnknownException,
validate_config_entry,
)
from .schemas import (
AVOID_SELECTOR,
LANGUAGE_SELECTOR,
TIME_TYPE_SELECTOR,
TRAFFIC_MODEL_SELECTOR,
TRANSIT_MODE_SELECTOR,
TRANSIT_ROUTING_PREFERENCE_SELECTOR,
UNITS_SELECTOR,
)
RECONFIGURE_SCHEMA = vol.Schema(
{
@@ -73,6 +75,13 @@ CONFIG_SCHEMA = RECONFIGURE_SCHEMA.extend(
OPTIONS_SCHEMA = vol.Schema(
{
vol.Optional(CONF_LANGUAGE): LANGUAGE_SELECTOR,
vol.Optional(CONF_AVOID): AVOID_SELECTOR,
vol.Optional(CONF_TRAFFIC_MODEL): TRAFFIC_MODEL_SELECTOR,
vol.Optional(CONF_TRANSIT_MODE): TRANSIT_MODE_SELECTOR,
vol.Optional(
CONF_TRANSIT_ROUTING_PREFERENCE
): TRANSIT_ROUTING_PREFERENCE_SELECTOR,
vol.Required(CONF_MODE): SelectSelector(
SelectSelectorConfig(
options=TRAVEL_MODES,
@@ -81,62 +90,9 @@ OPTIONS_SCHEMA = vol.Schema(
translation_key=CONF_MODE,
)
),
vol.Optional(CONF_LANGUAGE): SelectSelector(
SelectSelectorConfig(
options=sorted(ALL_LANGUAGES),
mode=SelectSelectorMode.DROPDOWN,
translation_key=CONF_LANGUAGE,
)
),
vol.Optional(CONF_AVOID): SelectSelector(
SelectSelectorConfig(
options=AVOID_OPTIONS,
sort=True,
mode=SelectSelectorMode.DROPDOWN,
translation_key=CONF_AVOID,
)
),
vol.Required(CONF_UNITS): SelectSelector(
SelectSelectorConfig(
options=UNITS,
sort=True,
mode=SelectSelectorMode.DROPDOWN,
translation_key=CONF_UNITS,
)
),
vol.Required(CONF_TIME_TYPE): SelectSelector(
SelectSelectorConfig(
options=TIME_TYPES,
sort=True,
mode=SelectSelectorMode.DROPDOWN,
translation_key=CONF_TIME_TYPE,
)
),
vol.Required(CONF_UNITS): UNITS_SELECTOR,
vol.Required(CONF_TIME_TYPE): TIME_TYPE_SELECTOR,
vol.Optional(CONF_TIME): TimeSelector(),
vol.Optional(CONF_TRAFFIC_MODEL): SelectSelector(
SelectSelectorConfig(
options=TRAFFIC_MODELS,
sort=True,
mode=SelectSelectorMode.DROPDOWN,
translation_key=CONF_TRAFFIC_MODEL,
)
),
vol.Optional(CONF_TRANSIT_MODE): SelectSelector(
SelectSelectorConfig(
options=TRANSPORT_TYPES,
sort=True,
mode=SelectSelectorMode.DROPDOWN,
translation_key=CONF_TRANSIT_MODE,
)
),
vol.Optional(CONF_TRANSIT_ROUTING_PREFERENCE): SelectSelector(
SelectSelectorConfig(
options=TRANSIT_PREFS,
sort=True,
mode=SelectSelectorMode.DROPDOWN,
translation_key=CONF_TRANSIT_ROUTING_PREFERENCE,
)
),
}
)
@@ -98,6 +98,7 @@ TRANSPORT_TYPES_TO_GOOGLE_SDK_ENUM = {
"rail": TransitPreferences.TransitTravelMode.RAIL,
}
TRAVEL_MODES = ["driving", "walking", "bicycling", "transit"]
TRAVEL_MODES_WITHOUT_TRANSIT = ["driving", "walking", "bicycling"]
TRAVEL_MODES_TO_GOOGLE_SDK_ENUM = {
"driving": RouteTravelMode.DRIVE,
"walking": RouteTravelMode.WALK,
@@ -1,5 +1,6 @@
"""Helpers for Google Time Travel integration."""
import datetime
import logging
from google.api_core.client_options import ClientOptions
@@ -12,11 +13,16 @@ from google.api_core.exceptions import (
)
from google.maps.routing_v2 import (
ComputeRoutesRequest,
ComputeRoutesResponse,
Location,
RouteModifiers,
RoutesAsyncClient,
RouteTravelMode,
RoutingPreference,
TransitPreferences,
Waypoint,
)
from google.protobuf import timestamp_pb2
from google.type import latlng_pb2
import voluptuous as vol
@@ -29,12 +35,40 @@ from homeassistant.helpers.issue_registry import (
async_delete_issue,
)
from homeassistant.helpers.location import find_coordinates
from homeassistant.util import dt as dt_util
from .const import DOMAIN
from .const import (
DOMAIN,
TRAFFIC_MODELS_TO_GOOGLE_SDK_ENUM,
TRANSIT_PREFS_TO_GOOGLE_SDK_ENUM,
TRANSPORT_TYPES_TO_GOOGLE_SDK_ENUM,
UNITS_TO_GOOGLE_SDK_ENUM,
)
_LOGGER = logging.getLogger(__name__)
def convert_time(time_str: str) -> timestamp_pb2.Timestamp:
"""Convert a string like '08:00' to a google pb2 Timestamp.
If the time is in the past, it will be shifted to the next day.
"""
parsed_time = dt_util.parse_time(time_str)
if parsed_time is None:
raise ValueError(f"Invalid time format: {time_str}")
start_of_day = dt_util.start_of_local_day()
combined = datetime.datetime.combine(
start_of_day,
parsed_time,
start_of_day.tzinfo,
)
if combined < dt_util.now():
combined = combined + datetime.timedelta(days=1)
timestamp = timestamp_pb2.Timestamp()
timestamp.FromDatetime(dt=combined)
return timestamp
def convert_to_waypoint(hass: HomeAssistant, location: str) -> Waypoint | None:
"""Convert a location to a Waypoint.
@@ -123,3 +157,78 @@ def create_routes_api_disabled_issue(hass: HomeAssistant, entry: ConfigEntry) ->
def delete_routes_api_disabled_issue(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Delete the issue for the Routes API being disabled."""
async_delete_issue(hass, DOMAIN, f"routes_api_disabled_{entry.entry_id}")
async def async_compute_routes(
client: RoutesAsyncClient,
origin: str,
destination: str,
hass: HomeAssistant,
travel_mode: int,
units: str,
language: str | None = None,
avoid: str | None = None,
traffic_model: str | None = None,
transit_mode: str | None = None,
transit_routing_preference: str | None = None,
departure_time: str | None = None,
arrival_time: str | None = None,
field_mask: str = "routes.duration,routes.distanceMeters,routes.localized_values",
) -> ComputeRoutesResponse | None:
"""Compute routes using Google Routes API."""
origin_waypoint = convert_to_waypoint(hass, origin)
destination_waypoint = convert_to_waypoint(hass, destination)
if origin_waypoint is None or destination_waypoint is None:
return None
route_modifiers = None
routing_preference = None
if travel_mode == RouteTravelMode.DRIVE:
routing_preference = RoutingPreference.TRAFFIC_AWARE_OPTIMAL
route_modifiers = RouteModifiers(
avoid_tolls=avoid == "tolls",
avoid_ferries=avoid == "ferries",
avoid_highways=avoid == "highways",
avoid_indoor=avoid == "indoor",
)
transit_preferences = None
if travel_mode == RouteTravelMode.TRANSIT:
transit_routing_pref = None
transit_travel_mode = (
TransitPreferences.TransitTravelMode.TRANSIT_TRAVEL_MODE_UNSPECIFIED
)
if transit_routing_preference is not None:
transit_routing_pref = TRANSIT_PREFS_TO_GOOGLE_SDK_ENUM[
transit_routing_preference
]
if transit_mode is not None:
transit_travel_mode = TRANSPORT_TYPES_TO_GOOGLE_SDK_ENUM[transit_mode]
transit_preferences = TransitPreferences(
routing_preference=transit_routing_pref,
allowed_travel_modes=[transit_travel_mode],
)
departure_timestamp = convert_time(departure_time) if departure_time else None
arrival_timestamp = convert_time(arrival_time) if arrival_time else None
request = ComputeRoutesRequest(
origin=origin_waypoint,
destination=destination_waypoint,
travel_mode=travel_mode,
routing_preference=routing_preference,
departure_time=departure_timestamp,
arrival_time=arrival_timestamp,
route_modifiers=route_modifiers,
language_code=language,
units=UNITS_TO_GOOGLE_SDK_ENUM[units],
traffic_model=TRAFFIC_MODELS_TO_GOOGLE_SDK_ENUM[traffic_model]
if traffic_model
else None,
transit_preferences=transit_preferences,
)
return await client.compute_routes(
request, metadata=[("x-goog-fieldmask", field_mask)]
)
@@ -0,0 +1,10 @@
{
"services": {
"get_transit_times": {
"service": "mdi:bus"
},
"get_travel_times": {
"service": "mdi:routes"
}
}
}
@@ -0,0 +1,137 @@
"""Schemas for the Google Travel Time integration."""
import voluptuous as vol
from homeassistant.const import ATTR_CONFIG_ENTRY_ID, CONF_LANGUAGE, CONF_MODE
from homeassistant.helpers.selector import (
ConfigEntrySelector,
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
TextSelector,
TimeSelector,
)
from .const import (
ALL_LANGUAGES,
AVOID_OPTIONS,
CONF_ARRIVAL_TIME,
CONF_AVOID,
CONF_DEPARTURE_TIME,
CONF_DESTINATION,
CONF_ORIGIN,
CONF_TIME_TYPE,
CONF_TRAFFIC_MODEL,
CONF_TRANSIT_MODE,
CONF_TRANSIT_ROUTING_PREFERENCE,
CONF_UNITS,
DOMAIN,
TIME_TYPES,
TRAFFIC_MODELS,
TRANSIT_PREFS,
TRANSPORT_TYPES,
TRAVEL_MODES_WITHOUT_TRANSIT,
UNITS,
UNITS_METRIC,
)
LANGUAGE_SELECTOR = SelectSelector(
SelectSelectorConfig(
options=sorted(ALL_LANGUAGES),
mode=SelectSelectorMode.DROPDOWN,
translation_key=CONF_LANGUAGE,
)
)
AVOID_SELECTOR = SelectSelector(
SelectSelectorConfig(
options=AVOID_OPTIONS,
sort=True,
mode=SelectSelectorMode.DROPDOWN,
translation_key=CONF_AVOID,
)
)
TRAFFIC_MODEL_SELECTOR = SelectSelector(
SelectSelectorConfig(
options=TRAFFIC_MODELS,
sort=True,
mode=SelectSelectorMode.DROPDOWN,
translation_key=CONF_TRAFFIC_MODEL,
)
)
TRANSIT_MODE_SELECTOR = SelectSelector(
SelectSelectorConfig(
options=TRANSPORT_TYPES,
sort=True,
mode=SelectSelectorMode.DROPDOWN,
translation_key=CONF_TRANSIT_MODE,
)
)
TRANSIT_ROUTING_PREFERENCE_SELECTOR = SelectSelector(
SelectSelectorConfig(
options=TRANSIT_PREFS,
sort=True,
mode=SelectSelectorMode.DROPDOWN,
translation_key=CONF_TRANSIT_ROUTING_PREFERENCE,
)
)
UNITS_SELECTOR = SelectSelector(
SelectSelectorConfig(
options=UNITS,
sort=True,
mode=SelectSelectorMode.DROPDOWN,
translation_key=CONF_UNITS,
)
)
TIME_TYPE_SELECTOR = SelectSelector(
SelectSelectorConfig(
options=TIME_TYPES,
sort=True,
mode=SelectSelectorMode.DROPDOWN,
translation_key=CONF_TIME_TYPE,
)
)
_SERVICE_BASE_SCHEMA = vol.Schema(
{
vol.Required(ATTR_CONFIG_ENTRY_ID): ConfigEntrySelector(
{"integration": DOMAIN}
),
vol.Required(CONF_ORIGIN): TextSelector(),
vol.Required(CONF_DESTINATION): TextSelector(),
vol.Optional(CONF_UNITS, default=UNITS_METRIC): UNITS_SELECTOR,
vol.Optional(CONF_LANGUAGE): LANGUAGE_SELECTOR,
}
)
SERVICE_GET_TRAVEL_TIMES_SCHEMA = _SERVICE_BASE_SCHEMA.extend(
{
vol.Optional(CONF_MODE, default="driving"): SelectSelector(
SelectSelectorConfig(
options=TRAVEL_MODES_WITHOUT_TRANSIT,
sort=True,
mode=SelectSelectorMode.DROPDOWN,
translation_key=CONF_MODE,
)
),
vol.Optional(CONF_AVOID): AVOID_SELECTOR,
vol.Optional(CONF_TRAFFIC_MODEL): TRAFFIC_MODEL_SELECTOR,
vol.Optional(CONF_DEPARTURE_TIME): TimeSelector(),
}
)
SERVICE_GET_TRANSIT_TIMES_SCHEMA = _SERVICE_BASE_SCHEMA.extend(
{
vol.Optional(CONF_TRANSIT_MODE): TRANSIT_MODE_SELECTOR,
vol.Optional(
CONF_TRANSIT_ROUTING_PREFERENCE
): TRANSIT_ROUTING_PREFERENCE_SELECTOR,
vol.Exclusive(CONF_DEPARTURE_TIME, "time"): TimeSelector(),
vol.Exclusive(CONF_ARRIVAL_TIME, "time"): TimeSelector(),
}
)
@@ -4,20 +4,11 @@ from __future__ import annotations
import datetime
import logging
from typing import TYPE_CHECKING, Any
from typing import Any
from google.api_core.client_options import ClientOptions
from google.api_core.exceptions import GoogleAPIError, PermissionDenied
from google.maps.routing_v2 import (
ComputeRoutesRequest,
Route,
RouteModifiers,
RoutesAsyncClient,
RouteTravelMode,
RoutingPreference,
TransitPreferences,
)
from google.protobuf import timestamp_pb2
from google.maps.routing_v2 import Route, RoutesAsyncClient
from homeassistant.components.sensor import (
SensorDeviceClass,
@@ -38,7 +29,6 @@ from homeassistant.core import CoreState, HomeAssistant
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.location import find_coordinates
from homeassistant.util import dt as dt_util
from .const import (
ATTRIBUTION,
@@ -53,14 +43,10 @@ from .const import (
CONF_UNITS,
DEFAULT_NAME,
DOMAIN,
TRAFFIC_MODELS_TO_GOOGLE_SDK_ENUM,
TRANSIT_PREFS_TO_GOOGLE_SDK_ENUM,
TRANSPORT_TYPES_TO_GOOGLE_SDK_ENUM,
TRAVEL_MODES_TO_GOOGLE_SDK_ENUM,
UNITS_TO_GOOGLE_SDK_ENUM,
)
from .helpers import (
convert_to_waypoint,
async_compute_routes,
create_routes_api_disabled_issue,
delete_routes_api_disabled_issue,
)
@@ -70,28 +56,6 @@ _LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = datetime.timedelta(minutes=10)
FIELD_MASK = "routes.duration,routes.localized_values"
def convert_time(time_str: str) -> timestamp_pb2.Timestamp | None:
"""Convert a string like '08:00' to a google pb2 Timestamp.
If the time is in the past, it will be shifted to the next day.
"""
parsed_time = dt_util.parse_time(time_str)
if TYPE_CHECKING:
assert parsed_time is not None
start_of_day = dt_util.start_of_local_day()
combined = datetime.datetime.combine(
start_of_day,
parsed_time,
start_of_day.tzinfo,
)
if combined < dt_util.now():
combined = combined + datetime.timedelta(days=1)
timestamp = timestamp_pb2.Timestamp()
timestamp.FromDatetime(dt=combined)
return timestamp
SENSOR_DESCRIPTIONS = [
SensorEntityDescription(
key="duration",
@@ -203,67 +167,6 @@ class GoogleTravelTimeSensor(SensorEntity):
self._config_entry.options[CONF_MODE]
]
if (
departure_time := self._config_entry.options.get(CONF_DEPARTURE_TIME)
) is not None:
departure_time = convert_time(departure_time)
if (
arrival_time := self._config_entry.options.get(CONF_ARRIVAL_TIME)
) is not None:
arrival_time = convert_time(arrival_time)
if travel_mode != RouteTravelMode.TRANSIT:
arrival_time = None
traffic_model = None
routing_preference = None
route_modifiers = None
if travel_mode == RouteTravelMode.DRIVE:
if (
options_traffic_model := self._config_entry.options.get(
CONF_TRAFFIC_MODEL
)
) is not None:
traffic_model = TRAFFIC_MODELS_TO_GOOGLE_SDK_ENUM[options_traffic_model]
routing_preference = RoutingPreference.TRAFFIC_AWARE_OPTIMAL
route_modifiers = RouteModifiers(
avoid_tolls=self._config_entry.options.get(CONF_AVOID) == "tolls",
avoid_ferries=self._config_entry.options.get(CONF_AVOID) == "ferries",
avoid_highways=self._config_entry.options.get(CONF_AVOID) == "highways",
avoid_indoor=self._config_entry.options.get(CONF_AVOID) == "indoor",
)
transit_preferences = None
if travel_mode == RouteTravelMode.TRANSIT:
transit_routing_preference = None
transit_travel_mode = (
TransitPreferences.TransitTravelMode.TRANSIT_TRAVEL_MODE_UNSPECIFIED
)
if (
option_transit_preferences := self._config_entry.options.get(
CONF_TRANSIT_ROUTING_PREFERENCE
)
) is not None:
transit_routing_preference = TRANSIT_PREFS_TO_GOOGLE_SDK_ENUM[
option_transit_preferences
]
if (
option_transit_mode := self._config_entry.options.get(CONF_TRANSIT_MODE)
) is not None:
transit_travel_mode = TRANSPORT_TYPES_TO_GOOGLE_SDK_ENUM[
option_transit_mode
]
transit_preferences = TransitPreferences(
routing_preference=transit_routing_preference,
allowed_travel_modes=[transit_travel_mode],
)
language = None
if (
options_language := self._config_entry.options.get(CONF_LANGUAGE)
) is not None:
language = options_language
self._resolved_origin = find_coordinates(self.hass, self._origin)
self._resolved_destination = find_coordinates(self.hass, self._destination)
_LOGGER.debug(
@@ -272,22 +175,24 @@ class GoogleTravelTimeSensor(SensorEntity):
self._resolved_destination,
)
if self._resolved_destination is not None and self._resolved_origin is not None:
request = ComputeRoutesRequest(
origin=convert_to_waypoint(self.hass, self._resolved_origin),
destination=convert_to_waypoint(self.hass, self._resolved_destination),
travel_mode=travel_mode,
routing_preference=routing_preference,
departure_time=departure_time,
arrival_time=arrival_time,
route_modifiers=route_modifiers,
language_code=language,
units=UNITS_TO_GOOGLE_SDK_ENUM[self._config_entry.options[CONF_UNITS]],
traffic_model=traffic_model,
transit_preferences=transit_preferences,
)
try:
response = await self._client.compute_routes(
request, metadata=[("x-goog-fieldmask", FIELD_MASK)]
response = await async_compute_routes(
client=self._client,
origin=self._resolved_origin,
destination=self._resolved_destination,
hass=self.hass,
travel_mode=travel_mode,
units=self._config_entry.options[CONF_UNITS],
language=self._config_entry.options.get(CONF_LANGUAGE),
avoid=self._config_entry.options.get(CONF_AVOID),
traffic_model=self._config_entry.options.get(CONF_TRAFFIC_MODEL),
transit_mode=self._config_entry.options.get(CONF_TRANSIT_MODE),
transit_routing_preference=self._config_entry.options.get(
CONF_TRANSIT_ROUTING_PREFERENCE
),
departure_time=self._config_entry.options.get(CONF_DEPARTURE_TIME),
arrival_time=self._config_entry.options.get(CONF_ARRIVAL_TIME),
field_mask=FIELD_MASK,
)
_LOGGER.debug("Received response: %s", response)
if response is not None and len(response.routes) > 0:
@@ -0,0 +1,167 @@
"""Services for the Google Travel Time integration."""
from typing import cast
from google.api_core.client_options import ClientOptions
from google.api_core.exceptions import GoogleAPIError, PermissionDenied
from google.maps.routing_v2 import RoutesAsyncClient
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_CONFIG_ENTRY_ID,
CONF_API_KEY,
CONF_LANGUAGE,
CONF_MODE,
)
from homeassistant.core import (
HomeAssistant,
ServiceCall,
ServiceResponse,
SupportsResponse,
callback,
)
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.service import async_get_config_entry
from .const import (
CONF_ARRIVAL_TIME,
CONF_AVOID,
CONF_DEPARTURE_TIME,
CONF_DESTINATION,
CONF_ORIGIN,
CONF_TRAFFIC_MODEL,
CONF_TRANSIT_MODE,
CONF_TRANSIT_ROUTING_PREFERENCE,
CONF_UNITS,
DOMAIN,
TRAVEL_MODES_TO_GOOGLE_SDK_ENUM,
)
from .helpers import (
async_compute_routes,
create_routes_api_disabled_issue,
delete_routes_api_disabled_issue,
)
from .schemas import SERVICE_GET_TRANSIT_TIMES_SCHEMA, SERVICE_GET_TRAVEL_TIMES_SCHEMA
SERVICE_GET_TRAVEL_TIMES = "get_travel_times"
SERVICE_GET_TRANSIT_TIMES = "get_transit_times"
def _build_routes_response(response) -> list[dict]:
"""Build the routes response from the API response."""
if response is None or not response.routes:
return []
return [
{
"duration": route.duration.seconds,
"duration_text": route.localized_values.duration.text,
"static_duration_text": route.localized_values.static_duration.text,
"distance_meters": route.distance_meters,
"distance_text": route.localized_values.distance.text,
}
for route in response.routes
]
def _raise_service_error(
hass: HomeAssistant, entry: ConfigEntry, exc: Exception
) -> None:
"""Raise a HomeAssistantError based on the exception."""
if isinstance(exc, PermissionDenied):
create_routes_api_disabled_issue(hass, entry)
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="permission_denied",
) from exc
if isinstance(exc, GoogleAPIError):
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="api_error",
translation_placeholders={"error": str(exc)},
) from exc
raise exc
@callback
def async_setup_services(hass: HomeAssistant) -> None:
"""Set up services for the Google Travel Time integration."""
async def async_get_travel_times_service(service: ServiceCall) -> ServiceResponse:
"""Handle the service call to get travel times (non-transit modes)."""
entry = async_get_config_entry(
service.hass, DOMAIN, service.data[ATTR_CONFIG_ENTRY_ID]
)
api_key = entry.data[CONF_API_KEY]
travel_mode = TRAVEL_MODES_TO_GOOGLE_SDK_ENUM[service.data[CONF_MODE]]
client_options = ClientOptions(api_key=api_key)
client = RoutesAsyncClient(client_options=client_options)
try:
response = await async_compute_routes(
client=client,
origin=service.data[CONF_ORIGIN],
destination=service.data[CONF_DESTINATION],
hass=hass,
travel_mode=travel_mode,
units=service.data[CONF_UNITS],
language=service.data.get(CONF_LANGUAGE),
avoid=service.data.get(CONF_AVOID),
traffic_model=service.data.get(CONF_TRAFFIC_MODEL),
departure_time=service.data.get(CONF_DEPARTURE_TIME),
)
except Exception as ex: # noqa: BLE001
_raise_service_error(hass, entry, ex)
delete_routes_api_disabled_issue(hass, entry)
return cast(ServiceResponse, {"routes": _build_routes_response(response)})
async def async_get_transit_times_service(service: ServiceCall) -> ServiceResponse:
"""Handle the service call to get transit times."""
entry = async_get_config_entry(
service.hass, DOMAIN, service.data[ATTR_CONFIG_ENTRY_ID]
)
api_key = entry.data[CONF_API_KEY]
client_options = ClientOptions(api_key=api_key)
client = RoutesAsyncClient(client_options=client_options)
try:
response = await async_compute_routes(
client=client,
origin=service.data[CONF_ORIGIN],
destination=service.data[CONF_DESTINATION],
hass=hass,
travel_mode=TRAVEL_MODES_TO_GOOGLE_SDK_ENUM["transit"],
units=service.data[CONF_UNITS],
language=service.data.get(CONF_LANGUAGE),
transit_mode=service.data.get(CONF_TRANSIT_MODE),
transit_routing_preference=service.data.get(
CONF_TRANSIT_ROUTING_PREFERENCE
),
departure_time=service.data.get(CONF_DEPARTURE_TIME),
arrival_time=service.data.get(CONF_ARRIVAL_TIME),
)
except Exception as ex: # noqa: BLE001
_raise_service_error(hass, entry, ex)
delete_routes_api_disabled_issue(hass, entry)
return cast(ServiceResponse, {"routes": _build_routes_response(response)})
hass.services.async_register(
DOMAIN,
SERVICE_GET_TRAVEL_TIMES,
async_get_travel_times_service,
SERVICE_GET_TRAVEL_TIMES_SCHEMA,
supports_response=SupportsResponse.ONLY,
)
hass.services.async_register(
DOMAIN,
SERVICE_GET_TRANSIT_TIMES,
async_get_transit_times_service,
SERVICE_GET_TRANSIT_TIMES_SCHEMA,
supports_response=SupportsResponse.ONLY,
)
@@ -0,0 +1,118 @@
get_travel_times:
fields:
config_entry_id:
required: true
selector:
config_entry:
integration: google_travel_time
origin:
required: true
example: "1600 Amphitheatre Parkway, Mountain View, CA"
selector:
text:
destination:
required: true
example: "1 Infinite Loop, Cupertino, CA"
selector:
text:
mode:
default: "driving"
selector:
select:
translation_key: mode
options:
- driving
- walking
- bicycling
units:
default: "metric"
selector:
select:
translation_key: units
options:
- metric
- imperial
language:
required: false
selector:
language:
avoid:
required: false
selector:
select:
translation_key: avoid
options:
- tolls
- highways
- ferries
- indoor
traffic_model:
required: false
selector:
select:
translation_key: traffic_model
options:
- best_guess
- pessimistic
- optimistic
departure_time:
required: false
selector:
time:
get_transit_times:
fields:
config_entry_id:
required: true
selector:
config_entry:
integration: google_travel_time
origin:
required: true
example: "1600 Amphitheatre Parkway, Mountain View, CA"
selector:
text:
destination:
required: true
example: "1 Infinite Loop, Cupertino, CA"
selector:
text:
units:
default: "metric"
selector:
select:
translation_key: units
options:
- metric
- imperial
language:
required: false
selector:
language:
transit_mode:
required: false
selector:
select:
translation_key: transit_mode
options:
- bus
- subway
- train
- tram
- rail
transit_routing_preference:
required: false
selector:
select:
translation_key: transit_routing_preference
options:
- less_walking
- fewer_transfers
departure_time:
required: false
selector:
time:
arrival_time:
required: false
selector:
time:
@@ -30,6 +30,14 @@
}
}
},
"exceptions": {
"api_error": {
"message": "Google API error: {error}"
},
"permission_denied": {
"message": "[%key:component::google_travel_time::config::error::permission_denied%]"
}
},
"issues": {
"routes_api_disabled": {
"description": "Your Google Travel Time integration `{entry_title}` uses an API key which does not have the Routes API enabled.\n\n Please follow the instructions to [enable the API for your project]({enable_api_url}) and make sure your [API key restrictions]({api_key_restrictions_url}) allow access to the Routes API.\n\n After enabling the API this issue will be resolved automatically.",
@@ -107,5 +115,91 @@
}
}
},
"services": {
"get_transit_times": {
"description": "Retrieves route alternatives and travel times between two locations using public transit.",
"fields": {
"arrival_time": {
"description": "The desired arrival time.",
"name": "Arrival time"
},
"config_entry_id": {
"description": "[%key:component::google_travel_time::services::get_travel_times::fields::config_entry_id::description%]",
"name": "[%key:component::google_travel_time::services::get_travel_times::fields::config_entry_id::name%]"
},
"departure_time": {
"description": "[%key:component::google_travel_time::services::get_travel_times::fields::departure_time::description%]",
"name": "[%key:component::google_travel_time::services::get_travel_times::fields::departure_time::name%]"
},
"destination": {
"description": "[%key:component::google_travel_time::services::get_travel_times::fields::destination::description%]",
"name": "[%key:component::google_travel_time::config::step::user::data::destination%]"
},
"language": {
"description": "[%key:component::google_travel_time::services::get_travel_times::fields::language::description%]",
"name": "[%key:common::config_flow::data::language%]"
},
"origin": {
"description": "[%key:component::google_travel_time::services::get_travel_times::fields::origin::description%]",
"name": "[%key:component::google_travel_time::config::step::user::data::origin%]"
},
"transit_mode": {
"description": "The preferred transit mode.",
"name": "[%key:component::google_travel_time::options::step::init::data::transit_mode%]"
},
"transit_routing_preference": {
"description": "The transit routing preference.",
"name": "[%key:component::google_travel_time::options::step::init::data::transit_routing_preference%]"
},
"units": {
"description": "[%key:component::google_travel_time::services::get_travel_times::fields::units::description%]",
"name": "[%key:component::google_travel_time::options::step::init::data::units%]"
}
},
"name": "Get transit times"
},
"get_travel_times": {
"description": "Retrieves route alternatives and travel times between two locations.",
"fields": {
"avoid": {
"description": "Features to avoid when calculating the route.",
"name": "[%key:component::google_travel_time::options::step::init::data::avoid%]"
},
"config_entry_id": {
"description": "The config entry to use for the service call.",
"name": "Config entry"
},
"departure_time": {
"description": "The desired departure time.",
"name": "Departure time"
},
"destination": {
"description": "The destination of the route.",
"name": "[%key:component::google_travel_time::config::step::user::data::destination%]"
},
"language": {
"description": "The language to use for the response.",
"name": "[%key:common::config_flow::data::language%]"
},
"mode": {
"description": "The mode of transportation.",
"name": "[%key:component::google_travel_time::options::step::init::data::mode%]"
},
"origin": {
"description": "The origin of the route.",
"name": "[%key:component::google_travel_time::config::step::user::data::origin%]"
},
"traffic_model": {
"description": "The traffic model to use when calculating driving routes.",
"name": "[%key:component::google_travel_time::options::step::init::data::traffic_model%]"
},
"units": {
"description": "Which unit system to use.",
"name": "[%key:component::google_travel_time::options::step::init::data::units%]"
}
},
"name": "Get travel times"
}
},
"title": "Google Maps Travel Time"
}
@@ -54,6 +54,10 @@ def routes_mock() -> Generator[AsyncMock]:
"homeassistant.components.google_travel_time.sensor.RoutesAsyncClient",
new=mock_client,
),
patch(
"homeassistant.components.google_travel_time.services.RoutesAsyncClient",
new=mock_client,
),
):
client_mock = mock_client.return_value
client_mock.compute_routes.return_value = ComputeRoutesResponse(
@@ -75,6 +79,7 @@ def routes_mock() -> Generator[AsyncMock]:
}
),
"duration": duration_pb2.Duration(seconds=1620),
"distance_meters": 21300,
}
)
]
@@ -0,0 +1,259 @@
"""Tests for Google Maps Travel Time services."""
from unittest.mock import AsyncMock
from google.api_core.exceptions import GoogleAPIError, PermissionDenied
import pytest
from homeassistant.components.google_travel_time.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from .const import DEFAULT_OPTIONS, MOCK_CONFIG
from tests.common import MockConfigEntry
@pytest.mark.parametrize(
("data", "options"),
[(MOCK_CONFIG, DEFAULT_OPTIONS)],
)
async def test_service_get_travel_times(
hass: HomeAssistant,
routes_mock: AsyncMock,
mock_config: MockConfigEntry,
) -> None:
"""Test service get_travel_times."""
response_data = await hass.services.async_call(
DOMAIN,
"get_travel_times",
{
"config_entry_id": mock_config.entry_id,
"origin": "location1",
"destination": "location2",
"mode": "driving",
"units": "metric",
},
blocking=True,
return_response=True,
)
assert response_data == {
"routes": [
{
"duration": 1620,
"duration_text": "27 mins",
"static_duration_text": "26 mins",
"distance_meters": 21300,
"distance_text": "21.3 km",
}
]
}
@pytest.mark.parametrize(
("data", "options"),
[(MOCK_CONFIG, DEFAULT_OPTIONS)],
)
async def test_service_get_travel_times_with_all_options(
hass: HomeAssistant,
routes_mock: AsyncMock,
mock_config: MockConfigEntry,
) -> None:
"""Test service get_travel_times with all optional parameters."""
response_data = await hass.services.async_call(
DOMAIN,
"get_travel_times",
{
"config_entry_id": mock_config.entry_id,
"origin": "location1",
"destination": "location2",
"mode": "driving",
"units": "imperial",
"language": "en",
"avoid": "tolls",
"traffic_model": "best_guess",
"departure_time": "08:00:00",
},
blocking=True,
return_response=True,
)
assert "routes" in response_data
assert len(response_data["routes"]) == 1
@pytest.mark.parametrize(
("data", "options"),
[(MOCK_CONFIG, DEFAULT_OPTIONS)],
)
async def test_service_get_travel_times_empty_response(
hass: HomeAssistant,
routes_mock: AsyncMock,
mock_config: MockConfigEntry,
) -> None:
"""Test service get_travel_times with empty response."""
routes_mock.compute_routes.return_value = None
response_data = await hass.services.async_call(
DOMAIN,
"get_travel_times",
{
"config_entry_id": mock_config.entry_id,
"origin": "location1",
"destination": "location2",
"mode": "driving",
"units": "metric",
},
blocking=True,
return_response=True,
)
assert response_data == {"routes": []}
@pytest.mark.parametrize(
("data", "options"),
[(MOCK_CONFIG, DEFAULT_OPTIONS)],
)
@pytest.mark.parametrize(
("exception", "error_message"),
[
(
PermissionDenied("test"),
"The Routes API is not enabled for this API key",
),
(GoogleAPIError("test"), "Google API error"),
],
)
async def test_service_get_travel_times_errors(
hass: HomeAssistant,
routes_mock: AsyncMock,
mock_config: MockConfigEntry,
exception: Exception,
error_message: str,
) -> None:
"""Test service get_travel_times error handling."""
routes_mock.compute_routes.side_effect = exception
with pytest.raises(
HomeAssistantError,
match=error_message,
):
await hass.services.async_call(
DOMAIN,
"get_travel_times",
{
"config_entry_id": mock_config.entry_id,
"origin": "location1",
"destination": "location2",
"mode": "driving",
"units": "metric",
},
blocking=True,
return_response=True,
)
@pytest.mark.parametrize(
("data", "options"),
[(MOCK_CONFIG, DEFAULT_OPTIONS)],
)
async def test_service_get_transit_times(
hass: HomeAssistant,
routes_mock: AsyncMock,
mock_config: MockConfigEntry,
) -> None:
"""Test service get_transit_times."""
response_data = await hass.services.async_call(
DOMAIN,
"get_transit_times",
{
"config_entry_id": mock_config.entry_id,
"origin": "location1",
"destination": "location2",
"units": "metric",
},
blocking=True,
return_response=True,
)
assert response_data == {
"routes": [
{
"duration": 1620,
"duration_text": "27 mins",
"static_duration_text": "26 mins",
"distance_meters": 21300,
"distance_text": "21.3 km",
}
]
}
@pytest.mark.parametrize(
("data", "options"),
[(MOCK_CONFIG, DEFAULT_OPTIONS)],
)
async def test_service_get_transit_times_with_all_options(
hass: HomeAssistant,
routes_mock: AsyncMock,
mock_config: MockConfigEntry,
) -> None:
"""Test service get_transit_times with all optional parameters."""
response_data = await hass.services.async_call(
DOMAIN,
"get_transit_times",
{
"config_entry_id": mock_config.entry_id,
"origin": "location1",
"destination": "location2",
"units": "imperial",
"language": "en",
"transit_mode": "bus",
"transit_routing_preference": "fewer_transfers",
"departure_time": "08:00:00",
},
blocking=True,
return_response=True,
)
assert "routes" in response_data
assert len(response_data["routes"]) == 1
@pytest.mark.parametrize(
("data", "options"),
[(MOCK_CONFIG, DEFAULT_OPTIONS)],
)
@pytest.mark.parametrize(
("exception", "error_message"),
[
(
PermissionDenied("test"),
"The Routes API is not enabled for this API key",
),
(GoogleAPIError("test"), "Google API error"),
],
)
async def test_service_get_transit_times_errors(
hass: HomeAssistant,
routes_mock: AsyncMock,
mock_config: MockConfigEntry,
exception: Exception,
error_message: str,
) -> None:
"""Test service get_transit_times error handling."""
routes_mock.compute_routes.side_effect = exception
with pytest.raises(
HomeAssistantError,
match=error_message,
):
await hass.services.async_call(
DOMAIN,
"get_transit_times",
{
"config_entry_id": mock_config.entry_id,
"origin": "location1",
"destination": "location2",
"units": "metric",
},
blocking=True,
return_response=True,
)