mirror of
https://github.com/home-assistant/core.git
synced 2025-07-23 05:07:41 +00:00
Update Tesla to use DataUpdateCoordinator (#38306)
* Update Tesla to use DataUpdateCoordinator * Update Tesla to use DataUpdateCoordinator * Fix linting errors * Apply suggestions from code review Co-authored-by: Chris Talkington <chris@talkingtontech.com> * Address requested changes * Apply suggestions from code review Co-authored-by: Chris Talkington <chris@talkingtontech.com> * Fix lint errors * Remove controller from hass.data Co-authored-by: Chris Talkington <chris@talkingtontech.com>
This commit is contained in:
parent
0d5e279509
commit
94b6d09b51
@ -1,8 +1,10 @@
|
||||
"""Support for Tesla cars."""
|
||||
import asyncio
|
||||
from collections import defaultdict
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
|
||||
import async_timeout
|
||||
from teslajsonpy import Controller as TeslaAPI, TeslaException
|
||||
import voluptuous as vol
|
||||
|
||||
@ -17,8 +19,10 @@ from homeassistant.const import (
|
||||
CONF_USERNAME,
|
||||
)
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
from homeassistant.helpers import aiohttp_client, config_validation as cv
|
||||
from homeassistant.helpers.entity import Entity
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
from homeassistant.util import slugify
|
||||
|
||||
from .config_flow import (
|
||||
@ -116,7 +120,6 @@ async def async_setup(hass, base_config):
|
||||
|
||||
async def async_setup_entry(hass, config_entry):
|
||||
"""Set up Tesla as config entry."""
|
||||
|
||||
hass.data.setdefault(DOMAIN, {})
|
||||
config = config_entry.data
|
||||
websession = aiohttp_client.async_get_clientsession(hass)
|
||||
@ -145,13 +148,22 @@ async def async_setup_entry(hass, config_entry):
|
||||
_LOGGER.error("Unable to communicate with Tesla API: %s", ex.message)
|
||||
return False
|
||||
_async_save_tokens(hass, config_entry, access_token, refresh_token)
|
||||
coordinator = TeslaDataUpdateCoordinator(
|
||||
hass, config_entry=config_entry, controller=controller
|
||||
)
|
||||
# Fetch initial data so we have data when entities subscribe
|
||||
entry_data = hass.data[DOMAIN][config_entry.entry_id] = {
|
||||
"controller": controller,
|
||||
"coordinator": coordinator,
|
||||
"devices": defaultdict(list),
|
||||
DATA_LISTENER: [config_entry.add_update_listener(update_listener)],
|
||||
}
|
||||
_LOGGER.debug("Connected to the Tesla API")
|
||||
all_devices = entry_data["controller"].get_homeassistant_components()
|
||||
|
||||
await coordinator.async_refresh()
|
||||
if not coordinator.last_update_success:
|
||||
raise ConfigEntryNotReady
|
||||
|
||||
all_devices = controller.get_homeassistant_components()
|
||||
|
||||
if not all_devices:
|
||||
return False
|
||||
@ -169,54 +181,87 @@ async def async_setup_entry(hass, config_entry):
|
||||
|
||||
async def async_unload_entry(hass, config_entry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
await asyncio.gather(
|
||||
*[
|
||||
hass.config_entries.async_forward_entry_unload(config_entry, component)
|
||||
for component in TESLA_COMPONENTS
|
||||
]
|
||||
unload_ok = all(
|
||||
await asyncio.gather(
|
||||
*[
|
||||
hass.config_entries.async_forward_entry_unload(config_entry, component)
|
||||
for component in TESLA_COMPONENTS
|
||||
]
|
||||
)
|
||||
)
|
||||
for listener in hass.data[DOMAIN][config_entry.entry_id][DATA_LISTENER]:
|
||||
listener()
|
||||
username = config_entry.title
|
||||
hass.data[DOMAIN].pop(config_entry.entry_id)
|
||||
_LOGGER.debug("Unloaded entry for %s", username)
|
||||
return True
|
||||
if unload_ok:
|
||||
hass.data[DOMAIN].pop(config_entry.entry_id)
|
||||
_LOGGER.debug("Unloaded entry for %s", username)
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
async def update_listener(hass, config_entry):
|
||||
"""Update when config_entry options update."""
|
||||
controller = hass.data[DOMAIN][config_entry.entry_id]["controller"]
|
||||
controller = hass.data[DOMAIN][config_entry.entry_id]["coordinator"].controller
|
||||
old_update_interval = controller.update_interval
|
||||
controller.update_interval = config_entry.options.get(CONF_SCAN_INTERVAL)
|
||||
_LOGGER.debug(
|
||||
"Changing scan_interval from %s to %s",
|
||||
old_update_interval,
|
||||
controller.update_interval,
|
||||
)
|
||||
if old_update_interval != controller.update_interval:
|
||||
_LOGGER.debug(
|
||||
"Changing scan_interval from %s to %s",
|
||||
old_update_interval,
|
||||
controller.update_interval,
|
||||
)
|
||||
|
||||
|
||||
class TeslaDataUpdateCoordinator(DataUpdateCoordinator):
|
||||
"""Class to manage fetching Tesla data."""
|
||||
|
||||
def __init__(self, hass, *, config_entry, controller):
|
||||
"""Initialize global Tesla data updater."""
|
||||
self.controller = controller
|
||||
self.config_entry = config_entry
|
||||
|
||||
update_interval = timedelta(seconds=MIN_SCAN_INTERVAL)
|
||||
|
||||
super().__init__(
|
||||
hass, _LOGGER, name=DOMAIN, update_interval=update_interval,
|
||||
)
|
||||
|
||||
async def _async_update_data(self):
|
||||
"""Fetch data from API endpoint."""
|
||||
if self.controller.is_token_refreshed():
|
||||
(refresh_token, access_token) = self.controller.get_tokens()
|
||||
_async_save_tokens(
|
||||
self.hass, self.config_entry, access_token, refresh_token
|
||||
)
|
||||
_LOGGER.debug("Saving new tokens in config_entry")
|
||||
|
||||
try:
|
||||
# Note: asyncio.TimeoutError and aiohttp.ClientError are already
|
||||
# handled by the data update coordinator.
|
||||
async with async_timeout.timeout(30):
|
||||
return await self.controller.update()
|
||||
except TeslaException as err:
|
||||
raise UpdateFailed(f"Error communicating with API: {err}")
|
||||
|
||||
|
||||
class TeslaDevice(Entity):
|
||||
"""Representation of a Tesla device."""
|
||||
|
||||
def __init__(self, tesla_device, controller, config_entry):
|
||||
def __init__(self, tesla_device, coordinator):
|
||||
"""Initialise the Tesla device."""
|
||||
self.tesla_device = tesla_device
|
||||
self.controller = controller
|
||||
self.config_entry = config_entry
|
||||
self._name = self.tesla_device.name
|
||||
self.tesla_id = slugify(self.tesla_device.uniq_name)
|
||||
self._attributes = {}
|
||||
self._icon = ICONS.get(self.tesla_device.type)
|
||||
self.coordinator = coordinator
|
||||
self._attributes = self.tesla_device.attrs.copy()
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the device."""
|
||||
return self._name
|
||||
return self.tesla_device.name
|
||||
|
||||
@property
|
||||
def unique_id(self) -> str:
|
||||
"""Return a unique ID."""
|
||||
return self.tesla_id
|
||||
return slugify(self.tesla_device.uniq_name)
|
||||
|
||||
@property
|
||||
def icon(self):
|
||||
@ -224,17 +269,22 @@ class TeslaDevice(Entity):
|
||||
if self.device_class:
|
||||
return None
|
||||
|
||||
return self._icon
|
||||
return ICONS.get(self.tesla_device.type)
|
||||
|
||||
@property
|
||||
def should_poll(self):
|
||||
"""Return the polling state."""
|
||||
return self.tesla_device.should_poll
|
||||
"""No need to poll. Coordinator notifies entity of updates."""
|
||||
return False
|
||||
|
||||
@property
|
||||
def available(self):
|
||||
"""Return if entity is available."""
|
||||
return self.coordinator.last_update_success
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return the state attributes of the device."""
|
||||
attr = self._attributes
|
||||
attr = self._attributes.copy()
|
||||
if self.tesla_device.has_battery():
|
||||
attr[ATTR_BATTERY_LEVEL] = self.tesla_device.battery_level()
|
||||
attr[ATTR_BATTERY_CHARGING] = self.tesla_device.battery_charging()
|
||||
@ -253,16 +303,21 @@ class TeslaDevice(Entity):
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
"""Register state update callback."""
|
||||
self.async_on_remove(self.coordinator.async_add_listener(self.refresh))
|
||||
|
||||
async def async_will_remove_from_hass(self):
|
||||
"""Prepare for unload."""
|
||||
|
||||
async def async_update(self):
|
||||
"""Update the state of the device."""
|
||||
if self.controller.is_token_refreshed():
|
||||
(refresh_token, access_token) = self.controller.get_tokens()
|
||||
_async_save_tokens(
|
||||
self.hass, self.config_entry, access_token, refresh_token
|
||||
)
|
||||
_LOGGER.debug("Saving new tokens in config_entry")
|
||||
await self.tesla_device.async_update()
|
||||
_LOGGER.debug("Updating state for: %s", self.name)
|
||||
await self.coordinator.async_request_refresh()
|
||||
self.refresh()
|
||||
|
||||
def refresh(self) -> None:
|
||||
"""Refresh the state of the device.
|
||||
|
||||
This assumes the coordinator has updated the controller.
|
||||
"""
|
||||
self.tesla_device.refresh()
|
||||
self.schedule_update_ha_state()
|
||||
|
@ -13,9 +13,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
async_add_entities(
|
||||
[
|
||||
TeslaBinarySensor(
|
||||
device,
|
||||
hass.data[TESLA_DOMAIN][config_entry.entry_id]["controller"],
|
||||
config_entry,
|
||||
device, hass.data[TESLA_DOMAIN][config_entry.entry_id]["coordinator"],
|
||||
)
|
||||
for device in hass.data[TESLA_DOMAIN][config_entry.entry_id]["devices"][
|
||||
"binary_sensor"
|
||||
@ -28,27 +26,16 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
class TeslaBinarySensor(TeslaDevice, BinarySensorEntity):
|
||||
"""Implement an Tesla binary sensor for parking and charger."""
|
||||
|
||||
def __init__(self, tesla_device, controller, config_entry):
|
||||
"""Initialise of a Tesla binary sensor."""
|
||||
super().__init__(tesla_device, controller, config_entry)
|
||||
self._state = None
|
||||
self._sensor_type = None
|
||||
if tesla_device.sensor_type in DEVICE_CLASSES:
|
||||
self._sensor_type = tesla_device.sensor_type
|
||||
|
||||
@property
|
||||
def device_class(self):
|
||||
"""Return the class of this binary sensor."""
|
||||
return self._sensor_type
|
||||
return (
|
||||
self.tesla_device.sensor_type
|
||||
if self.tesla_device.sensor_type in DEVICE_CLASSES
|
||||
else None
|
||||
)
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
"""Return the state of the binary sensor."""
|
||||
return self._state
|
||||
|
||||
async def async_update(self):
|
||||
"""Update the state of the device."""
|
||||
_LOGGER.debug("Updating sensor: %s", self._name)
|
||||
await super().async_update()
|
||||
self._state = self.tesla_device.get_value()
|
||||
self._attributes = self.tesla_device.attrs
|
||||
return self.tesla_device.get_value()
|
||||
|
@ -25,9 +25,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
async_add_entities(
|
||||
[
|
||||
TeslaThermostat(
|
||||
device,
|
||||
hass.data[TESLA_DOMAIN][config_entry.entry_id]["controller"],
|
||||
config_entry,
|
||||
device, hass.data[TESLA_DOMAIN][config_entry.entry_id]["coordinator"],
|
||||
)
|
||||
for device in hass.data[TESLA_DOMAIN][config_entry.entry_id]["devices"][
|
||||
"climate"
|
||||
@ -40,12 +38,6 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
class TeslaThermostat(TeslaDevice, ClimateEntity):
|
||||
"""Representation of a Tesla climate."""
|
||||
|
||||
def __init__(self, tesla_device, controller, config_entry):
|
||||
"""Initialize the Tesla device."""
|
||||
super().__init__(tesla_device, controller, config_entry)
|
||||
self._target_temperature = None
|
||||
self._temperature = None
|
||||
|
||||
@property
|
||||
def supported_features(self):
|
||||
"""Return the list of supported features."""
|
||||
@ -69,42 +61,33 @@ class TeslaThermostat(TeslaDevice, ClimateEntity):
|
||||
"""
|
||||
return SUPPORT_HVAC
|
||||
|
||||
async def async_update(self):
|
||||
"""Call by the Tesla device callback to update state."""
|
||||
_LOGGER.debug("Updating: %s", self._name)
|
||||
await super().async_update()
|
||||
self._target_temperature = self.tesla_device.get_goal_temp()
|
||||
self._temperature = self.tesla_device.get_current_temp()
|
||||
|
||||
@property
|
||||
def temperature_unit(self):
|
||||
"""Return the unit of measurement."""
|
||||
tesla_temp_units = self.tesla_device.measurement
|
||||
|
||||
if tesla_temp_units == "F":
|
||||
if self.tesla_device.measurement == "F":
|
||||
return TEMP_FAHRENHEIT
|
||||
return TEMP_CELSIUS
|
||||
|
||||
@property
|
||||
def current_temperature(self):
|
||||
"""Return the current temperature."""
|
||||
return self._temperature
|
||||
return self.tesla_device.get_current_temp()
|
||||
|
||||
@property
|
||||
def target_temperature(self):
|
||||
"""Return the temperature we try to reach."""
|
||||
return self._target_temperature
|
||||
return self.tesla_device.get_goal_temp()
|
||||
|
||||
async def async_set_temperature(self, **kwargs):
|
||||
"""Set new target temperatures."""
|
||||
temperature = kwargs.get(ATTR_TEMPERATURE)
|
||||
if temperature:
|
||||
_LOGGER.debug("%s: Setting temperature to %s", self._name, temperature)
|
||||
_LOGGER.debug("%s: Setting temperature to %s", self.name, temperature)
|
||||
await self.tesla_device.set_temperature(temperature)
|
||||
|
||||
async def async_set_hvac_mode(self, hvac_mode):
|
||||
"""Set new target hvac mode."""
|
||||
_LOGGER.debug("%s: Setting hvac mode to %s", self._name, hvac_mode)
|
||||
_LOGGER.debug("%s: Setting hvac mode to %s", self.name, hvac_mode)
|
||||
if hvac_mode == HVAC_MODE_OFF:
|
||||
await self.tesla_device.set_status(False)
|
||||
elif hvac_mode == HVAC_MODE_HEAT_COOL:
|
||||
|
@ -1,5 +1,6 @@
|
||||
"""Support for tracking Tesla cars."""
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
from homeassistant.components.device_tracker import SOURCE_TYPE_GPS
|
||||
from homeassistant.components.device_tracker.config_entry import TrackerEntity
|
||||
@ -13,9 +14,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
"""Set up the Tesla binary_sensors by config_entry."""
|
||||
entities = [
|
||||
TeslaDeviceEntity(
|
||||
device,
|
||||
hass.data[TESLA_DOMAIN][config_entry.entry_id]["controller"],
|
||||
config_entry,
|
||||
device, hass.data[TESLA_DOMAIN][config_entry.entry_id]["coordinator"],
|
||||
)
|
||||
for device in hass.data[TESLA_DOMAIN][config_entry.entry_id]["devices"][
|
||||
"devices_tracker"
|
||||
@ -27,44 +26,37 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
class TeslaDeviceEntity(TeslaDevice, TrackerEntity):
|
||||
"""A class representing a Tesla device."""
|
||||
|
||||
def __init__(self, tesla_device, controller, config_entry):
|
||||
def __init__(self, tesla_device, coordinator):
|
||||
"""Initialize the Tesla device scanner."""
|
||||
super().__init__(tesla_device, controller, config_entry)
|
||||
self._latitude = None
|
||||
self._longitude = None
|
||||
super().__init__(tesla_device, coordinator)
|
||||
self._attributes = {"trackr_id": self.unique_id}
|
||||
self._listener = None
|
||||
|
||||
async def async_update(self):
|
||||
"""Update the device info."""
|
||||
_LOGGER.debug("Updating device position: %s", self.name)
|
||||
await super().async_update()
|
||||
location = self.tesla_device.get_location()
|
||||
if location:
|
||||
self._latitude = location["latitude"]
|
||||
self._longitude = location["longitude"]
|
||||
self._attributes = {
|
||||
"trackr_id": self.unique_id,
|
||||
"heading": location["heading"],
|
||||
"speed": location["speed"],
|
||||
}
|
||||
|
||||
@property
|
||||
def latitude(self) -> float:
|
||||
def latitude(self) -> Optional[float]:
|
||||
"""Return latitude value of the device."""
|
||||
return self._latitude
|
||||
location = self.tesla_device.get_location()
|
||||
return self.tesla_device.get_location().get("latitude") if location else None
|
||||
|
||||
@property
|
||||
def longitude(self) -> float:
|
||||
def longitude(self) -> Optional[float]:
|
||||
"""Return longitude value of the device."""
|
||||
return self._longitude
|
||||
|
||||
@property
|
||||
def should_poll(self):
|
||||
"""Return whether polling is needed."""
|
||||
return True
|
||||
location = self.tesla_device.get_location()
|
||||
return self.tesla_device.get_location().get("longitude") if location else None
|
||||
|
||||
@property
|
||||
def source_type(self):
|
||||
"""Return the source type, eg gps or router, of the device."""
|
||||
return SOURCE_TYPE_GPS
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return the state attributes of the device."""
|
||||
attr = super().device_state_attributes.copy()
|
||||
location = self.tesla_device.get_location()
|
||||
if location:
|
||||
self._attributes = {
|
||||
"trackr_id": self.unique_id,
|
||||
"heading": location["heading"],
|
||||
"speed": location["speed"],
|
||||
}
|
||||
return attr
|
||||
|
@ -2,7 +2,6 @@
|
||||
import logging
|
||||
|
||||
from homeassistant.components.lock import LockEntity
|
||||
from homeassistant.const import STATE_LOCKED, STATE_UNLOCKED
|
||||
|
||||
from . import DOMAIN as TESLA_DOMAIN, TeslaDevice
|
||||
|
||||
@ -13,9 +12,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
"""Set up the Tesla binary_sensors by config_entry."""
|
||||
entities = [
|
||||
TeslaLock(
|
||||
device,
|
||||
hass.data[TESLA_DOMAIN][config_entry.entry_id]["controller"],
|
||||
config_entry,
|
||||
device, hass.data[TESLA_DOMAIN][config_entry.entry_id]["coordinator"],
|
||||
)
|
||||
for device in hass.data[TESLA_DOMAIN][config_entry.entry_id]["devices"]["lock"]
|
||||
]
|
||||
@ -25,28 +22,19 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
class TeslaLock(TeslaDevice, LockEntity):
|
||||
"""Representation of a Tesla door lock."""
|
||||
|
||||
def __init__(self, tesla_device, controller, config_entry):
|
||||
"""Initialise of the lock."""
|
||||
self._state = None
|
||||
super().__init__(tesla_device, controller, config_entry)
|
||||
|
||||
async def async_lock(self, **kwargs):
|
||||
"""Send the lock command."""
|
||||
_LOGGER.debug("Locking doors for: %s", self._name)
|
||||
_LOGGER.debug("Locking doors for: %s", self.name)
|
||||
await self.tesla_device.lock()
|
||||
|
||||
async def async_unlock(self, **kwargs):
|
||||
"""Send the unlock command."""
|
||||
_LOGGER.debug("Unlocking doors for: %s", self._name)
|
||||
_LOGGER.debug("Unlocking doors for: %s", self.name)
|
||||
await self.tesla_device.unlock()
|
||||
|
||||
@property
|
||||
def is_locked(self):
|
||||
"""Get whether the lock is in locked state."""
|
||||
return self._state == STATE_LOCKED
|
||||
|
||||
async def async_update(self):
|
||||
"""Update state of the lock."""
|
||||
_LOGGER.debug("Updating state for: %s", self._name)
|
||||
await super().async_update()
|
||||
self._state = STATE_LOCKED if self.tesla_device.is_locked() else STATE_UNLOCKED
|
||||
if self.tesla_device.is_locked() is None:
|
||||
return None
|
||||
return self.tesla_device.is_locked()
|
||||
|
@ -3,6 +3,6 @@
|
||||
"name": "Tesla",
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/tesla",
|
||||
"requirements": ["teslajsonpy==0.10.1"],
|
||||
"requirements": ["teslajsonpy==0.10.3"],
|
||||
"codeowners": ["@zabuldon", "@alandtse"]
|
||||
}
|
||||
|
@ -1,6 +1,8 @@
|
||||
"""Support for the Tesla sensors."""
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
from homeassistant.components.sensor import DEVICE_CLASSES
|
||||
from homeassistant.const import (
|
||||
LENGTH_KILOMETERS,
|
||||
LENGTH_MILES,
|
||||
@ -17,89 +19,96 @@ _LOGGER = logging.getLogger(__name__)
|
||||
|
||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
"""Set up the Tesla binary_sensors by config_entry."""
|
||||
controller = hass.data[TESLA_DOMAIN][config_entry.entry_id]["controller"]
|
||||
coordinator = hass.data[TESLA_DOMAIN][config_entry.entry_id]["coordinator"]
|
||||
entities = []
|
||||
for device in hass.data[TESLA_DOMAIN][config_entry.entry_id]["devices"]["sensor"]:
|
||||
if device.type == "temperature sensor":
|
||||
entities.append(TeslaSensor(device, controller, config_entry, "inside"))
|
||||
entities.append(TeslaSensor(device, controller, config_entry, "outside"))
|
||||
entities.append(TeslaSensor(device, coordinator, "inside"))
|
||||
entities.append(TeslaSensor(device, coordinator, "outside"))
|
||||
else:
|
||||
entities.append(TeslaSensor(device, controller, config_entry))
|
||||
entities.append(TeslaSensor(device, coordinator))
|
||||
async_add_entities(entities, True)
|
||||
|
||||
|
||||
class TeslaSensor(TeslaDevice, Entity):
|
||||
"""Representation of Tesla sensors."""
|
||||
|
||||
def __init__(self, tesla_device, controller, config_entry, sensor_type=None):
|
||||
def __init__(self, tesla_device, coordinator, sensor_type=None):
|
||||
"""Initialize of the sensor."""
|
||||
self.current_value = None
|
||||
self.units = None
|
||||
self.last_changed_time = None
|
||||
super().__init__(tesla_device, coordinator)
|
||||
self.type = sensor_type
|
||||
self._device_class = tesla_device.device_class
|
||||
super().__init__(tesla_device, controller, config_entry)
|
||||
|
||||
if self.type:
|
||||
self._name = f"{self.tesla_device.name} ({self.type})"
|
||||
@property
|
||||
def name(self) -> str:
|
||||
"""Return the name of the device."""
|
||||
return (
|
||||
self.tesla_device.name
|
||||
if not self.type
|
||||
else f"{self.tesla_device.name} ({self.type})"
|
||||
)
|
||||
|
||||
@property
|
||||
def unique_id(self) -> str:
|
||||
"""Return a unique ID."""
|
||||
if self.type:
|
||||
return f"{self.tesla_id}_{self.type}"
|
||||
return self.tesla_id
|
||||
return (
|
||||
super().unique_id if not self.type else f"{super().unique_id}_{self.type}"
|
||||
)
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
def state(self) -> Optional[float]:
|
||||
"""Return the state of the sensor."""
|
||||
return self.current_value
|
||||
|
||||
@property
|
||||
def unit_of_measurement(self):
|
||||
"""Return the unit_of_measurement of the device."""
|
||||
return self.units
|
||||
|
||||
@property
|
||||
def device_class(self):
|
||||
"""Return the device_class of the device."""
|
||||
return self._device_class
|
||||
|
||||
async def async_update(self):
|
||||
"""Update the state from the sensor."""
|
||||
_LOGGER.debug("Updating sensor: %s", self._name)
|
||||
await super().async_update()
|
||||
units = self.tesla_device.measurement
|
||||
|
||||
if self.tesla_device.type == "temperature sensor":
|
||||
if self.type == "outside":
|
||||
self.current_value = self.tesla_device.get_outside_temp()
|
||||
else:
|
||||
self.current_value = self.tesla_device.get_inside_temp()
|
||||
if units == "F":
|
||||
self.units = TEMP_FAHRENHEIT
|
||||
else:
|
||||
self.units = TEMP_CELSIUS
|
||||
elif self.tesla_device.type in ["range sensor", "mileage sensor"]:
|
||||
self.current_value = self.tesla_device.get_value()
|
||||
return self.tesla_device.get_outside_temp()
|
||||
return self.tesla_device.get_inside_temp()
|
||||
if self.tesla_device.type in ["range sensor", "mileage sensor"]:
|
||||
units = self.tesla_device.measurement
|
||||
if units == "LENGTH_MILES":
|
||||
self.units = LENGTH_MILES
|
||||
else:
|
||||
self.units = LENGTH_KILOMETERS
|
||||
self.current_value = round(
|
||||
convert(self.current_value, LENGTH_MILES, LENGTH_KILOMETERS), 2
|
||||
)
|
||||
elif self.tesla_device.type == "charging rate sensor":
|
||||
self.current_value = self.tesla_device.charging_rate
|
||||
self.units = units
|
||||
self._attributes = {
|
||||
"time_left": self.tesla_device.time_left,
|
||||
"added_range": self.tesla_device.added_range,
|
||||
"charge_energy_added": self.tesla_device.charge_energy_added,
|
||||
"charge_current_request": self.tesla_device.charge_current_request,
|
||||
"charger_actual_current": self.tesla_device.charger_actual_current,
|
||||
"charger_voltage": self.tesla_device.charger_voltage,
|
||||
}
|
||||
else:
|
||||
self.current_value = self.tesla_device.get_value()
|
||||
self.units = units
|
||||
return self.tesla_device.get_value()
|
||||
return round(
|
||||
convert(self.tesla_device.get_value(), LENGTH_MILES, LENGTH_KILOMETERS),
|
||||
2,
|
||||
)
|
||||
if self.tesla_device.type == "charging rate sensor":
|
||||
return self.tesla_device.charging_rate
|
||||
return self.tesla_device.get_value()
|
||||
|
||||
@property
|
||||
def unit_of_measurement(self) -> Optional[str]:
|
||||
"""Return the unit_of_measurement of the device."""
|
||||
units = self.tesla_device.measurement
|
||||
if units == "F":
|
||||
return TEMP_FAHRENHEIT
|
||||
if units == "C":
|
||||
return TEMP_CELSIUS
|
||||
if units == "LENGTH_MILES":
|
||||
return LENGTH_MILES
|
||||
if units == "LENGTH_KILOMETERS":
|
||||
return LENGTH_KILOMETERS
|
||||
return units
|
||||
|
||||
@property
|
||||
def device_class(self) -> Optional[str]:
|
||||
"""Return the device_class of the device."""
|
||||
return (
|
||||
self.tesla_device.device_class
|
||||
if self.tesla_device.device_class in DEVICE_CLASSES
|
||||
else None
|
||||
)
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return the state attributes of the device."""
|
||||
attr = self._attributes.copy()
|
||||
if self.tesla_device.type == "charging rate sensor":
|
||||
attr.update(
|
||||
{
|
||||
"time_left": self.tesla_device.time_left,
|
||||
"added_range": self.tesla_device.added_range,
|
||||
"charge_energy_added": self.tesla_device.charge_energy_added,
|
||||
"charge_current_request": self.tesla_device.charge_current_request,
|
||||
"charger_actual_current": self.tesla_device.charger_actual_current,
|
||||
"charger_voltage": self.tesla_device.charger_voltage,
|
||||
}
|
||||
)
|
||||
return attr
|
||||
|
@ -2,7 +2,7 @@
|
||||
import logging
|
||||
|
||||
from homeassistant.components.switch import SwitchEntity
|
||||
from homeassistant.const import STATE_OFF, STATE_ON
|
||||
from homeassistant.const import STATE_ON
|
||||
|
||||
from . import DOMAIN as TESLA_DOMAIN, TeslaDevice
|
||||
|
||||
@ -11,111 +11,95 @@ _LOGGER = logging.getLogger(__name__)
|
||||
|
||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
"""Set up the Tesla binary_sensors by config_entry."""
|
||||
controller = hass.data[TESLA_DOMAIN][config_entry.entry_id]["controller"]
|
||||
coordinator = hass.data[TESLA_DOMAIN][config_entry.entry_id]["coordinator"]
|
||||
entities = []
|
||||
for device in hass.data[TESLA_DOMAIN][config_entry.entry_id]["devices"]["switch"]:
|
||||
if device.type == "charger switch":
|
||||
entities.append(ChargerSwitch(device, controller, config_entry))
|
||||
entities.append(UpdateSwitch(device, controller, config_entry))
|
||||
entities.append(ChargerSwitch(device, coordinator))
|
||||
entities.append(UpdateSwitch(device, coordinator))
|
||||
elif device.type == "maxrange switch":
|
||||
entities.append(RangeSwitch(device, controller, config_entry))
|
||||
entities.append(RangeSwitch(device, coordinator))
|
||||
elif device.type == "sentry mode switch":
|
||||
entities.append(SentryModeSwitch(device, controller, config_entry))
|
||||
entities.append(SentryModeSwitch(device, coordinator))
|
||||
async_add_entities(entities, True)
|
||||
|
||||
|
||||
class ChargerSwitch(TeslaDevice, SwitchEntity):
|
||||
"""Representation of a Tesla charger switch."""
|
||||
|
||||
def __init__(self, tesla_device, controller, config_entry):
|
||||
"""Initialise of the switch."""
|
||||
self._state = None
|
||||
super().__init__(tesla_device, controller, config_entry)
|
||||
|
||||
async def async_turn_on(self, **kwargs):
|
||||
"""Send the on command."""
|
||||
_LOGGER.debug("Enable charging: %s", self._name)
|
||||
_LOGGER.debug("Enable charging: %s", self.name)
|
||||
await self.tesla_device.start_charge()
|
||||
|
||||
async def async_turn_off(self, **kwargs):
|
||||
"""Send the off command."""
|
||||
_LOGGER.debug("Disable charging for: %s", self._name)
|
||||
_LOGGER.debug("Disable charging for: %s", self.name)
|
||||
await self.tesla_device.stop_charge()
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
"""Get whether the switch is in on state."""
|
||||
return self._state == STATE_ON
|
||||
|
||||
async def async_update(self):
|
||||
"""Update the state of the switch."""
|
||||
_LOGGER.debug("Updating state for: %s", self._name)
|
||||
await super().async_update()
|
||||
self._state = STATE_ON if self.tesla_device.is_charging() else STATE_OFF
|
||||
if self.tesla_device.is_charging() is None:
|
||||
return None
|
||||
return self.tesla_device.is_charging() == STATE_ON
|
||||
|
||||
|
||||
class RangeSwitch(TeslaDevice, SwitchEntity):
|
||||
"""Representation of a Tesla max range charging switch."""
|
||||
|
||||
def __init__(self, tesla_device, controller, config_entry):
|
||||
"""Initialise the switch."""
|
||||
self._state = None
|
||||
super().__init__(tesla_device, controller, config_entry)
|
||||
|
||||
async def async_turn_on(self, **kwargs):
|
||||
"""Send the on command."""
|
||||
_LOGGER.debug("Enable max range charging: %s", self._name)
|
||||
_LOGGER.debug("Enable max range charging: %s", self.name)
|
||||
await self.tesla_device.set_max()
|
||||
|
||||
async def async_turn_off(self, **kwargs):
|
||||
"""Send the off command."""
|
||||
_LOGGER.debug("Disable max range charging: %s", self._name)
|
||||
_LOGGER.debug("Disable max range charging: %s", self.name)
|
||||
await self.tesla_device.set_standard()
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
"""Get whether the switch is in on state."""
|
||||
return self._state
|
||||
|
||||
async def async_update(self):
|
||||
"""Update the state of the switch."""
|
||||
_LOGGER.debug("Updating state for: %s", self._name)
|
||||
await super().async_update()
|
||||
self._state = bool(self.tesla_device.is_maxrange())
|
||||
if self.tesla_device.is_maxrange() is None:
|
||||
return None
|
||||
return bool(self.tesla_device.is_maxrange())
|
||||
|
||||
|
||||
class UpdateSwitch(TeslaDevice, SwitchEntity):
|
||||
"""Representation of a Tesla update switch."""
|
||||
|
||||
def __init__(self, tesla_device, controller, config_entry):
|
||||
def __init__(self, tesla_device, coordinator):
|
||||
"""Initialise the switch."""
|
||||
self._state = None
|
||||
tesla_device.type = "update switch"
|
||||
super().__init__(tesla_device, controller, config_entry)
|
||||
self._name = self._name.replace("charger", "update")
|
||||
self.tesla_id = self.tesla_id.replace("charger", "update")
|
||||
super().__init__(tesla_device, coordinator)
|
||||
self.controller = coordinator.controller
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the device."""
|
||||
return super().name.replace("charger", "update")
|
||||
|
||||
@property
|
||||
def unique_id(self) -> str:
|
||||
"""Return a unique ID."""
|
||||
return super().unique_id.replace("charger", "update")
|
||||
|
||||
async def async_turn_on(self, **kwargs):
|
||||
"""Send the on command."""
|
||||
_LOGGER.debug("Enable updates: %s %s", self._name, self.tesla_device.id())
|
||||
_LOGGER.debug("Enable updates: %s %s", self.name, self.tesla_device.id())
|
||||
self.controller.set_updates(self.tesla_device.id(), True)
|
||||
|
||||
async def async_turn_off(self, **kwargs):
|
||||
"""Send the off command."""
|
||||
_LOGGER.debug("Disable updates: %s %s", self._name, self.tesla_device.id())
|
||||
_LOGGER.debug("Disable updates: %s %s", self.name, self.tesla_device.id())
|
||||
self.controller.set_updates(self.tesla_device.id(), False)
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
"""Get whether the switch is in on state."""
|
||||
return self._state
|
||||
|
||||
async def async_update(self):
|
||||
"""Update the state of the switch."""
|
||||
car_id = self.tesla_device.id()
|
||||
_LOGGER.debug("Updating state for: %s %s", self._name, car_id)
|
||||
await super().async_update()
|
||||
self._state = bool(self.controller.get_updates(car_id))
|
||||
if self.controller.get_updates(self.tesla_device.id()) is None:
|
||||
return None
|
||||
return bool(self.controller.get_updates(self.tesla_device.id()))
|
||||
|
||||
|
||||
class SentryModeSwitch(TeslaDevice, SwitchEntity):
|
||||
@ -123,25 +107,17 @@ class SentryModeSwitch(TeslaDevice, SwitchEntity):
|
||||
|
||||
async def async_turn_on(self, **kwargs):
|
||||
"""Send the on command."""
|
||||
_LOGGER.debug("Enable sentry mode: %s", self._name)
|
||||
_LOGGER.debug("Enable sentry mode: %s", self.name)
|
||||
await self.tesla_device.enable_sentry_mode()
|
||||
|
||||
async def async_turn_off(self, **kwargs):
|
||||
"""Send the off command."""
|
||||
_LOGGER.debug("Disable sentry mode: %s", self._name)
|
||||
_LOGGER.debug("Disable sentry mode: %s", self.name)
|
||||
await self.tesla_device.disable_sentry_mode()
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
"""Get whether the switch is in on state."""
|
||||
if self.tesla_device.is_on() is None:
|
||||
return None
|
||||
return self.tesla_device.is_on()
|
||||
|
||||
@property
|
||||
def available(self):
|
||||
"""Indicate if Home Assistant is able to read the state and control the underlying device."""
|
||||
return self.tesla_device.available()
|
||||
|
||||
async def async_update(self):
|
||||
"""Update the state of the switch."""
|
||||
_LOGGER.debug("Updating state for: %s", self._name)
|
||||
await super().async_update()
|
||||
|
@ -2107,7 +2107,7 @@ temperusb==1.5.3
|
||||
tesla-powerwall==0.2.12
|
||||
|
||||
# homeassistant.components.tesla
|
||||
teslajsonpy==0.10.1
|
||||
teslajsonpy==0.10.3
|
||||
|
||||
# homeassistant.components.tensorflow
|
||||
# tf-models-official==2.2.1
|
||||
|
@ -936,7 +936,7 @@ tellduslive==0.10.11
|
||||
tesla-powerwall==0.2.12
|
||||
|
||||
# homeassistant.components.tesla
|
||||
teslajsonpy==0.10.1
|
||||
teslajsonpy==0.10.3
|
||||
|
||||
# homeassistant.components.toon
|
||||
toonapi==0.2.0
|
||||
|
Loading…
x
Reference in New Issue
Block a user