mirror of
https://github.com/home-assistant/core.git
synced 2025-07-23 13:17:32 +00:00
Use DataUpdateCoordinator in here_travel_time (#61398)
* Add DataUpdateCoordinator * Use TypedDict for extra_state_attributes * Extend DataUpdateCoordinator * Use platform enum * Use attribution property * Use relative imports * Revert native_value logic * Directly return result in build_hass_attribution * Correctly declare traffic_mode as bool * Use self._attr_* * Fix mypy issues * Update homeassistant/components/here_travel_time/__init__.py Co-authored-by: Allen Porter <allen.porter@gmail.com> * Update homeassistant/components/here_travel_time/__init__.py Co-authored-by: Allen Porter <allen.porter@gmail.com> * Update homeassistant/components/here_travel_time/sensor.py Co-authored-by: Allen Porter <allen.porter@gmail.com> * blacken * from datetime import time * remove none check * Move dataclasses to models.py * Set destination to now if None * Add mypy error code Co-authored-by: Allen Porter <allen.porter@gmail.com>
This commit is contained in:
parent
ad7a2c298b
commit
adbacdd5c2
@ -1 +1,168 @@
|
||||
"""The here_travel_time component."""
|
||||
"""The HERE Travel Time integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, time, timedelta
|
||||
import logging
|
||||
|
||||
import async_timeout
|
||||
from herepy import NoRouteFoundError, RouteMode, RoutingApi, RoutingResponse
|
||||
|
||||
from homeassistant.const import ATTR_ATTRIBUTION, CONF_UNIT_SYSTEM_IMPERIAL, Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.location import find_coordinates
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
from homeassistant.util import dt
|
||||
|
||||
from .const import (
|
||||
ATTR_DESTINATION,
|
||||
ATTR_DESTINATION_NAME,
|
||||
ATTR_DISTANCE,
|
||||
ATTR_DURATION,
|
||||
ATTR_DURATION_IN_TRAFFIC,
|
||||
ATTR_ORIGIN,
|
||||
ATTR_ORIGIN_NAME,
|
||||
ATTR_ROUTE,
|
||||
DEFAULT_SCAN_INTERVAL,
|
||||
DOMAIN,
|
||||
NO_ROUTE_ERROR_MESSAGE,
|
||||
TRAFFIC_MODE_ENABLED,
|
||||
TRAVEL_MODES_VEHICLE,
|
||||
)
|
||||
from .model import HERERoutingData, HERETravelTimeConfig
|
||||
|
||||
PLATFORMS = [Platform.SENSOR]
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class HereTravelTimeDataUpdateCoordinator(DataUpdateCoordinator):
|
||||
"""HERETravelTime DataUpdateCoordinator."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
api: RoutingApi,
|
||||
config: HERETravelTimeConfig,
|
||||
) -> None:
|
||||
"""Initialize."""
|
||||
super().__init__(
|
||||
hass,
|
||||
_LOGGER,
|
||||
name=DOMAIN,
|
||||
update_interval=timedelta(seconds=DEFAULT_SCAN_INTERVAL),
|
||||
)
|
||||
self._api = api
|
||||
self.config = config
|
||||
|
||||
async def _async_update_data(self) -> HERERoutingData | None:
|
||||
"""Get the latest data from the HERE Routing API."""
|
||||
try:
|
||||
async with async_timeout.timeout(10):
|
||||
return await self.hass.async_add_executor_job(self._update)
|
||||
except NoRouteFoundError as error:
|
||||
raise UpdateFailed(NO_ROUTE_ERROR_MESSAGE) from error
|
||||
|
||||
def _update(self) -> HERERoutingData | None:
|
||||
"""Get the latest data from the HERE Routing API."""
|
||||
if self.config.origin_entity_id is not None:
|
||||
origin = find_coordinates(self.hass, self.config.origin_entity_id)
|
||||
else:
|
||||
origin = self.config.origin
|
||||
|
||||
if self.config.destination_entity_id is not None:
|
||||
destination = find_coordinates(self.hass, self.config.destination_entity_id)
|
||||
else:
|
||||
destination = self.config.destination
|
||||
if destination is not None and origin is not None:
|
||||
here_formatted_destination = destination.split(",")
|
||||
here_formatted_origin = origin.split(",")
|
||||
arrival: str | None = None
|
||||
departure: str | None = None
|
||||
if self.config.arrival is not None:
|
||||
arrival = convert_time_to_isodate(self.config.arrival)
|
||||
if self.config.departure is not None:
|
||||
departure = convert_time_to_isodate(self.config.departure)
|
||||
|
||||
if arrival is None and departure is None:
|
||||
departure = "now"
|
||||
|
||||
_LOGGER.debug(
|
||||
"Requesting route for origin: %s, destination: %s, route_mode: %s, mode: %s, traffic_mode: %s, arrival: %s, departure: %s",
|
||||
here_formatted_origin,
|
||||
here_formatted_destination,
|
||||
RouteMode[self.config.route_mode],
|
||||
RouteMode[self.config.travel_mode],
|
||||
RouteMode[TRAFFIC_MODE_ENABLED],
|
||||
arrival,
|
||||
departure,
|
||||
)
|
||||
|
||||
response: RoutingResponse = self._api.public_transport_timetable(
|
||||
here_formatted_origin,
|
||||
here_formatted_destination,
|
||||
True,
|
||||
[
|
||||
RouteMode[self.config.route_mode],
|
||||
RouteMode[self.config.travel_mode],
|
||||
RouteMode[TRAFFIC_MODE_ENABLED],
|
||||
],
|
||||
arrival=arrival,
|
||||
departure=departure,
|
||||
)
|
||||
|
||||
_LOGGER.debug(
|
||||
"Raw response is: %s", response.response # pylint: disable=no-member
|
||||
)
|
||||
|
||||
attribution: str | None = None
|
||||
if "sourceAttribution" in response.response: # pylint: disable=no-member
|
||||
attribution = build_hass_attribution(
|
||||
response.response.get("sourceAttribution")
|
||||
) # pylint: disable=no-member
|
||||
route: list = response.response["route"] # pylint: disable=no-member
|
||||
summary: dict = route[0]["summary"]
|
||||
waypoint: list = route[0]["waypoint"]
|
||||
distance: float = summary["distance"]
|
||||
traffic_time: float = summary["baseTime"]
|
||||
if self.config.travel_mode in TRAVEL_MODES_VEHICLE:
|
||||
traffic_time = summary["trafficTime"]
|
||||
if self.config.units == CONF_UNIT_SYSTEM_IMPERIAL:
|
||||
# Convert to miles.
|
||||
distance = distance / 1609.344
|
||||
else:
|
||||
# Convert to kilometers
|
||||
distance = distance / 1000
|
||||
return HERERoutingData(
|
||||
{
|
||||
ATTR_ATTRIBUTION: attribution,
|
||||
ATTR_DURATION: summary["baseTime"] / 60, # type: ignore[misc]
|
||||
ATTR_DURATION_IN_TRAFFIC: traffic_time / 60,
|
||||
ATTR_DISTANCE: distance,
|
||||
ATTR_ROUTE: response.route_short,
|
||||
ATTR_ORIGIN: ",".join(here_formatted_origin),
|
||||
ATTR_DESTINATION: ",".join(here_formatted_destination),
|
||||
ATTR_ORIGIN_NAME: waypoint[0]["mappedRoadName"],
|
||||
ATTR_DESTINATION_NAME: waypoint[1]["mappedRoadName"],
|
||||
}
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def build_hass_attribution(source_attribution: dict) -> str | None:
|
||||
"""Build a hass frontend ready string out of the sourceAttribution."""
|
||||
if (suppliers := source_attribution.get("supplier")) is not None:
|
||||
supplier_titles = []
|
||||
for supplier in suppliers:
|
||||
if (title := supplier.get("title")) is not None:
|
||||
supplier_titles.append(title)
|
||||
joined_supplier_titles = ",".join(supplier_titles)
|
||||
return f"With the support of {joined_supplier_titles}. All information is provided without warranty of any kind."
|
||||
return None
|
||||
|
||||
|
||||
def convert_time_to_isodate(simple_time: time) -> str:
|
||||
"""Take a time like 08:00:00 and combine it with the current date."""
|
||||
combined = datetime.combine(dt.start_of_local_day(), simple_time)
|
||||
if combined < datetime.now():
|
||||
combined = combined + timedelta(days=1)
|
||||
return combined.isoformat()
|
||||
|
77
homeassistant/components/here_travel_time/const.py
Normal file
77
homeassistant/components/here_travel_time/const.py
Normal file
@ -0,0 +1,77 @@
|
||||
"""Constants for the HERE Travel Time integration."""
|
||||
from homeassistant.const import (
|
||||
CONF_UNIT_SYSTEM,
|
||||
CONF_UNIT_SYSTEM_IMPERIAL,
|
||||
CONF_UNIT_SYSTEM_METRIC,
|
||||
)
|
||||
|
||||
DOMAIN = "here_travel_time"
|
||||
DEFAULT_SCAN_INTERVAL = 300
|
||||
|
||||
CONF_DESTINATION = "destination"
|
||||
CONF_ORIGIN = "origin"
|
||||
CONF_TRAFFIC_MODE = "traffic_mode"
|
||||
CONF_ROUTE_MODE = "route_mode"
|
||||
CONF_ARRIVAL = "arrival"
|
||||
CONF_DEPARTURE = "departure"
|
||||
CONF_ARRIVAL_TIME = "arrival_time"
|
||||
CONF_DEPARTURE_TIME = "departure_time"
|
||||
CONF_TIME_TYPE = "time_type"
|
||||
CONF_TIME = "time"
|
||||
|
||||
ARRIVAL_TIME = "Arrival Time"
|
||||
DEPARTURE_TIME = "Departure Time"
|
||||
TIME_TYPES = [ARRIVAL_TIME, DEPARTURE_TIME]
|
||||
|
||||
DEFAULT_NAME = "HERE Travel Time"
|
||||
|
||||
TRACKABLE_DOMAINS = ["device_tracker", "sensor", "zone", "person"]
|
||||
|
||||
TRAVEL_MODE_BICYCLE = "bicycle"
|
||||
TRAVEL_MODE_CAR = "car"
|
||||
TRAVEL_MODE_PEDESTRIAN = "pedestrian"
|
||||
TRAVEL_MODE_PUBLIC = "publicTransport"
|
||||
TRAVEL_MODE_PUBLIC_TIME_TABLE = "publicTransportTimeTable"
|
||||
TRAVEL_MODE_TRUCK = "truck"
|
||||
TRAVEL_MODES = [
|
||||
TRAVEL_MODE_BICYCLE,
|
||||
TRAVEL_MODE_CAR,
|
||||
TRAVEL_MODE_PEDESTRIAN,
|
||||
TRAVEL_MODE_PUBLIC,
|
||||
TRAVEL_MODE_PUBLIC_TIME_TABLE,
|
||||
TRAVEL_MODE_TRUCK,
|
||||
]
|
||||
|
||||
TRAVEL_MODES_PUBLIC = [TRAVEL_MODE_PUBLIC, TRAVEL_MODE_PUBLIC_TIME_TABLE]
|
||||
TRAVEL_MODES_VEHICLE = [TRAVEL_MODE_CAR, TRAVEL_MODE_TRUCK]
|
||||
|
||||
TRAFFIC_MODE_ENABLED = "traffic_enabled"
|
||||
TRAFFIC_MODE_DISABLED = "traffic_disabled"
|
||||
TRAFFIC_MODES = [TRAFFIC_MODE_ENABLED, TRAFFIC_MODE_DISABLED]
|
||||
|
||||
ROUTE_MODE_FASTEST = "fastest"
|
||||
ROUTE_MODE_SHORTEST = "shortest"
|
||||
ROUTE_MODES = [ROUTE_MODE_FASTEST, ROUTE_MODE_SHORTEST]
|
||||
|
||||
ICON_BICYCLE = "mdi:bike"
|
||||
ICON_CAR = "mdi:car"
|
||||
ICON_PEDESTRIAN = "mdi:walk"
|
||||
ICON_PUBLIC = "mdi:bus"
|
||||
ICON_TRUCK = "mdi:truck"
|
||||
|
||||
UNITS = [CONF_UNIT_SYSTEM_METRIC, CONF_UNIT_SYSTEM_IMPERIAL]
|
||||
|
||||
ATTR_DURATION = "duration"
|
||||
ATTR_DISTANCE = "distance"
|
||||
ATTR_ROUTE = "route"
|
||||
ATTR_ORIGIN = "origin"
|
||||
ATTR_DESTINATION = "destination"
|
||||
|
||||
ATTR_UNIT_SYSTEM = CONF_UNIT_SYSTEM
|
||||
ATTR_TRAFFIC_MODE = CONF_TRAFFIC_MODE
|
||||
|
||||
ATTR_DURATION_IN_TRAFFIC = "duration_in_traffic"
|
||||
ATTR_ORIGIN_NAME = "origin_name"
|
||||
ATTR_DESTINATION_NAME = "destination_name"
|
||||
|
||||
NO_ROUTE_ERROR_MESSAGE = "HERE could not find a route based on the input"
|
35
homeassistant/components/here_travel_time/model.py
Normal file
35
homeassistant/components/here_travel_time/model.py
Normal file
@ -0,0 +1,35 @@
|
||||
"""Model Classes for here_travel_time."""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import time
|
||||
from typing import TypedDict
|
||||
|
||||
|
||||
class HERERoutingData(TypedDict):
|
||||
"""Routing information calculated from a herepy.RoutingResponse."""
|
||||
|
||||
ATTR_ATTRIBUTION: str | None
|
||||
ATTR_DURATION: float
|
||||
ATTR_DURATION_IN_TRAFFIC: float
|
||||
ATTR_DISTANCE: float
|
||||
ATTR_ROUTE: str
|
||||
ATTR_ORIGIN: str
|
||||
ATTR_DESTINATION: str
|
||||
ATTR_ORIGIN_NAME: str
|
||||
ATTR_DESTINATION_NAME: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class HERETravelTimeConfig:
|
||||
"""Configuration for HereTravelTimeDataUpdateCoordinator."""
|
||||
|
||||
origin: str | None
|
||||
destination: str | None
|
||||
origin_entity_id: str | None
|
||||
destination_entity_id: str | None
|
||||
travel_mode: str
|
||||
route_mode: str
|
||||
units: str
|
||||
arrival: time
|
||||
departure: time
|
@ -1,7 +1,7 @@
|
||||
"""Support for HERE travel time sensors."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, time, timedelta
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
|
||||
import herepy
|
||||
@ -18,15 +18,16 @@ from homeassistant.const import (
|
||||
CONF_UNIT_SYSTEM,
|
||||
CONF_UNIT_SYSTEM_IMPERIAL,
|
||||
CONF_UNIT_SYSTEM_METRIC,
|
||||
EVENT_HOMEASSISTANT_START,
|
||||
TIME_MINUTES,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.core import HomeAssistant
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.location import find_coordinates
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
from homeassistant.util import dt
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from . import HereTravelTimeDataUpdateCoordinator
|
||||
from .model import HERETravelTimeConfig
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
@ -175,21 +176,29 @@ async def async_setup_platform(
|
||||
destination = None
|
||||
destination_entity_id = config[CONF_DESTINATION_ENTITY_ID]
|
||||
|
||||
travel_mode = config[CONF_MODE]
|
||||
traffic_mode = config[CONF_TRAFFIC_MODE]
|
||||
route_mode = config[CONF_ROUTE_MODE]
|
||||
name = config[CONF_NAME]
|
||||
units = config.get(CONF_UNIT_SYSTEM, hass.config.units.name)
|
||||
arrival = config.get(CONF_ARRIVAL)
|
||||
departure = config.get(CONF_DEPARTURE)
|
||||
|
||||
here_data = HERETravelTimeData(
|
||||
here_client, travel_mode, traffic_mode, route_mode, units, arrival, departure
|
||||
here_travel_time_config = HERETravelTimeConfig(
|
||||
origin=origin,
|
||||
destination=destination,
|
||||
origin_entity_id=origin_entity_id,
|
||||
destination_entity_id=destination_entity_id,
|
||||
travel_mode=config[CONF_MODE],
|
||||
route_mode=config[CONF_ROUTE_MODE],
|
||||
units=config.get(CONF_UNIT_SYSTEM, hass.config.units.name),
|
||||
arrival=config.get(CONF_ARRIVAL),
|
||||
departure=config.get(CONF_DEPARTURE),
|
||||
)
|
||||
|
||||
sensor = HERETravelTimeSensor(
|
||||
name, origin, destination, origin_entity_id, destination_entity_id, here_data
|
||||
coordinator = HereTravelTimeDataUpdateCoordinator(
|
||||
hass,
|
||||
here_client,
|
||||
here_travel_time_config,
|
||||
)
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
|
||||
sensor = HERETravelTimeSensor(name, traffic_mode, coordinator)
|
||||
|
||||
async_add_entities([sensor])
|
||||
|
||||
@ -216,236 +225,66 @@ def _are_valid_client_credentials(here_client: herepy.RoutingApi) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
class HERETravelTimeSensor(SensorEntity):
|
||||
class HERETravelTimeSensor(SensorEntity, CoordinatorEntity):
|
||||
"""Representation of a HERE travel time sensor."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
name: str,
|
||||
origin: str,
|
||||
destination: str,
|
||||
origin_entity_id: str,
|
||||
destination_entity_id: str,
|
||||
here_data: HERETravelTimeData,
|
||||
traffic_mode: bool,
|
||||
coordinator: HereTravelTimeDataUpdateCoordinator,
|
||||
) -> None:
|
||||
"""Initialize the sensor."""
|
||||
self._name = name
|
||||
self._origin_entity_id = origin_entity_id
|
||||
self._destination_entity_id = destination_entity_id
|
||||
self._here_data = here_data
|
||||
self._unit_of_measurement = TIME_MINUTES
|
||||
self._attrs = {
|
||||
ATTR_UNIT_SYSTEM: self._here_data.units,
|
||||
ATTR_MODE: self._here_data.travel_mode,
|
||||
ATTR_TRAFFIC_MODE: self._here_data.traffic_mode,
|
||||
}
|
||||
if self._origin_entity_id is None:
|
||||
self._here_data.origin = origin
|
||||
|
||||
if self._destination_entity_id is None:
|
||||
self._here_data.destination = destination
|
||||
|
||||
async def async_added_to_hass(self) -> None:
|
||||
"""Delay the sensor update to avoid entity not found warnings."""
|
||||
|
||||
@callback
|
||||
def delayed_sensor_update(event):
|
||||
"""Update sensor after Home Assistant started."""
|
||||
self.async_schedule_update_ha_state(True)
|
||||
|
||||
self.hass.bus.async_listen_once(
|
||||
EVENT_HOMEASSISTANT_START, delayed_sensor_update
|
||||
)
|
||||
super().__init__(coordinator)
|
||||
self._traffic_mode = traffic_mode
|
||||
self._attr_native_unit_of_measurement = TIME_MINUTES
|
||||
self._attr_name = name
|
||||
|
||||
@property
|
||||
def native_value(self) -> str | None:
|
||||
"""Return the state of the sensor."""
|
||||
if self._here_data.traffic_mode and self._here_data.traffic_time is not None:
|
||||
return str(round(self._here_data.traffic_time / 60))
|
||||
if self._here_data.base_time is not None:
|
||||
return str(round(self._here_data.base_time / 60))
|
||||
|
||||
if self.coordinator.data is not None:
|
||||
return str(
|
||||
round(
|
||||
self.coordinator.data.get(
|
||||
ATTR_DURATION_IN_TRAFFIC
|
||||
if self._traffic_mode
|
||||
else ATTR_DURATION
|
||||
)
|
||||
)
|
||||
)
|
||||
return None
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
"""Get the name of the sensor."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def extra_state_attributes(
|
||||
self,
|
||||
) -> dict[str, None | float | str | bool] | None:
|
||||
"""Return the state attributes."""
|
||||
if self._here_data.base_time is None:
|
||||
return None
|
||||
|
||||
res = self._attrs
|
||||
if self._here_data.attribution is not None:
|
||||
res[ATTR_ATTRIBUTION] = self._here_data.attribution
|
||||
res[ATTR_DURATION] = self._here_data.base_time / 60
|
||||
res[ATTR_DISTANCE] = self._here_data.distance
|
||||
res[ATTR_ROUTE] = self._here_data.route
|
||||
res[ATTR_DURATION_IN_TRAFFIC] = self._here_data.traffic_time / 60
|
||||
res[ATTR_ORIGIN] = self._here_data.origin
|
||||
res[ATTR_DESTINATION] = self._here_data.destination
|
||||
res[ATTR_ORIGIN_NAME] = self._here_data.origin_name
|
||||
res[ATTR_DESTINATION_NAME] = self._here_data.destination_name
|
||||
return res
|
||||
if self.coordinator.data is not None:
|
||||
res = {
|
||||
ATTR_UNIT_SYSTEM: self.coordinator.config.units,
|
||||
ATTR_MODE: self.coordinator.config.travel_mode,
|
||||
ATTR_TRAFFIC_MODE: self._traffic_mode,
|
||||
**self.coordinator.data,
|
||||
}
|
||||
res.pop(ATTR_ATTRIBUTION)
|
||||
return res
|
||||
return None
|
||||
|
||||
@property
|
||||
def native_unit_of_measurement(self) -> str:
|
||||
"""Return the unit this state is expressed in."""
|
||||
return self._unit_of_measurement
|
||||
def attribution(self) -> str | None:
|
||||
"""Return the attribution."""
|
||||
return self.coordinator.data.get(ATTR_ATTRIBUTION)
|
||||
|
||||
@property
|
||||
def icon(self) -> str:
|
||||
"""Icon to use in the frontend depending on travel_mode."""
|
||||
if self._here_data.travel_mode == TRAVEL_MODE_BICYCLE:
|
||||
if self.coordinator.config.travel_mode == TRAVEL_MODE_BICYCLE:
|
||||
return ICON_BICYCLE
|
||||
if self._here_data.travel_mode == TRAVEL_MODE_PEDESTRIAN:
|
||||
if self.coordinator.config.travel_mode == TRAVEL_MODE_PEDESTRIAN:
|
||||
return ICON_PEDESTRIAN
|
||||
if self._here_data.travel_mode in TRAVEL_MODES_PUBLIC:
|
||||
if self.coordinator.config.travel_mode in TRAVEL_MODES_PUBLIC:
|
||||
return ICON_PUBLIC
|
||||
if self._here_data.travel_mode == TRAVEL_MODE_TRUCK:
|
||||
if self.coordinator.config.travel_mode == TRAVEL_MODE_TRUCK:
|
||||
return ICON_TRUCK
|
||||
return ICON_CAR
|
||||
|
||||
async def async_update(self) -> None:
|
||||
"""Update Sensor Information."""
|
||||
# Convert device_trackers to HERE friendly location
|
||||
if self._origin_entity_id is not None:
|
||||
self._here_data.origin = find_coordinates(self.hass, self._origin_entity_id)
|
||||
|
||||
if self._destination_entity_id is not None:
|
||||
self._here_data.destination = find_coordinates(
|
||||
self.hass, self._destination_entity_id
|
||||
)
|
||||
|
||||
await self.hass.async_add_executor_job(self._here_data.update)
|
||||
|
||||
|
||||
class HERETravelTimeData:
|
||||
"""HERETravelTime data object."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
here_client: herepy.RoutingApi,
|
||||
travel_mode: str,
|
||||
traffic_mode: bool,
|
||||
route_mode: str,
|
||||
units: str,
|
||||
arrival: datetime,
|
||||
departure: datetime,
|
||||
) -> None:
|
||||
"""Initialize herepy."""
|
||||
self.origin = None
|
||||
self.destination = None
|
||||
self.travel_mode = travel_mode
|
||||
self.traffic_mode = traffic_mode
|
||||
self.route_mode = route_mode
|
||||
self.arrival = arrival
|
||||
self.departure = departure
|
||||
self.attribution = None
|
||||
self.traffic_time = None
|
||||
self.distance = None
|
||||
self.route = None
|
||||
self.base_time = None
|
||||
self.origin_name = None
|
||||
self.destination_name = None
|
||||
self.units = units
|
||||
self._client = here_client
|
||||
self.combine_change = True
|
||||
|
||||
def update(self) -> None:
|
||||
"""Get the latest data from HERE."""
|
||||
if self.traffic_mode:
|
||||
traffic_mode = TRAFFIC_MODE_ENABLED
|
||||
else:
|
||||
traffic_mode = TRAFFIC_MODE_DISABLED
|
||||
|
||||
if self.destination is not None and self.origin is not None:
|
||||
# Convert location to HERE friendly location
|
||||
destination = self.destination.split(",")
|
||||
origin = self.origin.split(",")
|
||||
if (arrival := self.arrival) is not None:
|
||||
arrival = convert_time_to_isodate(arrival)
|
||||
if (departure := self.departure) is not None:
|
||||
departure = convert_time_to_isodate(departure)
|
||||
|
||||
if departure is None and arrival is None:
|
||||
departure = "now"
|
||||
|
||||
_LOGGER.debug(
|
||||
"Requesting route for origin: %s, destination: %s, route_mode: %s, mode: %s, traffic_mode: %s, arrival: %s, departure: %s",
|
||||
origin,
|
||||
destination,
|
||||
herepy.RouteMode[self.route_mode],
|
||||
herepy.RouteMode[self.travel_mode],
|
||||
herepy.RouteMode[traffic_mode],
|
||||
arrival,
|
||||
departure,
|
||||
)
|
||||
|
||||
try:
|
||||
response = self._client.public_transport_timetable(
|
||||
origin,
|
||||
destination,
|
||||
self.combine_change,
|
||||
[
|
||||
herepy.RouteMode[self.route_mode],
|
||||
herepy.RouteMode[self.travel_mode],
|
||||
herepy.RouteMode[traffic_mode],
|
||||
],
|
||||
arrival=arrival,
|
||||
departure=departure,
|
||||
)
|
||||
except herepy.NoRouteFoundError:
|
||||
# Better error message for cryptic no route error codes
|
||||
_LOGGER.error(NO_ROUTE_ERROR_MESSAGE)
|
||||
return
|
||||
|
||||
_LOGGER.debug("Raw response is: %s", response.response)
|
||||
|
||||
source_attribution = response.response.get("sourceAttribution")
|
||||
if source_attribution is not None:
|
||||
self.attribution = self._build_hass_attribution(source_attribution)
|
||||
route = response.response["route"]
|
||||
summary = route[0]["summary"]
|
||||
waypoint = route[0]["waypoint"]
|
||||
self.base_time = summary["baseTime"]
|
||||
if self.travel_mode in TRAVEL_MODES_VEHICLE:
|
||||
self.traffic_time = summary["trafficTime"]
|
||||
else:
|
||||
self.traffic_time = self.base_time
|
||||
distance = summary["distance"]
|
||||
if self.units == CONF_UNIT_SYSTEM_IMPERIAL:
|
||||
# Convert to miles.
|
||||
self.distance = distance / 1609.344
|
||||
else:
|
||||
# Convert to kilometers
|
||||
self.distance = distance / 1000
|
||||
self.route = response.route_short
|
||||
self.origin_name = waypoint[0]["mappedRoadName"]
|
||||
self.destination_name = waypoint[1]["mappedRoadName"]
|
||||
|
||||
@staticmethod
|
||||
def _build_hass_attribution(source_attribution: dict) -> str | None:
|
||||
"""Build a hass frontend ready string out of the sourceAttribution."""
|
||||
suppliers = source_attribution.get("supplier")
|
||||
if suppliers is not None:
|
||||
supplier_titles = []
|
||||
for supplier in suppliers:
|
||||
if (title := supplier.get("title")) is not None:
|
||||
supplier_titles.append(title)
|
||||
joined_supplier_titles = ",".join(supplier_titles)
|
||||
attribution = f"With the support of {joined_supplier_titles}. All information is provided without warranty of any kind."
|
||||
return attribution
|
||||
|
||||
|
||||
def convert_time_to_isodate(simple_time: time) -> str:
|
||||
"""Take a time like 08:00:00 and combine it with the current date."""
|
||||
combined = datetime.combine(dt.start_of_local_day(), simple_time)
|
||||
if combined < datetime.now():
|
||||
combined = combined + timedelta(days=1)
|
||||
return combined.isoformat()
|
||||
|
@ -49,16 +49,50 @@ PLATFORM = "here_travel_time"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"mode,icon,traffic_mode,unit_system",
|
||||
"mode,icon,traffic_mode,unit_system,expected_state,expected_distance,expected_duration_in_traffic",
|
||||
[
|
||||
(TRAVEL_MODE_CAR, ICON_CAR, True, "metric"),
|
||||
(TRAVEL_MODE_BICYCLE, ICON_BICYCLE, False, "metric"),
|
||||
(TRAVEL_MODE_PEDESTRIAN, ICON_PEDESTRIAN, False, "imperial"),
|
||||
(TRAVEL_MODE_PUBLIC_TIME_TABLE, ICON_PUBLIC, False, "imperial"),
|
||||
(TRAVEL_MODE_TRUCK, ICON_TRUCK, True, "metric"),
|
||||
(TRAVEL_MODE_CAR, ICON_CAR, True, "metric", "31", 23.903, 31.016666666666666),
|
||||
(TRAVEL_MODE_BICYCLE, ICON_BICYCLE, False, "metric", "30", 23.903, 30.05),
|
||||
(
|
||||
TRAVEL_MODE_PEDESTRIAN,
|
||||
ICON_PEDESTRIAN,
|
||||
False,
|
||||
"imperial",
|
||||
"30",
|
||||
14.852635608048994,
|
||||
30.05,
|
||||
),
|
||||
(
|
||||
TRAVEL_MODE_PUBLIC_TIME_TABLE,
|
||||
ICON_PUBLIC,
|
||||
False,
|
||||
"imperial",
|
||||
"30",
|
||||
14.852635608048994,
|
||||
30.05,
|
||||
),
|
||||
(
|
||||
TRAVEL_MODE_TRUCK,
|
||||
ICON_TRUCK,
|
||||
True,
|
||||
"metric",
|
||||
"31",
|
||||
23.903,
|
||||
31.016666666666666,
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_sensor(hass, mode, icon, traffic_mode, unit_system, valid_response):
|
||||
async def test_sensor(
|
||||
hass,
|
||||
mode,
|
||||
icon,
|
||||
traffic_mode,
|
||||
unit_system,
|
||||
expected_state,
|
||||
expected_distance,
|
||||
expected_duration_in_traffic,
|
||||
valid_response,
|
||||
):
|
||||
"""Test that sensor works."""
|
||||
config = {
|
||||
DOMAIN: {
|
||||
@ -85,25 +119,18 @@ async def test_sensor(hass, mode, icon, traffic_mode, unit_system, valid_respons
|
||||
sensor.attributes.get(ATTR_ATTRIBUTION)
|
||||
== "With the support of HERE Technologies. All information is provided without warranty of any kind."
|
||||
)
|
||||
if traffic_mode:
|
||||
assert sensor.state == "31"
|
||||
else:
|
||||
assert sensor.state == "30"
|
||||
assert sensor.state == expected_state
|
||||
|
||||
assert sensor.attributes.get(ATTR_DURATION) == 30.05
|
||||
if unit_system == "metric":
|
||||
assert sensor.attributes.get(ATTR_DISTANCE) == 23.903
|
||||
else:
|
||||
assert sensor.attributes.get(ATTR_DISTANCE) == 14.852635608048994
|
||||
assert sensor.attributes.get(ATTR_DISTANCE) == expected_distance
|
||||
assert sensor.attributes.get(ATTR_ROUTE) == (
|
||||
"US-29 - K St NW; US-29 - Whitehurst Fwy; "
|
||||
"I-495 N - Capital Beltway; MD-187 S - Old Georgetown Rd"
|
||||
)
|
||||
assert sensor.attributes.get(CONF_UNIT_SYSTEM) == unit_system
|
||||
if mode in TRAVEL_MODES_VEHICLE:
|
||||
assert sensor.attributes.get(ATTR_DURATION_IN_TRAFFIC) == 31.016666666666666
|
||||
else:
|
||||
assert sensor.attributes.get(ATTR_DURATION_IN_TRAFFIC) == 30.05
|
||||
assert (
|
||||
sensor.attributes.get(ATTR_DURATION_IN_TRAFFIC) == expected_duration_in_traffic
|
||||
)
|
||||
assert sensor.attributes.get(ATTR_ORIGIN) == ",".join(
|
||||
[CAR_ORIGIN_LATITUDE, CAR_ORIGIN_LONGITUDE]
|
||||
)
|
||||
@ -168,7 +195,7 @@ async def test_entity_ids(hass, valid_response):
|
||||
assert sensor.attributes.get(ATTR_DISTANCE) == 23.903
|
||||
|
||||
|
||||
async def test_route_not_found(hass, caplog, valid_response):
|
||||
async def test_route_not_found(hass, caplog):
|
||||
"""Test that route not found error is correctly handled."""
|
||||
config = {
|
||||
DOMAIN: {
|
||||
@ -181,12 +208,15 @@ async def test_route_not_found(hass, caplog, valid_response):
|
||||
"api_key": API_KEY,
|
||||
}
|
||||
}
|
||||
assert await async_setup_component(hass, DOMAIN, config)
|
||||
await hass.async_block_till_done()
|
||||
with patch(
|
||||
"homeassistant.components.here_travel_time.sensor._are_valid_client_credentials",
|
||||
return_value=True,
|
||||
), patch(
|
||||
"herepy.RoutingApi.public_transport_timetable",
|
||||
side_effect=NoRouteFoundError,
|
||||
):
|
||||
assert await async_setup_component(hass, DOMAIN, config)
|
||||
await hass.async_block_till_done()
|
||||
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
Loading…
x
Reference in New Issue
Block a user