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

Add config flow support to google_travel_time (#43509) 

* add config flow support to google_travel_time

* fix bugs and add strings

* fix import and add new test

* address comments in #43419 since this is a similar PR

* fix default name and test

* add unique ID and device info

* fix test

* feedback from waze PR

* continue incorporating feedback from waze PR

* final fixes and update tests

* call update in lambda

* Update homeassistant/components/google_travel_time/sensor.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* additional fixes

* validate config entry data during config flow and config entry setup

* don't store entity

* patch dependency instead of HA code

* fixes

* improve tests by moving all patching to fixtures

* use self.hass instead of setting self._hass

* invert if

* remove unnecessary else

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
pull/48017/head^2
Raman Gupta 2021-03-31 16:10:24 -04:00 committed by GitHub
parent bc5d828554
commit aae0ccc588
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 952 additions and 187 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.coveragerc
View File

@ -349,6 +349,8 @@ omit =
homeassistant/components/google/*
homeassistant/components/google_cloud/tts.py
homeassistant/components/google_maps/device_tracker.py
homeassistant/components/google_travel_time/__init__.py
homeassistant/components/google_travel_time/helpers.py
homeassistant/components/google_travel_time/sensor.py
homeassistant/components/gpmdp/media_player.py
homeassistant/components/gpsd/sensor.py

35
homeassistant/components/google_travel_time/__init__.py
View File

@ -1 +1,36 @@
"""The google_travel_time component."""
import asyncio
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
PLATFORMS = ["sensor"]
async def async_setup(hass: HomeAssistant, config: dict):
"""Set up the Google Maps Travel Time component."""
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Set up Google Maps Travel Time from a config entry."""
for component in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, component)
)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Unload a config entry."""
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(entry, component)
for component in PLATFORMS
]
)
)
return unload_ok

166
homeassistant/components/google_travel_time/config_flow.py Normal file
View File

@ -0,0 +1,166 @@
"""Config flow for Google Maps Travel Time integration."""
import logging
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_API_KEY, CONF_MODE, CONF_NAME
from homeassistant.core import callback
import homeassistant.helpers.config_validation as cv
from homeassistant.util import slugify
from .const import (
ALL_LANGUAGES,
ARRIVAL_TIME,
AVOID,
CONF_ARRIVAL_TIME,
CONF_AVOID,
CONF_DEPARTURE_TIME,
CONF_DESTINATION,
CONF_LANGUAGE,
CONF_ORIGIN,
CONF_TIME,
CONF_TIME_TYPE,
CONF_TRAFFIC_MODEL,
CONF_TRANSIT_MODE,
CONF_TRANSIT_ROUTING_PREFERENCE,
CONF_UNITS,
DEFAULT_NAME,
DEPARTURE_TIME,
DOMAIN,
TIME_TYPES,
TRANSIT_PREFS,
TRANSPORT_TYPE,
TRAVEL_MODE,
TRAVEL_MODEL,
UNITS,
)
from .helpers import is_valid_config_entry
_LOGGER = logging.getLogger(__name__)
class GoogleOptionsFlow(config_entries.OptionsFlow):
"""Handle an options flow for Google Travel Time."""
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Initialize google options flow."""
self.config_entry = config_entry
async def async_step_init(self, user_input=None):
"""Handle the initial step."""
if user_input is not None:
time_type = user_input.pop(CONF_TIME_TYPE)
if time := user_input.pop(CONF_TIME, None):
if time_type == ARRIVAL_TIME:
user_input[CONF_ARRIVAL_TIME] = time
else:
user_input[CONF_DEPARTURE_TIME] = time
return self.async_create_entry(title="", data=user_input)
if CONF_ARRIVAL_TIME in self.config_entry.options:
default_time_type = ARRIVAL_TIME
default_time = self.config_entry.options[CONF_ARRIVAL_TIME]
else:
default_time_type = DEPARTURE_TIME
default_time = self.config_entry.options.get(CONF_ARRIVAL_TIME)
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(
{
vol.Optional(
CONF_MODE, default=self.config_entry.options[CONF_MODE]
): vol.In(TRAVEL_MODE),
vol.Optional(
CONF_LANGUAGE,
default=self.config_entry.options.get(CONF_LANGUAGE),
): vol.In(ALL_LANGUAGES),
vol.Optional(
CONF_AVOID, default=self.config_entry.options.get(CONF_AVOID)
): vol.In(AVOID),
vol.Optional(
CONF_UNITS, default=self.config_entry.options[CONF_UNITS]
): vol.In(UNITS),
vol.Optional(CONF_TIME_TYPE, default=default_time_type): vol.In(
TIME_TYPES
),
vol.Optional(CONF_TIME, default=default_time): cv.string,
vol.Optional(
CONF_TRAFFIC_MODEL,
default=self.config_entry.options.get(CONF_TRAFFIC_MODEL),
): vol.In(TRAVEL_MODEL),
vol.Optional(
CONF_TRANSIT_MODE,
default=self.config_entry.options.get(CONF_TRANSIT_MODE),
): vol.In(TRANSPORT_TYPE),
vol.Optional(
CONF_TRANSIT_ROUTING_PREFERENCE,
default=self.config_entry.options.get(
CONF_TRANSIT_ROUTING_PREFERENCE
),
): vol.In(TRANSIT_PREFS),
}
),
)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Google Maps Travel Time."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
@staticmethod
@callback
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
) -> GoogleOptionsFlow:
"""Get the options flow for this handler."""
return GoogleOptionsFlow(config_entry)
async def async_step_user(self, user_input=None):
"""Handle the initial step."""
errors = {}
if user_input is not None:
if await self.hass.async_add_executor_job(
is_valid_config_entry,
self.hass,
_LOGGER,
user_input[CONF_API_KEY],
user_input[CONF_ORIGIN],
user_input[CONF_DESTINATION],
):
await self.async_set_unique_id(
slugify(
f"{DOMAIN}_{user_input[CONF_ORIGIN]}_{user_input[CONF_DESTINATION]}"
)
)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=user_input.get(
CONF_NAME,
(
f"{DEFAULT_NAME}: {user_input[CONF_ORIGIN]} -> "
f"{user_input[CONF_DESTINATION]}"
),
),
data=user_input,
)
# If we get here, it's because we couldn't connect
errors["base"] = "cannot_connect"
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_API_KEY): cv.string,
vol.Required(CONF_DESTINATION): cv.string,
vol.Required(CONF_ORIGIN): cv.string,
}
),
errors=errors,
)
async_step_import = async_step_user

89
homeassistant/components/google_travel_time/const.py Normal file
View File

@ -0,0 +1,89 @@
"""Constants for Google Travel Time."""
from homeassistant.const import CONF_UNIT_SYSTEM_IMPERIAL, CONF_UNIT_SYSTEM_METRIC
DOMAIN = "google_travel_time"
ATTRIBUTION = "Powered by Google"
CONF_DESTINATION = "destination"
CONF_OPTIONS = "options"
CONF_ORIGIN = "origin"
CONF_TRAVEL_MODE = "travel_mode"
CONF_LANGUAGE = "language"
CONF_AVOID = "avoid"
CONF_UNITS = "units"
CONF_ARRIVAL_TIME = "arrival_time"
CONF_DEPARTURE_TIME = "departure_time"
CONF_TRAFFIC_MODEL = "traffic_model"
CONF_TRANSIT_MODE = "transit_mode"
CONF_TRANSIT_ROUTING_PREFERENCE = "transit_routing_preference"
CONF_TIME_TYPE = "time_type"
CONF_TIME = "time"
ARRIVAL_TIME = "Arrival Time"
DEPARTURE_TIME = "Departure Time"
TIME_TYPES = [ARRIVAL_TIME, DEPARTURE_TIME]
DEFAULT_NAME = "Google Travel Time"
TRACKABLE_DOMAINS = ["device_tracker", "sensor", "zone", "person"]
ALL_LANGUAGES = [
"ar",
"bg",
"bn",
"ca",
"cs",
"da",
"de",
"el",
"en",
"es",
"eu",
"fa",
"fi",
"fr",
"gl",
"gu",
"hi",
"hr",
"hu",
"id",
"it",
"iw",
"ja",
"kn",
"ko",
"lt",
"lv",
"ml",
"mr",
"nl",
"no",
"pl",
"pt",
"pt-BR",
"pt-PT",
"ro",
"ru",
"sk",
"sl",
"sr",
"sv",
"ta",
"te",
"th",
"tl",
"tr",
"uk",
"vi",
"zh-CN",
"zh-TW",
]
AVOID = ["tolls", "highways", "ferries", "indoor"]
TRANSIT_PREFS = ["less_walking", "fewer_transfers"]
TRANSPORT_TYPE = ["bus", "subway", "train", "tram", "rail"]
TRAVEL_MODE = ["driving", "walking", "bicycling", "transit"]
TRAVEL_MODEL = ["best_guess", "pessimistic", "optimistic"]
UNITS = [CONF_UNIT_SYSTEM_METRIC, CONF_UNIT_SYSTEM_IMPERIAL]

72
homeassistant/components/google_travel_time/helpers.py Normal file
View File

@ -0,0 +1,72 @@
"""Helpers for Google Time Travel integration."""
from googlemaps import Client
from googlemaps.distance_matrix import distance_matrix
from googlemaps.exceptions import ApiError
from homeassistant.components.google_travel_time.const import TRACKABLE_DOMAINS
from homeassistant.const import ATTR_LATITUDE, ATTR_LONGITUDE
from homeassistant.helpers import location
def is_valid_config_entry(hass, logger, api_key, origin, destination):
"""Return whether the config entry data is valid."""
origin = resolve_location(hass, logger, origin)
destination = resolve_location(hass, logger, destination)
client = Client(api_key, timeout=10)
try:
distance_matrix(client, origin, destination, mode="driving")
except ApiError:
return False
return True
def resolve_location(hass, logger, loc):
"""Resolve a location."""
if loc.split(".", 1)[0] in TRACKABLE_DOMAINS:
return get_location_from_entity(hass, logger, loc)
return resolve_zone(hass, loc)
def get_location_from_entity(hass, logger, entity_id):
"""Get the location from the entity state or attributes."""
entity = hass.states.get(entity_id)
if entity is None:
logger.error("Unable to find entity %s", entity_id)
return None
# Check if the entity has location attributes
if location.has_location(entity):
return get_location_from_attributes(entity)
# Check if device is in a zone
zone_entity = hass.states.get("zone.%s" % entity.state)
if location.has_location(zone_entity):
logger.debug(
"%s is in %s, getting zone location", entity_id, zone_entity.entity_id
)
return get_location_from_attributes(zone_entity)
# If zone was not found in state then use the state as the location
if entity_id.startswith("sensor."):
return entity.state
# When everything fails just return nothing
return None
def get_location_from_attributes(entity):
"""Get the lat/long string from an entities attributes."""
attr = entity.attributes
return f"{attr.get(ATTR_LATITUDE)},{attr.get(ATTR_LONGITUDE)}"
def resolve_zone(hass, friendly_name):
"""Resolve a location from a zone's friendly name."""
entities = hass.states.all()
for entity in entities:
if entity.domain == "zone" and entity.name == friendly_name:
return get_location_from_attributes(entity)
return friendly_name

7
homeassistant/components/google_travel_time/manifest.json
View File

@ -2,6 +2,9 @@
"domain": "google_travel_time",
"name": "Google Maps Travel Time",
"documentation": "https://www.home-assistant.io/integrations/google_travel_time",
"requirements": ["googlemaps==2.5.1"],
"codeowners": []
"requirements": [
"googlemaps==2.5.1"
],
"codeowners": [],
"config_flow": true
}

335
homeassistant/components/google_travel_time/sensor.py
View File

@ -1,98 +1,60 @@
"""Support for Google travel time sensors."""
from __future__ import annotations
from datetime import datetime, timedelta
import logging
from typing import Callable
import googlemaps
from googlemaps import Client
from googlemaps.distance_matrix import distance_matrix
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
ATTR_ATTRIBUTION,
ATTR_LATITUDE,
ATTR_LONGITUDE,
CONF_API_KEY,
CONF_MODE,
CONF_NAME,
EVENT_HOMEASSISTANT_START,
TIME_MINUTES,
)
from homeassistant.helpers import location
from homeassistant.core import CoreState, HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
import homeassistant.helpers.config_validation as cv
import homeassistant.util.dt as dt_util
from .const import (
ALL_LANGUAGES,
ATTRIBUTION,
AVOID,
CONF_ARRIVAL_TIME,
CONF_AVOID,
CONF_DEPARTURE_TIME,
CONF_DESTINATION,
CONF_LANGUAGE,
CONF_OPTIONS,
CONF_ORIGIN,
CONF_TRAFFIC_MODEL,
CONF_TRANSIT_MODE,
CONF_TRANSIT_ROUTING_PREFERENCE,
CONF_TRAVEL_MODE,
CONF_UNITS,
DEFAULT_NAME,
DOMAIN,
TRACKABLE_DOMAINS,
TRANSIT_PREFS,
TRANSPORT_TYPE,
TRAVEL_MODE,
TRAVEL_MODEL,
UNITS,
)
from .helpers import get_location_from_entity, is_valid_config_entry, resolve_zone
_LOGGER = logging.getLogger(__name__)
ATTRIBUTION = "Powered by Google"
CONF_DESTINATION = "destination"
CONF_OPTIONS = "options"
CONF_ORIGIN = "origin"
CONF_TRAVEL_MODE = "travel_mode"
DEFAULT_NAME = "Google Travel Time"
SCAN_INTERVAL = timedelta(minutes=5)
ALL_LANGUAGES = [
"ar",
"bg",
"bn",
"ca",
"cs",
"da",
"de",
"el",
"en",
"es",
"eu",
"fa",
"fi",
"fr",
"gl",
"gu",
"hi",
"hr",
"hu",
"id",
"it",
"iw",
"ja",
"kn",
"ko",
"lt",
"lv",
"ml",
"mr",
"nl",
"no",
"pl",
"pt",
"pt-BR",
"pt-PT",
"ro",
"ru",
"sk",
"sl",
"sr",
"sv",
"ta",
"te",
"th",
"tl",
"tr",
"uk",
"vi",
"zh-CN",
"zh-TW",
]
AVOID = ["tolls", "highways", "ferries", "indoor"]
TRANSIT_PREFS = ["less_walking", "fewer_transfers"]
TRANSPORT_TYPE = ["bus", "subway", "train", "tram", "rail"]
TRAVEL_MODE = ["driving", "walking", "bicycling", "transit"]
TRAVEL_MODEL = ["best_guess", "pessimistic", "optimistic"]
UNITS = ["metric", "imperial"]
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_API_KEY): cv.string,
@ -105,23 +67,22 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
vol.Schema(
{
vol.Optional(CONF_MODE, default="driving"): vol.In(TRAVEL_MODE),
vol.Optional("language"): vol.In(ALL_LANGUAGES),
vol.Optional("avoid"): vol.In(AVOID),
vol.Optional("units"): vol.In(UNITS),
vol.Exclusive("arrival_time", "time"): cv.string,
vol.Exclusive("departure_time", "time"): cv.string,
vol.Optional("traffic_model"): vol.In(TRAVEL_MODEL),
vol.Optional("transit_mode"): vol.In(TRANSPORT_TYPE),
vol.Optional("transit_routing_preference"): vol.In(TRANSIT_PREFS),
vol.Optional(CONF_LANGUAGE): vol.In(ALL_LANGUAGES),
vol.Optional(CONF_AVOID): vol.In(AVOID),
vol.Optional(CONF_UNITS): vol.In(UNITS),
vol.Exclusive(CONF_ARRIVAL_TIME, "time"): cv.string,
vol.Exclusive(CONF_DEPARTURE_TIME, "time"): cv.string,
vol.Optional(CONF_TRAFFIC_MODEL): vol.In(TRAVEL_MODEL),
vol.Optional(CONF_TRANSIT_MODE): vol.In(TRANSPORT_TYPE),
vol.Optional(CONF_TRANSIT_ROUTING_PREFERENCE): vol.In(
TRANSIT_PREFS
),
}
),
),
}
)
TRACKABLE_DOMAINS = ["device_tracker", "sensor", "zone", "person"]
DATA_KEY = "google_travel_time"
def convert_time_to_utc(timestr):
"""Take a string like 08:00:00 and convert it to a unix timestamp."""
@ -133,63 +94,88 @@ def convert_time_to_utc(timestr):
return dt_util.as_timestamp(combined)
def setup_platform(hass, config, add_entities_callback, discovery_info=None):
"""Set up the Google travel time platform."""
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: Callable[[list[SensorEntity], bool], None],
) -> None:
"""Set up a Google travel time sensor entry."""
name = None
if not config_entry.options:
new_data = config_entry.data.copy()
options = new_data.pop(CONF_OPTIONS, {})
name = new_data.pop(CONF_NAME, None)
def run_setup(event):
"""
Delay the setup until Home Assistant is fully initialized.
if CONF_UNITS not in options:
options[CONF_UNITS] = hass.config.units.name
This allows any entities to be created already
"""
hass.data.setdefault(DATA_KEY, [])
options = config.get(CONF_OPTIONS)
if options.get("units") is None:
options["units"] = hass.config.units.name
travel_mode = config.get(CONF_TRAVEL_MODE)
mode = options.get(CONF_MODE)
if travel_mode is not None:
if CONF_TRAVEL_MODE in new_data:
wstr = (
"Google Travel Time: travel_mode is deprecated, please "
"add mode to the options dictionary instead!"
)
_LOGGER.warning(wstr)
if mode is None:
travel_mode = new_data.pop(CONF_TRAVEL_MODE)
if CONF_MODE not in options:
options[CONF_MODE] = travel_mode
titled_mode = options.get(CONF_MODE).title()
formatted_name = f"{DEFAULT_NAME} - {titled_mode}"
name = config.get(CONF_NAME, formatted_name)
api_key = config.get(CONF_API_KEY)
origin = config.get(CONF_ORIGIN)
destination = config.get(CONF_DESTINATION)
if CONF_MODE not in options:
options[CONF_MODE] = "driving"
sensor = GoogleTravelTimeSensor(
hass, name, api_key, origin, destination, options
hass.config_entries.async_update_entry(
config_entry, data=new_data, options=options
)
hass.data[DATA_KEY].append(sensor)
if sensor.valid_api_connection:
add_entities_callback([sensor])
api_key = config_entry.data[CONF_API_KEY]
origin = config_entry.data[CONF_ORIGIN]
destination = config_entry.data[CONF_DESTINATION]
name = name or f"{DEFAULT_NAME}: {origin} -> {destination}"
# Wait until start event is sent to load this component.
hass.bus.listen_once(EVENT_HOMEASSISTANT_START, run_setup)
if not await hass.async_add_executor_job(
is_valid_config_entry, hass, _LOGGER, api_key, origin, destination
):
raise ConfigEntryNotReady
client = Client(api_key, timeout=10)
sensor = GoogleTravelTimeSensor(
config_entry, name, api_key, origin, destination, client
)
async_add_entities([sensor], False)
async def async_setup_platform(
hass: HomeAssistant, config, add_entities_callback, discovery_info=None
):
"""Set up the Google travel time platform."""
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=config,
)
)
_LOGGER.warning(
"Your Google travel time configuration has been imported into the UI; "
"please remove it from configuration.yaml as support for it will be "
"removed in a future release"
)
class GoogleTravelTimeSensor(SensorEntity):
"""Representation of a Google travel time sensor."""
def __init__(self, hass, name, api_key, origin, destination, options):
def __init__(self, config_entry, name, api_key, origin, destination, client):
"""Initialize the sensor."""
self._hass = hass
self._name = name
self._options = options
self._config_entry = config_entry
self._unit_of_measurement = TIME_MINUTES
self._matrix = None
self.valid_api_connection = True
self._api_key = api_key
self._unique_id = config_entry.unique_id
self._client = client
# Check if location is a trackable entity
if origin.split(".", 1)[0] in TRACKABLE_DOMAINS:
@ -202,13 +188,14 @@ class GoogleTravelTimeSensor(SensorEntity):
else:
self._destination = destination
self._client = googlemaps.Client(api_key, timeout=10)
try:
self.update()
except googlemaps.exceptions.ApiError as exp:
_LOGGER.error(exp)
self.valid_api_connection = False
return
async def async_added_to_hass(self) -> None:
"""Handle when entity is added."""
if self.hass.state != CoreState.running:
self.hass.bus.async_listen_once(
EVENT_HOMEASSISTANT_START, self.first_update
)
else:
await self.first_update()
@property
def state(self):
@ -223,6 +210,20 @@ class GoogleTravelTimeSensor(SensorEntity):
return round(_data["duration"]["value"] / 60)
return None
@property
def device_info(self):
"""Return device specific attributes."""
return {
"name": DOMAIN,
"identifiers": {(DOMAIN, self._api_key)},
"entry_type": "service",
}
@property
def unique_id(self) -> str:
"""Return unique ID of entity."""
return self._unique_id
@property
def name(self):
"""Get the name of the sensor."""
@ -235,7 +236,8 @@ class GoogleTravelTimeSensor(SensorEntity):
return None
res = self._matrix.copy()
res.update(self._options)
options = self._config_entry.options.copy()
res.update(options)
del res["rows"]
_data = self._matrix["rows"][0]["elements"][0]
if "duration_in_traffic" in _data:
@ -254,78 +256,43 @@ class GoogleTravelTimeSensor(SensorEntity):
"""Return the unit this state is expressed in."""
return self._unit_of_measurement
async def first_update(self, _=None):
"""Run the first update and write the state."""
await self.hass.async_add_executor_job(self.update)
self.async_write_ha_state()
def update(self):
"""Get the latest data from Google."""
options_copy = self._options.copy()
dtime = options_copy.get("departure_time")
atime = options_copy.get("arrival_time")
options_copy = self._config_entry.options.copy()
dtime = options_copy.get(CONF_DEPARTURE_TIME)
atime = options_copy.get(CONF_ARRIVAL_TIME)
if dtime is not None and ":" in dtime:
options_copy["departure_time"] = convert_time_to_utc(dtime)
options_copy[CONF_DEPARTURE_TIME] = convert_time_to_utc(dtime)
elif dtime is not None:
options_copy["departure_time"] = dtime
options_copy[CONF_DEPARTURE_TIME] = dtime
elif atime is None:
options_copy["departure_time"] = "now"
options_copy[CONF_DEPARTURE_TIME] = "now"
if atime is not None and ":" in atime:
options_copy["arrival_time"] = convert_time_to_utc(atime)
options_copy[CONF_ARRIVAL_TIME] = convert_time_to_utc(atime)
elif atime is not None:
options_copy["arrival_time"] = atime
options_copy[CONF_ARRIVAL_TIME] = atime
# Convert device_trackers to google friendly location
if hasattr(self, "_origin_entity_id"):
self._origin = self._get_location_from_entity(self._origin_entity_id)
self._origin = get_location_from_entity(
self.hass, _LOGGER, self._origin_entity_id
)
if hasattr(self, "_destination_entity_id"):
self._destination = self._get_location_from_entity(
self._destination_entity_id
self._destination = get_location_from_entity(
self.hass, _LOGGER, self._destination_entity_id
)
self._destination = self._resolve_zone(self._destination)
self._origin = self._resolve_zone(self._origin)
self._destination = resolve_zone(self.hass, self._destination)
self._origin = resolve_zone(self.hass, self._origin)
if self._destination is not None and self._origin is not None:
self._matrix = self._client.distance_matrix(
self._origin, self._destination, **options_copy
self._matrix = distance_matrix(
self._client, self._origin, self._destination, **options_copy
)
def _get_location_from_entity(self, entity_id):
"""Get the location from the entity state or attributes."""
entity = self._hass.states.get(entity_id)
if entity is None:
_LOGGER.error("Unable to find entity %s", entity_id)
self.valid_api_connection = False
return None
# Check if the entity has location attributes
if location.has_location(entity):
return self._get_location_from_attributes(entity)
# Check if device is in a zone
zone_entity = self._hass.states.get("zone.%s" % entity.state)
if location.has_location(zone_entity):
_LOGGER.debug(
"%s is in %s, getting zone location", entity_id, zone_entity.entity_id
)
return self._get_location_from_attributes(zone_entity)
# If zone was not found in state then use the state as the location
if entity_id.startswith("sensor."):
return entity.state
# When everything fails just return nothing
return None
@staticmethod
def _get_location_from_attributes(entity):
"""Get the lat/long string from an entities attributes."""
attr = entity.attributes
return f"{attr.get(ATTR_LATITUDE)},{attr.get(ATTR_LONGITUDE)}"
def _resolve_zone(self, friendly_name):
entities = self._hass.states.all()
for entity in entities:
if entity.domain == "zone" and entity.name == friendly_name:
return self._get_location_from_attributes(entity)
return friendly_name

38
homeassistant/components/google_travel_time/strings.json Normal file
View File

@ -0,0 +1,38 @@
{
"title": "Google Maps Travel Time",
"config": {
"step": {
"user": {
"description": "When specifying the origin and destination, you can supply one or more locations separated by the pipe character, in the form of an address, latitude/longitude coordinates, or a Google place ID. When specifying the location using a Google place ID, the ID must be prefixed with `place_id:`.",
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]",
"origin": "Origin",
"destination": "Destination"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_location%]"
}
},
"options": {
"step": {
"init": {
"description": "You can optionally specify either a Departure Time or Arrival Time. If specifying a departure time, you can enter `now`, a Unix timestamp, or a 24 hour time string like `08:00:00`. If specifying an arrival time, you can use a Unix timestamp or a 24 hour time string like `08:00:00`",
"data": {
"mode": "Travel Mode",
"language": "Language",
"time_type": "Time Type",
"time": "Time",
"avoid": "Avoid",
"transit_mode": "Transit Mode",
"transit_routing_preference": "Transit Routing Preference",
"units": "Units"
}
}
}
}
}

32
homeassistant/components/google_travel_time/translations/en.json Normal file
View File

@ -0,0 +1,32 @@
{
"config": {
"abort": {
"already_configured": "Location is already configured"
},
"step": {
"options": {
"data": {
"arrival_time": "Arrival Time",
"avoid": "Avoid",
"departure_time": "Departure Time",
"language": "Language",
"mode": "Travel Mode",
"transit_mode": "Transit Mode",
"transit_routing_preference": "Transit Routing Preference",
"units": "Units"
},
"description": "You can either specify Departure Time or Arrival Time, but not both"
},
"user": {
"data": {
"api_key": "API Key",
"destination": "Destination",
"name": "Name",
"origin": "Origin"
},
"description": "When specifying the origin and destination, you can supply one or more locations separated by the pipe character, in the form of an address, latitude/longitude coordinates, or a Google place ID. When specifying the location using a Google place ID, the ID must be prefixed with `place_id:`."
}
}
},
"title": "Google Maps Travel Time"
}

1
homeassistant/generated/config_flows.py
View File

@ -84,6 +84,7 @@ FLOWS = [
"glances",
"goalzero",
"gogogate2",
"google_travel_time",
"gpslogger",
"gree",
"guardian",

3
requirements_test_all.txt
View File

@ -372,6 +372,9 @@ google-cloud-pubsub==2.1.0
# homeassistant.components.nest
google-nest-sdm==0.2.12
# homeassistant.components.google_travel_time
googlemaps==2.5.1
# homeassistant.components.gree
greeclimate==0.10.3

1
tests/components/google_travel_time/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the Google Maps Travel Time integration."""

59
tests/components/google_travel_time/conftest.py Normal file
View File

@ -0,0 +1,59 @@
"""Fixtures for Google Time Travel tests."""
from unittest.mock import Mock, patch
from googlemaps.exceptions import ApiError
import pytest
@pytest.fixture(name="skip_notifications", autouse=True)
def skip_notifications_fixture():
"""Skip notification calls."""
with patch("homeassistant.components.persistent_notification.async_create"), patch(
"homeassistant.components.persistent_notification.async_dismiss"
):
yield
@pytest.fixture(name="validate_config_entry")
def validate_config_entry_fixture():
"""Return valid config entry."""
with patch(
"homeassistant.components.google_travel_time.helpers.Client",
return_value=Mock(),
), patch(
"homeassistant.components.google_travel_time.helpers.distance_matrix",
return_value=None,
):
yield
@pytest.fixture(name="bypass_setup")
def bypass_setup_fixture():
"""Bypass entry setup."""
with patch(
"homeassistant.components.google_travel_time.async_setup", return_value=True
), patch(
"homeassistant.components.google_travel_time.async_setup_entry",
return_value=True,
):
yield
@pytest.fixture(name="bypass_update")
def bypass_update_fixture():
"""Bypass sensor update."""
with patch("homeassistant.components.google_travel_time.sensor.distance_matrix"):
yield
@pytest.fixture(name="invalidate_config_entry")
def invalidate_config_entry_fixture():
"""Return invalid config entry."""
with patch(
"homeassistant.components.google_travel_time.helpers.Client",
return_value=Mock(),
), patch(
"homeassistant.components.google_travel_time.helpers.distance_matrix",
side_effect=ApiError("test"),
):
yield

297
tests/components/google_travel_time/test_config_flow.py Normal file
View File

@ -0,0 +1,297 @@
"""Test the Google Maps Travel Time config flow."""
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.google_travel_time.const import (
ARRIVAL_TIME,
CONF_ARRIVAL_TIME,
CONF_AVOID,
CONF_DEPARTURE_TIME,
CONF_DESTINATION,
CONF_LANGUAGE,
CONF_OPTIONS,
CONF_ORIGIN,
CONF_TIME,
CONF_TIME_TYPE,
CONF_TRAFFIC_MODEL,
CONF_TRANSIT_MODE,
CONF_TRANSIT_ROUTING_PREFERENCE,
CONF_UNITS,
DEFAULT_NAME,
DEPARTURE_TIME,
DOMAIN,
)
from homeassistant.const import (
CONF_API_KEY,
CONF_MODE,
CONF_NAME,
CONF_UNIT_SYSTEM_IMPERIAL,
)
from tests.common import MockConfigEntry
async def test_minimum_fields(hass, validate_config_entry, bypass_setup):
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {}
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
},
)
assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result2["title"] == f"{DEFAULT_NAME}: location1 -> location2"
assert result2["data"] == {
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
}
async def test_invalid_config_entry(hass, invalidate_config_entry):
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {}
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
},
)
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result2["errors"] == {"base": "cannot_connect"}
async def test_options_flow(hass, validate_config_entry, bypass_update):
"""Test options flow."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
},
options={
CONF_MODE: "driving",
CONF_ARRIVAL_TIME: "test",
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(entry.entry_id, data=None)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_MODE: "driving",
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_TIME_TYPE: ARRIVAL_TIME,
CONF_TIME: "test",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == ""
assert result["data"] == {
CONF_MODE: "driving",
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_ARRIVAL_TIME: "test",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
}
assert entry.options == {
CONF_MODE: "driving",
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_ARRIVAL_TIME: "test",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
}
async def test_options_flow_departure_time(hass, validate_config_entry, bypass_update):
"""Test options flow wiith departure time."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(entry.entry_id, data=None)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_MODE: "driving",
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_TIME_TYPE: DEPARTURE_TIME,
CONF_TIME: "test",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == ""
assert result["data"] == {
CONF_MODE: "driving",
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_DEPARTURE_TIME: "test",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
}
assert entry.options == {
CONF_MODE: "driving",
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_DEPARTURE_TIME: "test",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
}
async def test_dupe_id(hass, validate_config_entry, bypass_setup):
"""Test setting up the same entry twice fails."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {}
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_API_KEY: "test",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
},
)
assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {}
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_API_KEY: "test",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
},
)
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result2["reason"] == "already_configured"
async def test_import_flow(hass, validate_config_entry, bypass_update):
"""Test import_flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_NAME: "test_name",
CONF_OPTIONS: {
CONF_MODE: "driving",
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_ARRIVAL_TIME: "test",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
},
},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "test_name"
assert result["data"] == {
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_NAME: "test_name",
CONF_OPTIONS: {
CONF_MODE: "driving",
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_ARRIVAL_TIME: "test",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
},
}
entry = hass.config_entries.async_entries(DOMAIN)[0]
assert entry.data == {
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
}
assert entry.options == {
CONF_MODE: "driving",
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_ARRIVAL_TIME: "test",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
}