jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-16 01:37:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add config flow for Waze Travel Time (#43419)

Browse Source 
* Add config flow for Waze Travel Time

* update translations

* setup entry is async

* fix update logic during setup

* support old config method in the interim

* fix requirements

* fix requirements

* add abort string

* changes based on @bdraco review

* fix tests

* add device identifier

* Update homeassistant/components/waze_travel_time/__init__.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* fix tests

* Update homeassistant/components/waze_travel_time/sensor.py

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

* log warning for deprecation message

* PR feedback

* fix tests and bugs

* re-add name to config schema to avoid breaking change

* handle if we get name from config in entry title

* fix name logic

* always set up options with defaults

* Update homeassistant/components/waze_travel_time/sensor.py

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

* Update config_flow.py

* Update sensor.py

* handle options updates by getting options on every update

* patch library instead of sensor

* fixes and make sure first update writes the state

* validate config entry data during config flow and entry setup

* fix input parameters

* fix tests

* invert if statement

* remove unnecessary else

* exclude helpers from coverage

* remove async_setup because it's no longer needed

* fix patch statements

Co-authored-by: J. Nick Koston <nick@koston.org>
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Raman Gupta 2021-04-05 19:25:52 -04:00 committed by GitHub
parent f3399aa8aa
commit 5305d083ec
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 813 additions and 166 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.coveragerc
View File

@ -1102,6 +1102,8 @@ omit =
homeassistant/components/waterfurnace/*
homeassistant/components/watson_iot/*
homeassistant/components/watson_tts/tts.py
homeassistant/components/waze_travel_time/__init__.py
homeassistant/components/waze_travel_time/helpers.py
homeassistant/components/waze_travel_time/sensor.py
homeassistant/components/webostv/*
homeassistant/components/whois/sensor.py

28
homeassistant/components/waze_travel_time/__init__.py
View File

@ -1 +1,29 @@
"""The waze_travel_time component."""
import asyncio
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
PLATFORMS = ["sensor"]
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Load the saved entities."""
for platform in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(config_entry, platform)
)
return True
async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(config_entry, platform)
for platform in PLATFORMS
]
)
)

149
homeassistant/components/waze_travel_time/config_flow.py Normal file
View File

@ -0,0 +1,149 @@
"""Config flow for Waze Travel Time integration."""
import logging
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_NAME, CONF_REGION
from homeassistant.core import callback
import homeassistant.helpers.config_validation as cv
from homeassistant.util import slugify
from .const import (
CONF_AVOID_FERRIES,
CONF_AVOID_SUBSCRIPTION_ROADS,
CONF_AVOID_TOLL_ROADS,
CONF_DESTINATION,
CONF_EXCL_FILTER,
CONF_INCL_FILTER,
CONF_ORIGIN,
CONF_REALTIME,
CONF_UNITS,
CONF_VEHICLE_TYPE,
DEFAULT_NAME,
DOMAIN,
REGIONS,
UNITS,
VEHICLE_TYPES,
)
from .helpers import is_valid_config_entry
_LOGGER = logging.getLogger(__name__)
class WazeOptionsFlow(config_entries.OptionsFlow):
"""Handle an options flow for Waze Travel Time."""
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Initialize waze 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:
return self.async_create_entry(title="", data=user_input)
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(
{
vol.Optional(
CONF_INCL_FILTER,
default=self.config_entry.options.get(CONF_INCL_FILTER),
): cv.string,
vol.Optional(
CONF_EXCL_FILTER,
default=self.config_entry.options.get(CONF_EXCL_FILTER),
): cv.string,
vol.Optional(
CONF_REALTIME,
default=self.config_entry.options[CONF_REALTIME],
): cv.boolean,
vol.Optional(
CONF_VEHICLE_TYPE,
default=self.config_entry.options[CONF_VEHICLE_TYPE],
): vol.In(VEHICLE_TYPES),
vol.Optional(
CONF_UNITS,
default=self.config_entry.options[CONF_UNITS],
): vol.In(UNITS),
vol.Optional(
CONF_AVOID_TOLL_ROADS,
default=self.config_entry.options[CONF_AVOID_TOLL_ROADS],
): cv.boolean,
vol.Optional(
CONF_AVOID_SUBSCRIPTION_ROADS,
default=self.config_entry.options[
CONF_AVOID_SUBSCRIPTION_ROADS
],
): cv.boolean,
vol.Optional(
CONF_AVOID_FERRIES,
default=self.config_entry.options[CONF_AVOID_FERRIES],
): cv.boolean,
}
),
)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Waze Travel Time."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
@staticmethod
@callback
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
) -> WazeOptionsFlow:
"""Get the options flow for this handler."""
return WazeOptionsFlow(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_ORIGIN],
user_input[CONF_DESTINATION],
user_input[CONF_REGION],
):
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_ORIGIN): cv.string,
vol.Required(CONF_DESTINATION): cv.string,
vol.Required(CONF_REGION): vol.In(REGIONS),
}
),
errors=errors,
)
async_step_import = async_step_user

40
homeassistant/components/waze_travel_time/const.py Normal file
View File

@ -0,0 +1,40 @@
"""Constants for waze_travel_time."""
from homeassistant.const import CONF_UNIT_SYSTEM_IMPERIAL, CONF_UNIT_SYSTEM_METRIC
DOMAIN = "waze_travel_time"
ATTR_DESTINATION = "destination"
ATTR_DURATION = "duration"
ATTR_DISTANCE = "distance"
ATTR_ORIGIN = "origin"
ATTR_ROUTE = "route"
ATTRIBUTION = "Powered by Waze"
CONF_DESTINATION = "destination"
CONF_ORIGIN = "origin"
CONF_INCL_FILTER = "incl_filter"
CONF_EXCL_FILTER = "excl_filter"
CONF_REALTIME = "realtime"
CONF_UNITS = "units"
CONF_VEHICLE_TYPE = "vehicle_type"
CONF_AVOID_TOLL_ROADS = "avoid_toll_roads"
CONF_AVOID_SUBSCRIPTION_ROADS = "avoid_subscription_roads"
CONF_AVOID_FERRIES = "avoid_ferries"
DEFAULT_NAME = "Waze Travel Time"
DEFAULT_REALTIME = True
DEFAULT_VEHICLE_TYPE = "car"
DEFAULT_AVOID_TOLL_ROADS = False
DEFAULT_AVOID_SUBSCRIPTION_ROADS = False
DEFAULT_AVOID_FERRIES = False
ICON = "mdi:car"
UNITS = [CONF_UNIT_SYSTEM_METRIC, CONF_UNIT_SYSTEM_IMPERIAL]
REGIONS = ["US", "NA", "EU", "IL", "AU"]
VEHICLE_TYPES = ["car", "taxi", "motorcycle"]
# Attempt to find entity_id without finding address with period.
ENTITY_ID_PATTERN = "(?<![a-zA-Z0-9 ])[a-z_]+[.][a-zA-Z0-9_]+"

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

@ -0,0 +1,72 @@
"""Helpers for Waze Travel Time integration."""
import re
from WazeRouteCalculator import WazeRouteCalculator, WRCError
from homeassistant.components.waze_travel_time.const import ENTITY_ID_PATTERN
from homeassistant.const import ATTR_LATITUDE, ATTR_LONGITUDE
from homeassistant.helpers import location
def is_valid_config_entry(hass, logger, origin, destination, region):
"""Return whether the config entry data is valid."""
origin = resolve_location(hass, logger, origin)
destination = resolve_location(hass, logger, destination)
try:
WazeRouteCalculator(origin, destination, region).calc_all_routes_info()
except WRCError:
return False
return True
def resolve_location(hass, logger, loc):
"""Resolve a location."""
if re.fullmatch(ENTITY_ID_PATTERN, loc):
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_id."""
state = hass.states.get(entity_id)
if state is None:
logger.error("Unable to find entity %s", entity_id)
return None
# Check if the entity has location attributes.
if location.has_location(state):
logger.debug("Getting %s location", entity_id)
return _get_location_from_attributes(state)
# Check if device is inside a zone.
zone_state = hass.states.get(f"zone.{state.state}")
if location.has_location(zone_state):
logger.debug(
"%s is in %s, getting zone location", entity_id, zone_state.entity_id
)
return _get_location_from_attributes(zone_state)
# If zone was not found in state then use the state as the location.
if entity_id.startswith("sensor."):
return state.state
# When everything fails just return nothing.
return None
def resolve_zone(hass, friendly_name):
"""Get a lat/long from a zones friendly_name."""
states = hass.states.all()
for state in states:
if state.domain == "zone" and state.name == friendly_name:
return _get_location_from_attributes(state)
return friendly_name
def _get_location_from_attributes(state):
"""Get the lat/long string from an states attributes."""
attr = state.attributes
return "{},{}".format(attr.get(ATTR_LATITUDE), attr.get(ATTR_LONGITUDE))

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

@ -2,6 +2,9 @@
"domain": "waze_travel_time",
"name": "Waze Travel Time",
"documentation": "https://www.home-assistant.io/integrations/waze_travel_time",
"requirements": ["WazeRouteCalculator==0.12"],
"codeowners": []
"requirements": [
"WazeRouteCalculator==0.12"
],
"codeowners": [],
"config_flow": true
}

341
homeassistant/components/waze_travel_time/sensor.py
View File

@ -1,61 +1,62 @@
"""Support for Waze travel time sensor."""
from __future__ import annotations
from datetime import timedelta
import logging
import re
from typing import Any, Callable
import WazeRouteCalculator
from WazeRouteCalculator import WazeRouteCalculator, WRCError
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_NAME,
CONF_REGION,
CONF_UNIT_SYSTEM_IMPERIAL,
CONF_UNIT_SYSTEM_METRIC,
EVENT_HOMEASSISTANT_START,
TIME_MINUTES,
)
from homeassistant.helpers import location
from homeassistant.core import Config, CoreState, HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
import homeassistant.helpers.config_validation as cv
from .const import (
ATTR_DESTINATION,
ATTR_DISTANCE,
ATTR_DURATION,
ATTR_ORIGIN,
ATTR_ROUTE,
ATTRIBUTION,
CONF_AVOID_FERRIES,
CONF_AVOID_SUBSCRIPTION_ROADS,
CONF_AVOID_TOLL_ROADS,
CONF_DESTINATION,
CONF_EXCL_FILTER,
CONF_INCL_FILTER,
CONF_ORIGIN,
CONF_REALTIME,
CONF_UNITS,
CONF_VEHICLE_TYPE,
DEFAULT_AVOID_FERRIES,
DEFAULT_AVOID_SUBSCRIPTION_ROADS,
DEFAULT_AVOID_TOLL_ROADS,
DEFAULT_NAME,
DEFAULT_REALTIME,
DEFAULT_VEHICLE_TYPE,
DOMAIN,
ENTITY_ID_PATTERN,
ICON,
REGIONS,
UNITS,
VEHICLE_TYPES,
)
from .helpers import get_location_from_entity, is_valid_config_entry, resolve_zone
_LOGGER = logging.getLogger(__name__)
ATTR_DESTINATION = "destination"
ATTR_DURATION = "duration"
ATTR_DISTANCE = "distance"
ATTR_ORIGIN = "origin"
ATTR_ROUTE = "route"
ATTRIBUTION = "Powered by Waze"
CONF_DESTINATION = "destination"
CONF_ORIGIN = "origin"
CONF_INCL_FILTER = "incl_filter"
CONF_EXCL_FILTER = "excl_filter"
CONF_REALTIME = "realtime"
CONF_UNITS = "units"
CONF_VEHICLE_TYPE = "vehicle_type"
CONF_AVOID_TOLL_ROADS = "avoid_toll_roads"
CONF_AVOID_SUBSCRIPTION_ROADS = "avoid_subscription_roads"
CONF_AVOID_FERRIES = "avoid_ferries"
DEFAULT_NAME = "Waze Travel Time"
DEFAULT_REALTIME = True
DEFAULT_VEHICLE_TYPE = "car"
DEFAULT_AVOID_TOLL_ROADS = False
DEFAULT_AVOID_SUBSCRIPTION_ROADS = False
DEFAULT_AVOID_FERRIES = False
ICON = "mdi:car"
UNITS = [CONF_UNIT_SYSTEM_METRIC, CONF_UNIT_SYSTEM_IMPERIAL]
REGIONS = ["US", "NA", "EU", "IL", "AU"]
VEHICLE_TYPES = ["car", "taxi", "motorcycle"]
SCAN_INTERVAL = timedelta(minutes=5)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
@ -63,7 +64,7 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
vol.Required(CONF_ORIGIN): cv.string,
vol.Required(CONF_DESTINATION): cv.string,
vol.Required(CONF_REGION): vol.In(REGIONS),
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_NAME): cv.string,
vol.Optional(CONF_INCL_FILTER): cv.string,
vol.Optional(CONF_EXCL_FILTER): cv.string,
vol.Optional(CONF_REALTIME, default=DEFAULT_REALTIME): cv.boolean,
@ -82,75 +83,119 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
)
def setup_platform(hass, config, add_entities, discovery_info=None):
async def async_setup_platform(
hass: HomeAssistant, config: Config, async_add_entities, discovery_info=None
):
"""Set up the Waze travel time sensor platform."""
destination = config.get(CONF_DESTINATION)
name = config.get(CONF_NAME)
origin = config.get(CONF_ORIGIN)
region = config.get(CONF_REGION)
incl_filter = config.get(CONF_INCL_FILTER)
excl_filter = config.get(CONF_EXCL_FILTER)
realtime = config.get(CONF_REALTIME)
vehicle_type = config.get(CONF_VEHICLE_TYPE)
avoid_toll_roads = config.get(CONF_AVOID_TOLL_ROADS)
avoid_subscription_roads = config.get(CONF_AVOID_SUBSCRIPTION_ROADS)
avoid_ferries = config.get(CONF_AVOID_FERRIES)
units = config.get(CONF_UNITS, hass.config.units.name)
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=config,
)
)
_LOGGER.warning(
"Your Waze configuration has been imported into the UI; "
"please remove it from configuration.yaml as support for it "
"will be removed in a future release"
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: Callable[[list[SensorEntity], bool], None],
) -> None:
"""Set up a Waze travel time sensor entry."""
defaults = {
CONF_REALTIME: DEFAULT_REALTIME,
CONF_VEHICLE_TYPE: DEFAULT_VEHICLE_TYPE,
CONF_UNITS: hass.config.units.name,
CONF_AVOID_FERRIES: DEFAULT_AVOID_FERRIES,
CONF_AVOID_SUBSCRIPTION_ROADS: DEFAULT_AVOID_SUBSCRIPTION_ROADS,
CONF_AVOID_TOLL_ROADS: DEFAULT_AVOID_TOLL_ROADS,
}
name = None
if not config_entry.options:
new_data = config_entry.data.copy()
name = new_data.pop(CONF_NAME, None)
options = {}
for key in [
CONF_INCL_FILTER,
CONF_EXCL_FILTER,
CONF_REALTIME,
CONF_VEHICLE_TYPE,
CONF_AVOID_TOLL_ROADS,
CONF_AVOID_SUBSCRIPTION_ROADS,
CONF_AVOID_FERRIES,
CONF_UNITS,
]:
if key in new_data:
options[key] = new_data.pop(key)
elif key in defaults:
options[key] = defaults[key]
hass.config_entries.async_update_entry(
config_entry, data=new_data, options=options
)
destination = config_entry.data[CONF_DESTINATION]
origin = config_entry.data[CONF_ORIGIN]
region = config_entry.data[CONF_REGION]
name = name or f"{DEFAULT_NAME}: {origin} -> {destination}"
if not await hass.async_add_executor_job(
is_valid_config_entry, hass, _LOGGER, origin, destination, region
):
raise ConfigEntryNotReady
data = WazeTravelTimeData(
None,
None,
region,
incl_filter,
excl_filter,
realtime,
units,
vehicle_type,
avoid_toll_roads,
avoid_subscription_roads,
avoid_ferries,
config_entry,
)
sensor = WazeTravelTime(name, origin, destination, data)
sensor = WazeTravelTime(config_entry.unique_id, name, origin, destination, data)
add_entities([sensor])
# Wait until start event is sent to load this component.
hass.bus.listen_once(EVENT_HOMEASSISTANT_START, lambda _: sensor.update())
def _get_location_from_attributes(state):
"""Get the lat/long string from an states attributes."""
attr = state.attributes
return "{},{}".format(attr.get(ATTR_LATITUDE), attr.get(ATTR_LONGITUDE))
async_add_entities([sensor], False)
class WazeTravelTime(SensorEntity):
"""Representation of a Waze travel time sensor."""
def __init__(self, name, origin, destination, waze_data):
def __init__(self, unique_id, name, origin, destination, waze_data):
"""Initialize the Waze travel time sensor."""
self._name = name
self._unique_id = unique_id
self._waze_data = waze_data
self._name = name
self._state = None
self._origin_entity_id = None
self._destination_entity_id = None
# Attempt to find entity_id without finding address with period.
pattern = "(?<![a-zA-Z0-9 ])[a-z_]+[.][a-zA-Z0-9_]+"
if re.fullmatch(pattern, origin):
cmpl_re = re.compile(ENTITY_ID_PATTERN)
if cmpl_re.fullmatch(origin):
_LOGGER.debug("Found origin source entity %s", origin)
self._origin_entity_id = origin
else:
self._waze_data.origin = origin
if re.fullmatch(pattern, destination):
if cmpl_re.fullmatch(destination):
_LOGGER.debug("Found destination source entity %s", destination)
self._destination_entity_id = destination
else:
self._waze_data.destination = destination
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 name(self):
"""Return the name of the sensor."""
@ -188,150 +233,118 @@ class WazeTravelTime(SensorEntity):
res[ATTR_DESTINATION] = self._waze_data.destination
return res
def _get_location_from_entity(self, entity_id):
"""Get the location from the entity_id."""
state = self.hass.states.get(entity_id)
if state is None:
_LOGGER.error("Unable to find entity %s", entity_id)
return None
# Check if the entity has location attributes.
if location.has_location(state):
_LOGGER.debug("Getting %s location", entity_id)
return _get_location_from_attributes(state)
# Check if device is inside a zone.
zone_state = self.hass.states.get(f"zone.{state.state}")
if location.has_location(zone_state):
_LOGGER.debug(
"%s is in %s, getting zone location", entity_id, zone_state.entity_id
)
return _get_location_from_attributes(zone_state)
# If zone was not found in state then use the state as the location.
if entity_id.startswith("sensor."):
return state.state
# When everything fails just return nothing.
return None
def _resolve_zone(self, friendly_name):
"""Get a lat/long from a zones friendly_name."""
states = self.hass.states.all()
for state in states:
if state.domain == "zone" and state.name == friendly_name:
return _get_location_from_attributes(state)
return friendly_name
async def first_update(self, _=None):
"""Run first update and write state."""
await self.hass.async_add_executor_job(self.update)
self.async_write_ha_state()
def update(self):
"""Fetch new state data for the sensor."""
_LOGGER.debug("Fetching Route for %s", self._name)
# Get origin latitude and longitude from entity_id.
if self._origin_entity_id is not None:
self._waze_data.origin = self._get_location_from_entity(
self._origin_entity_id
self._waze_data.origin = get_location_from_entity(
self.hass, _LOGGER, self._origin_entity_id
)
# Get destination latitude and longitude from entity_id.
if self._destination_entity_id is not None:
self._waze_data.destination = self._get_location_from_entity(
self._destination_entity_id
self._waze_data.destination = get_location_from_entity(
self.hass, _LOGGER, self._destination_entity_id
)
# Get origin from zone name.
self._waze_data.origin = self._resolve_zone(self._waze_data.origin)
self._waze_data.origin = resolve_zone(self.hass, self._waze_data.origin)
# Get destination from zone name.
self._waze_data.destination = self._resolve_zone(self._waze_data.destination)
self._waze_data.destination = resolve_zone(
self.hass, self._waze_data.destination
)
self._waze_data.update()
@property
def device_info(self) -> dict[str, Any] | None:
"""Return device specific attributes."""
return {
"name": "Waze",
"identifiers": {(DOMAIN, DOMAIN)},
"entry_type": "service",
}
@property
def unique_id(self) -> str:
"""Return unique ID of entity."""
return self._unique_id
class WazeTravelTimeData:
"""WazeTravelTime Data object."""
def __init__(
self,
origin,
destination,
region,
include,
exclude,
realtime,
units,
vehicle_type,
avoid_toll_roads,
avoid_subscription_roads,
avoid_ferries,
):
def __init__(self, origin, destination, region, config_entry):
"""Set up WazeRouteCalculator."""
self._calc = WazeRouteCalculator
self.origin = origin
self.destination = destination
self.region = region
self.include = include
self.exclude = exclude
self.realtime = realtime
self.units = units
self.config_entry = config_entry
self.duration = None
self.distance = None
self.route = None
self.avoid_toll_roads = avoid_toll_roads
self.avoid_subscription_roads = avoid_subscription_roads
self.avoid_ferries = avoid_ferries
# Currently WazeRouteCalc only supports PRIVATE, TAXI, MOTORCYCLE.
if vehicle_type.upper() == "CAR":
# Empty means PRIVATE for waze which translates to car.
self.vehicle_type = ""
else:
self.vehicle_type = vehicle_type.upper()
def update(self):
"""Update WazeRouteCalculator Sensor."""
if self.origin is not None and self.destination is not None:
# Grab options on every update
incl_filter = self.config_entry.options.get(CONF_INCL_FILTER)
excl_filter = self.config_entry.options.get(CONF_EXCL_FILTER)
realtime = self.config_entry.options[CONF_REALTIME]
vehicle_type = self.config_entry.options[CONF_VEHICLE_TYPE]
vehicle_type = "" if vehicle_type.upper() == "CAR" else vehicle_type.upper()
avoid_toll_roads = self.config_entry.options[CONF_AVOID_TOLL_ROADS]
avoid_subscription_roads = self.config_entry.options[
CONF_AVOID_SUBSCRIPTION_ROADS
]
avoid_ferries = self.config_entry.options[CONF_AVOID_FERRIES]
units = self.config_entry.options[CONF_UNITS]
try:
params = self._calc.WazeRouteCalculator(
params = WazeRouteCalculator(
self.origin,
self.destination,
self.region,
self.vehicle_type,
self.avoid_toll_roads,
self.avoid_subscription_roads,
self.avoid_ferries,
vehicle_type,
avoid_toll_roads,
avoid_subscription_roads,
avoid_ferries,
)
routes = params.calc_all_routes_info(real_time=self.realtime)
routes = params.calc_all_routes_info(real_time=realtime)
if self.include is not None:
if incl_filter is not None:
routes = {
k: v
for k, v in routes.items()
if self.include.lower() in k.lower()
if incl_filter.lower() in k.lower()
}
if self.exclude is not None:
if excl_filter is not None:
routes = {
k: v
for k, v in routes.items()
if self.exclude.lower() not in k.lower()
if excl_filter.lower() not in k.lower()
}
route = list(routes)[0]
self.duration, distance = routes[route]
if self.units == CONF_UNIT_SYSTEM_IMPERIAL:
if units == CONF_UNIT_SYSTEM_IMPERIAL:
# Convert to miles.
self.distance = distance / 1.609
else:
self.distance = distance
self.route = route
except self._calc.WRCError as exp:
except WRCError as exp:
_LOGGER.warning("Error on retrieving data: %s", exp)
return
except KeyError:

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

@ -0,0 +1,38 @@
{
"title": "Waze Travel Time",
"config": {
"step": {
"user": {
"description": "For Origin and Destination, enter the address or the GPS coordinates of the location (GPS coordinates has to be separated by a comma). You can also enter an entity id which provides this information in its state, an entity id with latitude and longitude attributes, or zone friendly name.",
"data": {
"origin": "Origin",
"destination": "Destination",
"region": "Region"
}
}
},
"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": "The `substring` inputs will allow you to force the integration to use a particular route or avoid a particular route in its time travel calculation.",
"data": {
"units": "Units",
"vehicle_type": "Vehicle Type",
"incl_filter": "Substring in Description of Selected Route",
"excl_filter": "Substring NOT in Description of Selected Route",
"realtime": "Realtime Travel Time?",
"avoid_toll_roads": "Avoid Toll Roads?",
"avoid_ferries": "Avoid Ferries?",
"avoid_subscription_roads": "Avoid Roads Needing a Vignette / Subscription?"
}
}
}
}
}

35
homeassistant/components/waze_travel_time/translations/en.json Normal file
View File

@ -0,0 +1,35 @@
{
"config": {
"abort": {
"already_configured": "Location is already configured"
},
"step": {
"user": {
"data": {
"destination": "Destination",
"origin": "Origin",
"region": "Region"
},
"description": "For Origin and Destination, enter the address or the GPS coordinates of the location (GPS coordinates has to be separated by a comma). You can also enter an entity id which provides this information in its state, an entity id with latitude and longitude attributes, or zone friendly name."
}
}
},
"options": {
"step": {
"init": {
"data": {
"avoid_ferries": "Avoid Ferries?",
"avoid_subscription_roads": "Avoid Roads Needing a Vignette / Subscription?",
"avoid_toll_roads": "Avoid Toll Roads?",
"excl_filter": "Substring NOT in Description of Selected Route",
"incl_filter": "Substring in Description of Selected Route",
"realtime": "Realtime Travel Time?",
"units": "Units",
"vehicle_type": "Vehicle Type"
},
"description": "The `substring` inputs will allow you to force the integration to use a particular route or avoid a particular route in its time travel calculation."
}
}
},
"title": "Waze Travel Time"
}

1
homeassistant/generated/config_flows.py
View File

@ -258,6 +258,7 @@ FLOWS = [
"vilfo",
"vizio",
"volumio",
"waze_travel_time",
"wemo",
"wiffi",
"wilight",

3
requirements_test_all.txt
View File

@ -38,6 +38,9 @@ RtmAPI==0.7.2
# homeassistant.components.onvif
WSDiscovery==2.0.0
# homeassistant.components.waze_travel_time
WazeRouteCalculator==0.12
# homeassistant.components.abode
abodepy==1.2.0

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

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

57
tests/components/waze_travel_time/conftest.py Normal file
View File

@ -0,0 +1,57 @@
"""Fixtures for Waze Travel Time tests."""
from unittest.mock import patch
from WazeRouteCalculator import WRCError
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.waze_travel_time.helpers.WazeRouteCalculator"
) as mock_wrc:
obj = mock_wrc.return_value
obj.calc_all_routes_info.return_value = None
yield
@pytest.fixture(name="bypass_setup")
def bypass_setup_fixture():
"""Bypass entry setup."""
with patch(
"homeassistant.components.waze_travel_time.async_setup_entry",
return_value=True,
):
yield
@pytest.fixture(name="mock_update")
def mock_update_fixture():
"""Mock an update to the sensor."""
with patch(
"homeassistant.components.waze_travel_time.sensor.WazeRouteCalculator.calc_all_routes_info",
return_value={"My route": (150, 300)},
):
yield
@pytest.fixture(name="invalidate_config_entry")
def invalidate_config_entry_fixture():
"""Return invalid config entry."""
with patch(
"homeassistant.components.waze_travel_time.helpers.WazeRouteCalculator"
) as mock_wrc:
obj = mock_wrc.return_value
obj.calc_all_routes_info.return_value = {}
obj.calc_all_routes_info.side_effect = WRCError("test")
yield

205
tests/components/waze_travel_time/test_config_flow.py Normal file
View File

@ -0,0 +1,205 @@
"""Test the Waze Travel Time config flow."""
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.waze_travel_time.const import (
CONF_AVOID_FERRIES,
CONF_AVOID_SUBSCRIPTION_ROADS,
CONF_AVOID_TOLL_ROADS,
CONF_DESTINATION,
CONF_EXCL_FILTER,
CONF_INCL_FILTER,
CONF_ORIGIN,
CONF_REALTIME,
CONF_UNITS,
CONF_VEHICLE_TYPE,
DEFAULT_NAME,
DOMAIN,
)
from homeassistant.const import CONF_REGION, 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_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_REGION: "US",
},
)
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result2["title"] == f"{DEFAULT_NAME}: location1 -> location2"
assert result2["data"] == {
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_REGION: "US",
}
async def test_options(hass, validate_config_entry, mock_update):
"""Test options flow."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_REGION: "US",
},
)
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_AVOID_FERRIES: True,
CONF_AVOID_SUBSCRIPTION_ROADS: True,
CONF_AVOID_TOLL_ROADS: True,
CONF_EXCL_FILTER: "exclude",
CONF_INCL_FILTER: "include",
CONF_REALTIME: False,
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_VEHICLE_TYPE: "taxi",
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == ""
assert result["data"] == {
CONF_AVOID_FERRIES: True,
CONF_AVOID_SUBSCRIPTION_ROADS: True,
CONF_AVOID_TOLL_ROADS: True,
CONF_EXCL_FILTER: "exclude",
CONF_INCL_FILTER: "include",
CONF_REALTIME: False,
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_VEHICLE_TYPE: "taxi",
}
assert entry.options == {
CONF_AVOID_FERRIES: True,
CONF_AVOID_SUBSCRIPTION_ROADS: True,
CONF_AVOID_TOLL_ROADS: True,
CONF_EXCL_FILTER: "exclude",
CONF_INCL_FILTER: "include",
CONF_REALTIME: False,
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_VEHICLE_TYPE: "taxi",
}
async def test_import(hass, validate_config_entry, mock_update):
"""Test import for config flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_REGION: "US",
CONF_AVOID_FERRIES: True,
CONF_AVOID_SUBSCRIPTION_ROADS: True,
CONF_AVOID_TOLL_ROADS: True,
CONF_EXCL_FILTER: "exclude",
CONF_INCL_FILTER: "include",
CONF_REALTIME: False,
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_VEHICLE_TYPE: "taxi",
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
await hass.async_block_till_done()
entry = hass.config_entries.async_entries(DOMAIN)[0]
assert entry.data == {
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_REGION: "US",
}
assert entry.options == {
CONF_AVOID_FERRIES: True,
CONF_AVOID_SUBSCRIPTION_ROADS: True,
CONF_AVOID_TOLL_ROADS: True,
CONF_EXCL_FILTER: "exclude",
CONF_INCL_FILTER: "include",
CONF_REALTIME: False,
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_VEHICLE_TYPE: "taxi",
}
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_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_REGION: "US",
},
)
await hass.async_block_till_done()
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_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_REGION: "US",
},
)
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result2["reason"] == "already_configured"
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_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_REGION: "US",
},
)
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result2["errors"] == {"base": "cannot_connect"}